168 lines
4.9 KiB
Python
168 lines
4.9 KiB
Python
# midi_engine.py
|
|
"""
|
|
MIDI Engine for Radio Panel
|
|
Numark DJ2Go controller support.
|
|
"""
|
|
|
|
import logging
|
|
import threading
|
|
from typing import Optional
|
|
|
|
import rtmidi
|
|
from PySide6.QtCore import QObject, Signal
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ─────────────────────────────────────────────
|
|
# MIDI Constants
|
|
# ─────────────────────────────────────────────
|
|
NOTE_ON = 0x90
|
|
NOTE_OFF = 0x80
|
|
CC = 0xB0
|
|
|
|
DECK1_PLAY = 0x3B
|
|
DECK1_CUE = 0x33
|
|
DECK1_JOG = 0x19
|
|
|
|
DECK2_PLAY = 0x42
|
|
DECK2_CUE = 0x3C
|
|
DECK2_JOG = 0x18
|
|
|
|
CART_NOTES = {
|
|
0x44: 0,
|
|
0x43: 1,
|
|
0x46: 2,
|
|
0x45: 3,
|
|
}
|
|
|
|
JOG_SCRUB_SECONDS = 5.0
|
|
JOG_DEAD_ZONE = 5 # Ignore movements below this threshold
|
|
|
|
|
|
def _decode_relative_jog(value: int) -> int:
|
|
"""Returns movement amount, or 0 if within dead zone."""
|
|
if 1 <= value <= 63:
|
|
return value if value >= JOG_DEAD_ZONE else 0
|
|
elif 65 <= value <= 127:
|
|
backward = 128 - value
|
|
return -backward if backward >= JOG_DEAD_ZONE else 0
|
|
return 0
|
|
|
|
|
|
class MidiEngine(QObject):
|
|
deck_play_pause = Signal(str)
|
|
deck_cue = Signal(str)
|
|
deck_jog = Signal(str, float)
|
|
cart_trigger = Signal(int)
|
|
controller_connected = Signal(bool, str)
|
|
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self._midi_in: Optional[rtmidi.MidiIn] = None
|
|
self._port_name: str = ""
|
|
self._running = False
|
|
|
|
def start(self, port_index: Optional[int] = None) -> bool:
|
|
try:
|
|
self._midi_in = rtmidi.MidiIn()
|
|
self._midi_in.ignore_types(sysex=True, timing=True, active_sense=True)
|
|
|
|
available_ports = self._midi_in.get_ports()
|
|
logger.info(f"Available MIDI ports: {available_ports}")
|
|
|
|
if not available_ports:
|
|
logger.error("No MIDI ports found.")
|
|
self.controller_connected.emit(False, "No MIDI ports found")
|
|
return False
|
|
|
|
if port_index is None:
|
|
port_index = self._find_numark_port(available_ports)
|
|
|
|
self._port_name = available_ports[port_index]
|
|
self._midi_in.open_port(port_index)
|
|
self._midi_in.set_callback(self._midi_callback)
|
|
|
|
self._running = True
|
|
logger.info(f"MIDI connected: {self._port_name}")
|
|
self.controller_connected.emit(True, self._port_name)
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.exception(f"Failed to open MIDI port: {e}")
|
|
self.controller_connected.emit(False, str(e))
|
|
return False
|
|
|
|
def stop(self):
|
|
self._running = False
|
|
if self._midi_in:
|
|
try:
|
|
self._midi_in.cancel_callback()
|
|
self._midi_in.close_port()
|
|
except Exception as e:
|
|
logger.warning(f"Error closing MIDI port: {e}")
|
|
finally:
|
|
del self._midi_in
|
|
self._midi_in = None
|
|
logger.info("MIDI engine stopped.")
|
|
|
|
def list_ports(self) -> list[str]:
|
|
try:
|
|
tmp = rtmidi.MidiIn()
|
|
ports = tmp.get_ports()
|
|
del tmp
|
|
return ports
|
|
except Exception:
|
|
return []
|
|
|
|
def _find_numark_port(self, ports: list[str]) -> int:
|
|
for i, name in enumerate(ports):
|
|
if "numark" in name.lower():
|
|
logger.info(f"Auto-detected Numark on port {i}: {name}")
|
|
return i
|
|
logger.warning("No Numark port found, defaulting to port 0.")
|
|
return 0
|
|
|
|
def _midi_callback(self, event, data=None):
|
|
if not self._running:
|
|
return
|
|
|
|
message, _ = event
|
|
|
|
if len(message) < 2:
|
|
return
|
|
|
|
status = message[0]
|
|
data_byte = message[1]
|
|
value = message[2] if len(message) > 2 else 0
|
|
|
|
status_type = status & 0xF0
|
|
|
|
if status_type == NOTE_ON and value > 0:
|
|
self._handle_note_on(data_byte)
|
|
elif status_type == CC:
|
|
self._handle_cc(data_byte, value)
|
|
|
|
def _handle_note_on(self, note: int):
|
|
if note == DECK1_PLAY:
|
|
self.deck_play_pause.emit('deck1')
|
|
elif note == DECK1_CUE:
|
|
self.deck_cue.emit('deck1')
|
|
elif note == DECK2_PLAY:
|
|
self.deck_play_pause.emit('deck2')
|
|
elif note == DECK2_CUE:
|
|
self.deck_cue.emit('deck2')
|
|
elif note in CART_NOTES:
|
|
self.cart_trigger.emit(CART_NOTES[note])
|
|
|
|
def _handle_cc(self, cc_num: int, value: int):
|
|
movement = _decode_relative_jog(value)
|
|
if movement == 0:
|
|
return
|
|
|
|
# Scale movement to seconds
|
|
delta = (movement / 10.0) * JOG_SCRUB_SECONDS
|
|
|
|
if cc_num == DECK1_JOG:
|
|
self.deck_jog.emit('deck1', delta)
|
|
elif cc_num == DECK2_JOG:
|
|
self.deck_jog.emit('deck2', delta) |