# 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 PITCH_BEND = 0xE0 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, is_relative: bool = False): key = (midi_type, midi_channel, midi_note) self.bindings[key] = { 'action': action, 'target_id': target_id, 'inverse': inverse, 'is_relative': is_relative, } 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'], 'is_relative': val.get('is_relative', False), }) 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), 'is_relative': entry.get('is_relative', 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) master_volume = Signal(float) controller_connected = Signal(bool, str) mapping_learned = Signal(str, object, int, int, int, bool) raw_midi_event = Signal(str, int, int, int) 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 = {} self._channel_volumes = {} self._master_knob_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 type_names = {0x90: "note_on", 0x80: "note_off", 0xB0: "cc", 0xE0: "pitch", 0xD0: "aftertouch"} type_name = type_names.get(status_type, f"0x{status_type:02X}") self.raw_midi_event.emit(type_name, channel, data_byte, value) if self._learn_mode and self._learn_target: action, target_id = self._learn_target mapping_type = status_type is_rel = (status_type == 0xB0 and (1 <= value <= 63 or 65 <= value <= 127)) note = 0 if status_type == PITCH_BEND else data_byte self.mapping.add(action, target_id, mapping_type, channel, note, is_relative=is_rel) self.mapping.save() self.mapping_learned.emit(action, target_id, mapping_type, channel, note, 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) elif status_type == PITCH_BEND: lsb = data_byte msb = message[2] if len(message) > 2 else 0 pb_value = (msb << 7) | lsb self._handle_learned_event(status_type, channel, 0, pb_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) is_rel = binding.get('is_relative', 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 == 'mixer_volume': if is_rel and 1 <= value <= 63: prev = self._channel_volumes.get(target_id, 1.0) new_vol = min(1.0, prev + value * 0.02) self._channel_volumes[target_id] = new_vol self.mixer_volume.emit(target_id, new_vol) elif is_rel and 65 <= value <= 127: prev = self._channel_volumes.get(target_id, 1.0) steps = value & 0x3F new_vol = max(0.0, prev - steps * 0.02) self._channel_volumes[target_id] = new_vol self.mixer_volume.emit(target_id, new_vol) else: if midi_type == PITCH_BEND: scaled = (16383 - value if inv else value) / 16383.0 else: scaled = (127 - value if inv else value) / 127.0 self._channel_volumes[target_id] = scaled 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': knob = str(target_id) if is_rel and 1 <= value <= 63: prev = self._master_knob_values.get(knob, 0.5) new_val = min(1.0, prev + value * 0.02) self._master_knob_values[knob] = new_val self.master_control.emit(knob, new_val) elif is_rel and 65 <= value <= 127: prev = self._master_knob_values.get(knob, 0.5) steps = value & 0x3F new_val = max(0.0, prev - steps * 0.02) self._master_knob_values[knob] = new_val self.master_control.emit(knob, new_val) else: if midi_type == PITCH_BEND: scaled = (16383 - value if inv else value) / 16383.0 else: scaled = (127 - value if inv else value) / 127.0 self._master_knob_values[knob] = scaled self.master_control.emit(knob, scaled) elif action == 'master_volume': if midi_type == PITCH_BEND: scaled = (16383 - value if inv else value) / 16383.0 else: scaled = (127 - value if inv else value) / 127.0 self.master_volume.emit(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)