gti_radiostudio/midi_engine.py

# 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)
    mixer_mic_on         = Signal(int, bool)
    mixer_gain           = Signal(int, float)
    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 == 'mixer_mic_on':
            on = (value > 63) if not inv else (value < 64)
            self.mixer_mic_on.emit(target_id, on)
        elif action == 'mixer_gain':
            if is_rel and 1 <= value <= 63:
                prev = self._channel_volumes.get(f"gain_{target_id}", 1.0)
                new_val = min(2.0, prev + value * 0.02)
                self._channel_volumes[f"gain_{target_id}"] = new_val
                self.mixer_gain.emit(target_id, new_val)
            elif is_rel and 65 <= value <= 127:
                prev = self._channel_volumes.get(f"gain_{target_id}", 1.0)
                steps = value & 0x3F
                new_val = max(0.0, prev - steps * 0.02)
                self._channel_volumes[f"gain_{target_id}"] = new_val
                self.mixer_gain.emit(target_id, 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
                scaled = scaled * 2.0  # map to 0-2 range
                self.mixer_gain.emit(target_id, scaled)
        elif action == 'mixer_gain':
            if is_rel and 1 <= value <= 63:
                prev = self._channel_volumes.get(f"gain_{target_id}", 1.0)
                new_val = min(2.0, prev + value * 0.02)
                self._channel_volumes[f"gain_{target_id}"] = new_val
                self.mixer_gain.emit(target_id, new_val)
            elif is_rel and 65 <= value <= 127:
                prev = self._channel_volumes.get(f"gain_{target_id}", 1.0)
                steps = value & 0x3F
                new_val = max(0.0, prev - steps * 0.02)
                self._channel_volumes[f"gain_{target_id}"] = new_val
                self.mixer_gain.emit(target_id, 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
                scaled = scaled * 2.0  # map to 0-2 range
                self.mixer_gain.emit(target_id, scaled)
        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}

        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)