From 6cc6596d411b752e4390c60fec1cf7eccc14b356 Mon Sep 17 00:00:00 2001 From: Cam Gorrie Date: Sun, 5 Jul 2026 11:05:02 -0400 Subject: [PATCH] WIP --- GUIDE.md | 3 +- common/token.py | 2 + modalapi/modhandler.py | 6 + pistomp/config.py | 32 ++- pistomp/hardware.py | 64 ++++- pistomp/midi_input_control.py | 70 +++++ pistomp/midi_input_manager.py | 123 +++++++++ setup/config_templates/default_config.yml | 27 +- .../default_config_pistompcore.yml | 23 +- .../default_config_pistomptre.yml | 27 +- tests/test_handler_cleanup.py | 2 + tests/test_hardware.py | 1 + tests/test_midi_input.py | 255 ++++++++++++++++++ typings/rtmidi/_rtmidi.pyi | 6 +- 14 files changed, 607 insertions(+), 34 deletions(-) create mode 100644 pistomp/midi_input_control.py create mode 100644 pistomp/midi_input_manager.py create mode 100644 tests/test_midi_input.py diff --git a/GUIDE.md b/GUIDE.md index e96de4839..c43db7411 100644 --- a/GUIDE.md +++ b/GUIDE.md @@ -384,7 +384,8 @@ mod-ui echoes back → ParamSetMessage → plugin.set_param_value() # caches va ## Coding principles -- **Type safety is mandatory** -- pyright zero! +- **Type safety is mandatory** -- zero pyright errors. +- **Keep things clean** -- ruff check must pass. - **Broad pyright ignores are a code smell** - **getattr and hasattr are banned** - **dependencies must form a DAG** diff --git a/common/token.py b/common/token.py index 622c66146..6a27ae48a 100755 --- a/common/token.py +++ b/common/token.py @@ -14,7 +14,9 @@ # along with pi-stomp. If not, see . ACTION = 'action' +ADC = 'adc' ADC_INPUT = 'adc_input' +ALSA = 'alsa' ANALOG_CONTROLLERS = 'analog_controllers' AUTOSYNC = 'autosync' BANK = 'bank' diff --git a/modalapi/modhandler.py b/modalapi/modhandler.py index 2cd7b8543..c8d230eae 100755 --- a/modalapi/modhandler.py +++ b/modalapi/modhandler.py @@ -41,6 +41,7 @@ from plugins.customization import lookup as plugin_lookup import modalapi.external_midi as ExternalMidi from modalapi.external_midi import EXTERNAL_INSTANCE_ID +from pistomp.midi_input_manager import MidiInputManager from modalapi.ethernet import EthernetManager from modalapi.jack_mute import JackMute from pistomp.lcd320x240 import Lcd @@ -176,6 +177,9 @@ def __init__(self, audiocard: Audiocard, homedir, data_dir="/home/pistomp/data") # External MIDI device synchronization self.external_midi = ExternalMidi.ExternalMidiManager() + # MIDI input from wireless/USB expression pedals (v2/v3 only) + self.midi_input_manager = MidiInputManager() + # Blend mode manager - multiple blend snapshots per pedalboard self.blend_modes: dict[str, Any] = {} # {snapshot_name: BlendMode} self.active_blend_mode: Any | None = None # Currently active blend mode @@ -192,6 +196,7 @@ def cleanup(self): if self._hardware is not None: self._hardware.cleanup() self.external_midi.close() + self.midi_input_manager.close() self.ws_bridge.stop() logging.info("WebSocket bridge stopped") self.ethernet_manager.shutdown() @@ -213,6 +218,7 @@ def _rest_post(self, url: str, *, json=None, data=None) -> Response | None: def add_hardware(self, hardware): self._hardware = hardware hardware.external_midi = self.external_midi + hardware.midi_input_manager = self.midi_input_manager self._controller_manager = ControllerManager(hardware) self.bind_volume_encoder() hardware.register_sink(self) diff --git a/pistomp/config.py b/pistomp/config.py index e587b5fd7..fea3adf2c 100644 --- a/pistomp/config.py +++ b/pistomp/config.py @@ -123,7 +123,32 @@ "type": "object", "properties": { "adc_input": { - "type": "integer" + "type": "integer", + "description": "Legacy top-level ADC channel; equivalent to input.adc" + }, + "input": { + "type": "object", + "description": "Input source: a physical ADC channel (adc) or a MIDI device (alsa)", + "properties": { + "adc": { + "type": "integer", + "description": "ADC channel of a physical control" + }, + "alsa": { + "oneOf": [ + {"type": "string"}, + {"type": "array", "items": {"type": "string"}, "minItems": 1} + ], + "description": "ALSA client name(s) of a MIDI device (USB or Bluetooth make no difference here; " + "list multiple names to try candidates in order, e.g. BLE name then USB name for " + "the same physical pedal)" + } + }, + "anyOf": [ + {"required": ["adc"]}, + {"required": ["alsa"]} + ], + "additionalProperties": False }, "id": { "type": "integer" @@ -148,8 +173,11 @@ } }, "required": [ - "adc_input", "midi_CC" + ], + "anyOf": [ + {"required": ["adc_input"]}, + {"required": ["input"]} ] } }, diff --git a/pistomp/hardware.py b/pistomp/hardware.py index f40e0390c..eb4eb3c69 100755 --- a/pistomp/hardware.py +++ b/pistomp/hardware.py @@ -24,6 +24,8 @@ import pistomp.encoder_controller as EncoderController import pistomp.footswitch as Footswitch import pistomp.taptempo as taptempo +from pistomp.midi_input_control import MidiInputControl +from pistomp.midi_input_manager import MidiInputManager from abc import ABC, abstractmethod from rtmidi import MidiOut @@ -52,7 +54,7 @@ def __init__(self, default_config, handler, midiout, refresh_callback): # Standard hardware objects (not required to exist) self.relay: Relay.Relay | None = None - self.analog_controls: list[AnalogMidiControl.AnalogMidiControl] = [] + self.analog_controls: list[AnalogMidiControl.AnalogMidiControl | MidiInputControl] = [] self.encoders = [] self.controllers: dict[str, Controller] = {} self.footswitches: list[Footswitch.Footswitch] = [] @@ -61,6 +63,9 @@ def __init__(self, default_config, handler, midiout, refresh_callback): self.ledstrip = None self.taptempo = taptempo.TapTempo(None) self.external_midi: ExternalMidiManager | None = None + # Owns rtmidi input ports for wireless/USB expression pedals. Assigned by the + # handler (v2/v3 only) in add_hardware, mirroring external_midi; None on v1. + self.midi_input_manager: MidiInputManager | None = None # control → destination; absent means internal (virtual/mod-host). # Rebuilt every reinit in __apply_midi_routing. Identity-keyed; controllers # are stable across reinit (mutated in place). @@ -160,6 +165,15 @@ def reinit(self, cfg): for fs in self.footswitches: self.handler.chord_helper.register(fs.longpress_groups) + # Wire wireless/USB MIDI inputs (channels/CCs are now finalized for this cfg). + self.__sync_midi_inputs() + + def __sync_midi_inputs(self) -> None: + if self.midi_input_manager is None: + return + controls = [c for c in self.analog_controls if isinstance(c, MidiInputControl)] + self.midi_input_manager.rebuild(controls) + @abstractmethod def init_analog_controls(self): ... @@ -280,30 +294,54 @@ def create_analog_controls(self, cfg): continue id = Util.DICT_GET(c, Token.ID) - adc_input = Util.DICT_GET(c, Token.ADC_INPUT) midi_cc = Util.DICT_GET(c, Token.MIDI_CC) threshold = Util.DICT_GET(c, Token.THRESHOLD) control_type = Util.DICT_GET(c, Token.TYPE) autosync = Util.DICT_GET(c, Token.AUTOSYNC) + # Input source: an `input:` block (adc / alsa) or the legacy top-level + # `adc_input`. ADC drives an AnalogMidiControl; alsa drives a + # MidiInputControl fed by MidiInputManager. + input_cfg = Util.DICT_GET(c, Token.INPUT) or {} + adc_input = Util.DICT_GET(input_cfg, Token.ADC) if adc_input is None: - logging.error("Config file error. Analog control specified without %s" % Token.ADC_INPUT) - continue + adc_input = Util.DICT_GET(c, Token.ADC_INPUT) + # alsa may be a single client name or an ordered list of candidates + # (e.g. try a BLE name before falling back to a USB name) + alsa = Util.DICT_GET(input_cfg, Token.ALSA) + if alsa is None: + device_candidates = [] + elif isinstance(alsa, str): + device_candidates = [alsa] + else: + device_candidates = list(alsa) + if midi_cc is None: logging.error("Config file error. Analog control specified without %s" % Token.MIDI_CC) continue - if threshold is None: - threshold = 16 # Default, 1024 is full scale - if autosync is None: - autosync = False # Default to False + if adc_input is None and not device_candidates: + logging.error("Config file error. Analog control specified without %s or an %s source" + % (Token.ADC_INPUT, Token.INPUT)) + continue + + control: AnalogMidiControl.AnalogMidiControl | MidiInputControl + if adc_input is not None: + if threshold is None: + threshold = 16 # Default, 1024 is full scale + if autosync is None: + autosync = False # Default to False + control = AnalogMidiControl.AnalogMidiControl(self.spi, adc_input, threshold, midi_cc, midi_channel, + control_type, id, c, autosync) + logging.debug("Created AnalogMidiControl Input: %d, Midi Chan: %d, CC: %d" % + (adc_input, midi_channel, midi_cc)) + else: + control = MidiInputControl(midi_channel, midi_cc, control_type, id, device_candidates, c) + logging.debug("Created MidiInputControl Devices: %s, Midi Chan: %d, CC: %d" % + (device_candidates, midi_channel, midi_cc)) - control = AnalogMidiControl.AnalogMidiControl(self.spi, adc_input, threshold, midi_cc, midi_channel, - control_type, id, c, autosync) self.analog_controls.append(control) key = format("%d:%d" % (midi_channel, midi_cc)) self.controllers[key] = control - logging.debug("Created AnalogMidiControl Input: %d, Midi Chan: %d, CC: %d" % - (adc_input, midi_channel, midi_cc)) @abstractmethod def add_encoder(self, id, type, callback, longpress_callback, midi_channel, midi_cc) -> EncoderController.EncoderController | None: @@ -428,7 +466,7 @@ def __init_midi(self, cfg): self.midi_channel = self.get_real_midi_channel(cfg) # TODO could iterate thru all objects here instead of handling in __init_footswitches for ac in self.analog_controls: - if isinstance(ac, AnalogMidiControl.AnalogMidiControl): + if isinstance(ac, (AnalogMidiControl.AnalogMidiControl, MidiInputControl)): ac.set_midi_channel(self.midi_channel) def __init_external_midi(self, cfg): diff --git a/pistomp/midi_input_control.py b/pistomp/midi_input_control.py new file mode 100644 index 000000000..2ad711827 --- /dev/null +++ b/pistomp/midi_input_control.py @@ -0,0 +1,70 @@ +# This file is part of pi-stomp. +# +# pi-stomp is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# pi-stomp is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with pi-stomp. If not, see . + +from typing import Any + +import pistomp.controller as controller +from pistomp.controller import AnalogDisplayInfo +from pistomp.input.event import AnalogEvent + + +class MidiInputControl(controller.Controller): + """An expression pedal (or knob) whose value arrives as MIDI CC from an + external device (USB or BLE) rather than a physical ADC channel. + + MidiInputManager receives the CC on the rtmidi callback thread and calls + feed_midi(), which only stores the value. The value is emitted as an + AnalogEvent on the poll thread in refresh() (called from + Hardware.poll_controls, same as AnalogMidiControl), keeping all dispatch — + and thus every LCD/blend touch — on the 10ms critical path.""" + + def __init__(self, midi_channel: int, midi_CC: int | None, type: str | None, + id: int | None = None, device_candidates: list[str] | None = None, + cfg: dict[str, Any] | None = None): + controller.Controller.__init__(self, midi_channel, midi_CC) + self.type = type + self.id = id + # ordered device names to try (bluetooth first, then usb); first match wins + self.device_candidates: list[str] = device_candidates or [] + self.cfg: dict[str, Any] = cfg or {} + self.last_read: int = 0 + self._pending: int | None = None # written by rtmidi callback thread, read by poll thread + + def set_midi_channel(self, midi_channel: int) -> None: + self.midi_channel = midi_channel + + def feed_midi(self, cc_value: int) -> None: + """Store the latest CC (rtmidi callback thread). A single int assignment + is atomic under the GIL; refresh() drains it on the poll thread.""" + self._pending = cc_value + + def refresh(self) -> None: + """Poll thread: emit an AnalogEvent if a new value has arrived. + Latest-wins, matching AnalogMidiControl's per-tick ADC read.""" + pending = self._pending + if pending is None or pending == self.last_read: + return + self.last_read = pending + self.midi_value = pending + self.sink.handle(AnalogEvent(controller=self, raw_value=pending, midi_value=pending)) + + def send_current_value(self) -> None: + """No-op: a MIDI input has no queryable position, so autosync doesn't apply.""" + + def get_normalized_value(self) -> float: + return self.last_read / 127.0 + + def get_display_info(self) -> AnalogDisplayInfo: + return {'type': self.type, 'id': self.id, 'category': None} diff --git a/pistomp/midi_input_manager.py b/pistomp/midi_input_manager.py new file mode 100644 index 000000000..fe5e0fb7e --- /dev/null +++ b/pistomp/midi_input_manager.py @@ -0,0 +1,123 @@ +# This file is part of pi-stomp. +# +# pi-stomp is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# pi-stomp is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with pi-stomp. If not, see . + +from __future__ import annotations + +import logging + +import rtmidi + +from pistomp.midi_input_control import MidiInputControl + +# MIDI status nibbles +_CONTROL_CHANGE = 0xB0 + + +def _client_name_of(port: str) -> str: + """ALSA client name prefix of an rtmidi port string ('Client:Port N:M').""" + return port.split(":")[0].strip() + + +class MidiInputManager: + """Opens rtmidi input ports for wireless/USB expression pedals and routes + incoming CC messages to the MidiInputControl bound to that port. + + One rtmidi.MidiIn is opened per unique ALSA client name and dispatches + straight to whichever control claimed that device — the incoming CC + number is not matched against anything; a pedal advertises whatever CC + its manufacturer picked (e.g. a Roland EV-1-WL sends CC 11) and that's + accepted as-is. The incoming channel is likewise ignored. `midi_CC` on + the control is purely the *output* CC used when the value is re-emitted + downstream (see Handler._emit_midi) — it plays no part in input + matching. Message delivery is via rtmidi's callback thread, but the + callback only stores into the control (see MidiInputControl.feed_midi) + — no dispatch happens off the poll thread. + """ + + def __init__(self): + self.ports: dict[str, rtmidi.MidiIn] = {} # client name -> open port + self.controls: dict[str, MidiInputControl] = {} # client name -> bound control + + def rebuild(self, controls: list[MidiInputControl]) -> None: + """(Re)bind each control to its device port. Called on every reinit() + since a fresh MidiInputControl is created per pedalboard load. + Ports persist across rebuilds; only the binding is refreshed.""" + for c in controls: + if c.device_candidates: + # FIXME: ports only (re)open here, i.e. on pedalboard switch. A BLE device + # that connects after boot won't be usable until the next reinit — add a + # periodic reopen. + self._ensure_open(c) + + def _ensure_open(self, control: MidiInputControl) -> bool: + """Bind control to the first candidate device with an open (or newly + opened) ALSA port.""" + for name in control.device_candidates: + if name in self.ports: + self.controls[name] = control + return True + if self._open(name, control): + return True + logging.warning("No MIDI input port found for any of: %s", control.device_candidates) + return False + + def _open(self, device_name: str, control: MidiInputControl) -> bool: + try: + probe = rtmidi.MidiIn() + ports = probe.get_ports() + del probe + except Exception as e: + logging.error(f"Failed to enumerate MIDI input ports: {e}") + return False + + for idx, port in enumerate(ports): + if _client_name_of(port).lower() != device_name.lower(): + continue + try: + midi_in = rtmidi.MidiIn() + midi_in.open_port(idx) + # rtmidi ignores sysex/timing/active-sense by default; we only act on CC anyway + midi_in.set_callback(self._on_message, device_name) + self.ports[device_name] = midi_in + self.controls[device_name] = control + logging.info(f"Opened MIDI input port: {port}") + return True + except Exception as e: + logging.error(f"Failed to open MIDI input port {device_name} (index {idx}): {e}") + return False + return False + + def _on_message(self, event, data: str | None = None) -> None: + """rtmidi callback thread. `data` is the device name bound at set_callback + time; feed whatever control currently claims that port. Any CC is accepted.""" + message, _delta = event + if len(message) < 3: + return + if (message[0] & 0xF0) != _CONTROL_CHANGE: # channel nibble ignored + return + control = self.controls.get(data) if data is not None else None + if control is not None: + control.feed_midi(message[2]) + + def close(self) -> None: + for name, midi_in in self.ports.items(): + try: + midi_in.close_port() + logging.debug(f"Closed MIDI input port: {name}") + except Exception as e: + logging.warning(f"Error closing MIDI input port {name}: {e}") + self.ports.clear() + self.controls = {} + logging.info("MIDI input manager closed") diff --git a/setup/config_templates/default_config.yml b/setup/config_templates/default_config.yml index c603c87f6..6f679d1c3 100755 --- a/setup/config_templates/default_config.yml +++ b/setup/config_templates/default_config.yml @@ -52,21 +52,36 @@ hardware: tap_tempo: set_mod_tap_tempo # analog_controllers: - # Each control definition is a list which starts with the adc_input - # adc_input: The analog input to which the control is connected (required) + # Each control has one input source and re-emits its value as a MIDI CC. # id: The id and position on the screen (starting with 0 on the left) # type: The control type, used to represent the control on the screen (optional) - # midi_CC: The MIDI CC message to be sent when the control is adjusted (optional) + # midi_CC: The MIDI CC message to be sent when the control is adjusted (required) + # input: The input source (required). One of: + # adc: A physical ADC channel + # alsa: ALSA client name of a MIDI device (USB or Bluetooth — pi-Stomp + # doesn't distinguish). Give a list to try candidates in order, + # e.g. a BLE name before a USB name for the same physical pedal. # midi_port: Send MIDI to this external port INSTEAD of the virtual MIDI Through port (optional) # Falls back to the virtual port only if the device is unavailable; must be the device name (e.g. 'Source Audio C4 Synth') - # autosync: Whether to send current value on pedalboard load (optional, default: false) + # autosync: Whether to send current value on pedalboard load (optional, default: false; ADC only) + # + # A MIDI-input pedal is re-emitted on this hardware's midi channel + midi_CC, exactly + # like an ADC pedal — the incoming device channel is ignored. Do NOT also enable the + # same device as a MIDI input inside MOD-UI, or mod-host will see both the device's + # native CC and our re-emitted CC (double events). Let pi-Stomp be the sole consumer. # #analog_controllers: - # - adc_input: 5 - # id: 0 + # - id: 0 # type: EXPRESSION # midi_CC: 75 + # input: + # adc: 5 # autosync: true + # - id: 1 # USB expression pedal + # type: EXPRESSION + # midi_CC: 77 + # input: + # alsa: "EV-1-WL USB-MIDI" # encoders: # Each encoder definition is a list which starts with the id diff --git a/setup/config_templates/default_config_pistompcore.yml b/setup/config_templates/default_config_pistompcore.yml index fba746d71..1c8959834 100755 --- a/setup/config_templates/default_config_pistompcore.yml +++ b/setup/config_templates/default_config_pistompcore.yml @@ -43,12 +43,27 @@ hardware: color: blue # analog_controllers: - # Each control definition is a list which starts with the adc_input - # adc_input: The analog input to which the control is connected (required) + # Each control has one input source and re-emits its value as a MIDI CC. # id: The id and position on the screen (starting with 0 on the left) # type: The control type, used to represent the control on the screen (optional) - # midi_CC: The MIDI CC message to be sent when the control is adjusted (optional) - # autosync: Whether to send current value on pedalboard load (optional, default: false) + # midi_CC: The MIDI CC message to be sent when the control is adjusted (required) + # input: The input source (required). One of: + # adc: A physical ADC channel + # alsa: ALSA client name of a MIDI device (USB or Bluetooth — pi-Stomp + # doesn't distinguish). Give a list to try candidates in order, + # e.g. a BLE name before a USB name for the same physical pedal. + # autosync: Whether to send current value on pedalboard load (optional, default: false; ADC only) + # + # A MIDI-input pedal is re-emitted on this hardware's midi channel + midi_CC, exactly + # like an ADC pedal — the incoming device channel is ignored. Do NOT also enable the + # same device as a MIDI input inside MOD-UI, or mod-host will see both the device's + # native CC and our re-emitted CC (double events). Let pi-Stomp be the sole consumer. + # + # - id: 3 # USB expression pedal + # type: EXPRESSION + # midi_CC: 75 + # input: + # alsa: "EV-1-WL USB-MIDI" # # analog_controllers: # - adc_input: 7 diff --git a/setup/config_templates/default_config_pistomptre.yml b/setup/config_templates/default_config_pistomptre.yml index 62645ada9..d5b5e20a8 100644 --- a/setup/config_templates/default_config_pistomptre.yml +++ b/setup/config_templates/default_config_pistomptre.yml @@ -54,21 +54,36 @@ hardware: tap_tempo: set_mod_tap_tempo # analog_controllers: - # Each control definition is a list which starts with the adc_input - # adc_input: The analog input to which the control is connected (required) + # Each control has one input source and re-emits its value as a MIDI CC. # id: The id and position on the screen (starting with 0 on the left) # type: The control type, used to represent the control on the screen (optional) - # midi_CC: The MIDI CC message to be sent when the control is adjusted (optional) + # midi_CC: The MIDI CC message to be sent when the control is adjusted (required) + # input: The input source (required). One of: + # adc: A physical ADC channel + # alsa: ALSA client name of a MIDI device (USB or Bluetooth — pi-Stomp + # doesn't distinguish). Give a list to try candidates in order, + # e.g. a BLE name before a USB name for the same physical pedal. # midi_port: Send MIDI to this external port INSTEAD of the virtual MIDI Through port (optional) # Falls back to the virtual port only if the device is unavailable; must be the device name (e.g. 'Source Audio C4 Synth') - # autosync: Whether to send current value on pedalboard load (optional, default: false) + # autosync: Whether to send current value on pedalboard load (optional, default: false; ADC only) + # + # A MIDI-input pedal is re-emitted on this hardware's midi channel + midi_CC, exactly + # like an ADC pedal — the incoming device channel is ignored. Do NOT also enable the + # same device as a MIDI input inside MOD-UI, or mod-host will see both the device's + # native CC and our re-emitted CC (double events). Let pi-Stomp be the sole consumer. # #analog_controllers: - # - adc_input: 5 - # id: 0 + # - id: 0 # physical expression pedal on ADC channel 5 # type: EXPRESSION # midi_CC: 75 + # input: + # adc: 5 # autosync: true + # - id: 1 # USB expression pedal + # type: EXPRESSION + # midi_CC: 77 + # input: + # alsa: "EV-1-WL USB-MIDI" # encoders: # Each encoder definition is a list which starts with the id diff --git a/tests/test_handler_cleanup.py b/tests/test_handler_cleanup.py index 3412d2dfc..e631c4788 100644 --- a/tests/test_handler_cleanup.py +++ b/tests/test_handler_cleanup.py @@ -26,7 +26,9 @@ def test_cleanup_closes_external_midi(self): h._lcd = None h._hardware = None h.external_midi = MagicMock() + h.midi_input_manager = MagicMock() h.ethernet_manager = MagicMock() h.ws_bridge = MagicMock() h.cleanup() h.external_midi.close.assert_called_once() + h.midi_input_manager.close.assert_called_once() diff --git a/tests/test_hardware.py b/tests/test_hardware.py index 502f630ef..6fd5c5603 100644 --- a/tests/test_hardware.py +++ b/tests/test_hardware.py @@ -149,6 +149,7 @@ def test_reinit_applies_routing_for_default_cfg(self, monkeypatch): hw.handler = MagicMock() hw.footswitches = [] # reinit registers longpress groups over these hw.external_routing = {} # __new__ bypasses __init__; reinit clears it + hw.midi_input_manager = None # __new__ bypasses __init__; reinit syncs it for name in ( "_Hardware__init_midi_default", diff --git a/tests/test_midi_input.py b/tests/test_midi_input.py new file mode 100644 index 000000000..620770f80 --- /dev/null +++ b/tests/test_midi_input.py @@ -0,0 +1,255 @@ +"""MidiInputControl / MidiInputManager: wireless & USB expression pedals. + +Covers the value pipeline (callback stores, poll thread emits), dispatch-by-port +that ignores the incoming channel *and* CC number (a pedal sends whatever CC its +manufacturer picked; `midi_CC` on the control is only the *output* CC used when +re-emitting downstream), and the lifecycle across a pedalboard switch: the same +control instance persists across reinit(), so a per-pedalboard midi_CC override +must not disturb its binding to the device port. +""" + +from typing import cast +from unittest.mock import MagicMock + +import common.token as Token +from pistomp.input.event import AnalogEvent +from pistomp.input.sink import InputSink +from pistomp.midi_input_control import MidiInputControl +from pistomp.midi_input_manager import MidiInputManager +from tests.test_hardware import _StubHardware + +CC = 0xB0 +NOTE_ON = 0x90 + + +class RecordingSink(InputSink): + def __init__(self): + self.events: list = [] + + def handle(self, event) -> bool: + self.events.append(event) + return True + + +# ── MidiInputControl ──────────────────────────────────────────────── + + +class TestMidiInputControl: + def test_refresh_emits_analog_event_on_change(self): + c = MidiInputControl(0, 75, Token.EXPRESSION, id=1) + sink = RecordingSink() + c.sink = sink + + c.feed_midi(64) + c.refresh() + assert len(sink.events) == 1 + e = sink.events[0] + assert isinstance(e, AnalogEvent) + assert e.controller is c + assert e.midi_value == 64 and e.raw_value == 64 + + def test_no_emit_without_new_value(self): + c = MidiInputControl(0, 75, Token.EXPRESSION) + sink = RecordingSink() + c.sink = sink + + c.refresh() # nothing fed + assert sink.events == [] + + c.feed_midi(64) + c.refresh() + c.refresh() # value unchanged since last emit + c.feed_midi(64) + c.refresh() # duplicate value + assert len(sink.events) == 1 + + c.feed_midi(0) + c.refresh() + assert len(sink.events) == 2 + + def test_normalized_and_display(self): + c = MidiInputControl(0, 75, Token.EXPRESSION, id=2) + c.sink = RecordingSink() + c.feed_midi(127) + c.refresh() + assert c.get_normalized_value() == 1.0 + assert c.get_display_info() == {"type": Token.EXPRESSION, "id": 2, "category": None} + + def test_send_current_value_is_noop(self): + # autosync doesn't apply to MIDI input; must not raise and must not need a sink + MidiInputControl(0, 75, Token.EXPRESSION).send_current_value() + + +# ── MidiInputManager dispatch ─────────────────────────────────────── + + +class TestMidiInputManagerDispatch: + def test_dispatch_by_port_ignores_channel_and_cc(self): + # No port scan involved here — bind the control to a device name directly + # and prove any incoming CC number (e.g. 11, as a Roland EV-1-WL sends) reaches it. + mgr = MidiInputManager() + c = MidiInputControl(0, 75, Token.EXPRESSION) # midi_CC 75 is the *output* CC only + mgr.controls["ev-1-wl"] = c + + mgr._on_message(([CC | 9, 11, 100], 0.0), "ev-1-wl") # channel 9, CC 11 + assert c._pending == 100 + + def test_non_cc_and_unbound_device_ignored(self): + mgr = MidiInputManager() + c = MidiInputControl(0, 75, Token.EXPRESSION) + mgr.controls["ev-1-wl"] = c + + mgr._on_message(([NOTE_ON, 11, 60], 0.0), "ev-1-wl") # not a CC + mgr._on_message(([CC, 11, 20], 0.0), "some-other-device") # unbound device + mgr._on_message(([CC, 11], 0.0), "ev-1-wl") # truncated + assert c._pending is None + + def test_control_without_device_candidates_skipped(self): + mgr = MidiInputManager() + mgr.rebuild([MidiInputControl(0, 75, Token.EXPRESSION)]) + assert mgr.controls == {} + + +# ── MidiInputManager port opening ─────────────────────────────────── + + +def _fake_rtmidi(monkeypatch, ports): + created: list[MagicMock] = [] + + def factory(): + m = MagicMock() + m.get_ports.return_value = ports + created.append(m) + return m + + monkeypatch.setattr("pistomp.midi_input_manager.rtmidi.MidiIn", factory) + return created + + +# Real ALSA client names observed on-device for a Roland EV-1-WL: the BLE GATT +# bridge and the USB class-compliant interface enumerate under different names. +_BT_PORT = "EV-1-WL:EV-1-WL Bluetooth 133:0" +_USB_PORT = "EV-1-WL USB-MIDI:EV-1-WL USB-MIDI MIDI 1 20:0" + + +class TestMidiInputManagerPorts: + def test_opens_matching_client_case_insensitive(self, monkeypatch): + _fake_rtmidi(monkeypatch, ["Other:in 12:0", _BT_PORT]) + mgr = MidiInputManager() + mgr.rebuild([MidiInputControl(0, 75, Token.EXPRESSION, device_candidates=["ev-1-wl"])]) + + assert "ev-1-wl" in mgr.ports + opened = cast(MagicMock, mgr.ports["ev-1-wl"]) + opened.open_port.assert_called_once_with(1) + assert opened.set_callback.called + + def test_candidate_priority_falls_back(self, monkeypatch): + _fake_rtmidi(monkeypatch, [_USB_PORT]) + mgr = MidiInputManager() + # BLE client name absent → falls back to the USB client name + mgr.rebuild([MidiInputControl(0, 75, Token.EXPRESSION, + device_candidates=["EV-1-WL", "EV-1-WL USB-MIDI"])]) + assert "EV-1-WL USB-MIDI" in mgr.ports and "EV-1-WL" not in mgr.ports + + def test_no_match_leaves_ports_and_controls_empty(self, monkeypatch): + _fake_rtmidi(monkeypatch, ["Something Else:0 20:0"]) + mgr = MidiInputManager() + mgr.rebuild([MidiInputControl(0, 75, Token.EXPRESSION, device_candidates=["Missing"])]) + assert mgr.ports == {} + assert mgr.controls == {} + # rebuild() runs again on every reinit(), so a device that appears later + # (e.g. BLE reconnect) still gets bound on a subsequent pedalboard switch. + + def test_close(self, monkeypatch): + _fake_rtmidi(monkeypatch, [_USB_PORT]) + mgr = MidiInputManager() + mgr.rebuild([MidiInputControl(0, 75, Token.EXPRESSION, device_candidates=["EV-1-WL USB-MIDI"])]) + port = cast(MagicMock, mgr.ports["EV-1-WL USB-MIDI"]) + mgr.close() + port.close_port.assert_called_once() + assert mgr.ports == {} and mgr.controls == {} + + +# ── Lifecycle across a pedalboard switch ──────────────────────────── + + +class TestPedalboardSwitchLifecycle: + def _hw(self, monkeypatch, ports=None): + _fake_rtmidi(monkeypatch, ports if ports is not None else []) + default_cfg = { + Token.HARDWARE: { + Token.VERSION: 3.0, + Token.MIDI: {Token.CHANNEL: 1}, + Token.ANALOG_CONTROLLERS: [ + { + Token.ID: 1, + Token.MIDI_CC: 75, + Token.TYPE: Token.EXPRESSION, + Token.INPUT: {Token.ALSA: ["EV-1-WL", "EV-1-WL USB-MIDI"]}, + } + ], + } + } + hw = _StubHardware(default_cfg, MagicMock(), MagicMock(name="midiout"), lambda *a, **k: None) + hw.midi_input_manager = MidiInputManager() + hw.create_analog_controls(default_cfg) + control = hw.analog_controls[0] + assert isinstance(control, MidiInputControl) + sink = RecordingSink() + control.sink = sink + hw.midi_input_manager.rebuild([control]) + return hw, control, sink + + def test_midi_cc_override_does_not_disturb_input_binding(self, monkeypatch): + # Only the USB name is present (no BLE bond yet), so the manager falls back to + # it. The device sends CC 11 (a Roland EV-1-WL's default); dispatch is by port, + # regardless of what midi_CC the pedalboard configures for output. + hw, control, sink = self._hw(monkeypatch, ports=[_USB_PORT]) + mgr = hw.midi_input_manager + assert mgr is not None + assert mgr.controls["EV-1-WL USB-MIDI"] is control + + # Pedalboard A remaps this pedal's *output* CC 75 → 77; input dispatch is unaffected. + pb_a = {Token.HARDWARE: {Token.ANALOG_CONTROLLERS: [{Token.ID: 1, Token.MIDI_CC: 77}]}} + hw.reinit(pb_a) + assert control.midi_CC == 77 + assert mgr.controls["EV-1-WL USB-MIDI"] is control + + mgr._on_message(([CC | 3, 11, 100], 0.0), "EV-1-WL USB-MIDI") # any channel, device's own CC 11 + control.refresh() + assert sink.events[-1].midi_value == 100 + + # Pedalboard B has no analog override → output CC reverts, binding still holds. + hw.reinit({Token.HARDWARE: {}}) + assert control.midi_CC == 75 + assert mgr.controls["EV-1-WL USB-MIDI"] is control + + mgr._on_message(([CC, 11, 40], 0.0), "EV-1-WL USB-MIDI") + control.refresh() + assert sink.events[-1].midi_value == 40 + + def test_ports_persist_across_switches(self, monkeypatch): + # A matching port opens once and is reused, not reopened, on each reinit. + created = _fake_rtmidi(monkeypatch, [_USB_PORT]) + default_cfg = { + Token.HARDWARE: { + Token.VERSION: 3.0, + Token.MIDI: {Token.CHANNEL: 1}, + Token.ANALOG_CONTROLLERS: [ + {Token.ID: 1, Token.MIDI_CC: 75, Token.TYPE: Token.EXPRESSION, + Token.INPUT: {Token.ALSA: "EV-1-WL USB-MIDI"}}, + ], + } + } + hw = _StubHardware(default_cfg, MagicMock(), MagicMock(), lambda *a, **k: None) + hw.midi_input_manager = MidiInputManager() + hw.create_analog_controls(default_cfg) + hw.analog_controls[0].sink = RecordingSink() + + hw.reinit({Token.HARDWARE: {}}) + port_after_first = hw.midi_input_manager.ports["EV-1-WL USB-MIDI"] + opens_after_first = sum(m.open_port.call_count for m in created) + + hw.reinit({Token.HARDWARE: {}}) + assert hw.midi_input_manager.ports["EV-1-WL USB-MIDI"] is port_after_first + assert sum(m.open_port.call_count for m in created) == opens_after_first diff --git a/typings/rtmidi/_rtmidi.pyi b/typings/rtmidi/_rtmidi.pyi index bb75672bb..7d8b5f755 100644 --- a/typings/rtmidi/_rtmidi.pyi +++ b/typings/rtmidi/_rtmidi.pyi @@ -1,4 +1,4 @@ -from typing import Callable, Sequence +from typing import Any, Callable, Sequence class MidiOut: def open_port(self, port: int = 0, port_name: str = "") -> None: ... @@ -19,7 +19,9 @@ class MidiIn: def get_port_count(self) -> int: ... def get_port_name(self, port: int) -> str: ... def get_message(self) -> tuple[list[int], float] | None: ... - def set_callback(self, func: Callable[[list[int], float], None]) -> None: ... + def set_callback( + self, func: Callable[[tuple[list[int], float], Any], None], data: Any = None + ) -> None: ... def cancel_callback(self) -> None: ... class RtMidiError(Exception): ...