uno-q-audio-synth/scripts/midi-bridge.py

#!/usr/bin/env python3
"""Bridge raw MIDI device to Pd via UDP. Auto-detects gadget vs host mode.

Sends parsed MIDI messages as Pd-style 'midi status data1 data2;' strings
to UDP ports for Pd (8081) and LED matrix visualization (8082).

Exposes Prometheus metrics on :9091/metrics.
"""
import os, socket, time, select, pathlib, threading, http.server

PD_HOST = os.environ.get("MIDI_BRIDGE_HOST", "127.0.0.1")
PD_PORT = int(os.environ.get("MIDI_BRIDGE_PORT", "8081"))
VIZ_PORT = int(os.environ.get("VIZ_PORT", "8082"))
METRICS_PORT = int(os.environ.get("MIDI_METRICS_PORT", "9091"))

# ── Metrics ──
midi_notes_on = 0
midi_notes_off = 0
midi_messages_total = 0
midi_errors = 0
last_status = 0
last_note = 0
last_velocity = 0
last_time = 0.0
_metrics_lock = threading.Lock()


def _find_midi_dev():
    """Auto-detect rawmidi device. Preference: gadget f_midi > env override > host mode."""
    dev = os.environ.get("MIDI_BRIDGE_DEV")
    if dev:
        return dev
    cards = pathlib.Path("/proc/asound").glob("card*")
    for card_dir in cards:
        cmi = card_dir / "id"
        if cmi.exists() and "midi" in cmi.read_text().strip():
            card_num = card_dir.name.replace("card", "")
            midi_dev = f"/dev/snd/midiC{card_num}D0"
            if os.path.exists(midi_dev):
                return midi_dev
    rawdevs = sorted(pathlib.Path("/dev/snd").glob("midiC?D0"))
    if rawdevs:
        return str(rawdevs[0])
    return "/dev/snd/midiC0D0"


def midi_msg(status, data1, data2=None):
    if data2 is not None:
        return f"midi {status} {data1} {data2};"
    return f"midi {status} {data1};"


def _open_midi():
    midi_dev = _find_midi_dev()
    while not os.path.exists(midi_dev):
        time.sleep(0.5)
        midi_dev = _find_midi_dev()
    fd = os.open(midi_dev, os.O_RDWR | os.O_NONBLOCK)
    os.set_blocking(fd, False)
    return fd


def _flush(sock, buf):
    global midi_notes_on, midi_notes_off, midi_messages_total, midi_errors
    global last_status, last_note, last_velocity, last_time

    status = buf[0]
    if (buf[0] & 0xF0) == 0xF0:
        return
    status = (status & 0xF0) | 0
    data1 = buf[1]
    data2 = buf[2] if len(buf) >= 3 else None
    msg = midi_msg(status, data1, data2)
    sock.sendto(msg.encode(), (PD_HOST, PD_PORT))
    sock.sendto(msg.encode(), (PD_HOST, VIZ_PORT))

    with _metrics_lock:
        midi_messages_total += 1
        last_status = status
        last_note = data1
        last_velocity = data2 if data2 is not None else 0
        last_time = time.time()
        msg_type = status & 0xF0
        if msg_type == 0x90 and (data2 or 0) > 0:
            midi_notes_on += 1
        elif msg_type == 0x80 or (msg_type == 0x90 and (data2 or 0) == 0):
            midi_notes_off += 1


# ── Prometheus HTTP ──

def _format_metrics():
    with _metrics_lock:
        lines = [
            "# HELP midi_messages_total Total MIDI messages received",
            "# TYPE midi_messages_total counter",
            f"midi_messages_total {midi_messages_total}",
            "",
            "# HELP midi_notes_on_total Total note-on messages",
            "# TYPE midi_notes_on_total counter",
            f"midi_notes_on_total {midi_notes_on}",
            "",
            "# HELP midi_notes_off_total Total note-off messages",
            "# TYPE midi_notes_off_total counter",
            f"midi_notes_off_total {midi_notes_off}",
            "",
            "# HELP midi_errors_total Total MIDI parse errors",
            "# TYPE midi_errors_total counter",
            f"midi_errors_total {midi_errors}",
            "",
            "# HELP midi_last_status Status byte of last message",
            "# TYPE midi_last_status gauge",
            f"midi_last_status {last_status}",
            "",
            "# HELP midi_last_note Note number of last message",
            "# TYPE midi_last_note gauge",
            f"midi_last_note {last_note}",
            "",
            "# HELP midi_last_velocity Velocity of last message",
            "# TYPE midi_last_velocity gauge",
            f"midi_last_velocity {last_velocity}",
            "",
            "# HELP midi_last_time_unix Epoch of last MIDI message",
            "# TYPE midi_last_time_unix gauge",
            f"midi_last_time_unix {last_time:.3f}",
            "",
        ]
    return "\n".join(lines) + "\n"


class _MetricsHandler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/metrics":
            body = _format_metrics().encode()
            self.send_response(200)
            self.send_header("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, format, *args):
        pass


def _start_metrics_server():
    server = http.server.HTTPServer(("0.0.0.0", METRICS_PORT), _MetricsHandler)
    t = threading.Thread(target=server.serve_forever, daemon=True)
    t.start()


def main():
    _start_metrics_server()

    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    running = True
    while running:
        try:
            fd = _open_midi()
            poll = select.poll()
            poll.register(fd, select.POLLIN)
            buf = bytearray()
            while running:
                try:
                    events = poll.poll(500)
                    for f, flags in events:
                        if flags & select.POLLIN:
                            data = os.read(fd, 1024)
                            if not data:
                                continue
                            for byte in data:
                                if byte & 0x80:
                                    if len(buf) > 0 and buf[0] >= 0x80:
                                        _flush(sock, buf)
                                    buf = bytearray([byte])
                                else:
                                    buf.append(byte)
                                    if len(buf) >= 3 and (buf[0] & 0xE0) in (0x80, 0x90, 0xA0, 0xB0, 0xE0):
                                        _flush(sock, buf)
                                        buf = bytearray()
                                    elif len(buf) >= 2 and (buf[0] & 0xE0) in (0xC0, 0xD0):
                                        _flush(sock, buf)
                                        buf = bytearray()
                    if not events:
                        time.sleep(0.01)
                except OSError as e:
                    if e.errno == 19:
                        os.close(fd)
                        time.sleep(1)
                        break
                    raise
                except BlockingIOError:
                    pass
        except KeyboardInterrupt:
            running = False

    sock.close()


if __name__ == "__main__":
    main()