Add 8-channel mixer with master compressor/limiter, MIDI learn system, separate deck outputs
- Mixer: 8 renamable stereo input channels with VU meter, volume fader, mute/solo - Master bus: compressor (threshold, ratio, attack, release, makeup) + limiter (ceiling), LUFS metering - Decks/carts keep their own JACK outputs, completely separate from mixer - MIDI learn: configurable bindings saved to ~/.gti_radiostudio/midi_map.json - mixer_widget.py: QDial-based DSP controls, scrollable channel strips
This commit is contained in:
+124
-31
@@ -1,10 +1,12 @@
|
||||
# midi_engine.py
|
||||
"""
|
||||
MIDI Engine for Radio Panel
|
||||
Numark DJ2Go controller support.
|
||||
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
|
||||
|
||||
@@ -13,34 +15,18 @@ 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,
|
||||
}
|
||||
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 # Ignore movements below this threshold
|
||||
JOG_DEAD_ZONE = 5
|
||||
|
||||
|
||||
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:
|
||||
@@ -49,12 +35,61 @@ def _decode_relative_jog(value: int) -> int:
|
||||
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)
|
||||
controller_connected = Signal(bool, str)
|
||||
mapping_learned = Signal(str, object, int, int, int, bool)
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
@@ -62,6 +97,30 @@ class MidiEngine(QObject):
|
||||
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()
|
||||
@@ -136,33 +195,67 @@ class MidiEngine(QObject):
|
||||
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 == '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)
|
||||
|
||||
def _handle_note_on(self, note: int):
|
||||
if note == DECK1_PLAY:
|
||||
# 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 == DECK1_CUE:
|
||||
elif note == 0x33:
|
||||
self.deck_cue.emit('deck1')
|
||||
elif note == DECK2_PLAY:
|
||||
elif note == 0x42:
|
||||
self.deck_play_pause.emit('deck2')
|
||||
elif note == DECK2_CUE:
|
||||
elif note == 0x3C:
|
||||
self.deck_cue.emit('deck2')
|
||||
elif note in CART_NOTES:
|
||||
self.cart_trigger.emit(CART_NOTES[note])
|
||||
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
|
||||
|
||||
# Scale movement to seconds
|
||||
delta = (movement / 10.0) * JOG_SCRUB_SECONDS
|
||||
|
||||
if cc_num == DECK1_JOG:
|
||||
if cc_num == 0x19:
|
||||
self.deck_jog.emit('deck1', delta)
|
||||
elif cc_num == DECK2_JOG:
|
||||
self.deck_jog.emit('deck2', delta)
|
||||
elif cc_num == 0x18:
|
||||
self.deck_jog.emit('deck2', delta)
|
||||
|
||||
Reference in New Issue
Block a user