"""Footswitch controller — debounced GPIO input for stomp switches. Handles multiple momentary footswitches with debouncing, long-press detection, and mode switching. Typical layout: [FS1] Preset Up / Tap Tempo [FS2] Preset Down / Hold for Tuner [FS3] Bypass [FS4] Bank Up [FS5] FX Select [FS6] Tap Tempo """ from __future__ import annotations import logging import threading import time from dataclasses import dataclass, field from enum import Enum from typing import Callable, Optional logger = logging.getLogger(__name__) DEBOUNCE_MS = 20 # Debounce window LONG_PRESS_MS = 500 # Long press threshold POLL_INTERVAL_S = 0.005 # 5ms poll for responsiveness class SwitchAction(Enum): """Actions triggered by footswitch events.""" PRESET_UP = "preset_up" PRESET_DOWN = "preset_down" BANK_UP = "bank_up" BANK_DOWN = "bank_down" BYPASS = "bypass" TAP_TEMPO = "tap_tempo" TUNER = "tuner" FX_PREV = "fx_prev" FX_NEXT = "fx_next" EXPRESSION_TOGGLE = "expression_toggle" MIDI_LEARN = "midi_learn" SNAPSHOT_SAVE = "snapshot_save" @dataclass class FootSwitch: """State of a single footswitch.""" gpio_pin: int action_default: SwitchAction action_long_press: Optional[SwitchAction] = None active_low: bool = True def _import_gpio(): """Import RPi.GPIO gracefully — returns None on non-RPi platforms.""" try: import RPi.GPIO as GPIO return GPIO except (ImportError, RuntimeError): return None class FootswitchController: """Debounced footswitch input monitor. In production, this reads RPi.GPIO. In testing/dev, it uses a virtual pin store that can be driven by simulate_press. """ def __init__(self, switches: list[FootSwitch] | None = None): self._switches = switches or self._default_layout() self._callbacks: dict[SwitchAction, list[Callable]] = {} self._running = False self._thread: Optional[threading.Thread] = None self._gpio = _import_gpio() # Per-pin state tracking — maps gpio_pin -> pin state self._pin_tracker: dict[int, _PinState] = {} for sw in self._switches: self._pin_tracker[sw.gpio_pin] = _PinState() def _default_layout(self) -> list[FootSwitch]: """4-switch default layout.""" return [ FootSwitch(17, SwitchAction.PRESET_UP, SwitchAction.TAP_TEMPO), FootSwitch(27, SwitchAction.PRESET_DOWN, SwitchAction.TUNER), FootSwitch(22, SwitchAction.BYPASS, SwitchAction.SNAPSHOT_SAVE), FootSwitch(23, SwitchAction.BANK_UP, SwitchAction.BANK_DOWN), ] def register_callback(self, action: SwitchAction, callback: Callable) -> None: """Register a callback for a switch action.""" self._callbacks.setdefault(action, []).append(callback) def _trigger(self, action: SwitchAction) -> None: """Fire callbacks for an action.""" for cb in self._callbacks.get(action, []): try: cb() except Exception as e: logger.error("Switch callback error for %s: %s", action.value, e) # --- GPIO abstraction layer --- def _read_pin(self, pin: int) -> bool: """Read a GPIO pin. Returns True = pressed, False = released. If RPi.GPIO is unavailable, reads from a virtual store (for testing). """ if self._gpio: raw = self._gpio.input(pin) # Find which switch maps to this pin so we can invert if active_low for sw in self._switches: if sw.gpio_pin == pin: return not raw if sw.active_low else bool(raw) return bool(raw) else: return self._pin_tracker[pin].virtual_level def _setup_pins(self) -> None: """Configure GPIO pins as inputs with pull-up/down.""" if not self._gpio: logger.info("No RPi.GPIO — running in virtual (test) mode") return self._gpio.setmode(self._gpio.BCM) for sw in self._switches: pull = self._gpio.PUD_UP if sw.active_low else self._gpio.PUD_DOWN self._gpio.setup(sw.gpio_pin, self._gpio.IN, pull_up_down=pull) def _cleanup_pins(self) -> None: """Release GPIO pin configuration.""" if self._gpio: self._gpio.cleanup() # --- Debounce engine --- def _poll_loop(self) -> None: """Background thread: poll pins with debounce and long-press detection.""" last_logged_state: dict[int, bool] = {} for sw in self._switches: last_logged_state[sw.gpio_pin] = False while self._running: now_ms = time.monotonic() * 1000 for sw in self._switches: pin = sw.gpio_pin tracker = self._pin_tracker[pin] raw = self._read_pin(pin) # --- Debounce --- if raw != tracker.unstable_level: tracker.unstable_level = raw tracker.last_change_ms = now_ms continue # Changed — wait for debounce window # Stable across debounce window? elapsed = now_ms - tracker.last_change_ms if elapsed < DEBOUNCE_MS: continue # Still within debounce window # Stable and beyond debounce window — commit stable state if raw != tracker.stable_level: tracker.stable_level = raw logger.debug("Pin %d debounced: %s", pin, "PRESSED" if raw else "released") if raw: # Just pressed tracker.press_start_ms = now_ms else: # Just released — check for short vs long press press_duration = now_ms - tracker.press_start_ms if press_duration >= LONG_PRESS_MS and sw.action_long_press: logger.debug("Pin %d LONG press (%dms) → %s", pin, int(press_duration), sw.action_long_press.value) self._trigger(sw.action_long_press) tracker.long_press_handled = True elif press_duration >= LONG_PRESS_MS and not sw.action_long_press: # No long-press action mapped — fall through to default logger.debug("Pin %d long press, no action mapped — triggering default", pin) self._trigger(sw.action_default) else: logger.debug("Pin %d short press (%dms) → %s", pin, int(press_duration), sw.action_default.value) self._trigger(sw.action_default) tracker.long_press_handled = False # If still pressed and past long-press threshold but no release yet # — don't repeatedly fire, just mark it if raw and tracker.stable_level and not tracker.long_press_handled: press_duration = now_ms - tracker.press_start_ms if press_duration >= LONG_PRESS_MS and sw.action_long_press: logger.debug("Pin %d LONG press triggered (no release needed)", pin) self._trigger(sw.action_long_press) tracker.long_press_handled = True # Reset long_press_handled when released if not raw: tracker.long_press_handled = False time.sleep(POLL_INTERVAL_S) # --- Lifecycle --- def start(self) -> None: """Start monitoring footswitches. In production, starts GPIO monitoring thread. """ self._setup_pins() self._running = True self._thread = threading.Thread(target=self._poll_loop, daemon=True, name="footswitch-poll") self._thread.start() logger.info("Footswitch controller started (%d switches)", len(self._switches)) def stop(self) -> None: """Stop monitoring.""" self._running = False if self._thread: self._thread.join(timeout=2.0) self._thread = None self._cleanup_pins() logger.info("Footswitch controller stopped") # --- Testing hooks --- def simulate_press(self, action: SwitchAction) -> None: """Simulate a footswitch press (for testing). Directly triggers the action without going through GPIO. """ logger.debug("SIMULATED press: %s", action.value) self._trigger(action) def simulate_gpio_change(self, pin: int, pressed: bool) -> None: """Set a virtual GPIO pin level for testing the debounce engine.""" tracker = self._pin_tracker.get(pin) if tracker: tracker.virtual_level = pressed # --- Internal state holder --- @dataclass class _PinState: """Per-pin debounce state.""" last_change_ms: float = 0.0 press_start_ms: float = 0.0 unstable_level: bool = False stable_level: bool = False long_press_handled: bool = False virtual_level: bool = False # For testing without RPi.GPIO