[WIP] Loop Jefe

common/contexts.py

@@ -83,7 +83,7 @@ class Effect:
 class ParamEffect(Effect):
     plugin: object  # PluginRef, resolved at pedalboard load
     symbol: Union[Symbol, type[SelectionSymbol]]
-    commit: bool = True  # WebSocket send_parameter on fire
+    commit: bool = True  # Declared: WebSocket send on fire. Dispatch does not read it. A no-op today.
     mirror: bool = True  # reconcile from inbound param_set echo
 
 
@@ -189,6 +189,12 @@ def add(self, decl: BindingDecl) -> None:
         key = (decl.control.cls, decl.event_kind)
         self.rows.setdefault(key, []).append(decl)
 
+    def remove(self, should_drop: Callable[[BindingDecl], bool]) -> None:
+        """Drop every row for which *should_drop* is true, across all buckets —
+        the mutation counterpart to add(), so callers never touch `rows`."""
+        for key, rows in list(self.rows.items()):
+            self.rows[key] = [d for d in rows if not should_drop(d)]
+
 
 # Per-class chain: which ContextKinds are consulted, top (highest precedence)
 # to bottom, for a given ControlClass. NAV is intentionally absent — it is an

common/loop_progress.py

@@ -0,0 +1,44 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+#
+# 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 Affero 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 Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with pi-stomp.  If not, see <https://www.gnu.org/licenses/>.
+
+"""How far a looping plugin is through its loop, as the footswitch strip draws it.
+
+Shared between the renderer that derives it (`modalapi.led_render`) and the
+widget that paints it (`uilib.footswitch`), which sit on opposite sides of the
+module DAG.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum, auto
+
+
+class LoopFill(Enum):
+    FILL = auto()  # determinate: `position` of the perimeter is behind us
+    STATIC = auto()  # the loop exists but isn't moving
+    CHASE = auto()  # length unknown: a head sweeping at `position`
+    FREE = auto()  # the loop runs, but no transport gives its position
+
+
+@dataclass(frozen=True)
+class LoopProgress:
+    mode: LoopFill
+    color: tuple[int, int, int]
+    segments: int  # bars in the loop; 0 when the length isn't known yet
+    position: float = 0.0  # turns, [0, 1)
+    pulse: float = 1.0  # brightness of the lit part; the beat envelope, 1.0 when free-running

common/parameter.py

@@ -34,6 +34,7 @@
 TTL_SCALEPOINTS = "scalePoints"
 TTL_TAPTEMPO = "tapTempo"
 TTL_TOGGLED = "toggled"
+TTL_TRIGGER = "trigger"
 
 # Identifies a Parameter: the key of plugin.parameters, ParamEffect.symbol,
 # edit_symbol(). Usually an LV2 port symbol (":bypass", "gain"); also an ALSA
@@ -110,13 +111,22 @@ def is_hidden_port(plugin_info: PortInfo) -> bool:
     return plugin_info.get("designation", "") in HIDDEN_DESIGNATIONS
 
 
+def _has_property(properties: list[str], name: str) -> bool:
+    """Match MOD-UI's short property names and raw LV2 URIs."""
+    return any(
+        property_name == name or property_name.rsplit("#", 1)[-1].rsplit("/", 1)[-1] == name
+        for property_name in properties
+    )
+
+
 class Type(Enum):
     DEFAULT = 0  # No explicitly defined type (eg. linear float)
     ENUMERATION = 1
     INTEGER = 2
     LOGARITHMIC = 3
     TAPTEMPO = 4
     TOGGLED = 5
+    TRIGGER = 6  # pprops:trigger — edge-triggered, self-clearing (momentary)
 
 
 class Parameter:
