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 5dbea7c83..cfd1ffccc 100755
--- a/modalapi/modhandler.py
+++ b/modalapi/modhandler.py
@@ -43,6 +43,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
@@ -181,6 +182,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
@@ -197,6 +201,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()
@@ -218,6 +223,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): ...