set bpm from encoder via ws

modalapi/modhandler.py

@@ -225,6 +225,10 @@ def __init__(self, audiocard: Audiocard, homedir, data_dir="/home/pistomp/data")
         # Suppress outbound WebSocket messages while a pedalboard change is in flight.
         self._is_pedalboard_loading = False
 
+        # Reactive BPM parameter observer state
+        self._bpm_unsub: Callable[[], None] | None = None
+        self._suppress_bpm_event: bool = False
+
         # Tuner state
         self._tuner_source_factory: TunerSourceFactory | None = None
         self._tuner_source_spec: str = "jack"
@@ -426,7 +430,11 @@ def _handle_encoder(self, event: EncoderEvent) -> bool:
         # Unconditional, and must stay that way: an unbound encoder has no row,
         # and this emit is the only way mod-ui sees its CC to MIDI-learn it.
         # Emission is hardware-level, below the table (see input/README.md).
-        self._emit_midi(c, emit_value)
+        # Transport parameters bypass 7-bit MIDI CC emission for high-precision WebSocket transport.
+        if c.parameter is None or not (
+            c.parameter.instance_id == Pedalboard.TRANSPORT_INSTANCE_ID and c.parameter.symbol == BPM_SYMBOL
+        ):
+            self._emit_midi(c, emit_value)
         return True
 
     def encoder_fallback(self, controller: EncoderController) -> int:
@@ -875,13 +883,13 @@ def _handle_ws_message(self, msg: WebSocketMessage):
             # MIDI slave, another HMI). :rolling's enum flips Playing/Stopped.
             if self._current is not None:
                 tp = self.current.pedalboard.transport_plugin
-                if tp is not None:
-                    if ROLLING_SYMBOL in tp.parameters:
-                        tp.set_param_value(ROLLING_SYMBOL, 1.0 if msg.rolling else 0.0)
-                    if BPM_SYMBOL in tp.parameters:
-                        tp.set_param_value(BPM_SYMBOL, msg.bpm)
-                    if BPB_SYMBOL in tp.parameters:
-                        tp.set_param_value(BPB_SYMBOL, msg.beats_per_bar)
+                tp.set_param_value(ROLLING_SYMBOL, 1.0 if msg.rolling else 0.0)
+                tp.set_param_value(BPB_SYMBOL, msg.beats_per_bar)
+                self._suppress_bpm_event = True
+                try:
+                    tp.set_param_value(BPM_SYMBOL, msg.bpm)
+                finally:
+                    self._suppress_bpm_event = False
             if self.hardware and self.hardware.taptempo:
                 self.hardware.taptempo.set_bpm(msg.bpm)
                 if self.hardware.taptempo.is_enabled():
@@ -1207,6 +1215,19 @@ def bind_current_pedalboard(self):
         # The pedalboard data has already been loaded, but this will overlay
         # any real time settings
         self._controller_manager.bind(self.current)
+        self._bind_transport_bpm_listener()
+
+    def _bind_transport_bpm_listener(self) -> None:
+        if self._bpm_unsub is not None:
+            self._bpm_unsub()
+            self._bpm_unsub = None
+        if self._current is not None:
+            bpm_param = self.current.pedalboard.transport_plugin.parameters[BPM_SYMBOL]
+            self._bpm_unsub = bpm_param.subscribe(self._on_bpm_param_changed)
+
+    def _on_bpm_param_changed(self, param: Parameter) -> None:
+        if not self._suppress_bpm_event:
+            self.set_mod_tap_tempo(param.value)
 
     def _redraw_after_binding(self, controller: Controller, is_footswitch: bool) -> None:
         if is_footswitch:
@@ -1442,7 +1463,9 @@ def parameter_value_commit(self, param, value):
                     self._emit_midi(controller, int(value))
                 return
 
-        if not self._is_pedalboard_loading:
+        if not self._is_pedalboard_loading and param.instance_id is not None and not (
+            param.instance_id == Pedalboard.TRANSPORT_INSTANCE_ID and param.symbol == BPM_SYMBOL
+        ):
             self.ws_bridge.send_parameter(param.instance_id, param.symbol, param.value)
 
     @property
@@ -1798,9 +1821,12 @@ def audio_parameter_commit(self, symbol, value):
     def get_callback(self, callback_name):
         return util.DICT_GET(self.callbacks, callback_name)
 
-    def set_mod_tap_tempo(self, bpm):
+    def set_mod_tap_tempo(self, bpm: float | None) -> None:
         if bpm is not None:
