#!/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())