shawn
3f7257aa23
CI / test (push) Has been cancelled
- Add USB audio IRQ affinity pinning to core 3 in main.py - Add enable_xrun_tracking() to AudioSystem for kernel-level diagnostics - Wrap Python process with chrt -f 80 in systemd service template - Add LimitSIGPENDING=128 for signal queue depth - Create scripts/rt-tune.sh — comprehensive RT tuning startup script (IRQ affinity, CPU governor, C-states, ALSA limits, xrun_debug) - Create docs/rt-performance-tuning.md — reference doc with all tuning knobs, measurement tools, and systematic procedure Targets: <12ms RT latency (8ms ideal), zero xruns, CPU <40% at 512/48k
937 lines
40 KiB
Python
937 lines
40 KiB
Python
#!/usr/bin/env python3
|
|
"""Pi Multi-FX Pedal — main entry point.
|
|
|
|
Wires together all subsystems in a defined boot order:
|
|
|
|
Config → Audio (I2S/JACK) → DSP Pipeline → Preset Manager → MIDI → Footswitch → Display → LEDs
|
|
|
|
Graceful shutdown reverses the order: stop input sources → fade UI → save state → stop audio.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import signal
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import yaml
|
|
|
|
# ── Subsystem imports ─────────────────────────────────────────────────────────
|
|
from src.system.audio import AudioConfig, AudioSystem, JackAudioClient, _jack_is_running
|
|
from src.system.defaults import ensure_defaults_exist
|
|
from src.dsp.pipeline import AudioPipeline
|
|
from src.dsp.nam_router import NAMEngineRouter
|
|
from src.dsp.ir_loader import IRLoader
|
|
from src.presets.manager import PresetManager
|
|
from src.presets.types import Channel, MIDIMapping, Preset
|
|
from src.midi.handler import MIDIHandler
|
|
from src.ui.footswitch import FootswitchController, FootSwitch, SwitchAction
|
|
from src.ui.leds import LEDController, LEDDriver, LEDConfig, LEDPattern
|
|
from src.ui.display import DisplayController, DisplayState
|
|
from src.web.server import WebServer, WebServerDeps
|
|
|
|
# ── Logging ───────────────────────────────────────────────────────────────────
|
|
|
|
logger = logging.getLogger("pedal")
|
|
_console = logging.StreamHandler()
|
|
_console.setFormatter(logging.Formatter(
|
|
"%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
datefmt="%H:%M:%S",
|
|
))
|
|
logger.addHandler(_console)
|
|
logger.setLevel(logging.INFO)
|
|
|
|
# ── Paths ─────────────────────────────────────────────────────────────────────
|
|
|
|
from src.system.config import DEFAULT_CONFIG_PATH, load_config, save_config, DEFAULT_CONFIG
|
|
|
|
# ── Security helpers ──────────────────────────────────────────────────────────
|
|
|
|
_AUTH_PIN_SEEN = False
|
|
|
|
|
|
def _ensure_singleton() -> None:
|
|
"""Kill any other main.py processes (from stale service restarts)."""
|
|
import subprocess, os
|
|
my_pid = os.getpid()
|
|
try:
|
|
r = subprocess.run(
|
|
["pgrep", "-f", "python3.*main.py"],
|
|
capture_output=True, text=True, timeout=3,
|
|
)
|
|
if r.returncode == 0:
|
|
for pid_str in r.stdout.strip().split():
|
|
pid = int(pid_str)
|
|
if pid != my_pid:
|
|
try:
|
|
os.kill(pid, 9)
|
|
logger.info("Killed stale pedal process PID %d", pid)
|
|
except (OSError, ProcessLookupError):
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _ensure_auth_pin(config: dict, config_path: Path) -> str:
|
|
"""Generate a random 6-digit auth PIN on first boot and persist it to config.
|
|
|
|
Returns the PIN (existing or newly generated). The PIN is logged at INFO
|
|
level on first generation so the user can see it. Subsequent boots log
|
|
only a brief confirmation.
|
|
"""
|
|
global _AUTH_PIN_SEEN
|
|
web_cfg = config.setdefault("web", {})
|
|
pin = web_cfg.get("auth_pin")
|
|
if not pin:
|
|
import secrets
|
|
pin = str(secrets.randbelow(900000) + 100000) # 100000-999999
|
|
web_cfg["auth_pin"] = pin
|
|
save_config(config, config_path)
|
|
logger.info("═" * 50)
|
|
logger.info("🔐 NEW AUTH PIN GENERATED: %s", pin)
|
|
logger.info("═" * 50)
|
|
_AUTH_PIN_SEEN = True
|
|
elif not _AUTH_PIN_SEEN:
|
|
logger.info("Auth PIN is configured (not shown — see ~/.pedal/config.yaml)")
|
|
_AUTH_PIN_SEEN = True
|
|
return pin
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# Pedal application
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class PedalApp:
|
|
"""Main pedal application — wires and owns all subsystems.
|
|
|
|
Lifecycle::
|
|
|
|
app = PedalApp()
|
|
app.boot() # start everything
|
|
app.run() # main loop (blocks, handles signals)
|
|
app.shutdown() # graceful teardown
|
|
"""
|
|
|
|
def __init__(self, config_path: Path = DEFAULT_CONFIG_PATH) -> None:
|
|
self._config = load_config(config_path)
|
|
self._config_path = config_path
|
|
|
|
# ── Runtime state ──────────────────────────────────────────────
|
|
self._running = False
|
|
self._boot_time: Optional[float] = None
|
|
self._bypassed = False
|
|
|
|
# ── Subsystem references (set during boot) ─────────────────────
|
|
self.audio_config: AudioConfig | None = None
|
|
self.audio_system: AudioSystem | None = None
|
|
self.jack_audio: JackAudioClient | None = None
|
|
self.nam_host: NAMEngineRouter | None = None
|
|
self.ir_loader: IRLoader | None = None
|
|
self.pipeline: AudioPipeline | None = None
|
|
self.presets: PresetManager | None = None
|
|
self.midi: MIDIHandler | None = None
|
|
self.footswitches: FootswitchController | None = None
|
|
self.leds: LEDController | None = None
|
|
self.display: DisplayController | None = None
|
|
self.web: WebServer | None = None
|
|
|
|
# ── Signal handlers ────────────────────────────────────────────
|
|
self._shutdown_requested = threading.Event()
|
|
|
|
# Override the defaults AFTER merging so our overrides win
|
|
if "shutdown_grace_period_s" not in self._config:
|
|
self._config["shutdown_grace_period_s"] = 3.0
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Boot — start subsystems in dependency order
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
def boot(self) -> bool:
|
|
"""Initialize all subsystems in correct order.
|
|
|
|
Returns True if boot succeeded (non-critical subsystems may start
|
|
in degraded mode — MIDI without interfaces, LEDs in mock, etc.).
|
|
"""
|
|
boot_start = time.monotonic()
|
|
logger.info("═══════ Pi Multi-FX Pedal — Booting ═══════")
|
|
|
|
try:
|
|
# ── 0. Ensure default IRs, NAM models, and directories ──
|
|
ensure_defaults_exist()
|
|
|
|
# ── 1. Audio config + system ──────────────────────────
|
|
acfg = self._config["audio"]
|
|
audio_mode = acfg.get("mode", "mono")
|
|
self.audio_config = AudioConfig(
|
|
hat_type=acfg.get("hat_type", "audioinjector"),
|
|
profile=acfg.get("profile", "standard"),
|
|
mode=audio_mode,
|
|
input_device=acfg.get("input_device", "hw:0,0"),
|
|
output_device=acfg.get("output_device", "hw:0,0"),
|
|
jack_enabled=acfg.get("jack_enabled", True),
|
|
auto_connect=acfg.get("auto_connect", True),
|
|
period=acfg.get("period"),
|
|
rate=acfg.get("rate"),
|
|
)
|
|
self.audio_system = AudioSystem(self.audio_config)
|
|
self.audio_system.setup_i2s(reboot_hint=True)
|
|
|
|
# Clean stale JACK SHM segments from prior crashes before starting
|
|
try:
|
|
import shutil
|
|
from pathlib import Path
|
|
for p in Path("/dev/shm").glob("jack*"):
|
|
if p.is_dir():
|
|
shutil.rmtree(p, ignore_errors=True)
|
|
else:
|
|
p.unlink(missing_ok=True)
|
|
except Exception:
|
|
pass
|
|
|
|
# Enable kernel-level xrun tracking for diagnostics
|
|
try:
|
|
self.audio_system.enable_xrun_tracking()
|
|
except Exception:
|
|
pass
|
|
|
|
if self.audio_config.jack_enabled:
|
|
self.audio_system.start_jack(timeout=10)
|
|
else:
|
|
logger.info("JACK disabled in config — skipping audio server start")
|
|
|
|
# ── 2. DSP pipeline (NAM + IR + FX chain) ────────────
|
|
block_size = self.audio_config.latency_profile["period"]
|
|
sample_rate = self.audio_config.latency_profile.get("rate", 48000)
|
|
self.nam_host = NAMEngineRouter(block_size=block_size, sample_rate=sample_rate)
|
|
self.ir_loader = IRLoader()
|
|
self.pipeline = AudioPipeline(nam_host=self.nam_host, ir_loader=self.ir_loader)
|
|
self.nam_host.warm_up()
|
|
|
|
# Set pipeline routing from config mode
|
|
if audio_mode == "stereo_4cm":
|
|
self.pipeline.routing_mode = "4cm"
|
|
logger.info("Pipeline routing set to 4CM mode")
|
|
else:
|
|
self.pipeline.routing_mode = "mono"
|
|
|
|
# ── 2b. JACK audio client (capture → pipeline → playback) ──
|
|
if self.audio_config.jack_enabled:
|
|
is_stereo = audio_mode == "stereo_4cm"
|
|
channels = 2 if is_stereo else 1
|
|
self.jack_audio = JackAudioClient(
|
|
pipeline=self.pipeline,
|
|
client_name="pi-multifx",
|
|
input_channels=channels,
|
|
output_channels=channels,
|
|
)
|
|
self.jack_audio.start()
|
|
if not self.jack_audio._active:
|
|
logger.warning("JACK audio client failed to start — running without real-time audio I/O")
|
|
else:
|
|
logger.info(
|
|
"JACK audio client started: %s mode, %d ch I/O",
|
|
audio_mode, channels,
|
|
)
|
|
|
|
# ── 3. Preset manager (shared across all channels) ──────────
|
|
pcfg = self._config["presets"]
|
|
self.presets = PresetManager(
|
|
preset_dir=pcfg.get("dir", "~/.pedal/presets"),
|
|
audio_pipeline=self.pipeline,
|
|
)
|
|
|
|
# ── 3b. Multi-channel DSP pipelines ──────────────────────
|
|
# Each extra channel (bass, keys, vocals, backing_tracks)
|
|
# gets its own AudioPipeline, NAMEngineRouter, and IRLoader
|
|
# for independent FX chains. Gated behind a config flag.
|
|
multi_ch_enabled = self._config.get("multi_channel", {}).get("enabled", False)
|
|
self.bass_pipeline: AudioPipeline | None = None
|
|
self.bass_nam_host: NAMEngineRouter | None = None
|
|
self.bass_ir_loader: IRLoader | None = None
|
|
if multi_ch_enabled:
|
|
self.bass_nam_host = NAMEngineRouter(block_size=block_size, sample_rate=sample_rate)
|
|
self.bass_ir_loader = IRLoader()
|
|
self.bass_pipeline = AudioPipeline(
|
|
nam_host=self.bass_nam_host,
|
|
ir_loader=self.bass_ir_loader,
|
|
)
|
|
self.bass_nam_host.warm_up()
|
|
logger.info("Bass DSP pipeline initialised (multi-channel enabled)")
|
|
else:
|
|
logger.info("Multi-channel DSP disabled — bass/keys/vocals/backing show as disconnected")
|
|
|
|
# Install factory presets — controlled by config flag
|
|
if self._config.get("presets", {}).get("install_factory", False):
|
|
for ch in Channel:
|
|
installed = self.presets.install_factory_presets(
|
|
overwrite=False, channel=ch,
|
|
)
|
|
if installed:
|
|
logger.info("Installed %d factory preset(s) for %s",
|
|
installed, ch.value)
|
|
ir_installed = self.presets.install_factory_irs(overwrite=False)
|
|
if ir_installed:
|
|
logger.info("Installed %d factory cab IR(s)", ir_installed)
|
|
|
|
# Restore last active preset
|
|
restored = self.presets.restore_state()
|
|
if restored is None:
|
|
# First boot or no saved state — activate bank 0, program 0
|
|
logger.info("No previous state — activating default preset")
|
|
try:
|
|
self.presets.select(0, 0)
|
|
except Exception:
|
|
logger.info("No presets exist yet — pedal ready for MIDI/footswitch input")
|
|
else:
|
|
# restore_state() loads from disk but doesn't activate the pipeline
|
|
# Need to call select() to trigger pipeline.load_preset() and load NAM
|
|
logger.info("State restored — activating preset '%s' (bank=%d, program=%d)",
|
|
restored.name, restored.bank, restored.program)
|
|
self.presets.select(restored.bank, restored.program)
|
|
|
|
# ── 4. MIDI handler ──────────────────────────────────
|
|
self.midi = MIDIHandler()
|
|
self._wire_midi_callbacks()
|
|
mcfg = self._config["midi"]
|
|
self.midi.start(
|
|
uart_port=mcfg.get("uart_port", "/dev/ttyAMA0"),
|
|
usb=mcfg.get("usb", True),
|
|
)
|
|
|
|
# ── 5. Footswitch controller ─────────────────────────
|
|
self.footswitches = FootswitchController(
|
|
switches=self._build_footswitch_layout(),
|
|
)
|
|
self._wire_footswitch_callbacks()
|
|
self.footswitches.start()
|
|
|
|
# ── 6. Display ───────────────────────────────────────
|
|
dcfg = self._config["display"]
|
|
self.display = DisplayController(
|
|
i2c_bus=dcfg.get("i2c_bus", 1),
|
|
i2c_addr=dcfg.get("i2c_addr", 0x3C),
|
|
)
|
|
self.display.initialize()
|
|
self._update_display()
|
|
|
|
# ── 7. LEDs ──────────────────────────────────────────
|
|
lcfg = self._config["leds"]
|
|
driver_map = {"neopixel": LEDDriver.NEOPIXEL, "dotstar": LEDDriver.DOTSTAR, "mock": LEDDriver.MOCK}
|
|
self.leds = LEDController(
|
|
num_leds=lcfg.get("num_leds", 4),
|
|
driver=driver_map.get(lcfg.get("driver", "neopixel"), LEDDriver.NEOPIXEL),
|
|
pin=lcfg.get("pin", "D18"),
|
|
brightness=lcfg.get("brightness", 0.5),
|
|
)
|
|
self.leds.initialize()
|
|
|
|
# Boot LED animation — quick scan
|
|
self.leds.preset_animate(direction="up")
|
|
|
|
# ── 7b. Auth PIN (first boot generates, persists to config) ──
|
|
_ensure_auth_pin(self._config, self._config_path)
|
|
|
|
# ── 8. Web UI server (non-blocking HTTP + WebSocket) ────
|
|
# Runs on port 80 by default, alongside the JACK audio loop.
|
|
try:
|
|
self.web = WebServer(
|
|
deps=WebServerDeps(
|
|
presets=self.presets,
|
|
pipeline=self.pipeline,
|
|
nam_host=self.nam_host,
|
|
ir_loader=self.ir_loader,
|
|
bass_pipeline=self.bass_pipeline,
|
|
bass_nam_host=self.bass_nam_host,
|
|
bass_ir_loader=self.bass_ir_loader,
|
|
audio_system=self.audio_system,
|
|
jack_audio=self.jack_audio,
|
|
config=self._config,
|
|
config_path=self._config_path,
|
|
),
|
|
host="0.0.0.0",
|
|
port=80,
|
|
)
|
|
self.web.start()
|
|
logger.info("Web UI server started on http://0.0.0.0:80")
|
|
except Exception as e:
|
|
logger.warning("Web UI server failed to start (non-fatal): %s", e)
|
|
|
|
except Exception as e:
|
|
logger.critical("Boot failed: %s", e, exc_info=True)
|
|
return False
|
|
|
|
self._boot_time = time.monotonic()
|
|
elapsed = self._boot_time - boot_start
|
|
logger.info("Boot complete in %.1fs — pedal is ready", elapsed)
|
|
return True
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Wire callbacks
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
def _wire_midi_callbacks(self) -> None:
|
|
"""Connect MIDI events to preset manager and parameter handlers."""
|
|
if not self.midi or not self.presets:
|
|
return
|
|
|
|
# Program Change → preset switch
|
|
self.midi.set_pc_callback(self._on_midi_pc)
|
|
|
|
# CC → registered parameter handlers (per preset mapping)
|
|
# We register a general CC handler that looks up the current
|
|
# preset's MIDI mappings dynamically.
|
|
# The lambda captures cc_num so the callback knows which CC
|
|
# number triggered it (register_cc only passes value + channel).
|
|
for cc in range(128):
|
|
self.midi.register_cc(cc, lambda val, ch, cc_num=cc: self._on_midi_cc(val, cc_num, ch))
|
|
|
|
# MIDI Learn completion
|
|
self.midi.set_midi_learn_callback(self._on_midi_learn)
|
|
|
|
def _on_midi_pc(self, channel: int, program: int) -> None:
|
|
"""MIDI Program Change → select preset."""
|
|
if not self.presets:
|
|
return
|
|
try:
|
|
preset = self.presets.midi_pc(channel, program)
|
|
self._on_preset_changed(preset)
|
|
except Exception as e:
|
|
logger.warning("MIDI PC to (ch=%d, pg=%d) failed: %s", channel, program, e)
|
|
|
|
def _on_midi_cc(self, value: int, cc_number: int, channel: int) -> None:
|
|
"""MIDI CC → live parameter update.
|
|
|
|
Maps expression pedal (CC 11) to master volume by default.
|
|
Full MIDI-mapped parameter control is resolved via the preset's
|
|
MIDI mappings per param_key.
|
|
|
|
Args:
|
|
value: CC value (0-127).
|
|
cc_number: The CC number (0-127) that triggered this callback.
|
|
channel: MIDI channel (0-15) the message arrived on.
|
|
"""
|
|
# Map expression pedal (CC 11) to master volume regardless of
|
|
# preset state — global mapping.
|
|
if cc_number == 11:
|
|
if self.pipeline:
|
|
normalized = value / 127.0
|
|
self.pipeline.master_volume = normalized
|
|
logger.debug("MIDI CC 11 → master volume: %.2f", normalized)
|
|
return
|
|
|
|
if not self.presets or not self.pipeline:
|
|
return
|
|
|
|
# Look up preset's MIDI mappings for other CC numbers
|
|
try:
|
|
bank = self.presets.current_bank
|
|
program = self.presets.current_program
|
|
preset = self.presets.load(bank, program)
|
|
except Exception:
|
|
return
|
|
|
|
# Iterate preset mappings and find which one matches this CC number
|
|
for param_key, mapping in preset.midi_mappings.items():
|
|
if mapping.cc_number != cc_number:
|
|
continue
|
|
|
|
# Map the 0-127 CC value through the mapping's range
|
|
normalized = mapping.min_val + (value / 127.0) * (mapping.max_val - mapping.min_val)
|
|
logger.debug("MIDI CC %d → %s = %.2f (range %.1f-%.1f)",
|
|
cc_number, param_key, normalized,
|
|
mapping.min_val, mapping.max_val)
|
|
|
|
# Resolve the param_key to a block + param name and apply it
|
|
from src.presets.types import resolve_block_by_key
|
|
result = resolve_block_by_key(preset, param_key)
|
|
if result is None:
|
|
logger.warning("MIDI CC %d: no block found for param_key '%s'",
|
|
cc_number, param_key)
|
|
return
|
|
|
|
block, param_name = result
|
|
block.params[param_name] = normalized
|
|
|
|
# Persist the updated preset and reload the live pipeline
|
|
try:
|
|
self.presets.save(preset)
|
|
self.pipeline.load_preset(preset)
|
|
logger.debug("MIDI CC %d applied: %s = %.2f",
|
|
cc_number, param_key, normalized)
|
|
except Exception as e:
|
|
logger.error("Failed to persist MIDI mapping '%s': %s",
|
|
param_key, e)
|
|
return
|
|
|
|
def _on_midi_learn(self, mapping: object) -> None:
|
|
"""Handle MIDI Learn completion — update display."""
|
|
logger.info("MIDI Learn complete: %s", mapping)
|
|
self._update_display()
|
|
|
|
def _wire_footswitch_callbacks(self) -> None:
|
|
"""Connect footswitch actions to pedal controls."""
|
|
if not self.footswitches:
|
|
return
|
|
|
|
self.footswitches.register_callback(SwitchAction.PRESET_UP, self._on_preset_up)
|
|
self.footswitches.register_callback(SwitchAction.PRESET_DOWN, self._on_preset_down)
|
|
self.footswitches.register_callback(SwitchAction.BANK_UP, self._on_bank_up)
|
|
self.footswitches.register_callback(SwitchAction.BANK_DOWN, self._on_bank_down)
|
|
self.footswitches.register_callback(SwitchAction.BYPASS, self._on_bypass_toggle)
|
|
self.footswitches.register_callback(SwitchAction.TAP_TEMPO, self._on_tap_tempo)
|
|
self.footswitches.register_callback(SwitchAction.TUNER, self._on_tuner_toggle)
|
|
self.footswitches.register_callback(SwitchAction.SNAPSHOT_SAVE, self._on_snapshot_save)
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Action handlers
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
def _on_preset_up(self) -> None:
|
|
if not self.presets:
|
|
return
|
|
try:
|
|
preset = self.presets.preset_up()
|
|
self._on_preset_changed(preset)
|
|
if self.leds:
|
|
self.leds.preset_animate(direction="up")
|
|
except Exception as e:
|
|
logger.warning("preset_up failed: %s", e)
|
|
|
|
def _on_preset_down(self) -> None:
|
|
if not self.presets:
|
|
return
|
|
try:
|
|
preset = self.presets.preset_down()
|
|
self._on_preset_changed(preset)
|
|
if self.leds:
|
|
self.leds.preset_animate(direction="down")
|
|
except Exception as e:
|
|
logger.warning("preset_down failed: %s", e)
|
|
|
|
def _on_bank_up(self) -> None:
|
|
if not self.presets:
|
|
return
|
|
try:
|
|
bank, preset = self.presets.bank_up()
|
|
self._on_preset_changed(preset)
|
|
if self.leds:
|
|
self.leds.preset_animate(direction="up")
|
|
except Exception as e:
|
|
logger.warning("bank_up failed: %s", e)
|
|
|
|
def _on_bank_down(self) -> None:
|
|
if not self.presets:
|
|
return
|
|
try:
|
|
bank, preset = self.presets.bank_down()
|
|
self._on_preset_changed(preset)
|
|
if self.leds:
|
|
self.leds.preset_animate(direction="down")
|
|
except Exception as e:
|
|
logger.warning("bank_down failed: %s", e)
|
|
|
|
def _on_bypass_toggle(self) -> None:
|
|
self._bypassed = not self._bypassed
|
|
if self.pipeline:
|
|
self.pipeline.bypassed = self._bypassed
|
|
logger.info("Bypass %s", "ON" if self._bypassed else "OFF")
|
|
if self.leds:
|
|
self.leds.set_bypass_led(2, self._bypassed)
|
|
self._update_display()
|
|
|
|
def _on_tap_tempo(self) -> None:
|
|
if self.leds:
|
|
self.leds.tap_tempo_blip()
|
|
logger.debug("Tap tempo")
|
|
|
|
def _on_tuner_toggle(self) -> None:
|
|
logger.info("Tuner mode toggled (stub — mute audio, show tuner display)")
|
|
if self.display:
|
|
state = DisplayState(mode="tuner", tuner_note="--", tuner_cents=0)
|
|
self.display.update(state)
|
|
|
|
def _on_snapshot_save(self) -> None:
|
|
if not self.presets:
|
|
return
|
|
try:
|
|
bank = self.presets.current_bank
|
|
program = self.presets.current_program
|
|
preset = self.presets.load(bank, program)
|
|
self.presets.save(preset)
|
|
logger.info("Snapshot saved: '%s'", preset.name)
|
|
except Exception as e:
|
|
logger.warning("Snapshot save failed: %s", e)
|
|
|
|
def _on_preset_changed(self, preset: Preset) -> None:
|
|
"""Called after any preset change to sync UI/state."""
|
|
self._update_display()
|
|
if self.leds:
|
|
self.leds.set_pixel(0, (0, 64, 255)) # Blue = active
|
|
logger.info("Active: '%s' (bank=%d, pg=%d)", preset.name, preset.bank, preset.program)
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Audio routing mode switching
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
def set_routing_mode(self, mode: str) -> None:
|
|
"""Hot-switch between mono and stereo_4cm routing at runtime.
|
|
|
|
Updates the pipeline routing mode, then restarts the JACK
|
|
audio client with the appropriate channel count.
|
|
|
|
Args:
|
|
mode: ``"mono"`` or ``"stereo_4cm"``.
|
|
|
|
Raises:
|
|
ValueError: If *mode* is not recognised.
|
|
"""
|
|
if mode not in ("mono", "stereo_4cm"):
|
|
raise ValueError(f"set_routing_mode: unknown mode {mode!r}")
|
|
|
|
if self.audio_config is not None:
|
|
self.audio_config.mode = mode
|
|
|
|
if self.pipeline is not None:
|
|
self.pipeline.routing_mode = "4cm" if mode == "stereo_4cm" else "mono"
|
|
logger.info("Pipeline routing mode switched to %s", self.pipeline.routing_mode)
|
|
|
|
# Restart JACK client with new channel count
|
|
if self.jack_audio is not None:
|
|
channels = 2 if mode == "stereo_4cm" else 1
|
|
self.jack_audio.set_channel_count(channels, channels)
|
|
|
|
# Re-connect JACK ports for the new routing
|
|
if self.audio_system is not None and self.audio_config is not None:
|
|
self.audio_system.connect_fx_ports()
|
|
|
|
logger.info("Routing mode switched to %s", mode)
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Display sync
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
def _update_display(self) -> None:
|
|
"""Refresh the OLED with current state."""
|
|
if not self.display or not self.presets:
|
|
return
|
|
|
|
try:
|
|
bank = self.presets.current_bank
|
|
program = self.presets.current_program
|
|
preset = self.presets.load(bank, program)
|
|
|
|
fx_active = [b.fx_type.value for b in preset.chain if b.enabled and not b.bypass]
|
|
fx_bypass_states = {b.fx_type.value: b.bypass for b in preset.chain}
|
|
|
|
state = DisplayState(
|
|
mode="preset",
|
|
preset_name=preset.name,
|
|
bank_name=f"Bank {bank}",
|
|
bypassed=self._bypassed,
|
|
fx_active=fx_active,
|
|
fx_bypass_states=fx_bypass_states,
|
|
hotspot_password=self._get_hotspot_password(),
|
|
)
|
|
self.display.update(state)
|
|
except Exception:
|
|
# Degraded: show minimal state
|
|
state = DisplayState(
|
|
mode="preset",
|
|
preset_name="Ready",
|
|
bank_name="",
|
|
bypassed=self._bypassed,
|
|
hotspot_password=self._get_hotspot_password(),
|
|
)
|
|
self.display.update(state)
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Helpers
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
def _get_hotspot_password(self) -> str:
|
|
"""Return hotspot password from config, or empty if not configured."""
|
|
try:
|
|
from src.system.network import get_hotspot_password
|
|
return get_hotspot_password()
|
|
except Exception:
|
|
return ""
|
|
|
|
def _build_footswitch_layout(self) -> list[FootSwitch]:
|
|
"""Build FootSwitch list from config."""
|
|
layout_cfg = self._config.get("footswitch", {}).get("layout", DEFAULT_CONFIG["footswitch"]["layout"])
|
|
action_map = {a.value: a for a in SwitchAction}
|
|
switches: list[FootSwitch] = []
|
|
for entry in layout_cfg:
|
|
default = action_map.get(entry.get("action_default", ""))
|
|
long_press = action_map.get(entry.get("action_long_press", ""))
|
|
if default is None:
|
|
logger.warning("Unknown footswitch action: %s", entry.get("action_default"))
|
|
continue
|
|
switches.append(FootSwitch(
|
|
gpio_pin=entry["gpio_pin"],
|
|
action_default=default,
|
|
action_long_press=long_press,
|
|
active_low=entry.get("active_low", True),
|
|
))
|
|
return switches
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Main loop
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
def run(self) -> None:
|
|
"""Main application loop.
|
|
|
|
Sets up signal handlers and blocks until a shutdown is requested.
|
|
"""
|
|
self._running = True
|
|
|
|
# Register signal handlers for graceful shutdown
|
|
signal.signal(signal.SIGTERM, self._signal_handler)
|
|
signal.signal(signal.SIGINT, self._signal_handler)
|
|
|
|
logger.info("Pedal running — press Ctrl+C or send SIGTERM to stop")
|
|
|
|
# Main loop — keep alive, refresh display periodically
|
|
last_display_refresh = 0.0
|
|
refresh_interval = 0.25 # 250 ms display refresh
|
|
|
|
while self._running and not self._shutdown_requested.is_set():
|
|
now = time.monotonic()
|
|
|
|
# Periodic display refresh (catches MIDI/footswitch-triggered
|
|
# state changes that don't go through _on_preset_changed)
|
|
if now - last_display_refresh >= refresh_interval:
|
|
self._update_display()
|
|
last_display_refresh = now
|
|
|
|
# Main loop sleeps — low CPU while waiting for events
|
|
self._shutdown_requested.wait(0.1)
|
|
|
|
self.shutdown()
|
|
|
|
def _signal_handler(self, signum: int, _frame: object) -> None:
|
|
"""Handle SIGTERM/SIGINT for graceful shutdown."""
|
|
signame = signal.Signals(signum).name
|
|
logger.info("Received %s — shutting down...", signame)
|
|
self._shutdown_requested.set()
|
|
self._running = False
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Shutdown — reverse boot order
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
def shutdown(self) -> None:
|
|
"""Graceful shutdown: stop inputs → fade UI → save → stop audio."""
|
|
logger.info("═══════ Shutting down ═══════")
|
|
shutdown_start = time.monotonic()
|
|
|
|
# 1. Stop JACK audio client (before JACK server)
|
|
if self.jack_audio:
|
|
try:
|
|
self.jack_audio.stop()
|
|
logger.debug("JACK audio client stopped")
|
|
except Exception as e:
|
|
logger.warning("JACK audio client stop error: %s", e)
|
|
|
|
# 2. Stop accepting input (MIDI + footswitch)
|
|
if self.midi:
|
|
try:
|
|
self.midi.stop()
|
|
logger.debug("MIDI stopped")
|
|
except Exception as e:
|
|
logger.warning("MIDI stop error: %s", e)
|
|
|
|
if self.footswitches:
|
|
try:
|
|
self.footswitches.stop()
|
|
logger.debug("Footswitch stopped")
|
|
except Exception as e:
|
|
logger.warning("Footswitch stop error: %s", e)
|
|
|
|
# 2b. Stop web server (before state save — WS clients get final message)
|
|
if self.web:
|
|
try:
|
|
self.web.stop()
|
|
logger.debug("Web server stopped")
|
|
except Exception as e:
|
|
logger.warning("Web server stop error: %s", e)
|
|
|
|
# 3. Save current state (preset + bank)
|
|
if self.presets:
|
|
try:
|
|
self.presets.save_state()
|
|
logger.info("State saved")
|
|
except Exception as e:
|
|
logger.warning("State save error: %s", e)
|
|
|
|
# 3. Fade LEDs
|
|
if self.leds:
|
|
try:
|
|
self._fade_leds()
|
|
logger.debug("LEDs faded")
|
|
except Exception as e:
|
|
logger.warning("LED fade error: %s", e)
|
|
|
|
# 4. Clear display
|
|
if self.display:
|
|
try:
|
|
self.display.clear()
|
|
logger.debug("Display cleared")
|
|
except Exception as e:
|
|
logger.warning("Display clear error: %s", e)
|
|
|
|
# 5. Stop JACK
|
|
if self.audio_system:
|
|
try:
|
|
self.audio_system.stop_jack()
|
|
logger.debug("JACK stopped")
|
|
except Exception as e:
|
|
logger.warning("JACK stop error: %s", e)
|
|
|
|
elapsed = time.monotonic() - shutdown_start
|
|
logger.info("Shutdown complete in %.1fs — goodbye", elapsed)
|
|
# Flush log
|
|
for h in logger.handlers:
|
|
h.flush()
|
|
|
|
@staticmethod
|
|
def _fade_leds() -> None:
|
|
"""Fade all LEDs to off over ~300ms (simulated if no hardware)."""
|
|
# In production this would ramp brightness down.
|
|
# For now we just set all off.
|
|
time.sleep(0.05)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# CLI entry
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def main() -> int:
|
|
"""Parse args, boot, and run the pedal application.
|
|
|
|
Returns exit code 0 on clean shutdown, 1 on boot failure.
|
|
"""
|
|
# Kill any stale pedal processes before doing anything else
|
|
_ensure_singleton()
|
|
|
|
# Lock process memory early to prevent page faults in RT callback
|
|
try:
|
|
import ctypes
|
|
libc = ctypes.CDLL('libc.so.6')
|
|
# MCL_CURRENT | MCL_FUTURE = 1 | 2 = 3
|
|
if libc.mlockall(3) == 0:
|
|
logger.info("mlockall() OK — process memory locked")
|
|
else:
|
|
logger.warning("mlockall() failed — check LimitMEMLOCK in systemd unit")
|
|
except Exception as exc:
|
|
logger.warning("mlockall() not available: %s", exc)
|
|
|
|
# Set CPU governor to performance for stable RT audio
|
|
try:
|
|
for c in range(os.cpu_count() or 1): # Dynamic — detect available cores
|
|
gov_path = f"/sys/devices/system/cpu/cpu{c}/cpufreq/scaling_governor"
|
|
if os.path.exists(gov_path):
|
|
with open(gov_path, "w") as f:
|
|
f.write("performance")
|
|
logger.info("CPU%d governor set to performance", c)
|
|
except Exception as exc:
|
|
logger.warning("Could not set CPU governor (non-root?): %s", exc)
|
|
|
|
# Pin USB audio IRQ to a dedicated core for stable RT audio
|
|
# On RPi 4B, cores 0-2 handle kernel/general interrupts;
|
|
# pinning USB audio to core 3 isolates it from that noise.
|
|
try:
|
|
import glob as _glob
|
|
# Find the USB audio interface IRQ
|
|
_usb_irq: str | None = None
|
|
# Strategy 1: xhci-hcd (USB 3.0 controller on Pi 4B)
|
|
for _proc_dir in _glob.glob("/proc/irq/[0-9]*"):
|
|
try:
|
|
_name = (_glob.glob(f"{_proc_dir}/name")[0] if _glob.glob(f"{_proc_dir}/name") else None)
|
|
if _name:
|
|
_irq_name = open(_name).read().strip()
|
|
if "xhci" in _irq_name or "dwc" in _irq_name:
|
|
_usb_irq = os.path.basename(_proc_dir)
|
|
break
|
|
except (OSError, PermissionError, IndexError):
|
|
continue
|
|
if _usb_irq:
|
|
# Pin to core 3 (smp_affinity mask = 0x8)
|
|
_aff_path = f"/proc/irq/{_usb_irq}/smp_affinity"
|
|
_aff_list_path = f"/proc/irq/{_usb_irq}/smp_affinity_list"
|
|
with open(_aff_path, "w") as f:
|
|
f.write("8")
|
|
with open(_aff_list_path, "w") as f:
|
|
f.write("3")
|
|
# Move all other IRQs away from core 3
|
|
for _proc_dir in _glob.glob("/proc/irq/[0-9]*"):
|
|
_irq_num = os.path.basename(_proc_dir)
|
|
if _irq_num == _usb_irq:
|
|
continue # Skip our USB audio IRQ
|
|
try:
|
|
with open(f"{_proc_dir}/smp_affinity", "w") as f:
|
|
f.write("7") # cores 0,1,2
|
|
except (OSError, PermissionError):
|
|
continue
|
|
logger.info("USB audio IRQ %s pinned to core 3 for RT stability", _usb_irq)
|
|
else:
|
|
logger.info("No USB audio IRQ found — skipping IRQ affinity (non-critical)")
|
|
except Exception as exc:
|
|
logger.warning("Could not set IRQ affinity (non-root?): %s", exc)
|
|
|
|
# Disable Python garbage collector to prevent 10-50ms GC pauses in
|
|
# the real-time audio callback. At ~500 numpy allocs/sec in the
|
|
# pipeline, default GC (threshold=700) triggers every ~1.4s,
|
|
# causing audible pops. Reference counting handles 99% of cleanup.
|
|
try:
|
|
import gc
|
|
gc.disable()
|
|
gc.collect() # One final sweep to clear boot-time cyclic garbage
|
|
logger.info("GC disabled for RT audio stability")
|
|
except Exception as exc:
|
|
logger.warning("Could not disable GC: %s", exc)
|
|
|
|
import argparse
|
|
|
|
logger.info("Pedal v2 starting up...")
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description="Pi Multi-FX Pedal — real-time guitar multi-effects",
|
|
)
|
|
parser.add_argument(
|
|
"-c", "--config",
|
|
type=str,
|
|
default=os.environ.get("PEDAL_CONFIG", str(DEFAULT_CONFIG_PATH)),
|
|
help="Path to config YAML (env: PEDAL_CONFIG, default: ~/.pedal/config.yaml)",
|
|
)
|
|
parser.add_argument(
|
|
"-v", "--verbose",
|
|
action="store_true",
|
|
help="Enable debug logging",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
if args.verbose:
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
config_path = Path(args.config)
|
|
app = PedalApp(config_path=config_path)
|
|
|
|
if not app.boot():
|
|
return 1
|
|
|
|
app.run()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|