Push a state snapshot only for events that change it

The hub sent every open tab a full snapshot on every StateEvent, LED writes
included - placing and removing a figure was 13 snapshots, 9 of them for LED
changes the snapshot does not even contain. List the events that matter, and make
a test fail when a new StateEvent is sorted into neither list.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

python-backend/musicmouse/services/web/hub.py

@@ -22,7 +22,19 @@ from typing import Any
 from fastapi import WebSocket
 
 from musicmouse.app import App
-from musicmouse.events import Event, StateEvent, ToggleClockRequested
+from musicmouse.events import (
+    ActiveFigureChanged,
+    ClockStateChanged,
+    ConnectionChanged,
+    Event,
+    FigureSelectionChanged,
+    LedEffectChanged,
+    PlaybackChanged,
+    StateEvent,
+    ToggleClockRequested,
+    TrackChanged,
+    VolumeChanged,
+)
 from musicmouse.services.web.state import snapshot
 
 _log = logging.getLogger(__name__)
@@ -33,6 +45,25 @@ __all__ = ["StateHub"]
 #: enough to leave running.
 POSITION_INTERVAL = 0.5
 
+#: The state events that change something in :func:`~musicmouse.services.web.state.snapshot`
+#: - each one costs every open tab a full snapshot. LED writes are the reason this is a
+#: list rather than "every StateEvent": a figure placed and removed is 9 LED writes,
+#: and the snapshot has no LED state in it.
+SNAPSHOT_EVENTS: tuple[type[StateEvent], ...] = (
+    PlaybackChanged,
+    TrackChanged,
+    VolumeChanged,
+    ActiveFigureChanged,
+    ConnectionChanged,
+    ClockStateChanged,
+)
+
+#: State events deliberately not pushed to the browser. Together with
+#: :data:`SNAPSHOT_EVENTS` this covers every :class:`StateEvent`; a test holds that, so
+#: a new state event has to be sorted into one or the other.
+#: The ring's album picker is shown on the LED ring, not in the browser.
+IGNORED_EVENTS: tuple[type[StateEvent], ...] = (LedEffectChanged, FigureSelectionChanged)
+
 
 class StateHub:
     def __init__(self, app: App) -> None:
@@ -44,7 +75,7 @@ class StateHub:
 
     def start(self) -> None:
         self._unsubscribe = [
-            self.app.bus.subscribe(StateEvent, self._on_state),
+            *(self.app.bus.subscribe(kind, self._on_state) for kind in SNAPSHOT_EVENTS),
             # The clock overlay is the kiosk's own; the remote's key only asks for it.
             self.app.bus.subscribe(ToggleClockRequested, self._on_clock_toggle),
         ]

python-backend/tests/test_web.py

@@ -21,15 +21,20 @@ from fastapi import FastAPI
 
 from musicmouse.app import ParentTimer
 from musicmouse.config import WebConfig, load_config
+from musicmouse.effects import OFF
 from musicmouse.events import (
     CancelTimerRequested,
     ClockStateChanged,
+    LedEffectChanged,
     SetTimerRequested,
+    StateEvent,
     ToggleClockRequested,
     ToggleScreenRequested,
 )
+from musicmouse.hardware import LedZone
 from musicmouse.library.analysis import BeatGrid, TrackCurves
 from musicmouse.library.models import track_key
+from musicmouse.services.web.hub import IGNORED_EVENTS, SNAPSHOT_EVENTS
 from musicmouse.services.web.service import build_app
 from musicmouse.simulator.harness import Simulation, build_simulation
 from tests.conftest import VALID_CONFIG, write_config
@@ -442,6 +447,30 @@ async def test_a_state_change_reaches_every_client(api: FastAPI, sim: Simulation
             assert message["state"]["active_figure"] == "fuchs"
 
 
+async def test_led_writes_are_not_pushed_to_the_clients(api: FastAPI, sim: Simulation) -> None:
+    """The snapshot has no LED state, so an LED write is no reason to send one."""
+    async with websocket_connect(api, "/api/ws") as socket:
+        await socket.next_json()
+
+        await sim.bus.emit_and_wait(
+            LedEffectChanged(zone=LedZone.RING, effect=OFF(), origin="mqtt")
+        )
+
+        assert socket.from_app.empty()
+
+
+def test_every_state_event_is_either_pushed_or_deliberately_ignored() -> None:
+    """A new StateEvent has to be sorted into SNAPSHOT_EVENTS or IGNORED_EVENTS."""
+
+    def subclasses(kind: type) -> set[type]:
+        found = set(kind.__subclasses__())
+        for sub in list(found):
+            found |= subclasses(sub)
+        return found
+
+    assert subclasses(StateEvent) == set(SNAPSHOT_EVENTS) | set(IGNORED_EVENTS)
+
+
 async def test_switching_albums_while_playing_reaches_every_client(
     api: FastAPI, client: Client, sim: Simulation
 ) -> None: