"""
MIDI device plumbing — discovering, opening, and registering hardware ports.
Provides interactive/automatic output and input device selection, the
multi-device registry used by the sequencer, and the ``bank_select()``
helper for addressing synth banks beyond the first 128 programs.
"""
import logging
import typing
import mido
logger = logging.getLogger(__name__)
# Type alias for device identifiers: index (int), name (str), or None (device 0).
DeviceId = typing.Union[int, str, None]
[docs]
class MidiDeviceRegistry:
"""Ordered registry of named MIDI ports (output or input).
Devices are stored in insertion order. Index 0 is always the first
(or only) device — the default for all APIs that do not specify a device.
Devices can be looked up by integer index or by name string.
``None`` always resolves to index 0.
The registry is intended to be append-only once playback has started.
All registered port objects must already be open.
"""
def __init__ (self) -> None:
"""
Create an empty registry; populate it with add().
"""
self._ports: typing.List[typing.Tuple[str, typing.Any]] = []
self._name_to_index: typing.Dict[str, int] = {}
# Per-device physical output latency in milliseconds, parallel to
# self._ports by index. Kept separate from the (name, port) tuple so
# replace() can swap the port object without disturbing latency.
self._latencies: typing.List[float] = []
[docs]
def add (self, name: str, port: typing.Any, latency_ms: float = 0.0) -> int:
"""Register a port under *name*. Returns the assigned integer index.
*latency_ms* is the device's physical output latency (non-negative);
see :meth:`set_latency`.
"""
idx = len(self._ports)
self._ports.append((name, port))
self._latencies.append(max(0.0, float(latency_ms)))
# First registration wins for name collisions.
if name not in self._name_to_index:
self._name_to_index[name] = idx
return idx
[docs]
def get (self, device: DeviceId = None) -> typing.Optional[typing.Any]:
"""Return the port for *device*, or ``None`` if the registry is empty.
``None`` → index 0. ``int`` → direct index. ``str`` → name lookup.
Returns ``None`` if the device cannot be resolved (empty registry,
out-of-range index, unknown name).
"""
if not self._ports:
return None
idx = self.index_of(device)
if idx < 0 or idx >= len(self._ports):
return None
return self._ports[idx][1]
[docs]
def index_of (self, device: DeviceId = None) -> int:
"""Resolve *device* to an integer index. Returns 0 for ``None``.
Returns -1 if the name is unknown or the index is out of range."""
if device is None:
return 0
if isinstance(device, int):
if 0 <= device < len(self._ports):
return device
return -1
# str
return self._name_to_index.get(device, -1)
[docs]
def replace (self, index: int, port: typing.Any) -> None:
"""Replace the port object at *index* without changing the name or index mapping.
Used by the backward-compat ``midi_out``/``midi_in`` setters to allow
test code to inject a fake port after the registry has been populated.
Raises ``IndexError`` if *index* is out of range.
"""
if index < 0 or index >= len(self._ports):
raise IndexError(f"MidiDeviceRegistry: index {index} out of range (size {len(self._ports)})")
name = self._ports[index][0]
self._ports[index] = (name, port)
# Latency is intentionally preserved — replace() is a pure port swap.
[docs]
def set_latency (self, device: DeviceId, latency_ms: float) -> None:
"""Set the physical output latency (milliseconds) for *device*.
*latency_ms* must be non-negative — a device cannot sound before it is
triggered, so a negative output latency is meaningless. Raises
``ValueError`` for a negative value or an unknown device.
"""
if latency_ms < 0:
raise ValueError(f"latency_ms must be non-negative — got {latency_ms}")
idx = self.index_of(device)
if idx < 0:
raise ValueError(f"Unknown output device: {device!r}")
self._latencies[idx] = float(latency_ms)
[docs]
def latency_of (self, device: DeviceId = None) -> float:
"""Return the latency (ms) for *device*, or 0.0 if it cannot be resolved.
Defensive on the hot dispatch path: an unknown device yields 0.0 rather
than raising, so a stray event can never crash the send loop.
"""
idx = self.index_of(device)
if idx < 0 or idx >= len(self._latencies):
return 0.0
return self._latencies[idx]
[docs]
def max_latency (self) -> float:
"""Return the largest latency across all registered devices (0.0 if empty)."""
return max(self._latencies, default=0.0)
[docs]
def close_all (self) -> None:
"""Close every registered port and clear the registry."""
for name, port in self._ports:
try:
port.close()
except (OSError, RuntimeError, AttributeError):
# Shutdown path: a failure on one port must not prevent closing the rest.
logger.exception(f"Error closing MIDI port '{name}'")
self._ports.clear()
self._name_to_index.clear()
self._latencies.clear()
def __len__ (self) -> int:
"""
Number of registered devices.
"""
return len(self._ports)
def __iter__ (self) -> typing.Iterator[typing.Any]:
"""Iterate over port objects (not names)."""
return (port for _, port in self._ports)
def __bool__ (self) -> bool:
"""
True if at least one device is registered.
"""
return bool(self._ports)
[docs]
def bank_select (bank: int) -> typing.Tuple[int, int]:
"""
Convert a 14-bit MIDI bank number to (MSB, LSB) for use with
``p.program_change()``.
MIDI bank select uses two control-change messages: CC 0 (Bank MSB) and
CC 32 (Bank LSB). Together they encode a 14-bit bank number in the
range 0–16,383:
MSB = bank // 128 (upper 7 bits, sent on CC 0)
LSB = bank % 128 (lower 7 bits, sent on CC 32)
Args:
bank: Integer bank number, 0–16,383. Values outside this range are
clamped.
Returns:
``(msb, lsb)`` tuple, each value in 0–127.
Example:
```python
msb, lsb = subsequence.bank_select(128) # → (1, 0)
p.program_change(48, bank_msb=msb, bank_lsb=lsb)
```
"""
bank = max(0, min(16383, bank))
return bank >> 7, bank & 0x7F
[docs]
def select_output_device (device_name: typing.Optional[str] = None) -> typing.Tuple[typing.Optional[str], typing.Optional[typing.Any]]:
"""
Select and open a MIDI output device.
If ``device_name`` is provided, attempts to open that specific device.
If ``device_name`` is None, auto-discovers available devices:
- If exactly one device exists, it is selected automatically.
- If multiple devices exist, prompts the user to choose one from the console.
- If no devices exist, logs an error and returns None.
Returns:
A tuple of (device_name, midi_out_object) or (None, None) on failure.
"""
try:
outputs = mido.get_output_names()
logger.info(f"Available MIDI outputs: {outputs}")
if not outputs:
logger.error("No MIDI output devices found.")
return None, None
# Explicit device requested
if device_name is not None:
if device_name in outputs:
midi_out = mido.open_output(device_name)
logger.info(f"Opened MIDI output: {device_name}")
return device_name, midi_out
else:
logger.error(
f"MIDI output device '{device_name}' not found. "
f"Available devices: {outputs}"
)
return None, None
# Auto-discover: one device - use it
if len(outputs) == 1:
selected_name = outputs[0]
midi_out = mido.open_output(selected_name)
logger.info(f"One MIDI output found - using '{selected_name}'")
return selected_name, midi_out
# Auto-discover: multiple devices - prompt user
print("\nAvailable MIDI output devices:\n")
for i, name in enumerate(outputs, 1):
print(f" {i}. {name}")
print()
while True:
try:
choice = int(input(f"Select a device (1-{len(outputs)}): "))
if 1 <= choice <= len(outputs):
break
except (ValueError, EOFError):
pass
print(f"Enter a number between 1 and {len(outputs)}.")
selected_name = outputs[choice - 1]
midi_out = mido.open_output(selected_name)
logger.info(f"Opened MIDI output: {selected_name}")
print(f"\nTip: To skip this prompt, pass the device name directly:\n")
print(f" Sequencer(output_device_name=\"{selected_name}\")")
print(f" Composition(output_device=\"{selected_name}\")\n")
return selected_name, midi_out
except (OSError, RuntimeError) as e:
logger.error(f"Failed to open MIDI output: {e}")
return None, None