-            self._rest_post(self.root_uri + "set_bpm", json={"value": bpm})
+            if self._ws_bridge is not None:
+                self.ws_bridge.send_bpm(bpm)
+            else:
+                self._rest_post(self.root_uri + "set_bpm", json={"value": bpm})
 
     def set_sync_mode(self, mode: SyncMode) -> None:
         """Optimistically switch the clock source; mod-ui's transport echo

modalapi/pedalboard.py

@@ -21,7 +21,7 @@
 from typing import Optional
 
 
-from common.parameter import BYPASS_SYMBOL, Parameter, PortInfo, Symbol, json_default
+from common.parameter import BYPASS_SYMBOL, TTL_INTEGER, Parameter, PortInfo, Symbol, json_default
 import modalapi.plugin as Plugin
 from modalapi.connections import Connection, build_connection
 from modalapi.plugin_customization import Customizer, default_customizer
@@ -57,6 +57,7 @@ def _transport_port_info(symbol: Symbol) -> PortInfo:
             symbol=":bpm",
             ranges={"minimum": _BPM_RANGE[0], "maximum": _BPM_RANGE[1]},
             units={"symbol": "BPM", "label": "beats per minute"},
+            properties=[TTL_INTEGER],
         )
     if symbol == BPB_SYMBOL:
         return PortInfo(
@@ -101,10 +102,8 @@ def __init__(self, title, bundle, root_uri="http://localhost:80/", customizer: C
         self.connections: list[Connection] = []
         self.hydrated = False
         # Synthetic /pedalboard pseudo-instance carrying :bpm/:bpb/:rolling.
-        # Built in hydrate() from timeInfo; None before the first hydrate and
-        # on boards with no transport metadata. Excluded from self.plugins so
-        # the effect-graph render never paints it.
-        self.transport_plugin: Plugin.Plugin | None = None
+        # Excluded from self.plugins so the effect-graph render never paints it.
+        self.transport_plugin: Plugin.Plugin = self._build_transport_plugin(None)
 
     def get_plugin_data(self, uri):
         url = self.root_uri + "effect/get?uri=" + urllib.parse.quote(uri)
@@ -237,45 +236,31 @@ def hydrate(self, plugin_dict) -> None:
 
         self.hydrated = True
 
-    def _build_transport_plugin(self, time_info: dict | None) -> Plugin.Plugin | None:
+    def _build_transport_plugin(self, time_info: dict | None) -> Plugin.Plugin:
         """The /pedalboard pseudo-instance carrying :bpm/:bpb/:rolling. Built
-        from mod-ui's timeInfo block. None when the board reports no transport
-        metadata (available == 0 or absent)."""
-        if not time_info:
-            return None
-        available = int(time_info.get("available", 0) or 0)
-        if available == 0:
-            return None
+        from mod-ui's timeInfo block (or default unbound parameters when absent)."""
+        time_info = time_info or {}
 
-        parameters: dict[Symbol, Parameter] = {}
-        # mod-ui's kPedalboardTimeAvailable* bit masks.
-        if available & 0x1:  # BPB
-            cc = time_info.get("bpbCC")
-            parameters[BPB_SYMBOL] = Parameter(
+        parameters: dict[Symbol, Parameter] = {
+            BPB_SYMBOL: Parameter(
                 _transport_port_info(BPB_SYMBOL),
-                float(time_info.get("bpb", _BPB_RANGE[0])),
-                self._binding(cc),
+                float(time_info.get("bpb", 4.0)),
+                self._binding(time_info.get("bpbCC")),
                 TRANSPORT_INSTANCE_ID,
-            )
-        if available & 0x2:  # BPM
-            cc = time_info.get("bpmCC")
-            parameters[BPM_SYMBOL] = Parameter(
+            ),
+            BPM_SYMBOL: Parameter(
                 _transport_port_info(BPM_SYMBOL),
-                float(time_info.get("bpm", _BPM_RANGE[0])),
-                self._binding(cc),
+                float(time_info.get("bpm", 120.0)),
+                self._binding(time_info.get("bpmCC")),
                 TRANSPORT_INSTANCE_ID,
-            )
-        if available & 0x4:  # Rolling
-            cc = time_info.get("rollingCC")
-            parameters[ROLLING_SYMBOL] = Parameter(
+            ),
+            ROLLING_SYMBOL: Parameter(
                 _transport_port_info(ROLLING_SYMBOL),
                 1.0 if time_info.get("rolling") else 0.0,
-                self._binding(cc),
+                self._binding(time_info.get("rollingCC")),
                 TRANSPORT_INSTANCE_ID,
-            )
-
-        if not parameters:
-            return None
+            ),
+        }
 
         # category drives footswitch color; "Utility" is the benign choice for
         # transport — no LV2 category exists for it. uri=urn:mod:pedalboard so
@@ -295,9 +280,8 @@ def find_plugin(self, instance_id: str) -> Plugin.Plugin | None:
         for p in self.plugins:
             if p.instance_id == instance_id:
                 return p
-        tp = self.transport_plugin
-        if tp is not None and tp.instance_id == instance_id:
-            return tp
+        if self.transport_plugin.instance_id == instance_id:
+            return self.transport_plugin
         return None
 
     def _build_plugin(self, instance_id: str, uri: str, x: float, y: float, info: dict) -> Optional[Plugin.Plugin]:

plugins/transport/__init__.py

@@ -9,7 +9,7 @@
 
 from __future__ import annotations
 
-from common.parameter import Parameter, Symbol
+from common.parameter import Parameter
 from modalapi.plugin_customization import PluginCustomization
 from modalapi.pedalboard import BPM_SYMBOL, BPB_SYMBOL, ROLLING_SYMBOL
 from plugins.customization import register

tests/integration/test_tap_tempo.py

@@ -6,16 +6,21 @@
 
 
 def test_set_mod_tap_tempo(modhandler_system: SystemFixture):
-    """set_mod_tap_tempo() POSTs to /set_bpm with the BPM value."""
+    """set_mod_tap_tempo() sends high-precision BPM via WS bridge (or REST fallback)."""
     handler = modhandler_system.handler
+    ws_bridge = modhandler_system.ws_bridge
     mock_post = modhandler_system.mock_post
 
     handler.set_mod_tap_tempo(120)
 
-    mock_post.assert_called_once()
-    call_args = mock_post.call_args
-    assert "set_bpm" in call_args.args[0]
-    assert call_args.kwargs.get("json", {}).get("value") == 120
+    if handler._ws_bridge is not None:
+        assert any("transport-bpm 120" in msg for msg in ws_bridge.sent)
+        mock_post.assert_not_called()
+    else:
+        mock_post.assert_called_once()
+        call_args = mock_post.call_args
+        assert "set_bpm" in call_args.args[0]
+        assert call_args.kwargs.get("json", {}).get("value") == 120
 
 
 def test_set_mod_tap_tempo_none(modhandler_system: SystemFixture):