Add Machine.set_memory_broadcast() for the unicast memory-snapshot stream

docs/memory.md

@@ -96,6 +96,33 @@ Run that a few times while changing exactly one thing on the machine; the
 offset that changes consistently is your address. (A length difference between
 snapshots shows up as changes versus `-1`.)
 
+## Live snapshot streaming: `set_memory_broadcast()`
+
+Instead of polling snapshots, the firmware can *push* them:
+`set_memory_broadcast(True)` makes the board send the whole SRAM data region
+as UDP packets to port 2040 on **one target IP**, over and over, until turned
+off. Each packet is a 4-byte big-endian offset header followed by up to 256
+bytes of data. This is the stream that live viewers like Warped Pinball's
+[Memory Mapper](https://github.com/warped-pinball/memory-mapper) listen to.
+
+```python
+m.set_memory_broadcast(True)                       # stream to this machine, every 100 ms
+m.set_memory_broadcast(True, frequency_ms=500)     # or at your own rate
+m.set_memory_broadcast(True, ip="192.168.1.20")    # or to another listener
+...
+m.set_memory_broadcast(False)                      # stop streaming
+```
+
+When `ip` is omitted the board streams back to whichever address made the
+request — usually exactly what you want, since the listener typically runs
+where your code runs. Over USB there is no requester IP, so pass `ip`
+explicitly. The stream goes only to that one address (never broadcast to the
+whole network), and the board keeps at most one stream target at a time.
+
+Enabling and disabling are authenticated. `frequency_ms` is clamped to the
+firmware's 10-60000 ms bounds. Streaming costs the board work on every tick,
+so turn it off when you're done watching.
+
 ## Calling the routes directly
 
 If you need something the wrappers don't do, `Machine.call()` reaches the raw

tests/test_machine.py

@@ -49,6 +49,11 @@ def make_machine(responses=None, streams=None, password="pw", requires_password=
     (lambda m: m.name_adjustment(1, "comp"), "/api/adjustments/name", True),
     (lambda m: m.peers(), "/api/network/peers", False),
     (lambda m: m.set_date(datetime.datetime(2026, 7, 13)), "/api/set_date", True),
+    (
+        lambda m: m.set_memory_broadcast(True),
+        "/api/memory/toggle-broadcast",
+        True,
+    ),
 ]
 
 
@@ -177,6 +182,22 @@ def test_read_decodes_int():
     assert machine.read(0x01, 2, byteorder="little") == 0x0201
 
 
+def test_set_memory_broadcast_bodies():
+    machine, transport = make_machine()
+    machine.set_memory_broadcast(True, frequency_ms=250)
+    machine.set_memory_broadcast(True, frequency_ms=1)  # below the firmware minimum
+    machine.set_memory_broadcast(True, frequency_ms=10**6)  # above the maximum
+    machine.set_memory_broadcast(True, ip="192.168.1.20")  # explicit target
+    machine.set_memory_broadcast(False)
+    assert [body for _, body, _ in transport.calls] == [
+        {"enable": True, "frequency_ms": 250},
+        {"enable": True, "frequency_ms": 10},
+        {"enable": True, "frequency_ms": 60000},
+        {"enable": True, "frequency_ms": 100, "ip": "192.168.1.20"},
+        {"enable": False},
+    ]
+
+
 def test_memory_snapshot_joins_stream():
     machine, transport = make_machine(streams={"/api/memory-snapshot": [b"abc", b"def"]})
     assert machine.memory_snapshot() == b"abcdef"

warpedpinball/machine.py

@@ -384,6 +384,37 @@ def memory_snapshot(self) -> bytes:
         """Full SRAM dump via the streamed ``/api/memory-snapshot`` route."""
         return b"".join(self.call_stream("/api/memory-snapshot"))
 
+    def set_memory_broadcast(
+        self, enabled: bool, frequency_ms: int = 100, ip: Optional[str] = None
+    ) -> Any:
+        """Toggle UDP streaming of memory snapshots to one client. Authenticated.
+
+        When enabled, the firmware sends the whole SRAM data region as UDP
+        packets to port 2040 on a single target IP every ``frequency_ms``
+        milliseconds (each packet is a 4-byte big-endian offset header
+        followed by up to 256 bytes of data) -- the live stream tools like
+        Warped Pinball's Memory Mapper listen to. The stream goes only to
+        that one address, never to the whole network, and the board keeps at
+        most one stream target at a time.
+
+        ``ip`` is the IPv4 address to stream to; when omitted the firmware
+        streams back to the requester (this machine), which is what you want
+        when the listener runs where this code runs. Over USB there is no
+        requester IP, so pass ``ip`` explicitly. ``frequency_ms`` is clamped
+        to the firmware's 10-60000 ms bounds; both extras are ignored when
+        disabling.
+        """
+        if enabled:
+            frequency_ms = max(10, min(60000, int(frequency_ms)))
+            body: Dict[str, Any] = {"enable": True, "frequency_ms": frequency_ms}
+            if ip is not None:
+                body["ip"] = ip
+        else:
+            body = {"enable": False}
+        return self._call_gated(
+            "/api/memory/toggle-broadcast", body=body, authenticated=True
+        )
+
     @staticmethod
     def diff_snapshots(a: bytes, b: bytes) -> List[Tuple[int, int, int]]:
         """Compare two snapshots; returns ``(offset, a_value, b_value)`` per