diff --git a/modalapi/modhandler.py b/modalapi/modhandler.py index f5d9f201b..687ad1d52 100755 --- a/modalapi/modhandler.py +++ b/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 diff --git a/modalapi/pedalboard.py b/modalapi/pedalboard.py index 7611a82d6..dc8127408 100755 --- a/modalapi/pedalboard.py +++ b/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]: diff --git a/plugins/transport/__init__.py b/plugins/transport/__init__.py index ef376e81b..46f210ea0 100644 --- a/plugins/transport/__init__.py +++ b/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 diff --git a/tests/integration/test_tap_tempo.py b/tests/integration/test_tap_tempo.py index 2702864e7..8643a186d 100644 --- a/tests/integration/test_tap_tempo.py +++ b/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): diff --git a/tests/snapshots/v3/test_transport_bindings/test_encoder_bpm_turn_parameter_dialog_snapshot/bpm_dialog_121.png b/tests/snapshots/v3/test_transport_bindings/test_encoder_bpm_turn_parameter_dialog_snapshot/bpm_dialog_121.png new file mode 100644 index 000000000..665e88e3e Binary files /dev/null and b/tests/snapshots/v3/test_transport_bindings/test_encoder_bpm_turn_parameter_dialog_snapshot/bpm_dialog_121.png differ diff --git a/tests/v3/test_transport_bindings.py b/tests/v3/test_transport_bindings.py index 006ea33f6..eca5afd4e 100644 --- a/tests/v3/test_transport_bindings.py +++ b/tests/v3/test_transport_bindings.py @@ -64,6 +64,99 @@ def test_transport_plugin_built_from_timeinfo(v3_system: SystemFixture): assert rolling.value == 0.0 +def test_transport_plugin_unconditional_fallback(v3_system: SystemFixture): + """When timeInfo is empty, None, or available=0, transport_plugin is unconditionally + created with default unbound parameters for :bpm, :bpb, and :rolling.""" + handler = v3_system.handler + assert handler.current is not None + + # Test empty dict/None + tp = handler.current.pedalboard._build_transport_plugin(None) + assert tp is not None + assert tp.instance_id == TRANSPORT_INSTANCE_ID + assert BPM_SYMBOL in tp.parameters + assert BPB_SYMBOL in tp.parameters + assert ROLLING_SYMBOL in tp.parameters + assert tp.parameters[BPM_SYMBOL].binding is None + assert tp.parameters[BPM_SYMBOL].value == 120.0 + assert tp.parameters[BPB_SYMBOL].value == 4.0 + assert tp.parameters[ROLLING_SYMBOL].value == 0.0 + + # Test available=0 + tp_zero = handler.current.pedalboard._build_transport_plugin({"available": 0}) + assert tp_zero is not None + assert BPM_SYMBOL in tp_zero.parameters + assert BPB_SYMBOL in tp_zero.parameters + assert ROLLING_SYMBOL in tp_zero.parameters + + +def test_reactive_bpm_parameter_change_triggers_set_mod_tap_tempo(v3_system: SystemFixture): + """Writing to transport_plugin.parameters[BPM_SYMBOL].value reactively notifies + subscribers and triggers set_mod_tap_tempo.""" + from unittest.mock import MagicMock + + handler = v3_system.handler + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + _attach_transport_plugin(handler) + ws_bridge.send_bpm = MagicMock(return_value=True) + + # Change BPM parameter value directly (e.g. via encoder or set_param_value) + tp = handler.current.pedalboard.transport_plugin + tp.set_param_value(BPM_SYMBOL, 148.0) + + # Verify reactive subscriber triggered send_bpm + ws_bridge.send_bpm.assert_called_once_with(148.0) + + +def test_transport_message_ws_suppresses_bpm_echo(v3_system: SystemFixture): + """An incoming WebSocket TransportMessage updates transport parameters without + echo-calling send_bpm back to mod-ui.""" + from unittest.mock import MagicMock + + handler = v3_system.handler + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + _attach_transport_plugin(handler) + ws_bridge.send_bpm = MagicMock(return_value=True) + + # Inject incoming WS TransportMessage from mod-ui: transport {rolling} {bpb} {bpm} {syncMode} + ws_bridge.inject("transport 1 4.0 155.0 Internal") + handler.poll_ws_messages() + + tp = handler.current.pedalboard.transport_plugin + assert tp.parameters[BPM_SYMBOL].value == 155.0 + assert tp.parameters[ROLLING_SYMBOL].value == 1.0 + assert tp.parameters[BPB_SYMBOL].value == 4.0 + + # Ensure send_bpm was NOT called because update originated from WS + ws_bridge.send_bpm.assert_not_called() + + +def test_partial_timeinfo_bitmask_creates_all_parameters(v3_system: SystemFixture): + """When timeInfo has a partial bitmask (e.g. only BPM), all 3 parameters + are still created, with missing ones remaining unbound.""" + handler = v3_system.handler + assert handler.current is not None + + # Available bitmask 0x2 = BPM only + tp = handler.current.pedalboard._build_transport_plugin({ + "available": 0x2, + "bpm": 130.0, + "bpmCC": {"channel": 0, "control": 10}, + }) + assert tp is not None + assert BPM_SYMBOL in tp.parameters + assert BPB_SYMBOL in tp.parameters + assert ROLLING_SYMBOL in tp.parameters + + assert tp.parameters[BPM_SYMBOL].binding == "0:10" + assert tp.parameters[BPB_SYMBOL].binding is None + assert tp.parameters[ROLLING_SYMBOL].binding is None + + def test_find_plugin_resolves_transport(v3_system: SystemFixture): """find_plugin returns the transport pseudo-plugin for the /pedalboard id, where the old `next(p for p in plugins)` lookup returned None.""" @@ -197,3 +290,305 @@ def test_load_time_binding_labels_on_load(v3_system: SystemFixture, make_plugin, # The encoder is bound to the Tempo parameter through the load-time row. assert enc1.parameter is tp.parameters[BPM_SYMBOL] snapshot("loaded") + + +def test_encoder_bpm_turn_sends_websocket(v3_system: SystemFixture, make_plugin): + """MIDI Learning /pedalboard :bpm to an encoder and turning it updates local tempo, + shows the dismissible parameter dialog, and dispatches a high-precision WebSocket packet.""" + handler = v3_system.handler + hw = v3_system.hw + ws_bridge = v3_system.ws_bridge + assert handler.current is not None and handler.lcd is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + _attach_transport_plugin(handler) + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + + # 1. Live MIDI Learn message from MOD-UI + ws_bridge.inject(f"midi_map /pedalboard :bpm {channel} {cc} 20.0 280.0") + handler.poll_ws_messages() + + tp = handler.current.pedalboard.transport_plugin + assert tp is not None + assert tp.parameters[BPM_SYMBOL].binding == f"{channel}:{cc}" + assert enc1.parameter is tp.parameters[BPM_SYMBOL] + + # 2. Turn encoder clockwise by 1 step (1.0 BPM per detent) + from pistomp.input.event import EncoderEvent + + event = EncoderEvent(controller=enc1, rotations=1, multiplier=1.0) + handler._handle_encoder(event) + + # 3. Tempo parameter updated (1 step clockwise from 120.0 -> 121.0 BPM) + assert tp.parameters[BPM_SYMBOL].value == 121.0 + + # 4. High-precision WebSocket transport-bpm packet queued for MOD-UI + sent_msgs = list(ws_bridge.sent) + assert any("transport-bpm 121.0" in m for m in sent_msgs) + + +def test_encoder_bpm_clamping_at_boundaries(v3_system: SystemFixture, make_plugin): + """BPM parameter edits clamp strictly at minimum (20.0) and maximum (280.0) limits.""" + from pistomp.input.event import EncoderEvent + + handler = v3_system.handler + hw = v3_system.hw + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + _attach_transport_plugin( + handler, + bpm_cc={"channel": int(channel), "control": int(cc), "hasRanges": True, "minimum": 20.0, "maximum": 280.0}, + ) + + tp = handler.current.pedalboard.transport_plugin + assert tp is not None + bpm_param = tp.parameters[BPM_SYMBOL] + + # Spin clockwise aggressively (+200 steps) + event_up = EncoderEvent(controller=enc1, rotations=200, multiplier=1.0) + handler._handle_encoder(event_up) + assert bpm_param.value == 280.0 + + # Spin counter-clockwise aggressively (-300 steps) + event_down = EncoderEvent(controller=enc1, rotations=-300, multiplier=1.0) + handler._handle_encoder(event_down) + assert bpm_param.value == 20.0 + + +def test_unbound_encoder_emits_midi_cc_for_learning(v3_system: SystemFixture, make_plugin): + """An unbound encoder emits fallback MIDI CC so MOD-UI can learn it, but once bound + to /pedalboard :bpm, turns bypass MIDI CC and send WebSocket messages instead.""" + from unittest.mock import MagicMock + from pistomp.input.event import EncoderEvent + + handler = v3_system.handler + hw = v3_system.hw + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + _attach_transport_plugin(handler) + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + + # Mock _emit_midi to observe MIDI CC output + handler._emit_midi = MagicMock() + + # 1. Turn unbound encoder -> emits MIDI CC for MOD-UI MIDI Learn + assert enc1.parameter is None + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=1.0)) + handler._emit_midi.assert_called_once() + handler._emit_midi.reset_mock() + + # 2. Bind encoder to /pedalboard :bpm via MIDI Learn + ws_bridge.inject(f"midi_map /pedalboard :bpm {channel} {cc} 20.0 280.0") + handler.poll_ws_messages() + assert enc1.parameter is not None + + # 3. Turn bound encoder -> bypasses _emit_midi and sends WebSocket transport-bpm + ws_bridge.sent.clear() + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=1.0)) + handler._emit_midi.assert_not_called() + assert any("transport-bpm" in msg for msg in ws_bridge.sent) + + +def test_parameter_value_commit_for_transport_bpm(v3_system: SystemFixture, make_plugin): + """parameter_value_commit for /pedalboard :bpm routes via set_mod_tap_tempo (WebSocket).""" + handler = v3_system.handler + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + _attach_transport_plugin(handler) + + tp = handler.current.pedalboard.transport_plugin + assert tp is not None + bpm_param = tp.parameters[BPM_SYMBOL] + + ws_bridge.sent.clear() + handler.parameter_value_commit(bpm_param, 135.0) + + assert bpm_param.value == 135.0 + assert any("transport-bpm 135.0" in m for m in ws_bridge.sent) + + +def test_encoder_bpm_fast_spin_acceleration(v3_system: SystemFixture, make_plugin): + """Slow encoder turn moves 1.0 BPM per detent; fast spin accelerates edit proportionally.""" + from pistomp.input.event import EncoderEvent + + handler = v3_system.handler + hw = v3_system.hw + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + _attach_transport_plugin( + handler, + bpm_cc={"channel": int(channel), "control": int(cc), "hasRanges": True, "minimum": 20.0, "maximum": 280.0}, + ) + + tp = handler.current.pedalboard.transport_plugin + assert tp is not None + bpm_param = tp.parameters[BPM_SYMBOL] + + # 1. Slow turn (multiplier=1.0) -> moves exactly 1.0 BPM (120.0 -> 121.0) + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=1.0)) + assert bpm_param.value == 121.0 + + # 2. Fast spin (multiplier=4.0) -> accelerates edit (>1.0 BPM) + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=4.0)) + assert bpm_param.value > 122.0 # Accelerated step (>1 BPM) + + +def test_incoming_transport_decimal_bpm_sync(v3_system: SystemFixture, make_plugin): + """External transport updates (e.g. Ableton Link) preserve decimal BPM values.""" + handler = v3_system.handler + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + _attach_transport_plugin(handler) + + tp = handler.current.pedalboard.transport_plugin + assert tp is not None + + # Inject external transport frame with decimal BPM (120.5) + ws_bridge.inject("transport 1 4.0 120.5 link") + handler.poll_ws_messages() + + assert tp.parameters[BPM_SYMBOL].value == 120.5 + + +def test_encoder_bpm_turn_without_websocket_bridge_falls_back_to_rest_post( + v3_system: SystemFixture, make_plugin +): + """If ws_bridge is None, encoder tempo turns execute REST POST fallback.""" + from unittest.mock import MagicMock + from pistomp.input.event import EncoderEvent + + handler = v3_system.handler + hw = v3_system.hw + mock_post = v3_system.mock_post + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + _attach_transport_plugin( + handler, + bpm_cc={"channel": int(channel), "control": int(cc), "hasRanges": True, "minimum": 20.0, "maximum": 280.0}, + ) + + # Set _ws_bridge to None to simulate missing WebSocket bridge + handler._ws_bridge = None + mock_post.reset_mock() + + # Turn encoder + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=1.0)) + + # Assert REST POST fallback was executed + mock_post.assert_called_once() + assert "set_bpm" in mock_post.call_args[0][0] + assert mock_post.call_args[1]["json"] == {"value": 121.0} + + +def test_encoder_bpm_turn_parameter_dialog_snapshot(v3_system: SystemFixture, make_plugin, snapshot): + """Turning a BPM-bound encoder 1 detent notch displays the parameter dialog on the LCD at 121 BPM.""" + from pistomp.input.event import EncoderEvent + + handler = v3_system.handler + hw = v3_system.hw + ws_bridge = v3_system.ws_bridge + assert handler.current is not None and handler.lcd is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + _attach_transport_plugin(handler) + + handler.lcd.link_data(handler.pedalboard_list, handler.current, hw.footswitches) + handler.lcd.draw_main_panel() + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + + ws_bridge.inject(f"midi_map /pedalboard :bpm {channel} {cc} 20.0 280.0") + handler.poll_ws_messages() + + # Turn encoder 1 detent (120.0 -> 121.0 BPM) + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=1.0)) + + # Capture LCD snapshot of 121 BPM parameter dialog badge + snapshot("bpm_dialog_121") + + +def test_encoder_rolling_or_bpb_emits_midi_cc(v3_system: SystemFixture, make_plugin): + """Transport parameters other than :bpm (e.g., :rolling, :bpb) still emit MIDI CC when turned.""" + from unittest.mock import MagicMock + from modalapi.pedalboard import ROLLING_SYMBOL + from pistomp.input.event import EncoderEvent + + handler = v3_system.handler + hw = v3_system.hw + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + _attach_transport_plugin(handler) + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + + ws_bridge.inject(f"midi_map /pedalboard :rolling {channel} {cc} 0.0 1.0") + handler.poll_ws_messages() + + tp = handler.current.pedalboard.transport_plugin + assert tp is not None + assert enc1.parameter is tp.parameters[ROLLING_SYMBOL] + + handler._emit_midi = MagicMock() + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=1.0)) + + # Assert _emit_midi was called for :rolling (unlike :bpm) + handler._emit_midi.assert_called_once() + + +def test_audio_parameter_commit_early_return(v3_system: SystemFixture): + """Audio parameters (instance_id is None) commit locally and return early without sending WS or MIDI CC.""" + from unittest.mock import MagicMock + + handler = v3_system.handler + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + param = handler._create_audio_parameter("Input Gain", "capture_volume", -19.75, 12) + assert param.instance_id is None + + ws_bridge.send_parameter = MagicMock() + handler._emit_midi = MagicMock() + + handler.parameter_value_commit(param, 0.0) + + # Audio param handled locally, no remote parameter set or MIDI CC emitted + ws_bridge.send_parameter.assert_not_called() + handler._emit_midi.assert_not_called() + + +