nickmin
7de0cc3de0
TopBarWidget already connects mapping_learned to _on_mapping_learned internally in set_midi_engine, so the external connection in main.py was both redundant and referenced a non-existent method.
295 lines
9.9 KiB
Python
295 lines
9.9 KiB
Python
# midi_engine.py
|
|
"""
|
|
MIDI Engine for GTI Radio Studio.
|
|
Supports MIDI learn and configurable control mapping.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
from typing import Optional
|
|
|
|
import rtmidi
|
|
from PySide6.QtCore import QObject, Signal
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
NOTE_ON = 0x90
|
|
NOTE_OFF = 0x80
|
|
CC = 0xB0
|
|
|
|
CONFIG_DIR = os.path.expanduser("~/.gti_radiostudio")
|
|
CONFIG_FILE = os.path.join(CONFIG_DIR, "midi_map.json")
|
|
|
|
JOG_SCRUB_SECONDS = 5.0
|
|
JOG_DEAD_ZONE = 5
|
|
|
|
|
|
def _decode_relative_jog(value: int) -> int:
|
|
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 MidiMapping:
|
|
def __init__(self):
|
|
self.bindings = {}
|
|
|
|
def add(self, action: str, target_id, midi_type: int, midi_channel: int,
|
|
midi_note: int, inverse: bool = False):
|
|
key = (midi_type, midi_channel, midi_note)
|
|
self.bindings[key] = {
|
|
'action': action,
|
|
'target_id': target_id,
|
|
'inverse': inverse,
|
|
}
|
|
|
|
def lookup(self, midi_type: int, midi_channel: int,
|
|
midi_note: int) -> Optional[dict]:
|
|
return self.bindings.get((midi_type, midi_channel, midi_note))
|
|
|
|
def save(self, filepath: str = CONFIG_FILE):
|
|
os.makedirs(os.path.dirname(filepath), exist_ok=True)
|
|
data = []
|
|
for key, val in self.bindings.items():
|
|
data.append({
|
|
'midi_type': key[0],
|
|
'midi_channel': key[1],
|
|
'midi_note': key[2],
|
|
'action': val['action'],
|
|
'target_id': val['target_id'],
|
|
'inverse': val['inverse'],
|
|
})
|
|
with open(filepath, 'w') as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
def load(self, filepath: str = CONFIG_FILE):
|
|
if not os.path.exists(filepath):
|
|
return
|
|
with open(filepath) as f:
|
|
data = json.load(f)
|
|
for entry in data:
|
|
key = (entry['midi_type'], entry['midi_channel'], entry['midi_note'])
|
|
self.bindings[key] = {
|
|
'action': entry['action'],
|
|
'target_id': entry['target_id'],
|
|
'inverse': entry.get('inverse', False),
|
|
}
|
|
|
|
|
|
class MidiEngine(QObject):
|
|
deck_play_pause = Signal(str)
|
|
deck_cue = Signal(str)
|
|
deck_jog = Signal(str, float)
|
|
cart_trigger = Signal(int)
|
|
mixer_volume = Signal(int, float)
|
|
mixer_mute = Signal(int, bool)
|
|
mixer_solo = Signal(int, bool)
|
|
mixer_pfl = Signal(int, bool)
|
|
master_control = Signal(str, float)
|
|
controller_connected = Signal(bool, str)
|
|
mapping_learned = Signal(str, object, int, int, int, bool)
|
|
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self._midi_in: Optional[rtmidi.MidiIn] = None
|
|
self._port_name: str = ""
|
|
self._running = False
|
|
|
|
self.mapping = MidiMapping()
|
|
self.mapping.load()
|
|
|
|
self._learn_target = None
|
|
self._learn_mode = False
|
|
|
|
self._last_cc_values = {}
|
|
|
|
@property
|
|
def learn_mode(self) -> bool:
|
|
return self._learn_mode
|
|
|
|
def set_learn_target(self, target: Optional[tuple]):
|
|
self._learn_target = target
|
|
|
|
def start_learn(self, action: str, target_id):
|
|
self._learn_mode = True
|
|
self._learn_target = (action, target_id)
|
|
logger.info(f"MIDI learn: waiting for {action} ({target_id})...")
|
|
|
|
def stop_learn(self):
|
|
self._learn_mode = False
|
|
self._learn_target = None
|
|
|
|
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
|
|
channel = status & 0x0F
|
|
|
|
# Learn mode: capture any MIDI event
|
|
if self._learn_mode and self._learn_target:
|
|
action, target_id = self._learn_target
|
|
mapping_type = status_type
|
|
self.mapping.add(action, target_id, mapping_type, channel, data_byte)
|
|
self.mapping.save()
|
|
self.mapping_learned.emit(action, target_id, mapping_type, channel, data_byte, value > 0)
|
|
self.stop_learn()
|
|
return
|
|
|
|
if status_type == NOTE_ON and value > 0:
|
|
self._handle_note_on(data_byte)
|
|
self._handle_learned_event(status_type, channel, data_byte, value)
|
|
elif status_type == NOTE_OFF or (status_type == NOTE_ON and value == 0):
|
|
pass
|
|
elif status_type == CC:
|
|
self._handle_cc(data_byte, value)
|
|
self._handle_learned_event(status_type, channel, data_byte, value)
|
|
|
|
def _handle_learned_event(self, midi_type: int, channel: int,
|
|
note: int, value: int):
|
|
binding = self.mapping.lookup(midi_type, channel, note)
|
|
if binding is None:
|
|
return
|
|
|
|
action = binding['action']
|
|
target_id = binding['target_id']
|
|
inv = binding.get('inverse', False)
|
|
|
|
if action == 'deck_play_pause' and isinstance(target_id, str):
|
|
self.deck_play_pause.emit(target_id)
|
|
elif action == 'deck_cue' and isinstance(target_id, str):
|
|
self.deck_cue.emit(target_id)
|
|
elif action == 'cart_trigger':
|
|
self.cart_trigger.emit(target_id)
|
|
elif action == 'deck_play_pause' and isinstance(target_id, str):
|
|
self.deck_play_pause.emit(target_id)
|
|
elif action == 'deck_cue' and isinstance(target_id, str):
|
|
self.deck_cue.emit(target_id)
|
|
elif action == 'cart_trigger':
|
|
self.cart_trigger.emit(target_id)
|
|
elif action == 'mixer_volume':
|
|
scaled = (127 - value if inv else value) / 127.0
|
|
self.mixer_volume.emit(target_id, scaled)
|
|
elif action == 'mixer_mute':
|
|
muted = (value < 64) if inv else (value > 63)
|
|
self.mixer_mute.emit(target_id, muted)
|
|
elif action == 'mixer_solo':
|
|
on = (value > 63) if not inv else (value < 64)
|
|
self.mixer_solo.emit(target_id, on)
|
|
elif action == 'mixer_pfl':
|
|
on = (value > 63) if not inv else (value < 64)
|
|
self.mixer_pfl.emit(target_id, on)
|
|
elif action == 'master_control':
|
|
scaled = (127 - value if inv else value) / 127.0
|
|
self.master_control.emit(str(target_id), scaled)
|
|
elif action == 'mixer_solo':
|
|
on = (value > 63) if not inv else (value < 64)
|
|
self.mixer_solo.emit(target_id, on)
|
|
elif action == 'mixer_pfl':
|
|
on = (value > 63) if not inv else (value < 64)
|
|
self.mixer_pfl.emit(target_id, on)
|
|
elif action == 'master_control':
|
|
scaled = (127 - value if inv else value) / 127.0
|
|
self.master_control.emit(str(target_id), scaled)
|
|
|
|
def _handle_note_on(self, note: int):
|
|
# Hardcoded Numark DJ2Go mapping (legacy)
|
|
NUMARK_CART = {0x44: 0, 0x43: 1, 0x46: 2, 0x45: 3}
|
|
|
|
if note == 0x3B:
|
|
self.deck_play_pause.emit('deck1')
|
|
elif note == 0x33:
|
|
self.deck_cue.emit('deck1')
|
|
elif note == 0x42:
|
|
self.deck_play_pause.emit('deck2')
|
|
elif note == 0x3C:
|
|
self.deck_cue.emit('deck2')
|
|
elif note in NUMARK_CART:
|
|
self.cart_trigger.emit(NUMARK_CART[note])
|
|
|
|
def _handle_cc(self, cc_num: int, value: int):
|
|
movement = _decode_relative_jog(value)
|
|
if movement == 0:
|
|
return
|
|
|
|
delta = (movement / 10.0) * JOG_SCRUB_SECONDS
|
|
|
|
if cc_num == 0x19:
|
|
self.deck_jog.emit('deck1', delta)
|
|
elif cc_num == 0x18:
|
|
self.deck_jog.emit('deck2', delta)
|