diff --git a/common/parameter.py b/common/parameter.py
index e7d493984..e3c546226 100644
--- a/common/parameter.py
+++ b/common/parameter.py
@@ -26,6 +26,7 @@
TTL_SCALEPOINTS = 'scalePoints'
TTL_TAPTEMPO = 'tapTempo'
TTL_TOGGLED = 'toggled'
+TTL_TRIGGER = 'trigger'
class Type(Enum):
DEFAULT = 0 # No explicitly defined type (eg. linear float)
@@ -34,6 +35,7 @@ class Type(Enum):
LOGARITHMIC = 3
TAPTEMPO = 4
TOGGLED = 5
+ TRIGGER = 6 # pprops:trigger — edge-triggered, self-clearing (momentary)
class Parameter:
@@ -57,7 +59,9 @@ def __init__(self, plugin_info, value: float, binding, instance_id=None):
properties = util.DICT_GET(plugin_info, TTL_PROPERTIES)
if properties is not None and len(properties) > 0:
- if TTL_ENUMERATION in properties:
+ if TTL_TRIGGER in properties:
+ self.type = Type.TRIGGER
+ elif TTL_ENUMERATION in properties:
self.enum_values = util.DICT_GET(plugin_info, TTL_SCALEPOINTS)
self.type = Type.ENUMERATION
elif TTL_INTEGER in properties:
@@ -69,6 +73,12 @@ def __init__(self, plugin_info, value: float, binding, instance_id=None):
elif TTL_TOGGLED in properties:
self.type = Type.TOGGLED
+ @property
+ def is_momentary(self) -> bool:
+ """True for edge-triggered, self-clearing ports (pprops:trigger) —
+ these need a one-shot 127 press rather than an absolute 127/0 toggle."""
+ return self.type == Type.TRIGGER
+
def get_enum_value_list(self):
ret = []
for v in self.enum_values:
diff --git a/modalapi/led_render.py b/modalapi/led_render.py
new file mode 100644
index 000000000..231163bab
--- /dev/null
+++ b/modalapi/led_render.py
@@ -0,0 +1,54 @@
+# This file is part of pi-stomp.
+#
+# pi-stomp is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# pi-stomp is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with pi-stomp. If not, see .
+
+"""Generic, data-driven footswitch-LED rendering.
+
+Pure function of (LedSpec, plugin.output_values) -> (color, style). No
+footswitch, beat, or plugin-instance coupling — the per-tick brightness
+envelope (pulse phase, downbeat emphasis) is applied uniformly by the
+handler's single LED-writing helper, not here.
+"""
+
+from __future__ import annotations
+
+from enum import Enum, auto
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from modalapi.plugin_customization import LedSpec
+
+
+class LedDisplayStyle(Enum):
+ SOLID = auto()
+ METRONOME = auto()
+
+
+def render_led_spec(
+ spec: LedSpec, output_values: dict[str, float]
+) -> tuple[tuple[int, int, int] | None, LedDisplayStyle]:
+ state = int(output_values.get(spec.state_symbol, 0))
+ if state in spec.off_states:
+ return None, LedDisplayStyle.SOLID
+ base = spec.colors.get(state)
+ if base is None:
+ return None, LedDisplayStyle.SOLID
+ if spec.downbeat_symbol is not None and int(output_values.get(spec.downbeat_symbol, -1)) == 0:
+ base = (
+ min(255, base[0] + spec.downbeat_tint),
+ min(255, base[1] + spec.downbeat_tint),
+ min(255, base[2] + spec.downbeat_tint),
+ )
+ style = LedDisplayStyle.METRONOME if (spec.pulse and state not in spec.steady_states) else LedDisplayStyle.SOLID
+ return base, style
diff --git a/modalapi/mod.py b/modalapi/mod.py
index 4e25c7135..5a5c42c8e 100755
--- a/modalapi/mod.py
+++ b/modalapi/mod.py
@@ -33,9 +33,10 @@
import modalapi.external_midi as ExternalMidi
from pistomp.encoder_controller import EncoderController
from pistomp.input.event import AnalogEvent, ControllerEvent, EncoderEvent, SwitchEvent, SwitchEventKind
-from rtmidi.midiconstants import CONTROL_CHANGE
from modalapi.ethernet import EthernetManager
from modalapi.jack_mute import JackMute
+from modalapi.led_render import render_led_spec
+from pistomp.category import get_category_color
from typing import Optional
from blend.snapshot import SnapshotManager
@@ -261,19 +262,6 @@ def _handle_switch_v1(self, event: SwitchEvent) -> bool:
return self._handle_footswitch(c, event.kind, event.timestamp)
return False
- def _emit_midi(self, controller, midi_value: int) -> None:
- if controller.midi_CC is None:
- return
- cc = [controller.midi_channel | CONTROL_CHANGE, controller.midi_CC, int(midi_value)]
- port_name = self.hardware.external_port_name(controller)
- if port_name is not None and self.external_midi is not None:
- try:
- if self.external_midi.send_raw(port_name, cc):
- return
- except Exception as e:
- logging.warning("External CC send failed on %s: %s", port_name, e)
- self.hardware.midiout.send_message(cc)
-
def add_lcd(self, lcd):
self.lcd = lcd
@@ -533,6 +521,35 @@ def poll_controls(self):
if self.universal_encoder_mode is not UniversalEncoderMode.LOADING:
self.hardware.poll_controls()
self._tick_chords()
+ self._drive_footswitch_leds()
+
+ def _drive_footswitch_leds(self) -> None:
+ """v1 has no beat_grid and no pixel (mono LCD) — just render each
+ footswitch's GPIO on/off state from its bound plugin's LedSpec (if
+ any) or the default toggle/category behavior. No metronome style."""
+ if self.hardware is None:
+ return
+ for fs in self.hardware.footswitches:
+ plugin = self._bound_plugin(fs)
+ if plugin is not None and plugin.customization.led_spec is not None:
+ color, _style = render_led_spec(plugin.customization.led_spec, plugin.output_values)
+ elif fs.toggled:
+ color = get_category_color(fs.category) if fs.category is not None else (255, 255, 255)
+ else:
+ color = None
+ if fs.led is not None:
+ if color is None:
+ fs.led.off()
+ else:
+ fs.led.on()
+
+ def _bound_plugin(self, fs):
+ if fs.parameter is None or self._current is None:
+ return None
+ for plugin in self.current.pedalboard.plugins:
+ if plugin.instance_id == fs.parameter.instance_id:
+ return plugin
+ return None
def poll_wifi(self):
self.wifi_manager.poll()
diff --git a/modalapi/modhandler.py b/modalapi/modhandler.py
index 2f5ea7c61..ec213fe4f 100755
--- a/modalapi/modhandler.py
+++ b/modalapi/modhandler.py
@@ -42,6 +42,8 @@
from plugins.customization import lookup as plugin_lookup
import modalapi.external_midi as ExternalMidi
from modalapi.external_midi import EXTERNAL_INSTANCE_ID
+from modalapi.led_render import LedDisplayStyle, render_led_spec
+from pistomp.category import get_category_color
from modalapi.ethernet import EthernetManager
from modalapi.jack_mute import JackMute
from pistomp.lcd320x240 import Lcd
@@ -53,9 +55,11 @@
parse_message,
LoadingEndMessage,
LoadingStartMessage,
+ OutputSetMessage,
PedalSnapshotMessage,
PluginBypassMessage,
TransportMessage,
+ BeatSyncMessage,
AddPluginMessage,
RemovePluginMessage,
ConnectMessage,
@@ -71,6 +75,7 @@
from pistomp.encoder_controller import EncoderController
from pistomp.footswitch import Footswitch
from pistomp.footswitch_chords import FootswitchChords
+from pistomp.beatsync import BeatGrid, TickState
from pistomp.input.event import (
AnalogEvent,
ControllerEvent,
@@ -82,10 +87,17 @@
from pistomp.tuner import TunerPanel, TunerSourceFactory
from pistomp.tuner.client import TunerClient
from pistomp.tuner.engine import TunerBackend, TunerEngine
-from rtmidi.midiconstants import CONTROL_CHANGE
from pathlib import Path
+_METRONOME_DOWNBEAT_RGB = (255, 255, 255)
+_METRONOME_BEAT_RGB = (180, 180, 180)
+
+
+def _now_us() -> int:
+ return int(time.clock_gettime(time.CLOCK_MONOTONIC) * 1_000_000)
+
+
class Modhandler(Handler):
__single = None
@@ -184,6 +196,9 @@ def __init__(self, audiocard: Audiocard, homedir, data_dir="/home/pistomp/data")
# Footswitch longpress/chord resolver (rebuilt on pedalboard change)
self.chord_helper = FootswitchChords()
+ self.beat_grid = BeatGrid()
+ self._taptempo_fs_cache: Footswitch | None = None
+
def cleanup(self):
if self._tuner_muted:
self.audiocard.set_output_muted(False)
@@ -310,17 +325,6 @@ def _handle_switch(self, event: SwitchEvent) -> bool:
return self._handle_footswitch(controller, event.kind, event.timestamp)
return False
- def _emit_midi(self, controller, midi_value: int) -> None:
- """Send a CC. Tries the external port if routed; falls back to virtual."""
- if controller.midi_CC is None:
- return
- cc = [controller.midi_channel | CONTROL_CHANGE, controller.midi_CC, int(midi_value)]
- port_name = self.hardware.external_port_name(controller)
- if port_name is not None and self.external_midi is not None:
- if self.external_midi.send_raw(port_name, cc):
- return
- self.hardware.midiout.send_message(cc)
-
def add_lcd(self, lcd):
self._lcd = lcd
@@ -333,11 +337,114 @@ def poll_controls(self):
if self.hardware:
self.hardware.poll_controls()
self._tick_chords()
+ # Drive footswitch LEDs in the same 10ms tick as the press so there's
+ # no latency between a state change and the LED reflecting it. Both
+ # fs.pixel and fs.led are written here — the single source of truth.
+ self._drive_footswitch_leds()
def poll_indicators(self):
if self.hardware:
self.hardware.poll_indicators()
+ def _taptempo_footswitch(self):
+ if self._taptempo_fs_cache is None and self.hardware is not None:
+ for fs in self.hardware.footswitches:
+ if fs.taptempo is not None:
+ self._taptempo_fs_cache = fs
+ break
+ return self._taptempo_fs_cache
+
+ def _drive_footswitch_leds(self, beat: TickState | None = None) -> None:
+ """Single per-tick LED driver: for each footswitch, get a (color, style)
+ frame from whichever renderer applies, then write it through the one
+ writer below. The taptempo footswitch is just another renderer — not a
+ special-cased branch — so ownership of "the pulse" lives in one place:
+ the brightness envelope in `_write_led`."""
+ if self.hardware is None:
+ return
+ if beat is None:
+ beat = self.beat_grid.tick(_now_us())
+ taptempo_fs = self._taptempo_footswitch()
+ for fs in self.hardware.footswitches:
+ if fs is taptempo_fs:
+ color, style = self._render_taptempo(fs, beat)
+ else:
+ color, style = self._render_footswitch(fs, beat)
+ self._write_led(fs, color, style, beat)
+
+ def _render_taptempo(
+ self, fs: Footswitch, beat: TickState
+ ) -> tuple[tuple[int, int, int] | None, LedDisplayStyle]:
+ """Built-in renderer for the taptempo footswitch: flashes from whichever
+ beat source is active — transport-anchored beat grid, or
+ taptempo.anchor + bpm blink when unanchored — else falls back to the
+ default per-footswitch renderer."""
+ if beat.is_anchored:
+ if beat.is_flashing:
+ return (_METRONOME_DOWNBEAT_RGB if beat.is_bar_start else _METRONOME_BEAT_RGB), LedDisplayStyle.SOLID
+ return None, LedDisplayStyle.SOLID
+ # Unanchored: if taptempo is enabled with a bpm, blink from the taptempo
+ # phase (on for the first ~100ms of each beat period). This replaces the
+ # old gpiozero hardware blink() with a 10ms-driver-computed on/off.
+ if fs.taptempo is not None and fs.taptempo.is_enabled() and fs.taptempo.get_bpm() > 0:
+ now_s = _now_us() / 1_000_000.0
+ period = 60.0 / fs.taptempo.get_bpm()
+ elapsed = now_s - fs.taptempo.anchor
+ phase_in_beat = elapsed % period
+ if phase_in_beat < 0.1: # 100ms on-window
+ return _METRONOME_BEAT_RGB, LedDisplayStyle.SOLID
+ return None, LedDisplayStyle.SOLID
+ # Taptempo disabled or no bpm yet: fall through to the default renderer.
+ return self._render_footswitch(fs, beat)
+
+ def _render_footswitch(
+ self, fs: Footswitch, beat: TickState # noqa: ARG002 - kept for renderer signature symmetry
+ ) -> tuple[tuple[int, int, int] | None, LedDisplayStyle]:
+ """Default per-footswitch renderer: a plugin's declarative LedSpec (read
+ from its generically-mirrored output_values) if bound and available,
+ else the built-in toggle + category color."""
+ plugin = self._bound_plugin(fs)
+ if plugin is not None and plugin.customization.led_spec is not None:
+ return render_led_spec(plugin.customization.led_spec, plugin.output_values)
+ if not fs.toggled:
+ return None, LedDisplayStyle.SOLID
+ color = get_category_color(fs.category) if fs.category is not None else (255, 255, 255)
+ return color, LedDisplayStyle.SOLID
+
+ def _bound_plugin(self, fs: Footswitch):
+ if fs.parameter is None or self._current is None:
+ return None
+ for plugin in self.current.pedalboard.plugins:
+ if plugin.instance_id == fs.parameter.instance_id:
+ return plugin
+ return None
+
+ @staticmethod
+ def _write_led(
+ fs: Footswitch,
+ color: tuple[int, int, int] | None,
+ style: LedDisplayStyle,
+ beat: TickState,
+ ) -> None:
+ """The one writer for fs.pixel/fs.led. Applies the metronome brightness
+ envelope uniformly for any METRONOME-style frame while transport is
+ anchored; everything else (including taptempo's already-final on/off
+ frames) is written as-is."""
+ if color is not None and style == LedDisplayStyle.METRONOME and beat.is_anchored:
+ brightness = 1.0 if beat.is_bar_start else 1.0 - (beat.beat_phase * 0.7)
+ color = (int(color[0] * brightness), int(color[1] * brightness), int(color[2] * brightness))
+ if color is None:
+ if fs.pixel is not None:
+ fs.pixel.set_enable(False)
+ if fs.led is not None:
+ fs.led.off()
+ return
+ if fs.pixel is not None:
+ fs.pixel.set_color(color)
+ fs.pixel.set_enable(True)
+ if fs.led is not None:
+ fs.led.on()
+
def poll_wifi(self):
self.wifi_manager.poll()
if self._lcd is not None and self.lcd.wifi_menu is not None:
@@ -614,6 +721,11 @@ def _handle_ws_message(self, msg: WebSocketMessage):
if self.hardware.taptempo.is_enabled():
fs = next((f for f in self.hardware.footswitches if f.taptempo is self.hardware.taptempo), None)
self.update_lcd_fs(footswitch=fs)
+ if not msg.rolling:
+ self.beat_grid.clear()
+
+ elif isinstance(msg, BeatSyncMessage):
+ self.beat_grid.on_anchor(msg)
elif isinstance(msg, ParamSetMessage):
# Mirror mod-ui's live value: refresh the cache (so a later edit opens
@@ -633,6 +745,13 @@ def _handle_ws_message(self, msg: WebSocketMessage):
# MIDI learn in mod-ui assigned a hardware control to a parameter.
self._apply_midi_binding(msg.instance, msg.symbol, msg.binding)
+ elif isinstance(msg, OutputSetMessage):
+ if self._current is not None:
+ for plugin in self.current.pedalboard.plugins:
+ if plugin.instance_id == msg.instance:
+ plugin.set_output_value(msg.symbol, msg.value)
+ break
+
def _handle_dynamic_plugin_add(self, msg: AddPluginMessage) -> None:
"""Handle an `add` WS message for a plugin not yet in the pedalboard model."""
assert self._current is not None
@@ -892,6 +1011,23 @@ 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._update_interesting_outputs()
+
+ def _update_interesting_outputs(self) -> None:
+ """Recompute the WS output_set subscription set from the pedalboard's
+ plugins (their own declared LedSpec outputs) — the plugin is the
+ natural owner of its output ports, not whichever footswitch happens to
+ be bound to it. Computed once at pedalboard load; a footswitch binding
+ change afterward can't add or remove monitored outputs since those are
+ fixed per plugin instance."""
+ if self._current is None:
+ self.ws_bridge.set_interesting_outputs(frozenset())
+ return
+ keys: set[str] = set()
+ for plugin in self.current.pedalboard.plugins:
+ for sym in plugin.monitored_output_symbols:
+ keys.add(f"{plugin.instance_id}/{sym}")
+ self.ws_bridge.set_interesting_outputs(frozenset(keys))
def _redraw_after_binding(self, controller, is_footswitch):
if is_footswitch:
diff --git a/modalapi/plugin.py b/modalapi/plugin.py
index bb5b6b93e..8c52e1b63 100755
--- a/modalapi/plugin.py
+++ b/modalapi/plugin.py
@@ -62,6 +62,10 @@ def __init__(
self.category: str | None = category
self.uri: str | None = uri
self.pedalboard_snapshot: dict[str, float] = {}
+ # Generic mirror of this plugin's subscribed lv2:OutputPort values (see
+ # `monitored_output_symbols`). Populated from WS `output_set` messages;
+ # consumed by the LED driver's LedSpec lookups. No footswitch involved.
+ self.output_values: dict[str, float] = {}
c: PluginCustomization = customization or PluginCustomization()
if extra_data is not None:
c = replace(c, extra_data=extra_data)
@@ -71,6 +75,22 @@ def __init__(
def extra_data(self) -> PluginExtraData | None:
return self.customization.extra_data
+ @property
+ def monitored_output_symbols(self) -> tuple[str, ...]:
+ """Output-port symbols this plugin wants mirrored via WS output_set,
+ derived from its LedSpec (if any). Generic — no footswitch involved."""
+ spec = self.customization.led_spec
+ if spec is None:
+ return ()
+ symbols = [spec.state_symbol]
+ if spec.downbeat_symbol is not None:
+ symbols.append(spec.downbeat_symbol)
+ return tuple(symbols)
+
+ def set_output_value(self, symbol: str, value: float) -> None:
+ """Cache a subscribed lv2:OutputPort value (from WS output_set)."""
+ self.output_values[symbol] = value
+
@property
def display_name(self) -> str:
c = self.customization
diff --git a/modalapi/plugin_customization.py b/modalapi/plugin_customization.py
index da8c98bc4..3782682aa 100644
--- a/modalapi/plugin_customization.py
+++ b/modalapi/plugin_customization.py
@@ -27,6 +27,29 @@ def extra_data_as(plugin: Plugin, kind: type[_TExtra]) -> _TExtra | None:
return data if isinstance(data, kind) else None
+@dataclass(frozen=True)
+class LedSpec:
+ """Declarative footswitch-LED rendering for a plugin, keyed off its own
+ (generically-mirrored) output ports. Interpreted by the handler's generic
+ LED driver — no per-plugin imperative code required.
+
+ state_symbol: the output port whose integer value selects `colors`.
+ downbeat_symbol: an optional second output port (e.g. loopjefe's
+ `measure_number`) whose value == 0 means "this is the loop's own
+ downbeat" — brightens the color by `downbeat_tint` per channel.
+ off_states / steady_states: state values that render as off, or as a
+ steady (non-pulsing) color even when `pulse` is True.
+ """
+
+ state_symbol: str
+ colors: dict[int, tuple[int, int, int]]
+ pulse: bool = False
+ off_states: frozenset[int] = frozenset()
+ steady_states: frozenset[int] = frozenset()
+ downbeat_symbol: str | None = None
+ downbeat_tint: int = 60
+
+
@dataclass(frozen=True)
class PluginCustomization:
panel_cls: type[PluginPanel] | None = None
@@ -37,6 +60,7 @@ class PluginCustomization:
tile_active_color: tuple[int, int, int] | None = None
tile_border: RectBorder | None = None
extra_data: PluginExtraData | None = None
+ led_spec: LedSpec | None = None
class Customizer(Protocol):
diff --git a/modalapi/websocket_bridge.py b/modalapi/websocket_bridge.py
index 9902f7bb1..3b21a2221 100644
--- a/modalapi/websocket_bridge.py
+++ b/modalapi/websocket_bridge.py
@@ -45,7 +45,7 @@ class WebSocketWorker:
"""
def __init__(
- self, ws_url: str, backpressure_threshold: int, command_queue: queue.Queue, received_queue: queue.Queue
+ self, ws_url: str, backpressure_threshold: int, command_queue: queue.Queue, received_queue: queue.Queue,
):
self.ws_url = ws_url
self.backpressure_threshold = backpressure_threshold
@@ -55,6 +55,12 @@ def __init__(
self.ws = None
self._loop: Optional[asyncio.AbstractEventLoop] = None
self._stop_event: asyncio.Event = asyncio.Event()
+ # Atomically-swappable set of "instance/symbol" keys whose output_set
+ # frames survive the prefix drop. Owned by the worker so it doesn't
+ # need a back-reference to the bridge. Swapped from the main thread
+ # via set_interesting_outputs; read here on the worker thread. The
+ # frozenset ref-swap is atomic under the GIL (CPython only).
+ self._interesting: frozenset[str] = frozenset()
# Metrics
self.messages_sent = 0
@@ -205,7 +211,20 @@ async def _receive_messages(self, ws):
await ws.send(message)
continue
elif message.startswith("output_set "):
- continue # audio-meter flood; nothing consumes it, drop before it floods the queue
+ # Keep only if a footswitch behavior subscribed to this output.
+ interesting = self._interesting
+ if interesting:
+ parts = message.split(" ", 3)
+ if len(parts) >= 3:
+ path = parts[1]
+ symbol = parts[2]
+ inst = path.removeprefix("/graph/")
+ key = f"{inst}/{symbol}"
+ if key in interesting:
+ self.received_queue.put(message)
+ self.messages_received += 1
+ logging.debug(f"Received subscribed output_set: {message[:100]}")
+ continue
self.received_queue.put(message)
self.messages_received += 1
logging.debug(f"Received message from server: {message[:100]}")
@@ -214,6 +233,12 @@ async def _receive_messages(self, ws):
except Exception as e:
logging.error(f"Error receiving message: {e}")
+ def set_interesting_outputs(self, keys: frozenset[str]) -> None:
+ """Atomically swap the set of 'instance/symbol' keys whose output_set
+ frames survive the prefix drop. Called from the main thread on
+ pedalboard load/rebind. Thread-safe under the GIL (frozenset ref swap)."""
+ self._interesting = keys
+
def _get_write_buffer_size(self, ws) -> int:
"""Return bytes waiting in the TCP write buffer, or 0 if unavailable."""
try:
@@ -286,6 +311,10 @@ def get_received_messages(self) -> list:
def get_queue_depth(self) -> int:
return self.command_queue.qsize()
+ def set_interesting_outputs(self, keys: frozenset[str]) -> None:
+ """Delegate to the worker, which owns the interesting-set."""
+ self._worker.set_interesting_outputs(keys)
+
def get_stats(self) -> dict:
stats = {
"queue_depth": self.get_queue_depth(),
diff --git a/modalapi/ws_protocol.py b/modalapi/ws_protocol.py
index 0d30e4cdd..654ebbccb 100644
--- a/modalapi/ws_protocol.py
+++ b/modalapi/ws_protocol.py
@@ -96,14 +96,31 @@ class TransportMessage:
bpm: float
+@dataclass
+class BeatSyncMessage:
+ """A sample of the transport clock (t_us=now, CLOCK_MONOTONIC) — not a
+ back-dated downbeat event. Consumers forward-extrapolate
+ pos(t) = beat_in_bar + (t - t_us) * bpm / 60, so cadence controls
+ tightness, never correctness; each sample fully replaces any prior
+ anchor. Emitted on a new bar (heartbeat) and on any discrete bpm/bpb
+ change while rolling. No absolute bar count — that's DAW-context mod-host
+ doesn't need to expose; only the fractional position within the current
+ bar matters for phase/downbeat math."""
+
+ t_us: int
+ bpm: float
+ bpb: float
+ beat_in_bar: float
+
+
@dataclass
class AddPluginMessage:
"""Plugin present in a (re)connect/load dump, or dynamically added (add ...)."""
instance: str # canonical bare form, e.g. "CollisionDrive"
- uri: str # LV2 plugin URI
- x: float # mod-ui canvas X
- y: float # mod-ui canvas Y
+ uri: str # LV2 plugin URI
+ x: float # mod-ui canvas X
+ y: float # mod-ui canvas Y
bypassed: bool
@@ -119,7 +136,7 @@ class ConnectMessage:
"""Two ports connected in the active pedalboard (connect ...)."""
port_from: str # e.g. "/graph/PluginA/out_L"
- port_to: str # e.g. "/graph/PluginB/in_L"
+ port_to: str # e.g. "/graph/PluginB/in_L"
@dataclass
@@ -139,6 +156,15 @@ class ParamSetMessage:
value: float
+@dataclass
+class OutputSetMessage:
+ """A plugin output-port value changed (output_set)."""
+
+ instance: str
+ symbol: str
+ value: float
+
+
@dataclass
class MidiMapMessage:
"""A MIDI binding was learned/assigned in mod-ui (midi_map ...)."""
@@ -172,11 +198,13 @@ class UnknownMessage:
TrueBypassMessage,
PluginBypassMessage,
TransportMessage,
+ BeatSyncMessage,
AddPluginMessage,
RemovePluginMessage,
ConnectMessage,
DisconnectMessage,
ParamSetMessage,
+ OutputSetMessage,
MidiMapMessage,
UnknownMessage,
]
@@ -281,6 +309,12 @@ def parse_message(raw_message: str) -> WebSocketMessage:
symbol, value_str = rest.split(" ", 1)
return ParamSetMessage(instance=instance, symbol=symbol, value=float(value_str))
+ # Format: output_set /graph/{instance} {symbol} {value}
+ case ["output_set", path, rest]:
+ instance = path.removeprefix("/graph/")
+ symbol, value_str = rest.split(" ", 1)
+ return OutputSetMessage(instance=instance, symbol=symbol, value=float(value_str))
+
# Format: midi_map /graph/{instance} {symbol} {channel} {controller} {min} {max}
case ["midi_map", path, rest]:
symbol, ch, ctrl = rest.split(" ")[:3]
@@ -304,6 +338,16 @@ def parse_message(raw_message: str) -> WebSocketMessage:
bpm = float(rest.split()[1])
return TransportMessage(rolling=rolling != "0", bpm=bpm)
+ # Format: beat_sync {t_us} {bpm} {bpb} {beat_in_bar}
+ case ["beat_sync", t_us, rest]:
+ bpm, bpb, beat_in_bar = rest.split(" ")
+ return BeatSyncMessage(
+ t_us=int(t_us),
+ bpm=float(bpm),
+ bpb=float(bpb),
+ beat_in_bar=float(beat_in_bar),
+ )
+
except (ValueError, IndexError) as e:
logging.warning(f"Failed to parse WebSocket message '{raw_message}': {e}")
return UnknownMessage(raw=raw_message)
diff --git a/pistomp/beatsync.py b/pistomp/beatsync.py
new file mode 100644
index 000000000..413b09a32
--- /dev/null
+++ b/pistomp/beatsync.py
@@ -0,0 +1,127 @@
+# This file is part of pi-stomp.
+#
+# pi-stomp is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# pi-stomp is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with pi-stomp. If not, see .
+
+from dataclasses import dataclass
+
+from modalapi.ws_protocol import BeatSyncMessage
+
+
+FLASH_US = 80_000
+STALE_AFTER_US = 5_000_000
+# A clock sample landing within this many beats of a boundary is treated as
+# the crossing itself (arms the flash/bar-start immediately) rather than
+# waiting for a later tick to detect it — this is what makes a downbeat
+# sample's own arrival distinguishable, fixing the old bug where the seeded
+# anchor position was never "crossed" because it was the modulo target itself.
+_ANCHOR_CROSSING_EPSILON_BEATS = 0.05
+
+
+@dataclass(frozen=True)
+class TickState:
+ is_anchored: bool
+ is_flashing: bool
+ is_bar_start: bool
+ bpm: float
+ bpb: float
+ beat_phase: float = 0.0 # normalized [0, 1) within the current beat
+
+
+class BeatGrid:
+ """Tracks the transport clock from a stream of `BeatSyncMessage` clock
+ samples: pos(t) = beat_in_bar + (t - t_us) * bpm / 60, anchored fresh from
+ each sample's own beat_in_bar (no cumulative bar count needed — mod-host
+ doesn't expose one). Downbeat is *computed* from this position
+ (`beat_index % bpb == 0`), not reconstructed from message-arrival timing —
+ so it's correct regardless of emission cadence, and self-healing: the
+ latest sample fully replaces any prior anchor, so a dropped/late one just
+ means more extrapolation, never a wrong lock."""
+
+ def __init__(self) -> None:
+ self._anchor_t_us: int | None = None
+ self._anchor_pos: float = 0.0
+ self._bpm: float = 120.0
+ self._bpb: float = 4.0
+ self._last_beat_idx: int = 0
+ self._flash_end_us: int | None = None
+ self._last_crossing_was_bar_start: bool = False
+
+ @property
+ def is_anchored(self) -> bool:
+ return self._anchor_t_us is not None
+
+ def on_anchor(self, msg: BeatSyncMessage) -> None:
+ if msg.bpm <= 0 or msg.bpb <= 0:
+ self.clear()
+ return
+ self._anchor_t_us = msg.t_us
+ self._anchor_pos = msg.beat_in_bar
+ self._bpm = msg.bpm
+ self._bpb = msg.bpb
+ self._flash_end_us = None
+ self._last_crossing_was_bar_start = False
+
+ current_beat_idx = int(self._anchor_pos // 1)
+ frac = self._anchor_pos - current_beat_idx
+ if frac < _ANCHOR_CROSSING_EPSILON_BEATS:
+ # This sample lands right at (or just past) a beat boundary — the
+ # crossing already happened at anchor time. Seed one beat behind
+ # so the very first tick() call (even at the anchor's own
+ # timestamp) detects the crossing and arms the flash/bar-start,
+ # instead of never detecting it because it *is* the modulo target.
+ self._last_beat_idx = current_beat_idx - 1
+ else:
+ self._last_beat_idx = current_beat_idx
+
+ def clear(self) -> None:
+ self._anchor_t_us = None
+ self._anchor_pos = 0.0
+ self._last_beat_idx = 0
+ self._flash_end_us = None
+ self._last_crossing_was_bar_start = False
+
+ def tick(self, now_us: int) -> TickState:
+ if self._anchor_t_us is None:
+ return TickState(False, False, False, self._bpm, self._bpb)
+
+ if self._bpm <= 0 or self._bpb <= 0:
+ self.clear()
+ return TickState(False, False, False, self._bpm, self._bpb)
+
+ if now_us - self._anchor_t_us > STALE_AFTER_US:
+ self.clear()
+ return TickState(False, False, False, self._bpm, self._bpb)
+
+ bpb_int = int(self._bpb)
+ delta_us = now_us - self._anchor_t_us
+ pos = self._anchor_pos + delta_us * self._bpm / 60_000_000.0
+ current_beat_idx = int(pos // 1)
+ beat_phase = pos - current_beat_idx # fractional part [0, 1)
+
+ if current_beat_idx > self._last_beat_idx:
+ self._last_beat_idx = current_beat_idx
+ self._flash_end_us = now_us + FLASH_US
+ self._last_crossing_was_bar_start = (current_beat_idx % bpb_int) == 0
+
+ is_flashing = (
+ self._flash_end_us is not None and now_us < self._flash_end_us
+ )
+ return TickState(
+ is_anchored=True,
+ is_flashing=is_flashing,
+ is_bar_start=is_flashing and self._last_crossing_was_bar_start,
+ bpm=self._bpm,
+ bpb=self._bpb,
+ beat_phase=beat_phase,
+ )
diff --git a/pistomp/config.py b/pistomp/config.py
index e587b5fd7..095f8a51a 100644
--- a/pistomp/config.py
+++ b/pistomp/config.py
@@ -25,6 +25,15 @@
DEFAULT_CONFIG_FILE = "default_config.yml"
+LONGPRESS_GROUP_NAMES = [
+ "next_snapshot",
+ "previous_snapshot",
+ "toggle_bypass",
+ "set_mod_tap_tempo",
+ "toggle_tap_tempo_enable",
+ "toggle_tuner_enable",
+]
+
schema = {
"$schema": "http://json-schema.org/draft-04/schema#",
"type": "object",
@@ -83,11 +92,30 @@
"type": "integer"
},
"longpress": {
- "type" : ["array", "string"],
- "items" : {
- "type" : "string",
- "enum" : ["next_snapshot", "previous_snapshot", "toggle_bypass", "set_mod_tap_tempo", "toggle_tap_tempo_enable"]
- }
+ "oneOf": [
+ {
+ "type": "string",
+ "enum": LONGPRESS_GROUP_NAMES
+ },
+ {
+ "type": "array",
+ "items": {
+ "type": "string",
+ "enum": LONGPRESS_GROUP_NAMES
+ }
+ },
+ {
+ "type": "object",
+ "description": "Send this CC (value 127) on longpress instead of resolving a named longpress group.",
+ "properties": {
+ "midi_CC": {
+ "type": "integer"
+ }
+ },
+ "required": ["midi_CC"],
+ "additionalProperties": False
+ }
+ ]
},
"midi_CC": {
"type": "integer"
diff --git a/pistomp/footswitch.py b/pistomp/footswitch.py
index c5cab07e0..d922f7faf 100755
--- a/pistomp/footswitch.py
+++ b/pistomp/footswitch.py
@@ -42,6 +42,7 @@ def __init__(self, id: int | None, led_pin, pixel, midi_CC, midi_channel, refres
self.category = None
self.pixel = pixel
self.longpress_groups = []
+ self.longpress_midi_CC = None
self.disabled = False
self.taptempo = taptempo
@@ -112,28 +113,27 @@ def toggle_relays(self, enabled: bool):
r.disable()
def set_led(self, enabled):
- if self.led is not None:
- if self.taptempo:
- tempo = self.taptempo.get_bpm()
- if tempo:
- period = 60/tempo
- on = 0.1
- self.led.blink(on_time=on, off_time=period - 0.1)
- elif enabled:
- self.led.on()
- else:
- self.led.off()
- if self.pixel:
- self.pixel.set_enable(enabled)
+ """Pure state update — flips fs.toggled only. The per-tick LED driver
+ (_drive_footswitch_leds in poll_controls) renders the new state to both
+ fs.pixel and fs.led on the next 10ms tick. No hardware writes here."""
+ self.toggled = enabled
def set_category(self, category):
self.category = category
- if self.pixel:
- self.pixel.set_color_by_category(category, self.toggled)
def set_lcd_color(self, color):
self.lcd_color = color
+ def set_longpress(self, value):
+ """Apply the raw 'longpress' config value: a group name, a list of
+ group names, a {midi_CC: N} object, or None."""
+ if isinstance(value, dict):
+ self.longpress_midi_CC = value.get(Token.MIDI_CC)
+ self.longpress_groups = []
+ else:
+ self.longpress_midi_CC = None
+ self.set_longpress_groups(value)
+
def set_longpress_groups(self, groups):
if groups is None:
self.longpress_groups = []
diff --git a/pistomp/handler.py b/pistomp/handler.py
index aaad27fb2..a5600dab0 100755
--- a/pistomp/handler.py
+++ b/pistomp/handler.py
@@ -16,8 +16,11 @@
from __future__ import annotations
+import logging
from typing import TYPE_CHECKING, Any
+from rtmidi.midiconstants import CONTROL_CHANGE
+
from pistomp.analogmidicontrol import AnalogMidiControl
from pistomp.current import Current
from pistomp.encoder_controller import EncoderController
@@ -95,6 +98,13 @@ def add_hardware(self, hardware):
def poll_controls(self):
raise NotImplementedError()
+ def _drive_footswitch_leds(self) -> None:
+ """Render footswitch LEDs from behaviors. Base implementation is a no-op;
+ Modhandler overrides with the beat-aware driver. Called from
+ poll_controls so the LED update happens in the same 10ms tick as the
+ press that triggered it."""
+ return
+
def poll_modui_changes(self):
raise NotImplementedError()
@@ -143,8 +153,11 @@ def _handle_footswitch(self, fs: "Footswitch", kind: SwitchEventKind, timestamp:
fs.toggle_relays(new_toggled)
fs.set_led(new_toggled)
self.update_lcd_fs(bypass_change=True)
+ elif fs.longpress_midi_CC is not None:
+ # Momentary trigger CC, distinct from the short-press binding
+ # (e.g. a plugin's "reset" port learned to this CC).
+ self._emit_midi(fs, 127, cc=fs.longpress_midi_CC)
else:
- # TODO: consider case where relay and longpress are specified
self.chord_helper.observe(fs, timestamp)
return True
@@ -159,9 +172,12 @@ def _handle_footswitch(self, fs: "Footswitch", kind: SwitchEventKind, timestamp:
fs.preset_callback()
return True
if fs.midi_CC is not None:
- fs.toggled = not fs.toggled
- fs.set_led(fs.toggled)
- self._emit_midi(fs, 127 if fs.toggled else 0)
+ if fs.parameter is not None and fs.parameter.is_momentary:
+ self._emit_midi(fs, 127)
+ else:
+ fs.toggled = not fs.toggled
+ fs.set_led(fs.toggled)
+ self._emit_midi(fs, 127 if fs.toggled else 0)
if fs.parameter is not None:
fs.parameter.value = not fs.toggled # FIXME: assumes mapped parameter is :bypass
self.update_lcd_fs(footswitch=fs)
@@ -174,8 +190,23 @@ def _tick_chords(self) -> None:
if cb:
cb()
- def _emit_midi(self, controller, midi_value: int) -> None:
- raise NotImplementedError()
+ def _emit_midi(self, controller, midi_value: int, cc: int | None = None) -> None:
+ """Send a CC via this controller's binding, or an explicit override CC
+ (e.g. a footswitch's longpress trigger, distinct from its short-press
+ binding). Tries the controller's routed external port; falls back to
+ the virtual MIDI Through port."""
+ cc_num = cc if cc is not None else controller.midi_CC
+ if cc_num is None:
+ return
+ message = [controller.midi_channel | CONTROL_CHANGE, cc_num, int(midi_value)]
+ port_name = self.hardware.external_port_name(controller)
+ if port_name is not None and self.hardware.external_midi is not None:
+ try:
+ if self.hardware.external_midi.send_raw(port_name, message):
+ return
+ except Exception as e:
+ logging.warning("External CC send failed on %s: %s", port_name, e)
+ self.hardware.midiout.send_message(message)
def cleanup(self):
raise NotImplementedError()
@@ -245,6 +276,8 @@ def _apply_midi_binding(self, instance, symbol, binding):
param.binding = binding
is_footswitch = self._bind_controller_to_param(plugin, param, controller)
self._redraw_after_binding(controller, is_footswitch)
+ if is_footswitch:
+ self._on_footswitch_binding_changed()
def _bind_controller_to_param(self, plugin, param, controller) -> bool:
# Wire a hardware controller to a plugin parameter. Returns True if the
@@ -267,6 +300,12 @@ def _bind_controller_to_param(self, plugin, param, controller) -> bool:
self.current.analog_controllers[key] = display_info
return False
+ def _on_footswitch_binding_changed(self) -> None:
+ """Hook fired after a live MIDI-learn binds a footswitch to a plugin.
+ Subclasses with output_set subscriptions (Modhandler) override to
+ recompute the WS interesting-set. Base no-op for v1 (Mod)."""
+ return
+
def _redraw_after_binding(self, controller, is_footswitch):
# Refresh the LCD after a learned binding. Subclasses redraw at their
# own granularity.
diff --git a/pistomp/hardware.py b/pistomp/hardware.py
index f40e0390c..4db79a9ff 100755
--- a/pistomp/hardware.py
+++ b/pistomp/hardware.py
@@ -512,9 +512,9 @@ def __init_footswitches(self, cfg):
if Token.COLOR in f:
fs.set_lcd_color(f[Token.COLOR])
- # Longpress and longpress groups
- if Token.LONGPRESS in f: # Can be a list or a single (string)
- fs.set_longpress_groups(Util.DICT_GET(f, Token.LONGPRESS))
+ # Longpress: a group name, a list of group names, or {midi_CC: N}
+ if Token.LONGPRESS in f:
+ fs.set_longpress(Util.DICT_GET(f, Token.LONGPRESS))
idx += 1
diff --git a/plugins/__init__.py b/plugins/__init__.py
index ab63af1e9..c3b9c65d0 100644
--- a/plugins/__init__.py
+++ b/plugins/__init__.py
@@ -49,6 +49,7 @@
import plugins.system_compressor # noqa: F401
import plugins.multiband_menu # noqa: F401 # MultibandWindow base
import plugins.layouts # noqa: F401 # Layout components
+import plugins.loopjefe # noqa: F401 # LoopJefe footswitch behavior
__all__ = [
"PluginCustomization",
diff --git a/plugins/loopjefe/__init__.py b/plugins/loopjefe/__init__.py
new file mode 100644
index 000000000..bda9e7809
--- /dev/null
+++ b/plugins/loopjefe/__init__.py
@@ -0,0 +1,50 @@
+"""LoopJefe multitrack looper plugin customization.
+
+Declarative footswitch-LED spec only: state colors + loop-downbeat tint,
+interpreted by the handler's generic LED driver (modalapi/led_render.py).
+Momentary press semantics come for free from `advance`/`reset` being
+`pprops:trigger` ports (common/parameter.py) — no plugin-specific input code.
+"""
+
+from __future__ import annotations
+
+from modalapi.plugin_customization import LedSpec, PluginCustomization
+from plugins.customization import register
+
+LOOPJEFE_URIS = (
+ "http://treefallsound.com/plugins/loopjefe",
+ "http://treefallsound.com/plugins/loopjefe-2x2",
+)
+
+# LoopJefePlugin state values (../loopjefe-lv2/src/types.h)
+_STATE_EMPTY = 0
+_STATE_STOPPED = 5
+
+_STATE_COLORS: dict[int, tuple[int, int, int]] = {
+ 1: (0, 80, 255), # Record Arm
+ 2: (255, 0, 0), # Recording
+ 3: (0, 80, 255), # Record Close
+ 4: (0, 255, 0), # Playback
+ _STATE_STOPPED: (80, 80, 80),
+ 6: (0, 80, 255), # Overdub Arm
+ 7: (255, 140, 0), # Overdub
+ 8: (0, 80, 255), # Overdub Close
+}
+
+_LOOPJEFE_LED_SPEC = LedSpec(
+ state_symbol="state",
+ colors=_STATE_COLORS,
+ pulse=True,
+ off_states=frozenset({_STATE_EMPTY}),
+ steady_states=frozenset({_STATE_STOPPED}),
+ downbeat_symbol="measure_number",
+ downbeat_tint=60,
+)
+
+register(
+ *LOOPJEFE_URIS,
+ customization=PluginCustomization(
+ display_name="LoopJefe",
+ led_spec=_LOOPJEFE_LED_SPEC,
+ ),
+)
diff --git a/tests/conftest.py b/tests/conftest.py
index 33e7f3e39..dd094493f 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -156,6 +156,7 @@ class FakeWebSocketBridge:
def __init__(self):
self.sent: list[str] = []
self._inbox: list[str] = []
+ self.interesting_calls: list[frozenset[str]] = []
def start(self) -> None:
pass
@@ -174,6 +175,9 @@ def send_bpm(self, bpm: float) -> bool:
def clear_queue(self) -> int:
return 0
+ def set_interesting_outputs(self, keys: frozenset[str]) -> None:
+ self.interesting_calls.append(keys)
+
def get_received_messages(self) -> list[str]:
msgs, self._inbox = self._inbox, []
return msgs
diff --git a/tests/input_router/test_handle_footswitch_longpress.py b/tests/input_router/test_handle_footswitch_longpress.py
new file mode 100644
index 000000000..ae96d5d7c
--- /dev/null
+++ b/tests/input_router/test_handle_footswitch_longpress.py
@@ -0,0 +1,135 @@
+"""Handler._handle_footswitch longpress dispatch: relay > longpress_midi_CC > chord group.
+
+Uses the real _handle_footswitch from pistomp.handler.Handler with a mocked
+Hardware, so the priority order between the three longpress behaviors is
+exercised as production code, not re-implemented.
+"""
+
+from unittest.mock import MagicMock
+
+from common.parameter import Parameter, Type
+from pistomp.footswitch import Footswitch
+from pistomp.handler import Handler
+from pistomp.input.event import SwitchEventKind
+
+
+class _TestHandler(Handler):
+ def __init__(self, hw):
+ super().__init__()
+ self.hardware = hw
+
+ def update_lcd_fs(self, footswitch=None, bypass_change=False):
+ pass
+
+ def get_callback(self, callback_name):
+ return None
+
+ def handle(self, event):
+ raise NotImplementedError
+
+
+def _make_handler():
+ hw = MagicMock()
+ hw.external_port_name.return_value = None
+ hw.external_midi = None
+ return _TestHandler(hw), hw
+
+
+def _make_footswitch(**kwargs):
+ return Footswitch(
+ id=kwargs.get("id", 1),
+ led_pin=None,
+ pixel=None,
+ midi_CC=kwargs.get("midi_CC", 10),
+ midi_channel=kwargs.get("midi_channel", 0),
+ refresh_callback=lambda **kw: None,
+ )
+
+
+class TestLongpressMidiCC:
+ def test_sends_longpress_cc_not_short_press_cc(self):
+ handler, hw = _make_handler()
+ fs = _make_footswitch(midi_CC=10)
+ fs.longpress_midi_CC = 65
+
+ handler._handle_footswitch(fs, SwitchEventKind.LONGPRESS, timestamp=1.0)
+
+ hw.midiout.send_message.assert_called_once()
+ message = hw.midiout.send_message.call_args[0][0]
+ assert message[1] == 65
+ assert message[2] == 127
+
+ def test_relay_takes_priority_over_longpress_midi_cc(self):
+ handler, hw = _make_handler()
+ fs = _make_footswitch()
+ fs.longpress_midi_CC = 65
+ relay = MagicMock()
+ relay.init_state.return_value = False
+ fs.add_relay(relay)
+
+ handler._handle_footswitch(fs, SwitchEventKind.LONGPRESS, timestamp=1.0)
+
+ hw.midiout.send_message.assert_not_called()
+ assert relay.enable.called or relay.disable.called
+
+ def test_no_longpress_midi_cc_falls_through_to_chord(self):
+ handler, hw = _make_handler()
+ fs = _make_footswitch()
+ fs.longpress_groups = ["toggle_bypass"]
+
+ handler._handle_footswitch(fs, SwitchEventKind.LONGPRESS, timestamp=1.0)
+
+ hw.midiout.send_message.assert_not_called()
+
+
+def _make_parameter(port_type: Type) -> Parameter:
+ info = {"symbol": "advance", "ranges": {"minimum": 0, "maximum": 1}}
+ p = Parameter(info, value=0.0, binding=None, instance_id="loopjefe")
+ p.type = port_type
+ return p
+
+
+class TestMomentaryShortPress:
+ """A footswitch bound to a pprops:trigger port (Parameter.is_momentary)
+ emits 127 every press (rising-edge trigger semantics) — no toggled flip.
+ Otherwise the existing toggle behavior (127/0 alternating) is preserved."""
+
+ def test_momentary_emits_127_every_press(self):
+ handler, hw = _make_handler()
+ fs = _make_footswitch(midi_CC=10)
+ fs.parameter = _make_parameter(Type.TRIGGER)
+
+ for _ in range(3):
+ handler._handle_footswitch(fs, SwitchEventKind.PRESS, timestamp=1.0)
+
+ messages = [c.args[0] for c in hw.midiout.send_message.call_args_list]
+ assert len(messages) == 3
+ assert all(m[2] == 127 for m in messages)
+ assert fs.toggled is False # never flipped
+
+ def test_non_momentary_toggles_as_before(self):
+ handler, hw = _make_handler()
+ fs = _make_footswitch(midi_CC=10)
+ fs.parameter = _make_parameter(Type.DEFAULT)
+
+ handler._handle_footswitch(fs, SwitchEventKind.PRESS, timestamp=1.0)
+ handler._handle_footswitch(fs, SwitchEventKind.PRESS, timestamp=2.0)
+
+ messages = [c.args[0] for c in hw.midiout.send_message.call_args_list]
+ assert len(messages) == 2
+ assert messages[0][2] == 127 # first press → toggled True → 127
+ assert messages[1][2] == 0 # second press → toggled False → 0
+ assert fs.toggled is False
+
+ def test_no_parameter_toggles_as_before(self):
+ """Regression guard: a footswitch with parameter=None (e.g. a preset
+ switch that never went through ControllerManager.bind) still toggles."""
+ handler, hw = _make_handler()
+ fs = _make_footswitch(midi_CC=10)
+ fs.parameter = None
+
+ handler._handle_footswitch(fs, SwitchEventKind.PRESS, timestamp=1.0)
+
+ message = hw.midiout.send_message.call_args.args[0]
+ assert message[2] == 127
+ assert fs.toggled is True
diff --git a/tests/integration/test_external_midi_loopback.py b/tests/integration/test_external_midi_loopback.py
index fe0d83ef2..cae0c13d7 100644
--- a/tests/integration/test_external_midi_loopback.py
+++ b/tests/integration/test_external_midi_loopback.py
@@ -24,12 +24,13 @@
class _FakeHardware(Hardware):
"""Minimal hardware stub: just enough for _emit_midi to resolve routing."""
- def __init__(self, port_name: str):
+ def __init__(self, port_name: str, mgr: ExternalMidiManager):
# Hardware.__init__ takes (default_config, handler, midiout, refresh_callback);
# skip it — this fake doesn't need a config or real handler/refresh.
self._port_name = port_name
self.midiout = MagicMock()
self.external_routing: dict = {}
+ self.external_midi = mgr
def external_port_name(self, controller: Controller) -> str | None:
info = self.external_routing.get(controller)
@@ -53,22 +54,11 @@ class _LoopbackHandler(Handler):
footswitch path exercises production code.
"""
- def __init__(self, hw: _FakeHardware, mgr: ExternalMidiManager):
+ def __init__(self, hw: _FakeHardware):
super().__init__()
self.hardware = hw
- self.external_midi = mgr
# chord_helper is set by Handler.__init__; we leave it as-is.
- def _emit_midi(self, controller, midi_value: int) -> None:
- if controller.midi_CC is None:
- return
- cc = [controller.midi_channel | CONTROL_CHANGE, controller.midi_CC, int(midi_value)]
- port_name = self.hardware.external_port_name(controller)
- if port_name is not None:
- if self.external_midi.send_raw(port_name, cc):
- return
- self.hardware.midiout.send_message(cc)
-
def update_lcd_fs(self, footswitch=None, bypass_change=False):
pass
@@ -208,8 +198,8 @@ def test_footswitch_press_reaches_real_port(self, loopback):
mgr = _manager_for(port_name)
mgr.open_port(port_name)
- hw = _FakeHardware(port_name)
- handler = _LoopbackHandler(hw, mgr)
+ hw = _FakeHardware(port_name, mgr)
+ handler = _LoopbackHandler(hw)
fs = Footswitch(id=0, led_pin=None, pixel=None, midi_CC=75,
midi_channel=0xB0, refresh_callback=lambda **kw: None)
@@ -226,8 +216,8 @@ def test_tweak_encoder_rotation_reaches_real_port(self, loopback):
mgr = _manager_for(port_name)
mgr.open_port(port_name)
- hw = _FakeHardware(port_name)
- handler = _LoopbackHandler(hw, mgr)
+ hw = _FakeHardware(port_name, mgr)
+ handler = _LoopbackHandler(hw)
enc = EncoderController(d_pin=None, clk_pin=None, midi_channel=0xB0, midi_CC=7)
hw.external_routing[enc] = RoutingInfo.external(port_name)
@@ -244,8 +234,8 @@ def test_expression_movement_reaches_real_port(self, loopback):
mgr = _manager_for(port_name)
mgr.open_port(port_name)
- hw = _FakeHardware(port_name)
- handler = _LoopbackHandler(hw, mgr)
+ hw = _FakeHardware(port_name, mgr)
+ handler = _LoopbackHandler(hw)
spi = MagicMock()
spi.readChannel.return_value = 512
diff --git a/tests/test_beatsync.py b/tests/test_beatsync.py
new file mode 100644
index 000000000..49a539aa9
--- /dev/null
+++ b/tests/test_beatsync.py
@@ -0,0 +1,225 @@
+"""BeatGrid — anchor + tick math for the metronome LED scheduler."""
+
+from modalapi.ws_protocol import BeatSyncMessage
+from pistomp.beatsync import FLASH_US, STALE_AFTER_US, BeatGrid, TickState
+
+
+def _anchor(t_us=0, bpm=120.0, bpb=4.0, beat_in_bar=0.0) -> BeatSyncMessage:
+ return BeatSyncMessage(t_us=t_us, bpm=bpm, bpb=bpb, beat_in_bar=beat_in_bar)
+
+
+class TestUnanchored:
+ def test_fresh_grid_is_not_anchored(self):
+ assert BeatGrid().is_anchored is False
+
+ def test_unanchored_tick_reports_unanchored(self):
+ state = BeatGrid().tick(now_us=1_000_000)
+ assert state.is_anchored is False
+ assert state.is_flashing is False
+ assert state.is_bar_start is False
+
+ def test_clear_is_idempotent(self):
+ g = BeatGrid()
+ g.clear()
+ g.clear()
+ assert g.is_anchored is False
+
+
+class TestAnchor:
+ def test_anchor_marks_anchored(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0))
+ assert g.is_anchored is True
+
+ def test_anchor_on_downbeat_flashes_and_marks_bar_start_immediately(self):
+ """The bug fix: a clock sample that *is* a downbeat (beat_in_bar=0)
+ must be visible at the anchor's own timestamp — waiting for a later
+ crossing would mean is_bar_start never fires (it was already the
+ modulo target, never something to cross into)."""
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0, beat_in_bar=0.0))
+ state = g.tick(now_us=1_000_000)
+ assert state.is_anchored is True
+ assert state.is_flashing is True
+ assert state.is_bar_start is True
+
+ def test_anchor_mid_bar_does_not_flash_immediately(self):
+ """A clock sample taken mid-bar (e.g. a bpm-change re-anchor) is not a
+ crossing — no flash until the next real beat boundary."""
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0, beat_in_bar=1.5))
+ state = g.tick(now_us=1_000_000)
+ assert state.is_anchored is True
+ assert state.is_flashing is False
+
+ def test_anchor_at_late_time_does_not_catch_up(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0))
+ state = g.tick(now_us=1_000_000 + 4 * 500_000)
+ assert state.is_anchored is True
+ assert state.is_flashing is True
+ # One flash, not four — verify the next tick is past the flash window
+ # and the *following* beat boundary fires exactly one more.
+ state = g.tick(now_us=1_000_000 + 4 * 500_000 + FLASH_US + 1)
+ assert state.is_flashing is False
+
+
+class TestFlash:
+ def test_first_beat_after_anchor_flashes(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0))
+ state = g.tick(now_us=1_000_000 + 500_000)
+ assert state.is_flashing is True
+ assert state.is_bar_start is False
+
+ def test_flash_expires_after_flash_us(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0))
+ g.tick(now_us=1_000_000 + 500_000)
+ state = g.tick(now_us=1_000_000 + 500_000 + FLASH_US)
+ assert state.is_flashing is False
+
+ def test_bar_start_marked_on_downbeat(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0))
+ g.tick(now_us=1_000_000 + 500_000)
+ g.tick(now_us=1_000_000 + 1_000_000)
+ g.tick(now_us=1_000_000 + 1_500_000)
+ state = g.tick(now_us=1_000_000 + 2_000_000)
+ assert state.is_flashing is True
+ assert state.is_bar_start is True
+
+ def test_subsequent_beats_flash_in_sequence(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0))
+ flashes = []
+ for i in range(8):
+ t = 1_000_000 + 500_000 * (i + 1)
+ state = g.tick(now_us=t)
+ flashes.append(state.is_flashing)
+ assert flashes == [True] * 8
+
+ def test_subsequent_bar_starts_every_bpb_beats(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=0, bpm=120.0, bpb=4.0))
+ bar_starts = []
+ for i in range(8):
+ t = 500_000 * (i + 1)
+ state = g.tick(now_us=t)
+ bar_starts.append(state.is_bar_start)
+ assert bar_starts == [False, False, False, True, False, False, False, True]
+
+
+class TestClear:
+ def test_clear_after_anchor(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0))
+ g.clear()
+ assert g.is_anchored is False
+ state = g.tick(now_us=2_000_000)
+ assert state.is_anchored is False
+
+ def test_clear_mid_flash(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0))
+ g.tick(now_us=1_000_000 + 500_000)
+ g.clear()
+ state = g.tick(now_us=1_000_000 + 600_000)
+ assert state.is_flashing is False
+
+
+class TestStaleTimeout:
+ def test_stale_anchor_clears_on_tick(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0))
+ state = g.tick(now_us=1_000_000 + STALE_AFTER_US + 1)
+ assert state.is_anchored is False
+
+ def test_freshly_anchored_is_not_stale(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0))
+ state = g.tick(now_us=1_000_000 + STALE_AFTER_US - 1)
+ assert state.is_anchored is True
+
+
+class TestInvalidAnchor:
+ def test_zero_bpm_clears_grid(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=0.0, bpb=4.0))
+ state = g.tick(now_us=1_000_000 + 500_000)
+ assert state.is_anchored is False
+
+ def test_zero_bpb_clears_grid(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=0.0))
+ state = g.tick(now_us=1_000_000 + 500_000)
+ assert state.is_anchored is False
+
+
+class TestReAnchor:
+ def test_re_anchor_resets_beat_counter(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0))
+ g.tick(now_us=1_000_000 + 1_500_000)
+ g.on_anchor(_anchor(t_us=10_000_000, bpm=120.0, bpb=4.0))
+ state = g.tick(now_us=10_000_000 + 500_000)
+ assert state.is_flashing is True
+ assert state.is_bar_start is False
+
+ def test_re_anchor_skips_missed_beats(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=1_000_000, bpm=120.0, bpb=4.0))
+ g.on_anchor(_anchor(t_us=1_000_000 + 4_000_000, bpm=120.0, bpb=4.0))
+ state = g.tick(now_us=1_000_000 + 4_000_000 + 500_000)
+ assert state.is_flashing is True
+ # First tick past the new anchor fires for the next live beat
+ # (beat 1, not a bar start). The 4 missed beats did not cause a
+ # flurry of catches-up.
+ assert state.is_bar_start is False
+
+
+class TestTickState:
+ def test_tick_state_is_immutable(self):
+ state = TickState(True, True, True, 120.0, 4.0)
+ try:
+ state.is_flashing = False # type: ignore[misc]
+ except Exception:
+ return
+ raise AssertionError("TickState should be frozen")
+
+
+class TestBeatPhase:
+ """beat_phase is the normalized [0, 1) within-beat position the
+ footswitch-LED driver uses to scale brightness."""
+
+ def test_phase_zero_at_beat_boundary(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=0, bpm=120.0, bpb=4.0)) # 120bpm → 500ms/beat
+ state = g.tick(now_us=500_000) # exactly beat 1
+ assert state.beat_phase == 0.0
+
+ def test_phase_advances_within_beat(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=0, bpm=120.0, bpb=4.0))
+ state = g.tick(now_us=125_000) # 1/4 of a 500ms beat
+ assert 0.0 <= state.beat_phase < 1.0
+ assert abs(state.beat_phase - 0.25) < 0.01
+
+ def test_phase_resets_across_beat_boundary(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=0, bpm=120.0, bpb=4.0))
+ g.tick(now_us=500_000) # beat 1
+ state = g.tick(now_us=750_000) # halfway through beat 2
+ assert abs(state.beat_phase - 0.5) < 0.01
+
+ def test_phase_in_range_zero_to_one(self):
+ g = BeatGrid()
+ g.on_anchor(_anchor(t_us=0, bpm=120.0, bpb=4.0))
+ for t_us in range(0, 2_000_000, 50_000):
+ state = g.tick(now_us=t_us)
+ assert 0.0 <= state.beat_phase < 1.0
+
+ def test_phase_is_zero_when_unanchored(self):
+ g = BeatGrid()
+ state = g.tick(now_us=1_000_000)
+ assert state.beat_phase == 0.0
diff --git a/tests/test_config_schema.py b/tests/test_config_schema.py
index 1223e86cb..282aa2c2b 100644
--- a/tests/test_config_schema.py
+++ b/tests/test_config_schema.py
@@ -60,3 +60,60 @@ def test_non_string_midi_port_rejected():
}
with pytest.raises(exceptions.ValidationError):
validate(instance=cfg, schema=schema)
+
+
+def test_longpress_group_name_accepted():
+ cfg = {
+ "hardware": {
+ "version": 3.0,
+ "midi": {"channel": 14},
+ "footswitches": [{"id": 0, "longpress": "toggle_bypass"}],
+ }
+ }
+ validate(instance=cfg, schema=schema)
+
+
+def test_longpress_group_name_list_accepted():
+ cfg = {
+ "hardware": {
+ "version": 3.0,
+ "midi": {"channel": 14},
+ "footswitches": [{"id": 0, "longpress": ["next_snapshot", "toggle_bypass"]}],
+ }
+ }
+ validate(instance=cfg, schema=schema)
+
+
+def test_longpress_midi_cc_object_accepted():
+ cfg = {
+ "hardware": {
+ "version": 3.0,
+ "midi": {"channel": 14},
+ "footswitches": [{"id": 0, "longpress": {"midi_CC": 65}}],
+ }
+ }
+ validate(instance=cfg, schema=schema)
+
+
+def test_longpress_unknown_group_name_rejected():
+ cfg = {
+ "hardware": {
+ "version": 3.0,
+ "midi": {"channel": 14},
+ "footswitches": [{"id": 0, "longpress": "not_a_real_group"}],
+ }
+ }
+ with pytest.raises(exceptions.ValidationError):
+ validate(instance=cfg, schema=schema)
+
+
+def test_longpress_object_without_midi_cc_rejected():
+ cfg = {
+ "hardware": {
+ "version": 3.0,
+ "midi": {"channel": 14},
+ "footswitches": [{"id": 0, "longpress": {}}],
+ }
+ }
+ with pytest.raises(exceptions.ValidationError):
+ validate(instance=cfg, schema=schema)
diff --git a/tests/test_footswitch.py b/tests/test_footswitch.py
index c0f939171..8f0b89803 100644
--- a/tests/test_footswitch.py
+++ b/tests/test_footswitch.py
@@ -59,6 +59,37 @@ def test_set_longpress_groups_none_clears(self):
assert fs.longpress_groups == []
+class TestSetLongpress:
+ """set_longpress is the config-facing entry point: dispatches between the
+ named-group form (str/list/None) and the {midi_CC: N} form."""
+
+ def test_string_sets_groups(self):
+ with _make_footswitch() as (fs, _sink):
+ fs.set_longpress("toggle_bypass")
+ assert fs.longpress_groups == ["toggle_bypass"]
+ assert fs.longpress_midi_CC is None
+
+ def test_list_sets_groups(self):
+ with _make_footswitch() as (fs, _sink):
+ fs.set_longpress(["next_snapshot", "toggle_bypass"])
+ assert fs.longpress_groups == ["next_snapshot", "toggle_bypass"]
+ assert fs.longpress_midi_CC is None
+
+ def test_none_clears_both(self):
+ with _make_footswitch() as (fs, _sink):
+ fs.set_longpress({"midi_CC": 65})
+ fs.set_longpress(None)
+ assert fs.longpress_groups == []
+ assert fs.longpress_midi_CC is None
+
+ def test_dict_sets_midi_cc_and_clears_groups(self):
+ with _make_footswitch() as (fs, _sink):
+ fs.set_longpress_groups(["toggle_bypass"])
+ fs.set_longpress({"midi_CC": 65})
+ assert fs.longpress_midi_CC == 65
+ assert fs.longpress_groups == []
+
+
class TestOnSwitch:
def test_short_press_dispatches_press_event(self):
with _make_footswitch() as (fs, sink):
diff --git a/tests/test_hardware.py b/tests/test_hardware.py
index 502f630ef..54469c37b 100644
--- a/tests/test_hardware.py
+++ b/tests/test_hardware.py
@@ -8,6 +8,7 @@
import common.token as Token
from modalapi.external_midi import ExternalMidiManager
+from pistomp.footswitch import Footswitch
from pistomp.hardware import Hardware
@@ -141,6 +142,38 @@ def test_external_port_opened_eagerly(self, routed_hw):
assert "My MIDI Device" in routed_hw.external_midi.midi_ports
+def _init_footswitches(hw, cfg):
+ hw._Hardware__init_footswitches(cfg)
+
+
+class TestInitFootswitchesLongpress:
+ """__init_footswitches dispatches the 'longpress' config value to
+ Footswitch.set_longpress: a group name (or list) vs. a {midi_CC: N} object."""
+
+ def _hw_with_footswitch(self, fs):
+ hw = object.__new__(_StubHardware)
+ hw.footswitches = [fs]
+ return hw
+
+ def test_group_name_sets_longpress_groups(self):
+ fs = Footswitch(id=0, led_pin=None, pixel=None, midi_CC=None, midi_channel=0,
+ refresh_callback=lambda **kw: None)
+ hw = self._hw_with_footswitch(fs)
+ cfg = {Token.HARDWARE: {Token.FOOTSWITCHES: [{Token.ID: 0, Token.LONGPRESS: "toggle_bypass"}]}}
+ _init_footswitches(hw, cfg)
+ assert fs.longpress_groups == ["toggle_bypass"]
+ assert fs.longpress_midi_CC is None
+
+ def test_midi_cc_object_sets_longpress_midi_cc(self):
+ fs = Footswitch(id=0, led_pin=None, pixel=None, midi_CC=None, midi_channel=0,
+ refresh_callback=lambda **kw: None)
+ hw = self._hw_with_footswitch(fs)
+ cfg = {Token.HARDWARE: {Token.FOOTSWITCHES: [{Token.ID: 0, Token.LONGPRESS: {"midi_CC": 65}}]}}
+ _init_footswitches(hw, cfg)
+ assert fs.longpress_midi_CC == 65
+ assert fs.longpress_groups == []
+
+
class TestReinitDefaultRouting:
def test_reinit_applies_routing_for_default_cfg(self, monkeypatch):
"""Routing is applied for the default config, not only for pedalboard cfg."""
diff --git a/tests/test_loopjefe_behavior.py b/tests/test_loopjefe_behavior.py
new file mode 100644
index 000000000..2d81b0530
--- /dev/null
+++ b/tests/test_loopjefe_behavior.py
@@ -0,0 +1,86 @@
+"""LoopJefe footswitch LED spec — state->color/style contract + registration.
+
+Pins:
+ - The loopjefe URIs are registered (plugins/__init__.py imports plugins.loopjefe).
+ - The registered LedSpec renders all 9 states correctly via the generic
+ render_led_spec driver, including the measure_number==0 loop-downbeat tint.
+ - Momentary press semantics come from the port (pprops:trigger on
+ advance/reset), not from anything here — not tested in this file.
+ - Brightness/pulse envelope is the driver's job — not tested here.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from modalapi.led_render import LedDisplayStyle, render_led_spec
+from modalapi.plugin_customization import LedSpec
+from plugins import lookup, registered_uris
+from plugins.loopjefe import LOOPJEFE_URIS
+
+
+def _spec() -> LedSpec:
+ spec = lookup(LOOPJEFE_URIS[0]).led_spec
+ assert spec is not None
+ return spec
+
+
+class TestRegistration:
+ def test_loopjefe_uris_are_registered(self):
+ registered = registered_uris()
+ for uri in LOOPJEFE_URIS:
+ assert uri in registered, f"{uri} not registered — plugins/__init__.py must import plugins.loopjefe"
+
+ def test_lookup_returns_loopjefe_led_spec(self):
+ for uri in LOOPJEFE_URIS:
+ cust = lookup(uri)
+ assert cust.led_spec is not None, f"lookup({uri!r}) did not return the loopjefe LedSpec"
+ assert cust.led_spec.state_symbol == "state"
+ assert cust.led_spec.downbeat_symbol == "measure_number"
+
+
+class TestStateColorAndStyle:
+ @pytest.mark.parametrize("state,expected_color", [
+ (0, None), # Empty -> off
+ (1, (0, 80, 255)), # Record Arm -> blue
+ (2, (255, 0, 0)), # Recording -> red
+ (3, (0, 80, 255)), # Record Close -> blue
+ (4, (0, 255, 0)), # Playback -> green
+ (5, (80, 80, 80)), # Stopped -> steady grey
+ (6, (0, 80, 255)), # Overdub Arm -> blue
+ (7, (255, 140, 0)), # Overdub -> orange
+ (8, (0, 80, 255)), # Overdub Close -> blue
+ ])
+ def test_state_color_with_nonzero_measure(self, state, expected_color):
+ color, _style = render_led_spec(_spec(), {"state": float(state), "measure_number": 1.0})
+ assert color == expected_color
+
+ @pytest.mark.parametrize("state,expected_style", [
+ (0, LedDisplayStyle.SOLID), # Empty -> off, solid
+ (5, LedDisplayStyle.SOLID), # Stopped -> steady grey
+ (1, LedDisplayStyle.METRONOME), # active -> pulse
+ (2, LedDisplayStyle.METRONOME),
+ (3, LedDisplayStyle.METRONOME),
+ (4, LedDisplayStyle.METRONOME),
+ (6, LedDisplayStyle.METRONOME),
+ (7, LedDisplayStyle.METRONOME),
+ (8, LedDisplayStyle.METRONOME),
+ ])
+ def test_state_style(self, state, expected_style):
+ _color, style = render_led_spec(_spec(), {"state": float(state), "measure_number": 1.0})
+ assert style == expected_style
+
+
+class TestLoopDownbeatTint:
+ def test_measure_zero_returns_distinct_color(self):
+ downbeat, _ = render_led_spec(_spec(), {"state": 2.0, "measure_number": 0.0}) # Recording -> red
+ normal, _ = render_led_spec(_spec(), {"state": 2.0, "measure_number": 2.0})
+ assert downbeat is not None and normal is not None
+ assert downbeat != normal
+ # The downbeat tint brightens each channel that wasn't already at 255
+ assert all(d >= n for d, n in zip(downbeat, normal))
+ assert any(d > n for d, n in zip(downbeat, normal))
+
+ def test_measure_zero_empty_state_still_off(self):
+ color, _style = render_led_spec(_spec(), {"state": 0.0, "measure_number": 0.0})
+ assert color is None
diff --git a/tests/test_websocket_bridge.py b/tests/test_websocket_bridge.py
index ff10395fa..3f7d88890 100644
--- a/tests/test_websocket_bridge.py
+++ b/tests/test_websocket_bridge.py
@@ -155,6 +155,82 @@ def test_receive_output_set_is_dropped():
assert ws._sent == []
+def test_receive_output_set_subscribed_survives():
+ worker = _make_worker()
+ worker.running = True
+ worker.set_interesting_outputs(frozenset({"loopjefe/state", "loopjefe/measure_number"}))
+ ws = _FakeWs(["output_set /graph/loopjefe state 2.0"])
+
+ asyncio.run(worker._receive_messages(ws))
+
+ msgs = []
+ while not worker.received_queue.empty():
+ msgs.append(worker.received_queue.get_nowait())
+ assert msgs == ["output_set /graph/loopjefe state 2.0"]
+ assert worker.messages_received == 1
+
+
+def test_receive_output_set_unsubscribed_is_dropped():
+ worker = _make_worker()
+ worker.running = True
+ worker.set_interesting_outputs(frozenset({"loopjefe/state"}))
+ ws = _FakeWs(["output_set /graph/Delay/meter 0.5"])
+
+ asyncio.run(worker._receive_messages(ws))
+
+ assert worker.received_queue.empty()
+ assert worker.messages_received == 0
+
+
+def test_receive_output_set_empty_interesting_drops_all():
+ """Regression: empty interesting-set must reproduce today's behavior —
+ every output_set is dropped before it floods the queue."""
+ worker = _make_worker()
+ worker.running = True
+ worker.set_interesting_outputs(frozenset())
+ ws = _FakeWs(["output_set /graph/loopjefe state 2.0", "output_set /graph/Amp/meter 0.9"])
+
+ asyncio.run(worker._receive_messages(ws))
+
+ assert worker.received_queue.empty()
+ assert worker.messages_received == 0
+
+
+def test_set_interesting_outputs_swaps_atomically():
+ """A subscription set swap takes effect immediately for subsequent frames;
+ the worker never sees a partially-updated set."""
+ worker = _make_worker()
+ worker.running = True
+ worker.set_interesting_outputs(frozenset({"loopjefe/state"}))
+ ws = _FakeWs([
+ "output_set /graph/loopjefe state 1.0", # subscribed → kept
+ "output_set /graph/loopjefe measure_number 0.0", # not subscribed → dropped
+ ])
+
+ asyncio.run(worker._receive_messages(ws))
+
+ msgs = []
+ while not worker.received_queue.empty():
+ msgs.append(worker.received_queue.get_nowait())
+ assert msgs == ["output_set /graph/loopjefe state 1.0"]
+
+
+def test_worker_does_not_hold_bridge_reference():
+ """Layering: the worker must not know about the bridge. The bridge owns the
+ worker, so a back-reference inverts the dependency and lets the worker reach
+ into bridge internals. The worker owns its own interesting-set instead."""
+ import inspect
+ worker = _make_worker()
+ sig = inspect.signature(WebSocketWorker.__init__)
+ assert "bridge" not in sig.parameters, (
+ "WebSocketWorker.__init__ must not take a bridge param — the worker "
+ "should own its interesting-set, not reach back into the bridge"
+ )
+ assert not hasattr(worker, "_bridge"), (
+ "WebSocketWorker must not store a _bridge reference"
+ )
+
+
def test_receive_mixed_messages_routes_correctly():
worker = _make_worker()
worker.running = True
diff --git a/tests/test_ws_protocol.py b/tests/test_ws_protocol.py
index 3810cfa69..7a7404afa 100644
--- a/tests/test_ws_protocol.py
+++ b/tests/test_ws_protocol.py
@@ -3,11 +3,13 @@
from modalapi.ws_protocol import (
AddHwPortMessage,
AddPluginMessage,
+ BeatSyncMessage,
ConnectMessage,
DisconnectMessage,
LoadingEndMessage,
LoadingStartMessage,
MidiMapMessage,
+ OutputSetMessage,
PedalSnapshotMessage,
ParamSetMessage,
PluginBypassMessage,
@@ -235,6 +237,43 @@ def test_transport_malformed_bpm_is_unknown():
)
+# ---------------------------------------------------------------------------
+# beat_sync (beat_sync {t_us} {bpm} {bpb} {beat_in_bar}) — a clock sample
+# (t_us=now), not a back-dated downbeat event. No absolute bar count — that's
+# DAW context mod-host doesn't need to expose.
+# ---------------------------------------------------------------------------
+
+
+def test_beat_sync_basic():
+ assert parse_message("beat_sync 1234567890 120.0 4 0.0") == BeatSyncMessage(
+ t_us=1234567890, bpm=120.0, bpb=4.0, beat_in_bar=0.0
+ )
+
+
+def test_beat_sync_zero_beat_in_bar():
+ assert parse_message("beat_sync 0 60.0 3 0.0") == BeatSyncMessage(
+ t_us=0, bpm=60.0, bpb=3.0, beat_in_bar=0.0
+ )
+
+
+def test_beat_sync_fractional_bpb():
+ assert parse_message("beat_sync 1000000 90.5 7 2.5") == BeatSyncMessage(
+ t_us=1000000, bpm=90.5, bpb=7.0, beat_in_bar=2.5
+ )
+
+
+def test_beat_sync_too_few_fields_is_unknown():
+ assert isinstance(parse_message("beat_sync 5 1234567890 120.0"), UnknownMessage)
+
+
+def test_beat_sync_non_int_t_us_is_unknown():
+ assert isinstance(parse_message("beat_sync 5 notanumber 120.0 4"), UnknownMessage)
+
+
+def test_beat_sync_non_float_bpm_is_unknown():
+ assert isinstance(parse_message("beat_sync 5 1234567890 notanumber 4"), UnknownMessage)
+
+
def test_plugin_bypass_nonzero_is_true():
msg = parse_message("param_set /graph/Reverb :bypass 0.5")
assert msg == PluginBypassMessage(instance="Reverb", bypassed=True)
@@ -353,3 +392,33 @@ def test_malformed_pedal_snapshot_non_int():
def test_empty_string():
msg = parse_message("")
assert isinstance(msg, UnknownMessage)
+
+
+# ---------------------------------------------------------------------------
+# output_set (output_set /graph/{instance} {symbol} {value})
+# ---------------------------------------------------------------------------
+
+
+def test_output_set_parses_to_output_set_message():
+ msg = parse_message("output_set /graph/loopjefe state 2.0")
+ assert msg == OutputSetMessage(instance="loopjefe", symbol="state", value=2.0)
+
+
+def test_output_set_integer_port():
+ msg = parse_message("output_set /graph/loopjefe measure_number 0.0")
+ assert msg == OutputSetMessage(instance="loopjefe", symbol="measure_number", value=0.0)
+
+
+def test_output_set_missing_value_is_unknown():
+ msg = parse_message("output_set /graph/loopjefe state")
+ assert isinstance(msg, UnknownMessage)
+
+
+def test_output_set_non_float_value_is_unknown():
+ msg = parse_message("output_set /graph/loopjefe state notanumber")
+ assert isinstance(msg, UnknownMessage)
+
+
+def test_output_set_strips_graph_prefix():
+ msg = parse_message("output_set /graph/loopjefe state 4.0")
+ assert msg == OutputSetMessage(instance="loopjefe", symbol="state", value=4.0)
diff --git a/tests/v3/test_footswitch_led_driver.py b/tests/v3/test_footswitch_led_driver.py
new file mode 100644
index 000000000..2d43cf59f
--- /dev/null
+++ b/tests/v3/test_footswitch_led_driver.py
@@ -0,0 +1,191 @@
+"""Unified footswitch LED driver — single source of truth for pixel + GPIO LED.
+
+The driver runs in poll_controls (10ms, same tick as the press) so there's no
+latency between a state change and the LED reflecting it. Both fs.pixel and
+fs.led are written from the same (color, style) frame in the same driver call
+— no separate set_led path.
+
+Covers:
+ - SOLID: shows the frame's color steadily (or off when color is None).
+ - METRONOME: scales brightness by beat_phase (bright at 0, dim toward 1).
+ - Unanchored METRONOME: steady color (no pulse).
+ - Off: color is None -> pixel disabled, GPIO LED off.
+ - Press renders in the same tick (poll_controls, not poll_indicators).
+ - set_led is a pure state update -- no hardware writes.
+ - Default per-footswitch renderer (no bound plugin / no LedSpec): toggle +
+ category color, falling back to off when not toggled.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from modalapi.led_render import LedDisplayStyle
+from modalapi.modhandler import Modhandler
+from pistomp.beatsync import TickState
+from pistomp.footswitch import Footswitch
+from tests.types import SystemFixture
+
+
+def _beat(beat_phase: float = 0.0, *, is_anchored: bool = True,
+ is_bar_start: bool = False, is_flashing: bool | None = None) -> TickState:
+ if is_flashing is None:
+ is_flashing = is_bar_start
+ return TickState(
+ is_anchored=is_anchored,
+ is_flashing=is_flashing,
+ is_bar_start=is_bar_start,
+ bpm=120.0,
+ bpb=4.0,
+ beat_phase=beat_phase,
+ )
+
+
+def _drive(handler: Modhandler, beat: TickState) -> None:
+ """Invoke the driver directly with a fabricated beat state."""
+ handler._drive_footswitch_leds(beat)
+
+
+def _fs_with_frame(v3_system: SystemFixture, color, style: LedDisplayStyle = LedDisplayStyle.SOLID) -> Footswitch:
+ """Stub the default per-footswitch renderer to return a fixed frame,
+ bypassing plugin-binding lookup entirely — isolates the writer/envelope
+ behavior under test from LedSpec rendering (covered in
+ tests/test_loopjefe_behavior.py)."""
+ fs = v3_system.hw.footswitches[0]
+ v3_system.handler._render_footswitch = MagicMock(return_value=(color, style)) # type: ignore[method-assign]
+ fs.pixel = MagicMock()
+ fs.led = MagicMock()
+ return fs
+
+
+class TestSolidStyle:
+ def test_solid_shows_color_steadily(self, v3_system: SystemFixture):
+ fs = _fs_with_frame(v3_system, (0, 255, 0), LedDisplayStyle.SOLID)
+ _drive(v3_system.handler, _beat(beat_phase=0.0))
+ fs.pixel.set_color.assert_called_once_with((0, 255, 0))
+ fs.pixel.set_enable.assert_called_once_with(True)
+ assert fs.led is not None
+ fs.led.on.assert_called_once() # type: ignore[unionAttr]
+
+ def test_solid_brightness_does_not_scale_with_phase(self, v3_system: SystemFixture):
+ fs = _fs_with_frame(v3_system, (100, 100, 100), LedDisplayStyle.SOLID)
+ _drive(v3_system.handler, _beat(beat_phase=0.9))
+ # SOLID must not scale — full color at any phase
+ fs.pixel.set_color.assert_called_once_with((100, 100, 100))
+
+ def test_solid_none_color_disables_pixel_and_led(self, v3_system: SystemFixture):
+ fs = _fs_with_frame(v3_system, None, LedDisplayStyle.SOLID)
+ _drive(v3_system.handler, _beat())
+ fs.pixel.set_enable.assert_called_once_with(False)
+ assert fs.led is not None
+ fs.led.off.assert_called_once() # type: ignore[unionAttr]
+
+
+class TestMetronomeStyle:
+ def test_metronome_bright_at_phase_zero(self, v3_system: SystemFixture):
+ fs = _fs_with_frame(v3_system, (100, 100, 100), LedDisplayStyle.METRONOME)
+ _drive(v3_system.handler, _beat(beat_phase=0.0))
+ # phase 0 → brightness 1.0 → unscaled color
+ fs.pixel.set_color.assert_called_once_with((100, 100, 100))
+
+ def test_metronome_dim_near_phase_one(self, v3_system: SystemFixture):
+ fs = _fs_with_frame(v3_system, (100, 100, 100), LedDisplayStyle.METRONOME)
+ _drive(v3_system.handler, _beat(beat_phase=0.9))
+ scaled = fs.pixel.set_color.call_args.args[0]
+ # phase 0.9 → brightness 1.0 - 0.9*0.7 = 0.37 → 100*0.37 = 37
+ assert scaled == (37, 37, 37)
+ assert all(c < 100 for c in scaled)
+
+ def test_metronome_brightest_on_bar_start(self, v3_system: SystemFixture):
+ fs = _fs_with_frame(v3_system, (100, 100, 100), LedDisplayStyle.METRONOME)
+ _drive(v3_system.handler, _beat(beat_phase=0.5, is_bar_start=True))
+ # bar start forces brightness 1.0 even though phase is mid-beat
+ fs.pixel.set_color.assert_called_once_with((100, 100, 100))
+
+ def test_metronome_unanchored_shows_steady_color(self, v3_system: SystemFixture):
+ fs = _fs_with_frame(v3_system, (100, 100, 100), LedDisplayStyle.METRONOME)
+ _drive(v3_system.handler, _beat(beat_phase=0.9, is_anchored=False))
+ # No pulse when unanchored — full color
+ fs.pixel.set_color.assert_called_once_with((100, 100, 100))
+
+
+class TestDefaultRendering:
+ """No bound plugin (or a bound plugin with no LedSpec) falls back to the
+ built-in toggle + category-color renderer."""
+
+ def test_unbound_untoggled_footswitch_is_off(self, v3_system: SystemFixture):
+ fs = v3_system.hw.footswitches[0]
+ fs.parameter = None
+ fs.toggled = False
+ fs.pixel = MagicMock()
+ fs.led = MagicMock()
+ _drive(v3_system.handler, _beat())
+ fs.pixel.set_enable.assert_called_once_with(False)
+ assert fs.led is not None
+ fs.led.off.assert_called_once() # type: ignore[unionAttr]
+
+ def test_unbound_toggled_footswitch_shows_category_or_white(self, v3_system: SystemFixture):
+ fs = v3_system.hw.footswitches[0]
+ fs.parameter = None
+ fs.toggled = True
+ fs.category = None
+ fs.pixel = MagicMock()
+ fs.led = MagicMock()
+ _drive(v3_system.handler, _beat())
+ fs.pixel.set_color.assert_called_once_with((255, 255, 255))
+ fs.pixel.set_enable.assert_called_once_with(True)
+
+
+class TestPixelAndLedSameSource:
+ """Both fs.pixel and fs.led are written from the same renderer query in the
+ same driver call — no separate set_led path fighting the driver."""
+
+ def test_solid_on_lights_both_pixel_and_led(self, v3_system: SystemFixture):
+ fs = _fs_with_frame(v3_system, (0, 255, 0), LedDisplayStyle.SOLID)
+ _drive(v3_system.handler, _beat())
+ fs.pixel.set_enable.assert_called_once_with(True)
+ assert fs.led is not None
+ fs.led.on.assert_called_once() # type: ignore[unionAttr]
+
+ def test_off_disables_both_pixel_and_led(self, v3_system: SystemFixture):
+ fs = _fs_with_frame(v3_system, None, LedDisplayStyle.SOLID)
+ _drive(v3_system.handler, _beat())
+ fs.pixel.set_enable.assert_called_once_with(False)
+ assert fs.led is not None
+ fs.led.off.assert_called_once() # type: ignore[unionAttr]
+
+ def test_set_led_does_not_touch_hardware(self, v3_system: SystemFixture):
+ """set_led is a pure state update — it flips fs.toggled only. The next
+ driver tick renders the new state to both pixel and LED."""
+ fs = v3_system.hw.footswitches[0]
+ fs.pixel = MagicMock()
+ fs.led = MagicMock()
+ fs.set_led(True)
+ assert fs.toggled is True
+ fs.pixel.set_enable.assert_not_called()
+ fs.pixel.set_color.assert_not_called()
+ assert fs.led is not None
+ fs.led.on.assert_not_called() # type: ignore[unionAttr]
+ fs.led.off.assert_not_called() # type: ignore[unionAttr]
+ fs.led.blink.assert_not_called() # type: ignore[unionAttr]
+
+
+class TestDriverRunsInPollControls:
+ """The LED driver runs in poll_controls (10ms), not poll_indicators (20ms),
+ so a press and its LED update happen in the same tick."""
+
+ def test_poll_controls_drives_leds(self, v3_system: SystemFixture):
+ fs = _fs_with_frame(v3_system, (0, 255, 0), LedDisplayStyle.SOLID)
+ with patch.object(v3_system.hw, "poll_controls"):
+ v3_system.handler.poll_controls()
+ fs.pixel.set_color.assert_called_once_with((0, 255, 0))
+ fs.pixel.set_enable.assert_called_once_with(True)
+
+ def test_poll_indicators_does_not_drive_footswitch_leds(self, v3_system: SystemFixture):
+ """poll_indicators still drives hardware.indicators (VU meters) but no
+ longer drives footswitch LEDs — that moved to poll_controls."""
+ fs = _fs_with_frame(v3_system, (0, 255, 0), LedDisplayStyle.SOLID)
+ with patch.object(v3_system.hw, "poll_indicators"):
+ v3_system.handler.poll_indicators()
+ fs.pixel.set_color.assert_not_called()
+ fs.pixel.set_enable.assert_not_called()
diff --git a/tests/v3/test_footswitch_presets.py b/tests/v3/test_footswitch_presets.py
index 72ab7a57d..95a40cbcf 100644
--- a/tests/v3/test_footswitch_presets.py
+++ b/tests/v3/test_footswitch_presets.py
@@ -10,15 +10,15 @@
ControllerManager.bind() can match it against an unrelated plugin's
MIDI-learned binding and steal fs.parameter.
3. The label survives even if `fs.parameter` still ends up set by some
- other path -- defense in depth on top of (2), so
- `draw_footswitches`/`update_footswitch` never let a plugin/param name
- clobber a preset label.
+ other path -- defense in depth on top of (2), so
+ `draw_footswitches`/`update_footswitch` never let a plugin/param name
+ clobber a preset label.
4. The footswitch's LED/indicator lights only when its mapped snapshot is
the currently active one.
"""
import yaml
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
from common.parameter import Parameter
from tests.types import SystemFixture
@@ -116,13 +116,21 @@ class TestPresetFootswitchIndicator:
def test_active_snapshot_footswitch_drives_physical_led(self, v3_system: SystemFixture):
"""A press never touches fs.toggled for preset footswitches (handler.py
returns early on the preset_callback branch), so the LCD redraw path
- is the only place that can also light the physical LED/pixel."""
+ is the only place that can also light the physical LED/pixel.
+
+ The per-tick LED driver (_drive_footswitch_leds) runs in poll_controls
+ and must light the pixel of the footswitch bound to the active preset.
+ Regression: stripping set_led/set_category's pixel calls left preset
+ footswitch pixels dark because preset switches never get a behavior
+ via ControllerManager.bind (no plugin parameter)."""
handler = v3_system.handler
hw = v3_system.hw
lcd = handler.lcd
fs0, fs1 = hw.footswitches[0], hw.footswitches[1]
fs0.pixel = MagicMock()
fs1.pixel = MagicMock()
+ fs0.led = MagicMock()
+ fs1.led = MagicMock()
fs0.add_preset(callback=handler.preset_set_and_change, callback_arg=0)
fs1.add_preset(callback=handler.preset_set_and_change, callback_arg=1)
handler.current.preset_index = 1
@@ -130,8 +138,16 @@ def test_active_snapshot_footswitch_drives_physical_led(self, v3_system: SystemF
lcd.link_data(handler.pedalboard_list, handler.current, hw.footswitches)
lcd.draw_main_panel()
- fs0.pixel.set_enable.assert_called_once_with(False)
- fs1.pixel.set_enable.assert_called_once_with(True)
+ # Drive one controls tick — the active preset's pixel must light.
+ # Patch hardware.poll_controls to skip the analog control refresh,
+ # which needs real SPI; we only want to exercise the handler's LED driver.
+ with patch.object(hw, "poll_controls"):
+ handler.poll_controls()
+
+ # fs1 is the active preset (index 1) → its pixel must be enabled.
+ fs1.pixel.set_enable.assert_called_with(True)
+ # fs0 is inactive → its pixel must be disabled.
+ fs0.pixel.set_enable.assert_called_with(False)
assert fs0.toggled is False
assert fs1.toggled is True
diff --git a/tests/v3/test_midi_learn.py b/tests/v3/test_midi_learn.py
index a21cad2e7..5111df68f 100644
--- a/tests/v3/test_midi_learn.py
+++ b/tests/v3/test_midi_learn.py
@@ -109,3 +109,94 @@ def test_v3_midi_learn_unknown_instance_is_ignored(v3_system: SystemFixture, mak
assert fs0.parameter is None
assert plugin.has_footswitch is False
+
+
+def _make_loopjefe_plugin_with_advance(make_parameter, instance_id="loopjefe"):
+ """Build a loopjefe plugin with an `advance` trigger parameter, using the
+ real registered loopjefe customization (plugins/__init__.py imports
+ plugins.loopjefe, which registers its LedSpec)."""
+ from common.parameter import Type
+ from modalapi.plugin import Plugin
+ from plugins import lookup
+ from plugins.loopjefe import LOOPJEFE_URIS
+
+ advance = make_parameter("advance", instance_id, value=0.0)
+ advance.type = Type.TRIGGER # pprops:trigger in loopjefe.ttl
+ uri = LOOPJEFE_URIS[0]
+ return Plugin(instance_id, {"advance": advance}, {}, "Looper",
+ uri=uri, customization=lookup(uri))
+
+
+class TestMidiLearnBindsMomentaryAndOutputs:
+ """Regression: the live MIDI-learn path (Handler._apply_midi_binding →
+ _bind_controller_to_param) must not need any plugin-specific input code —
+ momentary semantics come for free from the bound parameter's port type
+ (pprops:trigger → Type.TRIGGER), and the LED driver reads the plugin's own
+ generically-mirrored output_values (from its LedSpec), not anything cached
+ on the footswitch."""
+
+ def test_midi_learn_binds_trigger_parameter_as_momentary(self, v3_system: SystemFixture, make_parameter):
+ handler = v3_system.handler
+ hw = v3_system.hw
+ ws_bridge = v3_system.ws_bridge
+ assert handler.current
+
+ fs0 = hw.footswitches[0]
+ channel, cc = _binding_for(hw, fs0).split(":")
+
+ plugin = _make_loopjefe_plugin_with_advance(make_parameter)
+ handler.current.pedalboard.plugins = [plugin]
+
+ ws_bridge.inject(f"midi_map /graph/loopjefe advance {channel} {cc} 0.0 1.0")
+ handler.poll_ws_messages()
+
+ assert fs0.parameter is plugin.parameters["advance"]
+ assert fs0.parameter is not None
+ assert fs0.parameter.is_momentary is True, (
+ "advance is pprops:trigger — momentary must be derived from the "
+ "port type, with zero loopjefe-specific input code"
+ )
+
+ def test_update_interesting_outputs_derives_from_plugin_led_spec(
+ self, v3_system: SystemFixture, make_parameter
+ ):
+ """Monitored outputs are owned by the plugin (its LedSpec), not by
+ whichever footswitch happens to be bound to it."""
+ handler = v3_system.handler
+ assert handler.current
+
+ plugin = _make_loopjefe_plugin_with_advance(make_parameter)
+ handler.current.pedalboard.plugins = [plugin]
+
+ handler._update_interesting_outputs()
+
+ last = v3_system.ws_bridge.interesting_calls[-1]
+ assert "loopjefe/state" in last
+ assert "loopjefe/measure_number" in last
+
+ def test_output_set_updates_plugin_output_values_for_led_spec(
+ self, v3_system: SystemFixture, make_parameter
+ ):
+ """End-to-end: an output_set for loopjefe/state and measure_number
+ updates plugin.output_values generically, and the plugin's LedSpec
+ renders the right color/style from them — no footswitch involved."""
+ from modalapi.led_render import LedDisplayStyle, render_led_spec
+
+ handler = v3_system.handler
+ ws_bridge = v3_system.ws_bridge
+ assert handler.current
+
+ plugin = _make_loopjefe_plugin_with_advance(make_parameter)
+ handler.current.pedalboard.plugins = [plugin]
+
+ ws_bridge.inject("output_set /graph/loopjefe state 2.0")
+ ws_bridge.inject("output_set /graph/loopjefe measure_number 1.0")
+ handler.poll_ws_messages()
+
+ assert plugin.output_values["state"] == 2.0
+ assert plugin.output_values["measure_number"] == 1.0
+
+ assert plugin.customization.led_spec is not None
+ color, style = render_led_spec(plugin.customization.led_spec, plugin.output_values)
+ assert color == (255, 0, 0) # Recording → red
+ assert style == LedDisplayStyle.METRONOME
diff --git a/tests/v3/test_taptempo_led.py b/tests/v3/test_taptempo_led.py
new file mode 100644
index 000000000..781c647f6
--- /dev/null
+++ b/tests/v3/test_taptempo_led.py
@@ -0,0 +1,155 @@
+"""Taptempo footswitch LED — two metronome sources, one driver.
+
+The taptempo footswitch's LED flashes from whichever beat source is active:
+ - Transport anchored (beat_sync received): beat_grid drives the flash,
+ white on downbeat, grey on beat.
+ - Taptempo only (no beat_sync, but taptempo enabled with bpm): the LED
+ blinks from taptempo.anchor + bpm — on for the first ~100ms of each
+ beat period, off otherwise.
+ - Taptempo disabled: the footswitch behaves as a default toggle.
+
+The gpiozero hardware blink() is gone — the 10ms driver tick computes on/off
+from the taptempo phase, same as it does for the transport-anchored case.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from modalapi.modhandler import _METRONOME_BEAT_RGB, _METRONOME_DOWNBEAT_RGB
+from pistomp.beatsync import TickState
+from tests.types import SystemFixture
+
+
+def _find_taptempo_fs(v3_system: SystemFixture):
+ for fs in v3_system.hw.footswitches:
+ if fs.taptempo is not None:
+ return fs
+ raise AssertionError("No taptempo footswitch in v3 fixture")
+
+
+def _mock_fs(fs):
+ fs.pixel = MagicMock()
+ fs.led = MagicMock()
+
+
+class TestTransportAnchored:
+ """When beat_grid is anchored (beat_sync received), the taptempo footswitch
+ flashes beat-synced from the transport — same as the old _drive_metronome."""
+
+ def test_flashing_beat_shows_beat_color(self, v3_system: SystemFixture):
+ fs = _find_taptempo_fs(v3_system)
+ _mock_fs(fs)
+ beat = TickState(is_anchored=True, is_flashing=True, is_bar_start=False,
+ bpm=120.0, bpb=4.0, beat_phase=0.0)
+ v3_system.handler._drive_footswitch_leds(beat)
+ fs.pixel.set_color.assert_called_once_with(_METRONOME_BEAT_RGB)
+ fs.pixel.set_enable.assert_called_once_with(True)
+ assert fs.led is not None
+ fs.led.on.assert_called_once() # type: ignore[unionAttr]
+
+ def test_flashing_bar_start_shows_downbeat_color(self, v3_system: SystemFixture):
+ fs = _find_taptempo_fs(v3_system)
+ _mock_fs(fs)
+ beat = TickState(is_anchored=True, is_flashing=True, is_bar_start=True,
+ bpm=120.0, bpb=4.0, beat_phase=0.0)
+ v3_system.handler._drive_footswitch_leds(beat)
+ fs.pixel.set_color.assert_called_once_with(_METRONOME_DOWNBEAT_RGB)
+
+ def test_not_flashing_turns_off(self, v3_system: SystemFixture):
+ fs = _find_taptempo_fs(v3_system)
+ _mock_fs(fs)
+ beat = TickState(is_anchored=True, is_flashing=False, is_bar_start=False,
+ bpm=120.0, bpb=4.0, beat_phase=0.5)
+ v3_system.handler._drive_footswitch_leds(beat)
+ fs.pixel.set_enable.assert_called_once_with(False)
+ assert fs.led is not None
+ fs.led.off.assert_called_once() # type: ignore[unionAttr]
+
+
+class TestTaptempoBlink:
+ """When beat_grid is NOT anchored but taptempo is enabled with a bpm, the
+ LED blinks from taptempo.anchor + bpm — computed by the 10ms driver, not
+ gpiozero.blink()."""
+
+ def test_taptempo_blink_on_within_flash_window(self, v3_system: SystemFixture):
+ fs = _find_taptempo_fs(v3_system)
+ _mock_fs(fs)
+ assert fs.taptempo is not None
+ fs.taptempo.enable(True)
+ fs.taptempo.set_bpm(120.0) # 120bpm → 500ms period, 100ms on-window
+ fs.taptempo.anchor = 1000.0 # last tap at t=1000.0
+
+ # Now=1000.05 → 50ms into the beat → within the 100ms on-window → ON
+ with patch("modalapi.modhandler._now_us", return_value=int(1000.05 * 1_000_000)):
+ v3_system.handler._drive_footswitch_leds(
+ TickState(is_anchored=False, is_flashing=False, is_bar_start=False,
+ bpm=120.0, bpb=4.0, beat_phase=0.0)
+ )
+ fs.pixel.set_enable.assert_called_once_with(True)
+ assert fs.led is not None
+ fs.led.on.assert_called_once() # type: ignore[unionAttr]
+
+ def test_taptempo_blink_off_outside_flash_window(self, v3_system: SystemFixture):
+ fs = _find_taptempo_fs(v3_system)
+ _mock_fs(fs)
+ assert fs.taptempo is not None
+ fs.taptempo.enable(True)
+ fs.taptempo.set_bpm(120.0) # 500ms period, 100ms on-window
+ fs.taptempo.anchor = 1000.0
+
+ # Now=1000.3 → 300ms into the beat → past the 100ms on-window → OFF
+ with patch("modalapi.modhandler._now_us", return_value=int(1000.3 * 1_000_000)):
+ v3_system.handler._drive_footswitch_leds(
+ TickState(is_anchored=False, is_flashing=False, is_bar_start=False,
+ bpm=120.0, bpb=4.0, beat_phase=0.0)
+ )
+ fs.pixel.set_enable.assert_called_once_with(False)
+ assert fs.led is not None
+ fs.led.off.assert_called_once() # type: ignore[unionAttr]
+
+ def test_taptempo_zero_bpm_does_not_blink(self, v3_system: SystemFixture):
+ """No taps yet (bpm=0) → no blink; fall through to default behavior."""
+ fs = _find_taptempo_fs(v3_system)
+ _mock_fs(fs)
+ assert fs.taptempo is not None
+ fs.taptempo.enable(True)
+ fs.taptempo.set_bpm(0.0)
+ v3_system.handler._drive_footswitch_leds(
+ TickState(is_anchored=False, is_flashing=False, is_bar_start=False,
+ bpm=0.0, bpb=4.0, beat_phase=0.0)
+ )
+ # No blink — the default behavior takes over (off when not toggled)
+ fs.pixel.set_enable.assert_called_once_with(False)
+
+ def test_taptempo_disabled_falls_through_to_default(self, v3_system: SystemFixture):
+ """Taptempo disabled → the footswitch is a normal toggle; the driver
+ renders from the default behavior (toggled + category color)."""
+ fs = _find_taptempo_fs(v3_system)
+ _mock_fs(fs)
+ assert fs.taptempo is not None
+ fs.taptempo.enable(False)
+ fs.toggled = True
+ v3_system.handler._drive_footswitch_leds(
+ TickState(is_anchored=False, is_flashing=False, is_bar_start=False,
+ bpm=0.0, bpb=4.0, beat_phase=0.0)
+ )
+ # Default behavior: toggled=True → pixel on with category color
+ fs.pixel.set_enable.assert_called_once_with(True)
+
+ def test_no_gpiozero_blink_called(self, v3_system: SystemFixture):
+ """Regression: the gpiozero hardware blink() must not be called — the
+ driver computes on/off from the taptempo phase at 10ms granularity."""
+ fs = _find_taptempo_fs(v3_system)
+ _mock_fs(fs)
+ assert fs.taptempo is not None
+ fs.taptempo.enable(True)
+ fs.taptempo.set_bpm(120.0)
+ fs.taptempo.anchor = 1000.0
+ with patch("modalapi.modhandler._now_us", return_value=int(1000.05 * 1_000_000)):
+ v3_system.handler._drive_footswitch_leds(
+ TickState(is_anchored=False, is_flashing=False, is_bar_start=False,
+ bpm=120.0, bpb=4.0, beat_phase=0.0)
+ )
+ assert fs.led is not None
+ fs.led.blink.assert_not_called() # type: ignore[unionAttr]
\ No newline at end of file