@@ -171,20 +181,28 @@ def __init__(
 
         properties = plugin_info.get("properties") or []
         if len(properties) > 0:
-            if TTL_LOGARITHMIC in properties:
+            if _has_property(properties, TTL_LOGARITHMIC):
                 self.is_logarithmic = True
-            if TTL_ENUMERATION in properties:
+            if _has_property(properties, TTL_TRIGGER):
+                self.type = Type.TRIGGER
+            elif _has_property(properties, TTL_ENUMERATION):
                 self.enum_values = plugin_info.get("scalePoints") or []
                 self.type = Type.ENUMERATION
-            elif TTL_INTEGER in properties:
+            elif _has_property(properties, TTL_INTEGER):
                 self.type = Type.INTEGER
-            elif TTL_LOGARITHMIC in properties:
+            elif _has_property(properties, TTL_LOGARITHMIC):
                 self.type = Type.LOGARITHMIC
-            elif TTL_TAPTEMPO in properties:
+            elif _has_property(properties, TTL_TAPTEMPO):
                 self.type = Type.TAPTEMPO
-            elif TTL_TOGGLED in properties:
+            elif _has_property(properties, TTL_TOGGLED):
                 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
+
     @property
     def value(self) -> float:
         return self._value
@@ -216,6 +234,16 @@ def commit(self, value: float, sink: ParamSink | None) -> None:
             return
         self._notify_settled()
 
+    def pulse(self, sink: ParamSink | None) -> None:
+        """A self-clearing trigger edge (pprops:trigger): drive to the "on" edge,
+        publish it once through *sink*, then clear to rest — persists no value
+        and never reverts, so each press is a fresh rising edge."""
+        self._set(self.maximum)
+        if sink is not None:
+            sink(self)
+        self._set(self.minimum)
+        self._notify_settled()
+
     def _set(self, value: float) -> None:
         if value == self._value:
             return

modalapi/led_render.py

@@ -0,0 +1,113 @@
+# 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 <https://www.gnu.org/licenses/>.
+
+"""Generic, data-driven footswitch-LED rendering.
+
+Pure function of (LedSpec, plugin.output_values) -> (color, style). No
+footswitch, beat, or plugin-instance coupling — the handler applies the binary
+metronome state uniformly to the physical LED and loop perimeter.
+"""
+
+from __future__ import annotations
+
+from enum import Enum, auto
+from typing import TYPE_CHECKING
+
+from common.loop_progress import LoopFill, LoopProgress
+
+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
+
+
+def metronome_brightness(is_flashing: bool) -> float:
+    """Binary brightness shared by physical LEDs and the loop perimeter."""
+    return 1.0 if is_flashing else 0.0
+
+
+def state_label(spec: LedSpec, output_values: dict[str, float]) -> str | None:
+    """Short display name for the current state, or None when the plugin
+    declares no `labels`. Same lookup as `render_led_spec`, for the LCD."""
+    if spec.labels is None:
+        return None
+    return spec.labels.get(int(output_values.get(spec.state_symbol, 0)))
+
+
+def loop_progress(
+    spec: LedSpec,
+    output_values: dict[str, float],
+    bar_phase: float | None,
+    beat_brightness: float = 1.0,
+) -> LoopProgress | None:
+    """Where the plugin is through its loop, or None if it has no loop to be
+    through. `bar_phase` interpolates within the current bar — the plugin only
+    publishes a bar index, and a per-sample position port would be a monitored
+    output changing every process cycle. None means that the grid runs free. A
+    free grid gives a beat but no bar. There is then no position, and the loop
+    shows a full ring.
+    `beat_brightness` is 1.0 during the fixed metronome window and 0.0
+    otherwise, applied only to pulsing states."""
+    if spec.bars_symbol is None or spec.downbeat_symbol is None:
+        return None
+    color, style = render_led_spec(spec, output_values)
+    if color is None:
+        return None
+    pulse = beat_brightness if style is LedDisplayStyle.METRONOME else 1.0
+
+    state = int(output_values.get(spec.state_symbol, 0))
+    bars = int(output_values.get(spec.bars_symbol, 0))
+    measure = int(output_values.get(spec.downbeat_symbol, 0))
+
+    # Past the length it declared (an overdub that outgrew the head loop) is
+    # the same situation as a take still recording: a position, no denominator.
+    is_chasing = state in spec.chase_states or (bars > 0 and measure >= bars)
+
+    if bar_phase is None:
+        if bars <= 0 and not is_chasing:
+            return None
+        if state in spec.steady_states:
+            return LoopProgress(LoopFill.STATIC, color, bars)
+        return LoopProgress(LoopFill.FREE, color, 0 if is_chasing else bars, 0.0, pulse)
+
+    if is_chasing:
+        return LoopProgress(LoopFill.CHASE, color, 0, bar_phase, pulse)
+    if bars <= 0:
+        return None
+    if state in spec.steady_states:
+        return LoopProgress(LoopFill.STATIC, color, bars)
+    return LoopProgress(LoopFill.FILL, color, bars, (measure + bar_phase) / bars, pulse)