From d26bc15ea47ad95cc40dd331e8f9b813dac4cdb8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 16 Jul 2026 15:35:31 +0000 Subject: [PATCH 1/2] Add Machine.set_memory_broadcast() for the memory-snapshot UDP stream Wrap the firmware's authenticated /api/memory/toggle-broadcast route so clients (like Memory Mapper) can turn the live SRAM broadcast on and off without reaching for Machine.call(). Enabling takes a frequency_ms clamped to the firmware's 10-60000 ms bounds; disabling sends {"enable": false}. Documented in docs/memory.md and covered by the wrapper-route table plus a body/clamping test. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Xy1GfVJXuAkgZTMwUDDRTb --- docs/memory.md | 20 ++++++++++++++++++++ tests/test_machine.py | 19 +++++++++++++++++++ warpedpinball/machine.py | 19 +++++++++++++++++++ 3 files changed, 58 insertions(+) diff --git a/docs/memory.md b/docs/memory.md index d5d31fa..7bdf2d3 100644 --- a/docs/memory.md +++ b/docs/memory.md @@ -96,6 +96,26 @@ 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 broadcasting: `set_memory_broadcast()` + +Instead of polling snapshots, the firmware can *push* them: +`set_memory_broadcast(True)` makes the board broadcast the whole SRAM data +region as UDP packets to port 2040 on the local network, 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) # broadcast every 100 ms +m.set_memory_broadcast(True, frequency_ms=500) # or at your own rate +... +m.set_memory_broadcast(False) # stop broadcasting +``` + +Enabling and disabling are authenticated. `frequency_ms` is clamped to the +firmware's 10-60000 ms bounds. Broadcasting 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 diff --git a/tests/test_machine.py b/tests/test_machine.py index 46ab27e..84f0ff9 100644 --- a/tests/test_machine.py +++ b/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,20 @@ 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(False) + assert [body for _, body, _ in transport.calls] == [ + {"enable": True, "frequency_ms": 250}, + {"enable": True, "frequency_ms": 10}, + {"enable": True, "frequency_ms": 60000}, + {"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" diff --git a/warpedpinball/machine.py b/warpedpinball/machine.py index 30684da..2cf01d9 100644 --- a/warpedpinball/machine.py +++ b/warpedpinball/machine.py @@ -384,6 +384,25 @@ 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) -> Any: + """Toggle UDP broadcasting of memory snapshots. Authenticated. + + When enabled, the firmware broadcasts the whole SRAM data region as + UDP packets to port 2040 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. ``frequency_ms`` is clamped to the firmware's + 10-60000 ms bounds and ignored when disabling. + """ + if enabled: + frequency_ms = max(10, min(60000, int(frequency_ms))) + body: Dict[str, Any] = {"enable": True, "frequency_ms": frequency_ms} + 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 From 4f96160e030ee969e2b3ef73996a1c8135814e79 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 16 Jul 2026 18:06:03 +0000 Subject: [PATCH 2/2] Point set_memory_broadcast() at one client IP (firmware unicast pivot) The firmware no longer broadcasts memory snapshots to the whole network (vector#369); it streams to a single target IP, defaulting to the requesting client. Add the optional ip= parameter to set_memory_broadcast() and update the docs to describe the unicast behavior, including the USB caveat where ip must be passed explicitly. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Xy1GfVJXuAkgZTMwUDDRTb --- docs/memory.md | 27 +++++++++++++++++---------- tests/test_machine.py | 2 ++ warpedpinball/machine.py | 30 +++++++++++++++++++++--------- 3 files changed, 40 insertions(+), 19 deletions(-) diff --git a/docs/memory.md b/docs/memory.md index 7bdf2d3..a6bcf11 100644 --- a/docs/memory.md +++ b/docs/memory.md @@ -96,25 +96,32 @@ 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 broadcasting: `set_memory_broadcast()` +## Live snapshot streaming: `set_memory_broadcast()` Instead of polling snapshots, the firmware can *push* them: -`set_memory_broadcast(True)` makes the board broadcast the whole SRAM data -region as UDP packets to port 2040 on the local network, 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 +`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) # broadcast every 100 ms -m.set_memory_broadcast(True, frequency_ms=500) # or at your own rate +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 broadcasting +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. Broadcasting costs the board work on every -tick, so turn it off when you're done watching. +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 diff --git a/tests/test_machine.py b/tests/test_machine.py index 84f0ff9..4b50b9f 100644 --- a/tests/test_machine.py +++ b/tests/test_machine.py @@ -187,11 +187,13 @@ def test_set_memory_broadcast_bodies(): 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}, ] diff --git a/warpedpinball/machine.py b/warpedpinball/machine.py index 2cf01d9..4cae748 100644 --- a/warpedpinball/machine.py +++ b/warpedpinball/machine.py @@ -384,19 +384,31 @@ 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) -> Any: - """Toggle UDP broadcasting of memory snapshots. Authenticated. - - When enabled, the firmware broadcasts the whole SRAM data region as - UDP packets to port 2040 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. ``frequency_ms`` is clamped to the firmware's - 10-60000 ms bounds and ignored when disabling. + 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(