gti_radiostudio/midi_engine.py

# midi_engine.py
"""
MIDI Engine for Radio Panel
Numark DJ2Go controller support.

Confirmed mapping from hardware testing:
------------------------------------------
Play/Pause Deck 1  : Note 0x3B  ch=0
Play/Pause Deck 2  : Note 0x42  ch=0
Cue Deck 1         : Note 0x33  ch=0
Cue Deck 2         : Note 0x3C  ch=0
Jog Wheel Deck 1   : CC   0x19  ch=0
Jog Wheel Deck 2   : CC   0x18  ch=0
Cart 1             : Note 0x44  ch=0
Cart 2             : Note 0x43  ch=0
Cart 3             : Note 0x46  ch=0
Cart 4             : Note 0x45  ch=0
"""

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

# All controls on ch=0, differentiated by note/CC number only
DECK1_PLAY = 0x3B
DECK1_CUE  = 0x33
DECK1_JOG  = 0x19

DECK2_PLAY = 0x42
DECK2_CUE  = 0x3C
DECK2_JOG  = 0x18

# Cart buttons - confirmed from hardware test
CART_NOTES = {
    0x44: 0,   # Cart slot 1
    0x43: 1,   # Cart slot 2
    0x46: 2,   # Cart slot 3
    0x45: 3,   # Cart slot 4
}

JOG_SCRUB_SECONDS = 5.0


def _decode_relative_jog(value: int) -> int:
    if 1 <= value <= 63:
        return 1
    elif 65 <= value <= 127:
        return -1
    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
        channel     = status & 0x0F

        logger.debug(
            f"MIDI IN  status=0x{status:02X}  "
            f"data=0x{data_byte:02X}  "
            f"val={value}  "
            f"type=0x{status_type:02X}  ch={channel}"
        )

        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:
            logger.debug("DECK 1 PLAY/PAUSE")
            self.deck_play_pause.emit('deck1')

        elif note == DECK1_CUE:
            logger.debug("DECK 1 CUE")
            self.deck_cue.emit('deck1')

        elif note == DECK2_PLAY:
            logger.debug("DECK 2 PLAY/PAUSE")
            self.deck_play_pause.emit('deck2')

        elif note == DECK2_CUE:
            logger.debug("DECK 2 CUE")
            self.deck_cue.emit('deck2')

        elif note in CART_NOTES:
            slot = CART_NOTES[note]
            logger.debug(f"CART {slot + 1} TRIGGER")
            self.cart_trigger.emit(slot)

        else:
            logger.debug(f"Unmapped note: 0x{note:02X}")

    def _handle_cc(self, cc_num: int, value: int):
        direction = _decode_relative_jog(value)
        if direction == 0:
            return

        delta = direction * JOG_SCRUB_SECONDS

        if cc_num == DECK1_JOG:
            logger.debug(f"DECK 1 JOG  delta={delta:+.1f}s")
            self.deck_jog.emit('deck1', delta)

        elif cc_num == DECK2_JOG:
            logger.debug(f"DECK 2 JOG  delta={delta:+.1f}s")
            self.deck_jog.emit('deck2', delta)

        else:
            logger.debug(f"Unmapped CC: cc=0x{cc_num:02X} val={value}")