#!/usr/bin/env python3 import os, json, time, threading, re from pathlib import Path from queue import Queue, Empty from threading import Event import yaml import mido from paho.mqtt.client import Client as MQTT # ---------- Config via env ---------- MQTT_HOST = os.environ.get("MQTT_HOST", "localhost") MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883")) MQTT_USER = os.environ.get("MQTT_USER") or None MQTT_PASS = os.environ.get("MQTT_PASS") or None BASE_TOPIC = os.environ.get("MIXER_BASE_TOPIC", "mixer").rstrip("/") DISC_PREFIX = os.environ.get("HA_DISCOVERY_PREFIX", "homeassistant").rstrip("/") CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", "midi-mqtt-bridge") DEVICE_NAME = os.environ.get("MIXER_DEVICE_NAME", "Roland M-380") MAP_PATH = Path(os.environ.get("MIXER_MAP_PATH", "/config/m380_map.yaml")) DEBUG = bool(int(os.environ.get("DEBUG", "0"))) FADE_STEP = int(os.environ.get("FADE_STEP", "1")) # % per step FADE_INTERVAL = float(os.environ.get("FADE_INTERVAL", "0.10")) # seconds between steps # publish throttles PUBLISH_MIN_PERIOD = 0.05 # seconds per topic # ---------- Globals ---------- _map = {} _rev_index = {} # (midi_ch, cc) -> ("fader"/"mute", ch_key) _in_port = None msg_q: Queue = Queue() mqtt = None _last_pub_ts = {} # topic -> last time _state_cache = {} # topic -> last payload (string) for resync _chan_state = {} # ch_key -> {"brightness": int 0..100, "muted": bool} _out_port = None _out_lock = threading.Lock() _fade_jobs = {} # ch_key -> {"stop": Event, "thread": Thread} def stop_active_fade(ch_key: str, wait=False): job = _fade_jobs.get(ch_key) if not job: return job["stop"].set() t = job.get("thread") if wait and t and t.is_alive(): try: t.join(timeout=0.5) except Exception: pass _fade_jobs.pop(ch_key, None) def _run_fade(ch_key: str, cfg: dict, target: int): """ Linear fade: step FADE_STEP every FADE_INTERVAL until we hit target (inclusive). Publishes sensor state each step and sends MIDI CC. Aborts if stop event is set. """ if DEBUG: print(f"[fade] start ch={ch_key} target={target} step={FADE_STEP} interval={FADE_INTERVAL}") stop = _fade_jobs.get(ch_key, {}).get("stop") or Event() _fade_jobs[ch_key] = {"stop": stop} # ensure present target = max(0, min(100, int(target))) f = (cfg.get("fader") or {}) port = _map.get("port_out") if not (f.get("midi_type") == "cc" and port): return while not stop.is_set(): st = _current_ch_state(ch_key) cur = int(st["level"]) if cur == target: publish_channel_sensors(ch_key, cfg) # ensure final state seen break # step toward target delta = target - cur step = FADE_STEP if delta > 0 else -FADE_STEP if abs(delta) < abs(step): step = delta # final small step new_level = cur + step new_level = max(0, min(100, new_level)) # send CC + publish ensure_unmuted(ch_key, cfg) midi_send_cc(port, f["midi_channel"], f["controller"], pct_to_cc(new_level)) publish_channel_sensors(ch_key, cfg, level=new_level) # sleep per tick time.sleep(FADE_INTERVAL) def start_fade(ch_key: str, cfg: dict, target: int): # cancel existing fade on this channel stop_active_fade(ch_key, wait=False) ev = Event() _fade_jobs[ch_key] = {"stop": ev} t = threading.Thread(target=_run_fade, args=(ch_key, cfg, target), daemon=True) _fade_jobs[ch_key]["thread"] = t t.start() # ---------- Helpers ---------- def slugify(s: str) -> str: s = (s or "").strip().lower() return re.sub(r"[^a-z0-9]+", "_", s).strip("_") or "unnamed" def chan_slug(ch_key: str, cfg: dict) -> str: nm = str((cfg or {}).get("name") or "").strip() return slugify(nm) if nm else f"ch_{ch_key}" def pct_to_cc(p: float) -> int: p = max(0.0, min(100.0, float(p))) return int(round(127 * (p / 100.0))) def cc_to_pct(v: int) -> int: v = max(0, min(127, int(v))) return int(round((v / 127.0) * 100.0)) def midi_send_cc(port_name: str, midi_ch_1_16: int, controller: int, value: int): global _out_port ch0 = max(1, min(16, int(midi_ch_1_16))) - 1 msg = mido.Message("control_change", channel=ch0, control=int(controller), value=int(value)) with _out_lock: for _ in (1, 2): # try once, then reopen-and-try try: if _out_port is None: open_midi_out() if DEBUG: print(f"[MIDI TX] ch={midi_ch_1_16} cc={controller} val={value}") _out_port.send(msg) return except Exception as e: print(f"[MIDI TX] reopen due to: {e}") time.sleep(0.05) _out_port = None def publish(topic: str, payload: str, retain=True, force=False): now = time.monotonic() last_t = _last_pub_ts.get(topic, 0.0) if not force and (now - last_t) < PUBLISH_MIN_PERIOD: return _last_pub_ts[topic] = now _state_cache[topic] = payload if DEBUG: print(f"[MQTT] {topic} = {payload} (retain={retain})") mqtt.publish(topic, payload=payload, qos=1, retain=retain) def adjust_level_step(ch_key: str, cfg: dict, delta: int): """Adjust channel level by delta (usually ±1), clamp 0..100, send CC, publish.""" st = _current_ch_state(ch_key) new_level = max(0, min(100, st["level"] + int(delta))) if new_level == st["level"]: # Already at boundary; still publish current so external UIs stay in sync if they asked. publish_channel_sensors(ch_key, cfg) return f = (cfg.get("fader") or {}) port = _map.get("port_out") if f.get("midi_type") == "cc" and port: midi_send_cc(port, f["midi_channel"], f["controller"], pct_to_cc(new_level)) publish_channel_sensors(ch_key, cfg, level=new_level) AUTO_UNMUTE_ON_LEVEL = bool(int(os.environ.get("AUTO_UNMUTE_ON_LEVEL", "0"))) def ensure_unmuted(ch_key: str, cfg: dict): if not AUTO_UNMUTE_ON_LEVEL: return st = _current_ch_state(ch_key) if not st["muted"]: return mu = (cfg.get("mute") or {}) port = _map.get("port_out") if mu.get("midi_type") == "cc" and port: on_v, off_v = int(mu.get("on_value", 1)), int(mu.get("off_value", 0)) midi_send_cc(port, mu["midi_channel"], mu["controller"], off_v) publish_channel_sensors(ch_key, cfg, muted=False) def open_midi_out(): global _out_port name = _map.get("port_out") if not name: raise RuntimeError("port_out missing in YAML") try: if _out_port: _out_port.close() except Exception: pass _out_port = mido.open_output(name) print(f"[midi] output on '{name}'") # ---------- Map / reverse index ---------- def load_map(): global _map, _rev_index, _chan_state _map = yaml.safe_load(MAP_PATH.read_text()) or {} _rev_index.clear() chans = _map.get("channels") or {} for ch_key, cfg in chans.items(): f = (cfg or {}).get("fader") or {} m = (cfg or {}).get("mute") or {} if f.get("midi_type") == "cc": _rev_index[(int(f["midi_channel"]), int(f["controller"]))] = ("fader", ch_key) if m.get("midi_type") == "cc": _rev_index[(int(m["midi_channel"]), int(m["controller"]))] = ("mute", ch_key) _chan_state.setdefault(ch_key, {"level": 0, "muted": False}) print(f"[bridge] map loaded: {len(chans)} channels | in='{_map.get('port_in')}' out='{_map.get('port_out')}'") # ---------- Discovery ---------- def publish_discovery(): dev = { "identifiers": [f"midi_mqtt_bridge:{DEVICE_NAME}"], "name": DEVICE_NAME, "manufacturer": "Roland", "model": DEVICE_NAME, } chans = _map.get("channels") or {} for ch_key, cfg in chans.items(): disp = cfg.get("name") or f"CH {ch_key}" slug = chan_slug(ch_key, cfg) uid_base = f"mixer_{ch_key}_{slug}" t, _ = topics_for_channel(ch_key, cfg) lvl_cfg = { "name": f"{disp} Level", "uniq_id": f"{uid_base}_level", "stat_t": t["level_state"], "unit_of_meas": "%", "stat_cla": "measurement", "avty_t": f"{BASE_TOPIC}/availability", "pl_avail": "online", "pl_not_avail": "offline", "device": dev, } mqtt.publish(f"{DISC_PREFIX}/sensor/{uid_base}_level/config", json.dumps(lvl_cfg), qos=1, retain=True) mute_cfg = { "name": f"{disp} Mute", "uniq_id": f"{uid_base}_mute", "stat_t": t["mute_state"], "payload_on": "ON", # ON = muted "payload_off": "OFF", # OFF = unmuted "avty_t": f"{BASE_TOPIC}/availability", "pl_avail": "online", "pl_not_avail": "offline", "device": dev, } mqtt.publish(f"{DISC_PREFIX}/binary_sensor/{uid_base}_mute/config", json.dumps(mute_cfg), qos=1, retain=True) publish(f"{BASE_TOPIC}/availability", "online", retain=True, force=True) # ---------- MIDI input ---------- def midi_callback(msg): try: msg_q.put_nowait(msg) except Exception: pass def open_midi_in(): global _in_port try: if _in_port is not None: _in_port.close() except Exception: pass name = _map.get("port_in") if not name: raise RuntimeError("port_in missing in YAML") _in_port = mido.open_input(name, callback=midi_callback) print(f"[midi] listening on '{name}'") def topics_for_channel(ch_key: str, cfg: dict): slug = chan_slug(ch_key, cfg) if cfg.get("name") else None t_num = { "level_state": f"{BASE_TOPIC}/{ch_key}/level", "mute_state": f"{BASE_TOPIC}/{ch_key}/mute", "level_cmd": f"{BASE_TOPIC}/{ch_key}/fader/set", "mute_cmd": f"{BASE_TOPIC}/{ch_key}/mute/set", } t_name = None if slug: t_name = { "level_state": f"{BASE_TOPIC}/by-name/{slug}/level", "mute_state": f"{BASE_TOPIC}/by-name/{slug}/mute", "level_cmd": f"{BASE_TOPIC}/by-name/{slug}/fader/set", "mute_cmd": f"{BASE_TOPIC}/by-name/{slug}/mute/set", } return t_num, t_name def _current_ch_state(ch_key: str): st = _chan_state.setdefault(ch_key, {"level": 0, "muted": False}) st["level"] = int(max(0, min(100, int(st.get("level", 0))))) st["muted"] = bool(st.get("muted", False)) return st def publish_channel_sensors(ch_key: str, cfg: dict, level: int | None = None, muted: bool | None = None, force=False): st = _current_ch_state(ch_key) if level is not None: st["level"] = int(max(0, min(100, level))) if muted is not None: st["muted"] = bool(muted) t_num, t_name = topics_for_channel(ch_key, cfg) publish(t_num["level_state"], str(int(st["level"])), retain=True, force=force) publish(t_num["mute_state"], "ON" if st["muted"] else "OFF", retain=True, force=force) if t_name: publish(t_name["level_state"], str(int(st["level"])), retain=True, force=force) publish(t_name["mute_state"], "ON" if st["muted"] else "OFF", retain=True, force=force) def resync_all_states(force=False): chans = _map.get("channels") or {} for ch_key, cfg in chans.items(): publish_channel_sensors(ch_key, cfg, force=force) # ---------- MQTT handlers ---------- def on_connect(client, userdata, flags, rc, properties=None): print(f"[mqtt] connected rc={rc}") # New command topics mqtt.subscribe(f"{BASE_TOPIC}/+/fader/set") mqtt.subscribe(f"{BASE_TOPIC}/+/mute/set") mqtt.subscribe(f"{BASE_TOPIC}/by-name/+/fader/set") mqtt.subscribe(f"{BASE_TOPIC}/by-name/+/mute/set") # Nudge topics (no payload) mqtt.subscribe(f"{BASE_TOPIC}/+/fader/up") mqtt.subscribe(f"{BASE_TOPIC}/+/fader/down") mqtt.subscribe(f"{BASE_TOPIC}/by-name/+/fader/up") mqtt.subscribe(f"{BASE_TOPIC}/by-name/+/fader/down") # Fade topics mqtt.subscribe(f"{BASE_TOPIC}/+/fader/fade") mqtt.subscribe(f"{BASE_TOPIC}/by-name/+/fader/fade") # Admin mqtt.subscribe(f"{BASE_TOPIC}/cmd/#") publish_discovery() publish(f"{BASE_TOPIC}/availability", "online", retain=True, force=True) def parse_target(parts): """ Accepts: number-based: [BASE, , 'fader', 'set' ] -> 'fader' [BASE, , 'fader', 'up' ] -> 'fader_up' [BASE, , 'fader', 'down'] -> 'fader_down' [BASE, , 'fader', 'fade'] -> 'fade' [BASE, , 'mute', 'set' ] -> 'mute' by-name: [BASE, 'by-name', , 'fader', 'set' ] -> 'fader' [BASE, 'by-name', , 'fader', 'up' ] -> 'fader_up' [BASE, 'by-name', , 'fader', 'down'] -> 'fader_down' [BASE, 'by-name', , 'fader', 'fade'] -> 'fade' [BASE, 'by-name', , 'mute', 'set' ] -> 'mute' Returns (kind, ch_key, cfg) or (None, None, None) """ chans = _map.get("channels") or {} # number-based if len(parts) == 4 and parts[0] == BASE_TOPIC: ch_key, mid, tail = parts[1], parts[2], parts[3] if ch_key in chans: if mid == "fader" and tail in ("set", "up", "down", "fade"): return ("fader" if tail == "set" else f"fader_{tail}", ch_key, chans[ch_key]) if mid == "mute" and tail == "set": return "mute", ch_key, chans[ch_key] # by-name if len(parts) == 5 and parts[0] == BASE_TOPIC and parts[1] == "by-name": slug, mid, tail = parts[2], parts[3], parts[4] for ck, cfg in chans.items(): if cfg.get("name") and slugify(cfg["name"]) == slug: if mid == "fader" and tail in ("set", "up", "down", "fade"): return ("fader" if tail == "set" else f"fader_{tail}", ck, cfg) if mid == "mute" and tail == "set": return "mute", ck, cfg return None, None, None def on_message(client, userdata, msg): try: topic = msg.topic if topic.startswith(f"{BASE_TOPIC}/cmd/"): handle_cmd(topic.split("/", 2)[2]) return parts = topic.split("/") kind, ch_key, cfg = parse_target(parts) if not cfg: if DEBUG: print(f"[mqtt] ignoring topic (no cfg): {topic}") return port = _map.get("port_out") if not port: if DEBUG: print("[mqtt] no port_out configured; dropping command") return raw = (msg.payload or b"").decode().strip() if DEBUG: print(f"[mqtt] got {topic} = {raw!r} -> kind={kind} ch={ch_key}") # Any direct command cancels an in-flight fade on this channel if kind in ("fader", "fader_up", "fader_down", "mute"): stop_active_fade(ch_key, wait=False) if kind == "fader": # expects numeric 0..100 try: pct = int(float(raw)) except Exception: if DEBUG: print(f"[mqtt] bad fader payload for {topic}: {raw!r}") return pct = max(0, min(100, pct)) f = (cfg.get("fader") or {}) if f.get("midi_type") == "cc": ensure_unmuted(ch_key, cfg) midi_send_cc(port, f["midi_channel"], f["controller"], pct_to_cc(pct)) publish_channel_sensors(ch_key, cfg, level=pct) elif kind in ("fader_up", "fader_down"): delta = 1 if kind == "fader_up" else -1 st = _current_ch_state(ch_key) new_level = max(0, min(100, int(st["level"]) + delta)) f = (cfg.get("fader") or {}) if f.get("midi_type") == "cc": ensure_unmuted(ch_key, cfg) midi_send_cc(port, f["midi_channel"], f["controller"], pct_to_cc(new_level)) publish_channel_sensors(ch_key, cfg, level=new_level) elif kind in ("fade", "fader_fade"): # <— new fade handler # payload = target 0..100 try: target = int(float(raw)) except Exception: if DEBUG: print(f"[mqtt] bad fade payload for {topic}: {raw!r}") return target = max(0, min(100, target)) start_fade(ch_key, cfg, target) elif kind == "fade": # payload = target 0..100 try: target = int(float(raw)) except Exception: if DEBUG: print(f"[mqtt] bad fade payload for {topic}: {raw!r}") return target = max(0, min(100, target)) start_fade(ch_key, cfg, target) elif kind == "mute": # accepts ON/OFF or 1/0 v = raw.upper() if v in ("ON", "OFF"): want_muted = (v == "ON") else: try: want_muted = bool(int(raw)) except Exception: if DEBUG: print(f"[mqtt] bad mute payload for {topic}: {raw!r}") return mu = (cfg.get("mute") or {}) if mu.get("midi_type") == "cc": on_v, off_v = int(mu.get("on_value", 1)), int(mu.get("off_value", 0)) midi_send_cc(port, mu["midi_channel"], mu["controller"], on_v if want_muted else off_v) publish_channel_sensors(ch_key, cfg, muted=want_muted) except Exception as e: print(f"[mqtt] msg error: {e}") def handle_cmd(tail: str): if tail == "reload_map": load_map() open_midi_in() publish_discovery() resync_all_states(force=True) elif tail == "republish_discovery": publish_discovery() elif tail == "resync_states": resync_all_states(force=True) # ---------- MIDI -> MQTT worker ---------- def worker_midi_to_mqtt(): while True: try: for _ in range(64): msg = msg_q.get_nowait() if getattr(msg, "type", "") != "control_change": continue ch1 = msg.channel + 1 key = (ch1, msg.control) mapping = _rev_index.get(key) if not mapping: continue kind, ch_key = mapping cfg = _map["channels"][ch_key] if kind == "fader": pct = cc_to_pct(msg.value) publish_channel_sensors(ch_key, cfg, level=pct) else: mu = cfg.get("mute") or {} on_v, off_v = int(mu.get("on_value", 1)), int(mu.get("off_value", 0)) v = int(msg.value) muted = None if v == on_v or (on_v in (1,127) and v in (1,127) and v != off_v): muted = True elif v == off_v or (off_v in (0,127) and v in (0,127) and v != on_v): muted = False if muted is not None: publish_channel_sensors(ch_key, cfg, muted=muted) except Empty: pass except Exception as e: print(f"[worker] {e}") time.sleep(0.04) # ---------- Main ---------- def main(): global mqtt load_map() open_midi_in() open_midi_out() mqtt = MQTT(client_id=CLIENT_ID, clean_session=True) if MQTT_USER: mqtt.username_pw_set(MQTT_USER, MQTT_PASS) mqtt.will_set(f"{BASE_TOPIC}/availability", "offline", qos=1, retain=True) mqtt.on_connect = on_connect mqtt.on_message = on_message mqtt.connect(MQTT_HOST, MQTT_PORT, keepalive=30) mqtt.loop_start() t = threading.Thread(target=worker_midi_to_mqtt, daemon=True) t.start() print("[bridge] running; CTRL+C to stop") try: while True: time.sleep(1) except KeyboardInterrupt: pass finally: try: mqtt.publish(f"{BASE_TOPIC}/availability", "offline", qos=1, retain=True) except Exception: pass try: mqtt.loop_stop() except Exception: pass try: if _in_port: _in_port.close() except Exception: pass if __name__ == "__main__": main()