shawn/raspberry-pi-mixer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 1-4: Audio stack, mixer engine, MIDI, and network API
Lint & Validate / lint (push) Has been cancelled
Details

Browse Source 
P2-R1: ALSA + JACK2 low-latency config (scripts, quirks, tuning)
P2-R2: Carla integration (build scripts, 8ch rack config, NAM LV2 support)
P2-R3: Plugin manager, categories, blacklist, NAM model support
P3-R1: Mixer DSP engine (channel strip, routing matrix, bus mgr, automation)
P4-R1: MIDI engine (learn mode, clock sync, HID discovery, mapping store)
P4-R2: Network API (OSC server, FastAPI REST, WebSocket, auth, rate limiter)
P5-R1: Touchscreen UI evaluation + main entry point
docs: Audio stack, Carla integration, MIDI support, UI evaluation
tests: Full test suite (292 passing)
This commit is contained in:
Shawn
2026-05-19 20:39:17 -04:00
parent 7b123762b5
commit 9cd8292acc
97 changed files with 26545 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/midi/__init__.py
+53
View File
@@ -0,0 +1,53 @@
"""MIDI subsystem for Raspberry Pi RT Audio Mixer.
Provides USB MIDI device discovery, mapping engine, learn mode,
clock sync, and JACK MIDI bridge.
"""
from .types import (
MIDIMessage,
MIDIMessageType,
MIDIMapping,
MixerParameter,
ParameterCategory,
ParameterType,
)
from .midi_engine import MIDIEngine
from .midi_learn import MIDILearn, LearnState
from .midi_clock import MIDIClock
from .mapping_store import (
save_mappings,
load_mappings,
list_sessions,
delete_session,
)
from .jack_midi_bridge import JACKMIDIBridge
from .device_discovery import discover_all, MIDIDevice
from .controllers import (
KNOWN_CONTROLLERS,
find_controller,
find_controller_by_alsa,
)
__all__ = [
"MIDIMessage",
"MIDIMessageType",
"MIDIMapping",
"MixerParameter",
"ParameterCategory",
"ParameterType",
"MIDIEngine",
"MIDILearn",
"LearnState",
"MIDIClock",
"save_mappings",
"load_mappings",
"list_sessions",
"delete_session",
"JACKMIDIBridge",
"discover_all",
"MIDIDevice",
"KNOWN_CONTROLLERS",
"find_controller",
"find_controller_by_alsa",
]
src/midi/controllers.py
+412
View File
@@ -0,0 +1,412 @@
"""Supported MIDI controller database.
Pre-defined profiles for common USB MIDI controllers. Each profile
maps physical controls to MIDI CC/NRPN numbers. Profiles are used
to auto-populate initial mappings and for device identification.
"""
from __future__ import annotations
from dataclasses import dataclass, field
@dataclass
class ControllerControl:
"""Describes one physical control on a MIDI controller."""
name: str
cc_number: int # MIDI CC number (or -1 for NRPN)
channel: int = 0 # MIDI channel (0–15, -1 = omni)
control_type: str = "fader" # fader, knob, button, encoder, transport
is_nrpn: bool = False
nrpn_number: int = 0
notes: str = ""
@dataclass
class ControllerProfile:
"""Profile describing a known MIDI controller's control layout."""
manufacturer: str
model: str
usb_vendor_id: int # USB VID (hex)
usb_product_id: int # USB PID (hex)
alsa_client_pattern: str = "" # Substring to match in ALSA client name
num_faders: int = 0
num_knobs: int = 0
num_buttons: int = 0
has_transport: bool = False
has_jog_wheel: bool = False
has_scribble_strips: bool = False
controls: list[ControllerControl] = field(default_factory=list)
midi_channel_count: int = 1
notes: str = ""
def find_control(self, cc: int, channel: int = 0) -> ControllerControl | None:
for ctrl in self.controls:
if ctrl.cc_number == cc and (ctrl.channel == channel or ctrl.channel < 0):
return ctrl
return None
# ── Known controllers ───────────────────────────────────────────────────────
KNOWN_CONTROLLERS: list[ControllerProfile] = [
ControllerProfile(
manufacturer="Behringer",
model="X-Touch Compact",
usb_vendor_id=0x1397,
usb_product_id=0x00B4,
alsa_client_pattern="X-TOUCH COMPACT",
num_faders=9,
num_knobs=16,
num_buttons=39,
has_transport=True,
has_jog_wheel=False,
midi_channel_count=1,
notes="MCU-compatible mode preferred. Use MIDI channel 1.",
controls=[
# 9 motorised faders
ControllerControl("Fader 1", 70, 0, "fader"),
ControllerControl("Fader 2", 71, 0, "fader"),
ControllerControl("Fader 3", 72, 0, "fader"),
ControllerControl("Fader 4", 73, 0, "fader"),
ControllerControl("Fader 5", 74, 0, "fader"),
ControllerControl("Fader 6", 75, 0, "fader"),
ControllerControl("Fader 7", 76, 0, "fader"),
ControllerControl("Fader 8", 77, 0, "fader"),
ControllerControl("Master", 78, 0, "fader"),
# 16 rotary encoders (top row)
ControllerControl("Encoder 1", 10, 0, "encoder"),
ControllerControl("Encoder 2", 11, 0, "encoder"),
ControllerControl("Encoder 3", 12, 0, "encoder"),
ControllerControl("Encoder 4", 13, 0, "encoder"),
ControllerControl("Encoder 5", 14, 0, "encoder"),
ControllerControl("Encoder 6", 15, 0, "encoder"),
ControllerControl("Encoder 7", 16, 0, "encoder"),
ControllerControl("Encoder 8", 17, 0, "encoder"),
# Bottom encoders
ControllerControl("Encoder 9", 18, 0, "encoder"),
ControllerControl("Encoder 10", 19, 0, "encoder"),
ControllerControl("Encoder 11", 20, 0, "encoder"),
ControllerControl("Encoder 12", 21, 0, "encoder"),
ControllerControl("Encoder 13", 22, 0, "encoder"),
ControllerControl("Encoder 14", 23, 0, "encoder"),
ControllerControl("Encoder 15", 24, 0, "encoder"),
ControllerControl("Encoder 16", 25, 0, "encoder"),
# Buttons per channel
ControllerControl("Rec 1", 0, 0, "button"),
ControllerControl("Rec 2", 1, 0, "button"),
ControllerControl("Rec 3", 2, 0, "button"),
ControllerControl("Rec 4", 3, 0, "button"),
ControllerControl("Rec 5", 4, 0, "button"),
ControllerControl("Rec 6", 5, 0, "button"),
ControllerControl("Rec 7", 6, 0, "button"),
ControllerControl("Rec 8", 7, 0, "button"),
ControllerControl("Solo 1", 8, 0, "button"),
ControllerControl("Solo 2", 9, 0, "button"),
ControllerControl("Solo 3", 10, 0, "button"),
ControllerControl("Solo 4", 11, 0, "button"),
# Transport
ControllerControl("Play", 114, 0, "transport"),
ControllerControl("Stop", 115, 0, "transport"),
ControllerControl("Rec", 116, 0, "transport"),
ControllerControl("Rewind", 117, 0, "transport"),
ControllerControl("FFwd", 118, 0, "transport"),
ControllerControl("Loop", 119, 0, "transport"),
],
),
ControllerProfile(
manufacturer="Behringer",
model="X-Touch (MCU mode)",
usb_vendor_id=0x1397,
usb_product_id=0x00B5,
alsa_client_pattern="X-TOUCH",
num_faders=9,
num_knobs=8,
num_buttons=92,
has_transport=True,
has_jog_wheel=True,
has_scribble_strips=True,
midi_channel_count=1,
notes="Standard Mackie Control Universal protocol. 9 motorised faders, V-Pots, jog wheel.",
controls=[
ControllerControl("Fader 1", 0, 0, "fader"),
ControllerControl("Fader 2", 1, 0, "fader"),
ControllerControl("Fader 3", 2, 0, "fader"),
ControllerControl("Fader 4", 3, 0, "fader"),
ControllerControl("Fader 5", 4, 0, "fader"),
ControllerControl("Fader 6", 5, 0, "fader"),
ControllerControl("Fader 7", 6, 0, "fader"),
ControllerControl("Fader 8", 7, 0, "fader"),
ControllerControl("Master", 8, 0, "fader"),
# V-Pots (relative encoders)
ControllerControl("V-Pot 1", 16, 0, "encoder"),
ControllerControl("V-Pot 2", 17, 0, "encoder"),
ControllerControl("V-Pot 3", 18, 0, "encoder"),
ControllerControl("V-Pot 4", 19, 0, "encoder"),
ControllerControl("V-Pot 5", 20, 0, "encoder"),
ControllerControl("V-Pot 6", 21, 0, "encoder"),
ControllerControl("V-Pot 7", 22, 0, "encoder"),
ControllerControl("V-Pot 8", 23, 0, "encoder"),
# Transport
ControllerControl("Play", 94, 0, "transport"),
ControllerControl("Stop", 93, 0, "transport"),
ControllerControl("Rec", 95, 0, "transport"),
ControllerControl("Rewind", 91, 0, "transport"),
ControllerControl("FFwd", 92, 0, "transport"),
],
),
ControllerProfile(
manufacturer="FaderFox",
model="UC4",
usb_vendor_id=0x16D0,
usb_product_id=0x0D27,
alsa_client_pattern="Faderfox UC4",
num_faders=8,
num_knobs=0,
num_buttons=32,
has_transport=False,
midi_channel_count=1,
notes="Class-compliant USB MIDI. No drivers needed on Linux.",
controls=[
ControllerControl("Fader 1", 16, 0, "fader"),
ControllerControl("Fader 2", 17, 0, "fader"),
ControllerControl("Fader 3", 18, 0, "fader"),
ControllerControl("Fader 4", 19, 0, "fader"),
ControllerControl("Fader 5", 20, 0, "fader"),
ControllerControl("Fader 6", 21, 0, "fader"),
ControllerControl("Fader 7", 22, 0, "fader"),
ControllerControl("Fader 8", 23, 0, "fader"),
],
),
ControllerProfile(
manufacturer="Akai",
model="MIDImix",
usb_vendor_id=0x09E8,
usb_product_id=0x0031,
alsa_client_pattern="MIDImix",
num_faders=9,
num_knobs=24,
num_buttons=16,
has_transport=False,
midi_channel_count=1,
notes="8 channel strips + master. 3 knobs per channel. Bank buttons send CC.",
controls=[
# 8 channel faders
ControllerControl("Fader 1", 19, 0, "fader"),
ControllerControl("Fader 2", 23, 0, "fader"),
ControllerControl("Fader 3", 27, 0, "fader"),
ControllerControl("Fader 4", 31, 0, "fader"),
ControllerControl("Fader 5", 49, 0, "fader"),
ControllerControl("Fader 6", 53, 0, "fader"),
ControllerControl("Fader 7", 57, 0, "fader"),
ControllerControl("Fader 8", 61, 0, "fader"),
ControllerControl("Master", 62, 0, "fader"),
# Knobs row 1 per channel
ControllerControl("Knob 1A", 16, 0, "knob"),
ControllerControl("Knob 2A", 20, 0, "knob"),
ControllerControl("Knob 3A", 24, 0, "knob"),
ControllerControl("Knob 4A", 28, 0, "knob"),
ControllerControl("Knob 5A", 46, 0, "knob"),
ControllerControl("Knob 6A", 50, 0, "knob"),
ControllerControl("Knob 7A", 54, 0, "knob"),
ControllerControl("Knob 8A", 58, 0, "knob"),
# Knobs row 2
ControllerControl("Knob 1B", 17, 0, "knob"),
ControllerControl("Knob 2B", 21, 0, "knob"),
ControllerControl("Knob 3B", 25, 0, "knob"),
ControllerControl("Knob 4B", 29, 0, "knob"),
ControllerControl("Knob 5B", 47, 0, "knob"),
ControllerControl("Knob 6B", 51, 0, "knob"),
ControllerControl("Knob 7B", 55, 0, "knob"),
ControllerControl("Knob 8B", 59, 0, "knob"),
# Knobs row 3
ControllerControl("Knob 1C", 18, 0, "knob"),
ControllerControl("Knob 2C", 22, 0, "knob"),
ControllerControl("Knob 3C", 26, 0, "knob"),
ControllerControl("Knob 4C", 30, 0, "knob"),
ControllerControl("Knob 5C", 48, 0, "knob"),
ControllerControl("Knob 6C", 52, 0, "knob"),
ControllerControl("Knob 7C", 56, 0, "knob"),
ControllerControl("Knob 8C", 60, 0, "knob"),
# Mute/Solo buttons (momentary)
ControllerControl("Mute 1", 1, 0, "button"),
ControllerControl("Mute 2", 4, 0, "button"),
ControllerControl("Mute 3", 7, 0, "button"),
ControllerControl("Mute 4", 10, 0, "button"),
ControllerControl("Mute 5", 13, 0, "button"),
ControllerControl("Mute 6", 16, 0, "button"),
ControllerControl("Mute 7", 19, 0, "button"),
ControllerControl("Mute 8", 22, 0, "button"),
ControllerControl("Solo 1", 2, 0, "button"),
ControllerControl("Solo 2", 3, 0, "button"),
ControllerControl("Solo 3", 8, 0, "button"),
ControllerControl("Solo 4", 9, 0, "button"),
ControllerControl("Solo 5", 14, 0, "button"),
ControllerControl("Solo 6", 15, 0, "button"),
ControllerControl("Solo 7", 20, 0, "button"),
ControllerControl("Solo 8", 21, 0, "button"),
],
),
ControllerProfile(
manufacturer="Arturia",
model="BeatStep",
usb_vendor_id=0x1C75,
usb_product_id=0x0208,
alsa_client_pattern="Arturia BeatStep",
num_faders=0,
num_knobs=17,
num_buttons=16,
has_transport=True,
midi_channel_count=1,
notes="17 endless encoders + 16 pads. Good for transport and parameter tweaking.",
controls=[
ControllerControl("Knob 1", 74, 0, "encoder"),
ControllerControl("Knob 2", 71, 0, "encoder"),
ControllerControl("Knob 3", 76, 0, "encoder"),
ControllerControl("Knob 4", 77, 0, "encoder"),
ControllerControl("Knob 5", 93, 0, "encoder"),
ControllerControl("Knob 6", 73, 0, "encoder"),
ControllerControl("Knob 7", 75, 0, "encoder"),
ControllerControl("Knob 8", 79, 0, "encoder"),
ControllerControl("Knob 9", 72, 0, "encoder"),
ControllerControl("Knob 10", 80, 0, "encoder"),
ControllerControl("Knob 11", 81, 0, "encoder"),
ControllerControl("Knob 12", 82, 0, "encoder"),
ControllerControl("Knob 13", 83, 0, "encoder"),
ControllerControl("Knob 14", 84, 0, "encoder"),
ControllerControl("Knob 15", 85, 0, "encoder"),
ControllerControl("Knob 16", 86, 0, "encoder"),
ControllerControl("Big Knob", 112, 0, "encoder"),
# Transport
ControllerControl("Play", 116, 0, "transport"),
ControllerControl("Stop", 117, 0, "transport"),
ControllerControl("Rec", 118, 0, "transport"),
],
),
ControllerProfile(
manufacturer="Novation",
model="Launch Control XL",
usb_vendor_id=0x1235,
usb_product_id=0x0061,
alsa_client_pattern="Launch Control XL",
num_faders=8,
num_knobs=24,
num_buttons=16,
has_transport=False,
midi_channel_count=1,
notes="8 faders, 24 knobs (3 rows x 8), 16 buttons. Highly customisable.",
controls=[
ControllerControl("Fader 1", 77, 0, "fader"),
ControllerControl("Fader 2", 78, 0, "fader"),
ControllerControl("Fader 3", 79, 0, "fader"),
ControllerControl("Fader 4", 80, 0, "fader"),
ControllerControl("Fader 5", 81, 0, "fader"),
ControllerControl("Fader 6", 82, 0, "fader"),
ControllerControl("Fader 7", 83, 0, "fader"),
ControllerControl("Fader 8", 84, 0, "fader"),
# Knobs row 1
ControllerControl("Knob 1A", 13, 0, "knob"),
ControllerControl("Knob 2A", 14, 0, "knob"),
ControllerControl("Knob 3A", 15, 0, "knob"),
ControllerControl("Knob 4A", 16, 0, "knob"),
ControllerControl("Knob 5A", 17, 0, "knob"),
ControllerControl("Knob 6A", 18, 0, "knob"),
ControllerControl("Knob 7A", 19, 0, "knob"),
ControllerControl("Knob 8A", 20, 0, "knob"),
# Knobs row 2
ControllerControl("Knob 1B", 29, 0, "knob"),
ControllerControl("Knob 2B", 30, 0, "knob"),
ControllerControl("Knob 3B", 31, 0, "knob"),
ControllerControl("Knob 4B", 32, 0, "knob"),
ControllerControl("Knob 5B", 33, 0, "knob"),
ControllerControl("Knob 6B", 34, 0, "knob"),
ControllerControl("Knob 7B", 35, 0, "knob"),
ControllerControl("Knob 8B", 36, 0, "knob"),
],
),
ControllerProfile(
manufacturer="Korg",
model="nanoKONTROL2",
usb_vendor_id=0x0944,
usb_product_id=0x0117,
alsa_client_pattern="nanoKONTROL2",
num_faders=8,
num_knobs=8,
num_buttons=32,
has_transport=True,
midi_channel_count=1,
notes="Compact USB controller. 8 faders, 8 knobs, 24 buttons, transport. Class-compliant.",
controls=[
ControllerControl("Fader 1", 0, 0, "fader"),
ControllerControl("Fader 2", 1, 0, "fader"),
ControllerControl("Fader 3", 2, 0, "fader"),
ControllerControl("Fader 4", 3, 0, "fader"),
ControllerControl("Fader 5", 4, 0, "fader"),
ControllerControl("Fader 6", 5, 0, "fader"),
ControllerControl("Fader 7", 6, 0, "fader"),
ControllerControl("Fader 8", 7, 0, "fader"),
ControllerControl("Knob 1", 16, 0, "knob"),
ControllerControl("Knob 2", 17, 0, "knob"),
ControllerControl("Knob 3", 18, 0, "knob"),
ControllerControl("Knob 4", 19, 0, "knob"),
ControllerControl("Knob 5", 20, 0, "knob"),
ControllerControl("Knob 6", 21, 0, "knob"),
ControllerControl("Knob 7", 22, 0, "knob"),
ControllerControl("Knob 8", 23, 0, "knob"),
ControllerControl("Solo 1", 32, 0, "button"),
ControllerControl("Solo 2", 33, 0, "button"),
ControllerControl("Solo 3", 34, 0, "button"),
ControllerControl("Solo 4", 35, 0, "button"),
ControllerControl("Solo 5", 36, 0, "button"),
ControllerControl("Solo 6", 37, 0, "button"),
ControllerControl("Solo 7", 38, 0, "button"),
ControllerControl("Solo 8", 39, 0, "button"),
ControllerControl("Mute 1", 48, 0, "button"),
ControllerControl("Mute 2", 49, 0, "button"),
ControllerControl("Mute 3", 50, 0, "button"),
ControllerControl("Mute 4", 51, 0, "button"),
ControllerControl("Mute 5", 52, 0, "button"),
ControllerControl("Mute 6", 53, 0, "button"),
ControllerControl("Mute 7", 54, 0, "button"),
ControllerControl("Mute 8", 55, 0, "button"),
ControllerControl("Rec 1", 64, 0, "button"),
ControllerControl("Rec 2", 65, 0, "button"),
ControllerControl("Rec 3", 66, 0, "button"),
ControllerControl("Rec 4", 67, 0, "button"),
ControllerControl("Rec 5", 68, 0, "button"),
ControllerControl("Rec 6", 69, 0, "button"),
ControllerControl("Rec 7", 70, 0, "button"),
ControllerControl("Rec 8", 71, 0, "button"),
# Transport
ControllerControl("Play", 41, 0, "transport"),
ControllerControl("Stop", 42, 0, "transport"),
ControllerControl("Rec", 43, 0, "transport"),
ControllerControl("Rewind", 44, 0, "transport"),
ControllerControl("FFwd", 45, 0, "transport"),
ControllerControl("Loop", 46, 0, "transport"),
],
),
]
def find_controller(usb_vid: int, usb_pid: int) -> ControllerProfile | None:
"""Look up a known controller by USB vendor/product IDs."""
for c in KNOWN_CONTROLLERS:
if c.usb_vendor_id == usb_vid and c.usb_product_id == usb_pid:
return c
return None
def find_controller_by_alsa(client_name: str) -> ControllerProfile | None:
"""Look up a known controller by ALSA client name substring match."""
name_upper = client_name.upper()
for c in KNOWN_CONTROLLERS:
if c.alsa_client_pattern and c.alsa_client_pattern.upper() in name_upper:
return c
return None
src/midi/device_discovery.py
+313
View File
@@ -0,0 +1,313 @@
"""USB MIDI device discovery, enumeration, and hotplug notification.
Uses udev for hotplug detection and pyudev for programmatic device
enumeration. Falls back to ALSA sequencer client listing if pyudev
is unavailable.
"""
from __future__ import annotations
import logging
import os
import time
from dataclasses import dataclass, field
from typing import Callable
logger = logging.getLogger(__name__)
@dataclass
class MIDIDevice:
"""Represents a discovered USB MIDI device.
Populated from udev or ALSA sequencer enumeration.
"""
device_node: str # e.g. /dev/snd/midiC1D0 or /dev/midi1
manufacturer: str = ""
product: str = ""
serial: str = ""
usb_vendor_id: int = 0
usb_product_id: int = 0
alsa_client_id: int = -1 # ALSA sequencer client ID
alsa_client_name: str = ""
num_ports: int = 0
is_online: bool = True
@property
def display_name(self) -> str:
parts = []
if self.manufacturer:
parts.append(self.manufacturer)
if self.product:
parts.append(self.product)
if parts:
return " ".join(parts)
if self.alsa_client_name:
return self.alsa_client_name
return os.path.basename(self.device_node)
# ── udev-based discovery ────────────────────────────────────────────────────
def _parse_udev_device(dev) -> MIDIDevice | None:
"""Extract MIDIDevice from a pyudev Device object."""
try:
node = dev.device_node
except Exception:
return None
# USB parent info
vid, pid = 0, 0
manufacturer, product, serial = "", "", ""
try:
usb_dev = dev.find_parent("usb", "usb_device")
if usb_dev:
vid = int(usb_dev.get("ID_VENDOR_ID", "0"), 16)
pid = int(usb_dev.get("ID_MODEL_ID", "0"), 16)
manufacturer = usb_dev.get("ID_VENDOR_FROM_DATABASE", "") or usb_dev.get("ID_VENDOR", "")
product = usb_dev.get("ID_MODEL_FROM_DATABASE", "") or usb_dev.get("ID_MODEL", "")
serial = usb_dev.get("ID_SERIAL_SHORT", "")
except Exception:
pass
return MIDIDevice(
device_node=node,
manufacturer=manufacturer,
product=product,
serial=serial,
usb_vendor_id=vid,
usb_product_id=pid,
)
def discover_udev() -> list[MIDIDevice]:
"""Enumerate USB MIDI devices using pyudev.
Returns all raw MIDI device nodes (subsystem=sound with midi capability).
"""
try:
import pyudev
except ImportError:
logger.debug("pyudev not available, falling back to ALSA enumeration")
return []
devices: list[MIDIDevice] = []
context = pyudev.Context()
for dev in context.list_devices(subsystem="sound"):
devtype = dev.get("DEVTYPE", "")
# ALSA rawmidi devices
if devtype == "rawmidi" or "midi" in dev.sys_name.lower():
d = _parse_udev_device(dev)
if d:
devices.append(d)
# Also check /dev/midi* devices
for dev in context.list_devices(subsystem="sound"):
if "midi" in dev.get("DEVNAME", "").lower():
d = _parse_udev_device(dev)
if d and d not in devices:
devices.append(d)
return devices
# ── ALSA sequencer-based discovery ──────────────────────────────────────────
def discover_alsa_seq() -> list[MIDIDevice]:
"""Enumerate MIDI devices via ALSA sequencer clients.
Parses /proc/asound/seq/clients (no external deps).
"""
devices: list[MIDIDevice] = []
clients_path = "/proc/asound/seq/clients"
if not os.path.exists(clients_path):
return devices
try:
with open(clients_path) as fh:
for line in fh:
line = line.strip()
if not line or line.startswith("Client"):
continue
# Format: "Client 128 : "USB MIDI Device" [type=kernel]"
parts = line.split(":", 1)
if len(parts) < 2:
continue
client_part = parts[0].strip()
if not client_part.startswith("Client "):
continue
try:
client_id = int(client_part.split()[1])
except (IndexError, ValueError):
continue
name_part = parts[1].strip().strip('"')
# Extract port count from the bracket part
num_ports = 1
if "[" in name_part:
name_part, _ = name_part.rsplit("[", 1)
name_part = name_part.strip()
devices.append(MIDIDevice(
device_node=f"alsa:client:{client_id}",
alsa_client_id=client_id,
alsa_client_name=name_part,
num_ports=num_ports,
))
except Exception as exc:
logger.warning("Failed to read ALSA seq clients: %s", exc)
return devices
def discover_all() -> list[MIDIDevice]:
"""Enumerate all available MIDI input devices.
Tries udev first (richer metadata), falls back to ALSA sequencer.
"""
devices = discover_udev()
if not devices:
devices = discover_alsa_seq()
# Enrich ALSA-only devices with seq client IDs
for dev in devices:
if dev.alsa_client_id < 0:
_enrich_alsa_id(dev)
return devices
def _enrich_alsa_id(dev: MIDIDevice) -> None:
"""Try to find ALSA sequencer client ID for a udev-discovered device."""
seq_devs = discover_alsa_seq()
for sd in seq_devs:
if sd.alsa_client_name and (
dev.product in sd.alsa_client_name
or dev.manufacturer in sd.alsa_client_name
):
dev.alsa_client_id = sd.alsa_client_id
dev.alsa_client_name = sd.alsa_client_name
break
# ── Hotplug monitoring ──────────────────────────────────────────────────────
DeviceCallback = Callable[[MIDIDevice], None] # called on add
DeviceRemoveCallback = Callable[[str], None] # called with device_node on remove
class MIDIHotplugMonitor:
"""Monitors udev for USB MIDI device add/remove events.
Uses pyudev.Monitor when available; polls /proc/asound/seq/clients
as a fallback.
"""
def __init__(self):
self._monitor = None
self._observer = None
self._on_add: list[DeviceCallback] = []
self._on_remove: list[DeviceRemoveCallback] = []
self._running = False
self._known_devices: dict[str, MIDIDevice] = {}
def on_add(self, callback: DeviceCallback) -> None:
self._on_add.append(callback)
def on_remove(self, callback: DeviceRemoveCallback) -> None:
self._on_remove.append(callback)
def start(self) -> None:
"""Start monitoring for hotplug events (non-blocking)."""
self._running = True
try:
import pyudev
self._monitor = pyudev.Monitor.from_netlink(pyudev.Context())
self._monitor.filter_by(subsystem="sound")
import pyudev_monitor # noqa: F811
from pyudev import MonitorObserver
self._observer = MonitorObserver(self._monitor, self._handle_udev_event)
self._observer.start()
logger.info("MIDI hotplug monitor started (pyudev)")
except ImportError:
logger.info("pyudev not available; hotplug requires polling")
self._monitor = None
def stop(self) -> None:
self._running = False
if self._observer:
self._observer.stop()
self._observer = None
logger.info("MIDI hotplug monitor stopped")
def _handle_udev_event(self, device) -> None:
action = device.get("ACTION", "")
if action == "add":
d = _parse_udev_device(device)
if d and d.device_node not in self._known_devices:
self._known_devices[d.device_node] = d
for cb in self._on_add:
try:
cb(d)
except Exception as exc:
logger.error("Hotplug add callback failed: %s", exc)
elif action == "remove":
node = device.get("DEVNAME", "")
if node:
self._known_devices.pop(node, None)
for cb in self._on_remove:
try:
cb(node)
except Exception as exc:
logger.error("Hotplug remove callback failed: %s", exc)
def poll_once(self) -> list[MIDIDevice]:
"""Poll-based check for new/removed devices. Returns newly added devices.
Call periodically (~1-2 Hz) as fallback when udev monitoring isn't available.
"""
current = discover_all()
current_nodes = {d.device_node for d in current}
known_nodes = set(self._known_devices.keys())
added = [d for d in current if d.device_node not in known_nodes]
removed = known_nodes - current_nodes
self._known_devices = {d.device_node: d for d in current}
for d in added:
for cb in self._on_add:
try:
cb(d)
except Exception as exc:
logger.error("Poll add callback failed: %s", exc)
for node in removed:
for cb in self._on_remove:
try:
cb(node)
except Exception as exc:
logger.error("Poll remove callback failed: %s", exc)
return added
# ── Convenience ─────────────────────────────────────────────────────────────
def print_devices() -> None:
"""Print a human-readable device list to stdout."""
devices = discover_all()
if not devices:
print("No MIDI devices found.")
return
print(f"{'Node':<30} {'Name':<40} {'VID:PID':<12} {'ALSA':<8}")
print("-" * 90)
for d in devices:
vidpid = f"{d.usb_vendor_id:04x}:{d.usb_product_id:04x}" if d.usb_vendor_id else "-"
alsa = str(d.alsa_client_id) if d.alsa_client_id > 0 else "-"
name = d.display_name[:38]
print(f"{d.device_node:<30} {name:<40} {vidpid:<12} {alsa:<8}")
src/midi/jack_midi_bridge.py
+235
View File
@@ -0,0 +1,235 @@
"""JACK MIDI bridge — create JACK MIDI ports for Carla consumption.
Bridges mapped MIDI events from the engine into JACK MIDI output ports.
Carla (or any JACK MIDI-aware application) can connect to these ports
and route MIDI to soft-synths, samplers, and effects.
Two modes:
1. Direct JACK client (requires jack-client Python module)
2. ALSA sequencer bridge (uses a2jmidid for ALSA → JACK bridging)
The direct JACK client mode is preferred as it avoids the extra a2jmidid
daemon, but requires the `jack` Python package (pip install JACK-Client).
"""
from __future__ import annotations
import logging
import threading
import time
from typing import Any
logger = logging.getLogger(__name__)
# Maximum number of JACK MIDI ports to create
DEFAULT_NUM_PORTS = 4
class JACKMIDIBridge:
"""Creates JACK MIDI output ports and writes MIDI events to them.
Usage:
bridge = JACKMIDIBridge(client_name="rpi-mixer")
bridge.start()
# ... after engine processes a mapping ...
bridge.send_midi(port_index=0, event=[0x90, 60, 100])
"""
def __init__(
self,
client_name: str = "rpi-mixer",
num_ports: int = DEFAULT_NUM_PORTS,
port_name_prefix: str = "midi_out",
):
self.client_name = client_name
self.num_ports = num_ports
self.port_name_prefix = port_name_prefix
self._client = None
self._ports: list[Any] = []
self._alsa_client = None
self._alsa_ports: list[int] = []
self._running = False
self._activated = False
self._lock = threading.Lock()
# ── Lifecycle ────────────────────────────────────────────────────────
def start(self) -> bool:
"""Activate JACK client and register MIDI output ports.
Returns True on success, False if JACK is unavailable.
"""
try:
import jack
self._client = jack.Client(self.client_name)
except ImportError:
logger.warning(
"jack-client Python module not available. "
"Install with: pip install JACK-Client. "
"Falling back to ALSA sequencer output."
)
return self._start_alsa_fallback()
except Exception as exc:
logger.error("Failed to create JACK client: %s", exc)
return self._start_alsa_fallback()
# Register MIDI output ports
with self._lock:
for i in range(self.num_ports):
port_name = f"{self.port_name_prefix}_{i}"
try:
port = self._client.midi_outports.register(port_name)
self._ports.append(port)
logger.info("Registered JACK MIDI port: %s:%s", self.client_name, port_name)
except Exception as exc:
logger.error("Failed to register port %s: %s", port_name, exc)
if not self._ports:
logger.error("No JACK MIDI ports registered")
return self._start_alsa_fallback()
# Activate the client
try:
self._client.activate()
self._activated = True
except Exception as exc:
logger.error("Failed to activate JACK client: %s", exc)
return self._start_alsa_fallback()
self._running = True
logger.info("JACK MIDI bridge active: %d ports on '%s'", len(self._ports), self.client_name)
return True
def _start_alsa_fallback(self) -> bool:
"""Fallback: use ALSA sequencer output (bridged via a2jmidid).
This creates ALSA sequencer ports that a2jmidid automatically
bridges to JACK MIDI. No direct JACK dependency needed.
"""
try:
import alsaseq
except ImportError:
logger.warning("alsaseq not available either — MIDI output disabled")
return False
try:
self._alsa_client = alsaseq.SequencerClient(
self.client_name,
client_type=alsaseq.SEQ_CLIENT_DUPLEX
)
with self._lock:
for i in range(self.num_ports):
port_name = f"{self.port_name_prefix}_{i}"
port_id = self._alsa_client.create_port(
port_name,
caps=alsaseq.SEQ_PORT_CAP_WRITE | alsaseq.SEQ_PORT_CAP_SUBS_WRITE,
type=alsaseq.SEQ_PORT_TYPE_MIDI_GENERIC,
)
self._alsa_ports.append(port_id)
logger.info("Created ALSA seq port: %s:%s (id=%d)", self.client_name, port_name, port_id)
except Exception as exc:
logger.error("Failed to create ALSA sequencer client: %s", exc)
return False
self._running = True
logger.info("ALSA sequencer MIDI bridge active (connect via a2jmidid)")
return True
def stop(self) -> None:
"""Deactivate the JACK client and clean up ports."""
self._running = False
if self._client and self._activated:
try:
self._client.deactivate()
except Exception as exc:
logger.warning("Error deactivating JACK client: %s", exc)
self._client = None
self._ports.clear()
self._activated = False
logger.info("JACK MIDI bridge stopped")
# ── MIDI output ──────────────────────────────────────────────────────
def send_midi(self, port_index: int, event: bytes | list[int], timestamp: float = 0.0) -> bool:
"""Send a raw MIDI event to a JACK MIDI port.
Args:
port_index: Which output port (0..num_ports-1).
event: Raw MIDI bytes (e.g., [0x90, 0x3C, 0x64] for note on).
timestamp: JACK frame time offset (0.0 = as soon as possible).
Returns True if the event was queued successfully.
"""
if not self._running:
return False
if isinstance(event, list):
event = bytes(event)
with self._lock:
if port_index < 0 or port_index >= len(self._ports):
logger.debug("Invalid port index %d (have %d ports)", port_index, len(self._ports))
return False
try:
port = self._ports[port_index]
port.write_midi_event(timestamp, event)
return True
except Exception as exc:
logger.error("MIDI write to port %d failed: %s", port_index, exc)
return False
def send_midi_burst(self, port_index: int, events: list[bytes | list[int]]) -> int:
"""Send multiple MIDI events atomically (same JACK cycle).
Returns number of events successfully queued.
"""
sent = 0
for event in events:
if self.send_midi(port_index, event):
sent += 1
return sent
# ── Port info ────────────────────────────────────────────────────────
@property
def port_count(self) -> int:
return len(self._ports)
def port_name(self, index: int) -> str:
if 0 <= index < len(self._ports):
return f"{self.client_name}:{self.port_name_prefix}_{index}"
return ""
@property
def is_active(self) -> bool:
return self._running and self._activated
# ── Utility: generate common MIDI messages ──────────────────────────────────
def midi_note_on(channel: int, note: int, velocity: int = 100) -> bytes:
"""Build a Note On message (3 bytes)."""
return bytes([0x90 | (channel & 0x0F), note & 0x7F, velocity & 0x7F])
def midi_note_off(channel: int, note: int) -> bytes:
"""Build a Note Off message (3 bytes)."""
return bytes([0x80 | (channel & 0x0F), note & 0x7F, 0])
def midi_cc(channel: int, controller: int, value: int) -> bytes:
"""Build a Control Change message (3 bytes)."""
return bytes([0xB0 | (channel & 0x0F), controller & 0x7F, value & 0x7F])
def midi_pitch_bend(channel: int, value: int) -> bytes:
"""Build a Pitch Bend message (3 bytes, 14-bit value)."""
value = max(0, min(16383, value))
lsb = value & 0x7F
msb = (value >> 7) & 0x7F
return bytes([0xE0 | (channel & 0x0F), lsb, msb])
def midi_program_change(channel: int, program: int) -> bytes:
"""Build a Program Change message (2 bytes)."""
return bytes([0xC0 | (channel & 0x0F), program & 0x7F])
src/midi/mapping_store.py
+163
View File
@@ -0,0 +1,163 @@
"""MIDI mapping persistence — save/load mappings as JSON.
Mappings are stored per-session (a "session" is one mapping configuration
file). The file format is designed for human readability and version control
friendliness.
"""
from __future__ import annotations
import json
import logging
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from .types import MIDIMapping, MIDIMessageType, ParameterType
logger = logging.getLogger(__name__)
DEFAULT_MAPPINGS_DIR = Path.home() / ".config" / "rpi-mixer" / "mappings"
DEFAULT_SESSION_NAME = "default"
# ── Serialisation ───────────────────────────────────────────────────────────
def mapping_to_dict(m: MIDIMapping) -> dict[str, Any]:
"""Serialise one mapping to a plain dict."""
return {
"midi": {
"type": m.msg_type.name,
"channel": m.channel,
"controller": m.controller,
"is_nrpn": m.is_nrpn,
"nrpn_number": m.nrpn_number,
"source_device": m.source_device or None,
},
"target": {
"parameter": m.param_type.value,
"channel": m.param_channel,
},
"value": {
"midi_min": m.midi_min,
"midi_max": m.midi_max,
"param_min": m.param_min,
"param_max": m.param_max,
"invert": m.invert,
"curve": m.curve,
},
"meta": {
"label": m.label or None,
"enabled": m.enabled,
},
}
def mapping_from_dict(d: dict[str, Any]) -> MIDIMapping:
"""Deserialise one mapping from a dict."""
midi = d.get("midi", {})
target = d.get("target", {})
value = d.get("value", {})
meta = d.get("meta", {})
msg_type = MIDIMessageType[midi.get("type", "CONTROL_CHANGE")]
return MIDIMapping(
msg_type=msg_type,
channel=midi.get("channel", 0),
controller=midi.get("controller", 0),
is_nrpn=midi.get("is_nrpn", False),
nrpn_number=midi.get("nrpn_number", 0),
source_device=midi.get("source_device") or "",
param_type=ParameterType(target.get("parameter", "volume")),
param_channel=target.get("channel", -1),
midi_min=value.get("midi_min", 0),
midi_max=value.get("midi_max", 127),
param_min=value.get("param_min", 0.0),
param_max=value.get("param_max", 1.0),
invert=value.get("invert", False),
curve=value.get("curve", "linear"),
label=meta.get("label") or "",
enabled=meta.get("enabled", True),
)
# ── Session file I/O ────────────────────────────────────────────────────────
def _session_path(session_name: str = DEFAULT_SESSION_NAME) -> Path:
"""Resolve the JSON file path for a session."""
return DEFAULT_MAPPINGS_DIR / f"{session_name}.json"
def save_mappings(
mappings: list[MIDIMapping],
session_name: str = DEFAULT_SESSION_NAME,
metadata: dict[str, Any] | None = None,
) -> Path:
"""Persist mapping list to JSON.
Returns the path written.
"""
path = _session_path(session_name)
path.parent.mkdir(parents=True, exist_ok=True)
doc: dict[str, Any] = {
"version": 1,
"session": session_name,
"updated": datetime.now(timezone.utc).isoformat(),
"metadata": metadata or {},
"mappings": [mapping_to_dict(m) for m in mappings],
}
# Atomic write: write to temp then rename
tmp = path.with_suffix(".tmp")
with open(tmp, "w") as fh:
json.dump(doc, fh, indent=2, sort_keys=True)
os.replace(tmp, path)
logger.info("Saved %d mappings to %s", len(mappings), path)
return path
def load_mappings(session_name: str = DEFAULT_SESSION_NAME) -> list[MIDIMapping]:
"""Load mapping list from JSON.
Returns empty list if the file doesn't exist.
"""
path = _session_path(session_name)
if not path.exists():
logger.info("No mapping file at %s", path)
return []
with open(path) as fh:
doc = json.load(fh)
version = doc.get("version", 0)
if version != 1:
logger.warning("Unknown mapping format version %d in %s", version, path)
raw = doc.get("mappings", [])
mappings = [mapping_from_dict(m) for m in raw]
logger.info("Loaded %d mappings from %s", len(mappings), path)
return mappings
def list_sessions() -> list[str]:
"""Return names of all saved mapping sessions."""
if not DEFAULT_MAPPINGS_DIR.exists():
return []
return sorted(
p.stem for p in DEFAULT_MAPPINGS_DIR.glob("*.json")
if p.stem != "default" or True # Include default
)
def delete_session(session_name: str) -> bool:
"""Delete a mapping session file. Returns True if successful."""
path = _session_path(session_name)
if path.exists():
path.unlink()
logger.info("Deleted mapping session: %s", session_name)
return True
return False
src/midi/midi_clock.py
+217
View File
@@ -0,0 +1,217 @@
"""MIDI Clock synchronisation.
Receives MIDI clock (timing clock) messages and derives tempo (BPM)
from the timing pulse stream. MIDI clock sends 24 pulses per quarter
note (PPQN). Also handles START, STOP, CONTINUE for transport sync.
Supports:
- BPM detection from clock pulse timing
- Tempo averaging (moving window) to smooth jitter
- Song Position Pointer (SPP) for locate
- Transport state tracking (stopped, playing)
- MIDI clock output for master clock generation
"""
from __future__ import annotations
import collections
import logging
import time
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
PPQN = 24 # MIDI clock pulses per quarter note
DEFAULT_WINDOW = 96 # 4 quarter notes at 24 PPQN
@dataclass
class MIDIClockState:
"""Current state of MIDI clock reception."""
running: bool = False # Transport is playing
bpm: float = 120.0 # Current estimated tempo
bpm_raw: float = 120.0 # Raw instantaneous BPM (no smoothing)
bpm_stable: bool = False # Tempo estimate has converged
song_position: float = 0.0 # Current song position in beats
last_pulse_time: float = 0.0 # monotonic timestamp of last clock pulse
pulse_count: int = 0 # Total clock pulses received
ppqn: int = PPQN
# Moving window for tempo averaging
_pulse_intervals: collections.deque = field(default_factory=lambda: collections.deque(maxlen=DEFAULT_WINDOW))
def reset(self) -> None:
self.running = False
self.bpm = 120.0
self.bpm_raw = 120.0
self.bpm_stable = False
self.song_position = 0.0
self.last_pulse_time = 0.0
self.pulse_count = 0
self._pulse_intervals.clear()
class MIDIClock:
"""MIDI clock receiver and tempo estimator.
Feed clock messages via process_message(). BPM is derived from
the inter-pulse timing using a moving average window.
Usage:
clock = MIDIClock()
clock.on_tempo_change = lambda bpm: print(f"Tempo: {bpm:.1f}")
clock.on_transport = lambda event: print(f"Transport: {event}")
# Feed messages:
clock.process_message(msg)
"""
def __init__(self, window_size: int = DEFAULT_WINDOW):
self.state = MIDIClockState()
self.state._pulse_intervals = collections.deque(maxlen=window_size)
self._on_tempo_change: list = []
self._on_transport: list = []
self._on_beat: list = [] # Fires every quarter note (every 24 pulses)
# ── Callbacks ────────────────────────────────────────────────────────
def on_tempo_change(self, callback):
"""callback(bpm: float, raw_bpm: float, stable: bool)."""
self._on_tempo_change.append(callback)
def on_transport(self, callback):
"""callback(event: str) where event is 'start', 'stop', 'continue'."""
self._on_transport.append(callback)
def on_beat(self, callback):
"""callback(beat: int) — fires every quarter note (24 pulses)."""
self._on_beat.append(callback)
# ── Message processing ───────────────────────────────────────────────
def process_message(self, status_byte: int) -> None:
"""Process a MIDI system realtime message.
Args:
status_byte: The raw MIDI status byte (0xF8–0xFC).
"""
now = time.monotonic()
if status_byte == 0xF8: # Timing Clock
self._handle_clock(now)
elif status_byte == 0xFA: # Start
self._handle_start(now)
elif status_byte == 0xFB: # Continue
self._handle_continue(now)
elif status_byte == 0xFC: # Stop
self._handle_stop()
elif status_byte == 0xF2: # Song Position Pointer (handled separately)
pass # SPP requires two data bytes, handled upstream
def process_song_position(self, beats: int) -> None:
"""Handle Song Position Pointer (SPP).
Args:
beats: Song position in MIDI beats (0–16383, where 1 beat = 6 clock pulses).
"""
self.state.song_position = beats
logger.debug("Song position: %.1f beats", beats)
# ── Internal handlers ────────────────────────────────────────────────
def _handle_clock(self, now: float) -> None:
s = self.state
s.pulse_count += 1
if s.last_pulse_time > 0:
interval = now - s.last_pulse_time
if interval > 0 and interval < 2.0: # Sanity: ignore >2s gaps
s._pulse_intervals.append(interval)
if len(s._pulse_intervals) >= 4: # Need at least a few pulses
avg_interval = sum(s._pulse_intervals) / len(s._pulse_intervals)
if avg_interval > 0:
# 24 pulses per quarter note, BPM = 60 / (seconds per quarter note)
# seconds per quarter note = avg_interval * 24
s.bpm_raw = 60.0 / (avg_interval * s.ppqn)
s.bpm = s.bpm_raw
s.bpm_stable = len(s._pulse_intervals) >= ((s._pulse_intervals.maxlen or DEFAULT_WINDOW) // 2)
s.last_pulse_time = now
# Beat callback every 24 pulses
if s.pulse_count % s.ppqn == 0:
beat = s.pulse_count // s.ppqn
for cb in self._on_beat:
try:
cb(beat)
except Exception as exc:
logger.error("Beat callback failed: %s", exc)
# Fire tempo change on beat boundaries (not every pulse)
if s.bpm_stable:
for cb in self._on_tempo_change:
try:
cb(s.bpm, s.bpm_raw, s.bpm_stable)
except Exception as exc:
logger.error("Tempo callback failed: %s", exc)
def _handle_start(self, now: float) -> None:
self.state.running = True
self.state.song_position = 0.0
self.state.pulse_count = 0
self.state.last_pulse_time = now
self.state._pulse_intervals.clear()
logger.info("MIDI START — transport running")
for cb in self._on_transport:
try:
cb("start")
except Exception as exc:
logger.error("Transport callback failed: %s", exc)
def _handle_continue(self, now: float) -> None:
self.state.running = True
self.state.last_pulse_time = now
logger.info("MIDI CONTINUE — transport running")
for cb in self._on_transport:
try:
cb("continue")
except Exception as exc:
logger.error("Transport callback failed: %s", exc)
def _handle_stop(self) -> None:
self.state.running = False
logger.info("MIDI STOP — transport stopped")
for cb in self._on_transport:
try:
cb("stop")
except Exception as exc:
logger.error("Transport callback failed: %s", exc)
# ── Master clock output (for generating MIDI clock) ──────────────────
def generate_clock_pulse(self) -> float | None:
"""Generate next clock pulse interval for master output.
Returns:
Seconds until next pulse, or None if transport not running.
"""
if not self.state.running:
return None
if self.state.bpm <= 0:
return None
# Seconds per quarter note = 60 / BPM
# Interval per pulse = (60 / BPM) / PPQN
return (60.0 / self.state.bpm) / self.state.ppqn
# ── Query ────────────────────────────────────────────────────────────
@property
def tempo(self) -> float:
return self.state.bpm
@property
def is_running(self) -> bool:
return self.state.running
def reset(self) -> None:
self.state.reset()
src/midi/midi_engine.py
+272
View File
@@ -0,0 +1,272 @@
"""Core MIDI processing engine.
Reads MIDI events from input ports (ALSA sequencer or rtmidi),
applies mapping rules, dispatches parameter changes downstream,
and optionally forwards mapped MIDI to output ports.
Architecture:
USB MIDI device → ALSA seq / rtmidi input → MappingEngine
→ ParameterRegistry callbacks (DSP / UI)
→ [optional] JACK MIDI / ALSA seq output ports
"""
from __future__ import annotations
import logging
import threading
import time
from collections import defaultdict
from typing import Callable
from .types import (
MIDIMessage,
MIDIMessageType,
MIDIMapping,
ParameterType,
)
logger = logging.getLogger(__name__)
# Callback shape: (mapping: MIDIMapping, scaled_value: float, raw_msg: MIDIMessage) -> None
MappingCallback = Callable[[MIDIMapping, float, MIDIMessage], None]
class MIDIEngine:
"""Core MIDI routing and mapping engine.
Maintains a list of active mappings, reads MIDI events from one or
more backends, matches events against mappings, and dispatches
scaled parameter values to registered callbacks.
Thread-safe for concurrent event input and mapping updates.
"""
def __init__(self, name: str = "rpi-mixer-midi"):
self.name = name
self._mappings: list[MIDIMapping] = []
self._mappings_lock = threading.Lock()
self._callbacks: list[MappingCallback] = []
self._running = False
# NRPN state machine per (source_device, channel)
# NRPN uses CC 99 (MSB), CC 98 (LSB), CC 6 (data MSB), CC 38 (data LSB), CC 96 (increment), CC 97 (decrement)
self._nrpn_state: dict[tuple[str, int], dict[str, int | None]] = defaultdict(
lambda: {"msb": None, "lsb": None, "data_msb": None}
)
# Stats
self.event_count: int = 0
self.mapped_count: int = 0
self.uptime_start: float = 0.0
# ── Mapping CRUD ────────────────────────────────────────────────────
def set_mappings(self, mappings: list[MIDIMapping]) -> None:
"""Replace the entire mapping table atomically."""
with self._mappings_lock:
self._mappings = list(mappings)
logger.info("Loaded %d mappings", len(mappings))
def add_mapping(self, mapping: MIDIMapping) -> None:
with self._mappings_lock:
self._mappings.append(mapping)
logger.info("Added mapping: %s", mapping.label or f"CC{mapping.controller}→{mapping.param_type.value}")
def remove_mapping(self, mapping: MIDIMapping) -> bool:
with self._mappings_lock:
for i, m in enumerate(self._mappings):
if (
m.msg_type == mapping.msg_type
and m.channel == mapping.channel
and m.controller == mapping.controller
and m.param_type == mapping.param_type
and m.param_channel == mapping.param_channel
):
self._mappings.pop(i)
logger.info("Removed mapping: %s", mapping.label or f"CC{mapping.controller}")
return True
return False
def get_mappings(self) -> list[MIDIMapping]:
with self._mappings_lock:
return list(self._mappings)
def find_mappings_for(self, param_type: ParameterType, channel: int = -1) -> list[MIDIMapping]:
"""Return all mappings targeting a specific parameter."""
with self._mappings_lock:
return [m for m in self._mappings if m.param_type == param_type and m.param_channel == channel]
# ── Callback management ──────────────────────────────────────────────
def on_mapped(self, callback: MappingCallback) -> None:
"""Register a callback for mapped parameter changes."""
self._callbacks.append(callback)
def remove_callback(self, callback: MappingCallback) -> None:
if callback in self._callbacks:
self._callbacks.remove(callback)
# ── Event processing ─────────────────────────────────────────────────
def process_event(self, msg: MIDIMessage) -> list[tuple[MIDIMapping, float]]:
"""Process one MIDI event through the mapping table.
Handles NRPN state machine transparently: when a complete NRPN
value arrives (via CC 6/38/96/97 after parameter select via
CC 98/99), constructs a synthetic MIDIMessage with is_nrpn=True
and the full 14-bit value.
Returns list of (mapping, scaled_value) for all matched mappings.
"""
self.event_count += 1
results: list[tuple[MIDIMapping, float]] = []
# Handle NRPN state machine
resolved = self._handle_nrpn_state(msg)
if resolved is None:
return results # Intermediate NRPN message, not a value
msg = resolved
with self._mappings_lock:
mappings = list(self._mappings)
for mapping in mappings:
if not mapping.enabled:
continue
if mapping.matches(msg) and msg.value is not None:
scaled = mapping.scale_value(msg.value)
results.append((mapping, scaled))
self.mapped_count += 1
# Fire callbacks
for mapping, scaled in results:
for cb in self._callbacks:
try:
cb(mapping, scaled, msg)
except Exception as exc:
logger.error("Mapping callback failed: %s", exc)
return results
def _handle_nrpn_state(self, msg: MIDIMessage) -> MIDIMessage | None:
"""Track NRPN parameter selection and return synthetic value messages.
NRPN protocol:
1. CC 99 (NRPN MSB) — parameter number high 7 bits
2. CC 98 (NRPN LSB) — parameter number low 7 bits
3. CC 6 (Data Entry MSB) — value high 7 bits
4. CC 38 (Data Entry LSB) — value low 7 bits (optional)
Or CC 96 (Increment) / CC 97 (Decrement) for relative changes
Returns None for intermediate messages (99, 98, 38 without data MSB).
Returns a synthetic MIDIMessage with is_nrpn=True when a complete
value arrives.
"""
if msg.msg_type != MIDIMessageType.CONTROL_CHANGE or msg.controller is None:
return msg # Not a CC message
key = (msg.source_device, msg.channel)
state = self._nrpn_state[key]
if msg.controller == 99: # NRPN MSB
state["msb"] = msg.value
return None
elif msg.controller == 98: # NRPN LSB
state["lsb"] = msg.value
return None
elif msg.controller == 6: # Data Entry MSB
# A value just arrived — build synthetic NRPN message
nrpn_msb = state.get("msb")
nrpn_lsb = state.get("lsb")
if nrpn_msb is None or nrpn_lsb is None:
# NRPN parameter not fully selected yet
state["data_msb"] = msg.value
return None
# Combine MSB + LSB for 14-bit value (if CC 38 follows, it refines LSB)
value = (msg.value << 7) if msg.value is not None else 0
state["data_msb"] = msg.value
return None # Wait for CC 38 or treat as 7-bit
elif msg.controller == 38: # Data Entry LSB
# Completes the 14-bit value
nrpn_msb = state.get("msb")
nrpn_lsb = state.get("lsb")
data_msb = state.get("data_msb")
if nrpn_msb is None or nrpn_lsb is None:
return None
value = ((data_msb or 0) << 7) | (msg.value or 0)
return MIDIMessage(
msg_type=MIDIMessageType.CONTROL_CHANGE,
channel=msg.channel,
is_nrpn=True,
nrpn_msb=nrpn_msb,
nrpn_lsb=nrpn_lsb,
value=value,
timestamp=msg.timestamp,
source_device=msg.source_device,
)
elif msg.controller == 96: # Data Increment
nrpn_msb = state.get("msb")
nrpn_lsb = state.get("lsb")
if nrpn_msb is None or nrpn_lsb is None:
return None
return MIDIMessage(
msg_type=MIDIMessageType.CONTROL_CHANGE,
channel=msg.channel,
is_nrpn=True,
nrpn_msb=nrpn_msb,
nrpn_lsb=nrpn_lsb,
value=1, # Increment by 1
timestamp=msg.timestamp,
source_device=msg.source_device,
)
elif msg.controller == 97: # Data Decrement
nrpn_msb = state.get("msb")
nrpn_lsb = state.get("lsb")
if nrpn_msb is None or nrpn_lsb is None:
return None
return MIDIMessage(
msg_type=MIDIMessageType.CONTROL_CHANGE,
channel=msg.channel,
is_nrpn=True,
nrpn_msb=nrpn_msb,
nrpn_lsb=nrpn_lsb,
value=-1, # Decrement by 1
timestamp=msg.timestamp,
source_device=msg.source_device,
)
else:
# Regular CC — reset NRPN state for this channel
state["msb"] = None
state["lsb"] = None
state["data_msb"] = None
return msg # Regular CC, pass through
return msg # Shouldn't reach here, but just in case
# ── Lifecycle ────────────────────────────────────────────────────────
def start(self) -> None:
self._running = True
self.uptime_start = time.monotonic()
logger.info("MIDI engine started")
def stop(self) -> None:
self._running = False
logger.info("MIDI engine stopped (events: %d, mapped: %d)", self.event_count, self.mapped_count)
@property
def running(self) -> bool:
return self._running
@property
def stats(self) -> dict:
uptime = time.monotonic() - self.uptime_start if self.uptime_start else 0
return {
"events_total": self.event_count,
"events_mapped": self.mapped_count,
"mappings_active": len(self._mappings),
"uptime_seconds": round(uptime, 1),
"events_per_second": round(self.event_count / uptime, 1) if uptime > 0 else 0,
}
src/midi/midi_learn.py
+308
View File
@@ -0,0 +1,308 @@
"""MIDI Learn Mode — state machine for interactive mapping assignment.
Learn mode listens for incoming MIDI messages and associates the
last-received message pattern with a user-selected mixer parameter.
State machine:
IDLE → LISTENING (user selects parameter to learn) → LISTENING (waits for MIDI)
→ CAPTURED (message received, awaiting confirmation)
→ IDLE (confirmed/discarded)
Supports:
- Single learn: assign one CC to one parameter
- Batch learn: rapid-fire assignment (activate, wiggle knob, next param, repeat)
- NRPN learn: detects NRPN parameter numbers automatically
"""
from __future__ import annotations
import enum
import logging
from dataclasses import dataclass, field
from .types import (
MIDIMessage,
MIDIMessageType,
MIDIMapping,
ParameterType,
)
logger = logging.getLogger(__name__)
class LearnState(enum.Enum):
IDLE = "idle"
LISTENING = "listening" # Waiting for MIDI input
CAPTURED = "captured" # MIDI received, awaiting confirmation
BATCH = "batch" # Batch mode: auto-confirm and stay listening
@dataclass
class LearnSession:
"""Tracks the state of one learn operation."""
state: LearnState = LearnState.IDLE
# What parameter are we learning for?
param_type: ParameterType | None = None
param_channel: int = -1
param_label: str = ""
# Captured MIDI source
captured_msg: MIDIMessage | None = None
captured_cc: int = -1
captured_channel: int = 0
captured_is_nrpn: bool = False
captured_nrpn: int = 0
captured_device: str = ""
# Batch mode
batch_index: int = 0 # Current position in the batch parameter list
batch_params: list[tuple[ParameterType, int, str]] = field(default_factory=list)
batch_mappings: list[MIDIMapping] = field(default_factory=list)
def reset(self) -> None:
"""Reset to idle, discarding any captured state."""
self.state = LearnState.IDLE
self.captured_msg = None
self.captured_cc = -1
self.captured_channel = 0
self.captured_is_nrpn = False
self.captured_nrpn = 0
self.captured_device = ""
self.batch_index = 0
self.batch_params.clear()
self.batch_mappings.clear()
class MIDILearn:
"""Learn mode manager.
Usage:
learn = MIDILearn()
learn.on_mapped = engine.on_mapped # Forward learned mappings
# Single learn
learn.start_learn(ParameterType.VOLUME, channel=0)
# ... wait for MIDI event (call learn.feed_message(msg))
mapping = learn.confirm() # → MIDIMapping if captured
# Batch learn
learn.start_batch([
(ParameterType.VOLUME, 0, "CH1 Vol"),
(ParameterType.PAN, 0, "CH1 Pan"),
(ParameterType.MUTE, 0, "CH1 Mute"),
])
# feed each message... each capture auto-advances
"""
def __init__(self):
self.session = LearnSession()
self._on_learned: list = [] # callbacks receiving (mapping,)
def on_learned(self, callback):
"""Register callback: callback(MIDIMapping) on confirmation."""
self._on_learned.append(callback)
# ── Learn lifecycle ──────────────────────────────────────────────────
def start_learn(self, param_type: ParameterType, channel: int = -1, label: str = "") -> None:
"""Begin single learn mode for one parameter."""
self.session.reset()
self.session.state = LearnState.LISTENING
self.session.param_type = param_type
self.session.param_channel = channel
self.session.param_label = label
logger.info("Learn mode: listening for %s CH%d", param_type.value, channel)
def start_batch(self, params: list[tuple[ParameterType, int, str]]) -> None:
"""Begin batch learn mode for a sequence of parameters.
Each incoming MIDI message is auto-mapped to the current parameter
and the session advances to the next. Confirmation is automatic.
"""
self.session.reset()
self.session.state = LearnState.BATCH
self.session.batch_params = list(params)
self.session.batch_index = 0
self._set_batch_target()
logger.info("Learn mode: batch of %d params", len(params))
def _set_batch_target(self) -> None:
"""Update the current batch target."""
if self.session.batch_index < len(self.session.batch_params):
pt, ch, label = self.session.batch_params[self.session.batch_index]
self.session.param_type = pt
self.session.param_channel = ch
self.session.param_label = label
def feed_message(self, msg: MIDIMessage) -> MIDIMapping | None:
"""Accept an incoming MIDI message for learn capture.
Returns a MIDIMapping if the message triggered a mapping confirmation
(single confirm or batch auto-assign). Returns None if still listening
or if the message was ignored.
"""
if self.session.state not in (LearnState.LISTENING, LearnState.BATCH, LearnState.CAPTURED):
return None
# Only care about CC and NRPN messages (and notes for transport triggers)
if msg.msg_type not in (MIDIMessageType.CONTROL_CHANGE, MIDIMessageType.NOTE_ON, MIDIMessageType.NOTE_OFF):
return None
if self.session.state in (LearnState.LISTENING, LearnState.BATCH):
self._capture(msg)
if self.session.state == LearnState.CAPTURED:
# In single mode, we now have the capture; caller must confirm()
return None
if self.session.state == LearnState.BATCH:
# Auto-confirm and advance
return self._batch_confirm()
return None
def _capture(self, msg: MIDIMessage) -> None:
"""Store the incoming MIDI message as the captured source."""
# Determine CC or NRPN
if msg.is_nrpn:
cc = -1
is_nrpn = True
nrpn = msg.nrpn_number or 0
else:
cc = msg.controller if msg.controller is not None else -1
is_nrpn = False
nrpn = 0
self.session.captured_msg = msg
self.session.captured_cc = cc
self.session.captured_channel = msg.channel
self.session.captured_is_nrpn = is_nrpn
self.session.captured_nrpn = nrpn
self.session.captured_device = msg.source_device
if self.session.state == LearnState.LISTENING:
self.session.state = LearnState.CAPTURED
logger.info(
"Learn captured: ch=%d %s=%d device=%s",
msg.channel,
"NRPN" if is_nrpn else "CC",
nrpn if is_nrpn else cc,
msg.source_device,
)
def confirm(self) -> MIDIMapping | None:
"""Confirm the captured mapping in single-learn mode.
Returns the created MIDIMapping, or None if nothing captured.
"""
if self.session.state != LearnState.CAPTURED:
return None
mapping = self._build_mapping()
self.session.reset()
logger.info("Learn confirmed: %s", mapping.label or f"CC{mapping.controller}")
self._notify(mapping)
return mapping
def _batch_confirm(self) -> MIDIMapping | None:
"""Auto-confirm in batch mode and advance to next parameter."""
mapping = self._build_mapping()
self.session.batch_mappings.append(mapping)
logger.info("Batch learn [%d/%d]: %s",
self.session.batch_index + 1,
len(self.session.batch_params),
mapping.label or f"CC{mapping.controller}")
self._notify(mapping)
self.session.batch_index += 1
if self.session.batch_index >= len(self.session.batch_params):
# Batch complete
self.session.state = LearnState.IDLE
logger.info("Batch learn complete: %d mappings", len(self.session.batch_mappings))
return mapping # Return last one
self._set_batch_target()
return mapping
def discard(self) -> None:
"""Discard the current capture and return to idle."""
self.session.reset()
logger.info("Learn discarded")
def cancel(self) -> None:
"""Alias for discard."""
self.discard()
# ── Internal ─────────────────────────────────────────────────────────
def _build_mapping(self) -> MIDIMapping:
"""Construct a MIDIMapping from captured state."""
s = self.session
msg_type = s.captured_msg.msg_type if s.captured_msg else MIDIMessageType.CONTROL_CHANGE
return MIDIMapping(
msg_type=msg_type,
channel=s.captured_channel,
controller=s.captured_cc,
is_nrpn=s.captured_is_nrpn,
nrpn_number=s.captured_nrpn,
source_device=s.captured_device,
param_type=s.param_type or ParameterType.VOLUME,
param_channel=s.param_channel,
label=s.param_label or self._auto_label(),
enabled=True,
)
def _auto_label(self) -> str:
"""Generate a human-readable label for the mapping."""
s = self.session
pt = s.param_type.value if s.param_type else "param"
ch = f"CH{s.param_channel}" if s.param_channel >= 0 else "Master"
src = f"CC{s.captured_cc}" if not s.captured_is_nrpn else f"NRPN{s.captured_nrpn}"
return f"{ch} {pt} ← {src}"
def _notify(self, mapping: MIDIMapping) -> None:
for cb in self._on_learned:
try:
cb(mapping)
except Exception as exc:
logger.error("Learn callback failed: %s", exc)
# ── Query state ──────────────────────────────────────────────────────
@property
def is_idle(self) -> bool:
return self.session.state == LearnState.IDLE
@property
def is_listening(self) -> bool:
return self.session.state == LearnState.LISTENING
@property
def is_captured(self) -> bool:
return self.session.state == LearnState.CAPTURED
@property
def is_batch(self) -> bool:
return self.session.state == LearnState.BATCH
@property
def status_text(self) -> str:
s = self.session
pt = s.param_type
pt_val = pt.value if pt else "?"
pl = s.param_label or pt_val
if s.state == LearnState.IDLE:
return "Idle"
if s.state == LearnState.LISTENING:
return f"Listening for {pl}..."
if s.state == LearnState.CAPTURED:
src = f"CC{s.captured_cc}" if not s.captured_is_nrpn else f"NRPN{s.captured_nrpn}"
return f"Captured {src} -> {pl} (confirm/discard)"
if s.state == LearnState.BATCH:
idx = s.batch_index + 1
total = len(self.session.batch_params)
return f"Batch [{idx}/{total}] {pl}..."
return "Unknown"
src/midi/types.py
+256
View File
@@ -0,0 +1,256 @@
"""MIDI types, enums, and message classes for the Raspberry Pi RT Audio Mixer.
Defines the canonical MIDI message representation, mixer parameter
taxonomy, and mapping data structures used throughout the MIDI subsystem.
"""
from __future__ import annotations
import enum
from dataclasses import dataclass, field
from typing import Optional
# ── MIDI protocol constants ────────────────────────────────────────────────
class MIDIMessageType(enum.IntEnum):
"""MIDI channel voice message status nibbles (upper 4 bits)."""
NOTE_OFF = 0x8
NOTE_ON = 0x9
POLY_PRESSURE = 0xA
CONTROL_CHANGE = 0xB # CC
PROGRAM_CHANGE = 0xC
CHANNEL_PRESSURE = 0xD
PITCH_BEND = 0xE
@classmethod
def from_status(cls, status_byte: int) -> MIDIMessageType:
return cls(status_byte >> 4)
@staticmethod
def channel(status_byte: int) -> int:
return status_byte & 0x0F
class MIDISystemMessage(enum.IntEnum):
"""System message types (status byte when channel nibble is ignored)."""
SYSEX_START = 0xF0
MTC_QUARTER = 0xF1 # MIDI Time Code quarter frame
SONG_POSITION = 0xF2
SONG_SELECT = 0xF3
TUNE_REQUEST = 0xF6
SYSEX_END = 0xF7
TIMING_CLOCK = 0xF8
START = 0xFA
CONTINUE = 0xFB
STOP = 0xFC
ACTIVE_SENSE = 0xFE
SYSTEM_RESET = 0xFF
# ── Mixer parameter taxonomy ────────────────────────────────────────────────
class ParameterCategory(enum.StrEnum):
"""Top-level categories of mixer parameters."""
CHANNEL = "channel"
MASTER = "master"
FX = "fx"
ROUTING = "routing"
TRANSPORT = "transport"
UTILITY = "utility"
class ParameterType(enum.StrEnum):
"""Concrete mixer parameter types.
Each type implies a value range and interpolation model.
"""
# Channel strip
VOLUME = "volume" # dB or linear 0.0–1.0
PAN = "pan" # -1.0 (L) to 1.0 (R)
MUTE = "mute" # boolean
SOLO = "solo" # boolean
GAIN = "gain" # dB pre-fader
PHASE_INVERT = "phase_invert" # boolean
# EQ (3-band parametric)
EQ_LOW_FREQ = "eq_low_freq"
EQ_LOW_GAIN = "eq_low_gain"
EQ_LOW_Q = "eq_low_q"
EQ_MID_FREQ = "eq_mid_freq"
EQ_MID_GAIN = "eq_mid_gain"
EQ_MID_Q = "eq_mid_q"
EQ_HIGH_FREQ = "eq_high_freq"
EQ_HIGH_GAIN = "eq_high_gain"
EQ_HIGH_Q = "eq_high_q"
EQ_ENABLE = "eq_enable"
# Dynamics
COMP_THRESHOLD = "comp_threshold"
COMP_RATIO = "comp_ratio"
COMP_ATTACK = "comp_attack"
COMP_RELEASE = "comp_release"
COMP_GAIN = "comp_gain"
GATE_THRESHOLD = "gate_threshold"
GATE_RANGE = "gate_range"
# FX sends
FX_SEND_A = "fx_send_a"
FX_SEND_B = "fx_send_b"
FX_SEND_C = "fx_send_c"
FX_SEND_D = "fx_send_d"
FX_RETURN_A = "fx_return_a"
FX_RETURN_B = "fx_return_b"
FX_RETURN_C = "fx_return_c"
FX_RETURN_D = "fx_return_d"
# Master
MASTER_VOLUME = "master_volume"
MASTER_MUTE = "master_mute"
MASTER_DIM = "master_dim"
MONITOR_VOLUME = "monitor_volume"
PHONES_VOLUME = "phones_volume"
# Transport
PLAY = "play"
STOP = "stop"
RECORD = "record"
LOOP = "loop"
TEMPO = "tempo"
TAP_TEMPO = "tap_tempo"
# Utility
SNAPSHOT_LOAD = "snapshot_load"
SNAPSHOT_SAVE = "snapshot_save"
SCENE_NEXT = "scene_next"
SCENE_PREV = "scene_prev"
# ── Data classes ────────────────────────────────────────────────────────────
@dataclass(frozen=True, slots=True)
class MIDIMessage:
"""Canonical MIDI message.
Normalises CC, NRPN, note, pitch bend, and system realtime messages
into a single representation for the mapping engine.
"""
msg_type: MIDIMessageType
channel: int # 0–15
controller: int | None = None # CC number (0–127) or note number
value: int | None = None # 0–127 for CC, 0–16383 for pitch bend
is_nrpn: bool = False
nrpn_msb: int | None = None # NRPN parameter number MSB (CC 99)
nrpn_lsb: int | None = None # NRPN parameter number LSB (CC 98)
timestamp: float = 0.0 # seconds (monotonic)
source_device: str = "" # ALSA client name or USB device node
@property
def is_cc(self) -> bool:
return self.msg_type == MIDIMessageType.CONTROL_CHANGE and not self.is_nrpn
@property
def nrpn_number(self) -> int | None:
"""Full 14-bit NRPN number (MSB << 7 | LSB)."""
if self.is_nrpn and self.nrpn_msb is not None and self.nrpn_lsb is not None:
return (self.nrpn_msb << 7) | self.nrpn_lsb
return None
@property
def value_normalised(self) -> float:
"""Value scaled to 0.0–1.0 for CC, -1.0–1.0 for pitch bend."""
if self.value is None:
return 0.0
if self.msg_type == MIDIMessageType.PITCH_BEND:
return (self.value - 8192) / 8192.0 # -1.0 to ~1.0
return self.value / 127.0
def __repr__(self) -> str:
base = f"ch={self.channel}"
if self.is_nrpn:
base += f" NRPN={self.nrpn_number}"
elif self.controller is not None:
base += f" CC={self.controller}"
if self.value is not None:
base += f" val={self.value}"
if self.source_device:
base += f" src={self.source_device}"
return f"<MIDI {self.msg_type.name} {base}>"
@dataclass(slots=True)
class MixerParameter:
"""A bindable mixer parameter."""
param_type: ParameterType
category: ParameterCategory
channel: int = -1 # -1 = master / global
label: str = ""
min_val: float = 0.0
max_val: float = 1.0
default_val: float = 0.5
step: float = 0.0 # 0 = continuous; >0 = stepped
@property
def full_label(self) -> str:
if self.channel >= 0:
return f"CH{self.channel} {self.label or self.param_type.value}"
return self.label or self.param_type.value
@dataclass(slots=True)
class MIDIMapping:
"""A single mapping from a MIDI message pattern to a mixer parameter."""
# MIDI source pattern (match criteria)
msg_type: MIDIMessageType = MIDIMessageType.CONTROL_CHANGE
channel: int = 0
controller: int = 0 # CC number or note number
is_nrpn: bool = False
nrpn_number: int = 0 # 14-bit NRPN parameter number
source_device: str = "" # empty = any device
# Mixer target
param_type: ParameterType = ParameterType.VOLUME
param_channel: int = -1
# Value mapping
midi_min: int = 0
midi_max: int = 127
param_min: float = 0.0
param_max: float = 1.0
invert: bool = False
curve: str = "linear" # linear, logarithmic, exponential
# Metadata
label: str = ""
enabled: bool = True
def scale_value(self, midi_value: int) -> float:
"""Convert raw MIDI value to the parameter's output range."""
raw = (midi_value - self.midi_min) / max(1, self.midi_max - self.midi_min)
raw = max(0.0, min(1.0, raw))
if self.invert:
raw = 1.0 - raw
if self.curve == "logarithmic":
import math
raw = math.log10(1 + 9 * raw)
elif self.curve == "exponential":
raw = raw ** 2
return self.param_min + raw * (self.param_max - self.param_min)
def matches(self, msg: MIDIMessage) -> bool:
"""Return True if msg matches this mapping's source criteria."""
if msg.msg_type != self.msg_type:
return False
if self.channel >= 0 and msg.channel != self.channel:
return False
if self.is_nrpn:
if not msg.is_nrpn or msg.nrpn_number != self.nrpn_number:
return False
else:
if msg.controller != self.controller:
return False
if self.source_device and msg.source_device != self.source_device:
return False
return True