#!/usr/bin/env python3 import socket, msgpack, time, os, select, json, subprocess VIZ_PORT = int(os.environ.get("VIZ_PORT", "8082")) ROUTER_SOCK = os.environ.get("ROUTER_SOCK", "/var/run/arduino-router.sock") COLS = 13 ROWS = 8 REFRESH_INTERVAL = 1.0 STATS_FILE = "/tmp/viz-stats.json" DIAG_INTERVAL = 10.0 DIAG_DURATION = 0.3 _poll = select.poll() POLLIN = select.POLLIN SOCK_BUFFER = 4096 SERVICES = ["midi-bridge", "pd-synth-onboard", "led-matrix-viz", "prometheus-exporter"] STATUS_CACHE = {} LAST_STATUS_CHECK = 0.0 def _check_services(): global STATUS_CACHE, LAST_STATUS_CHECK now = time.monotonic() if now - LAST_STATUS_CHECK < 5.0: return STATUS_CACHE for svc in SERVICES: try: r = subprocess.run(["systemctl", "is-active", svc], capture_output=True, text=True, timeout=3) STATUS_CACHE[svc] = r.stdout.strip() == "active" except Exception: STATUS_CACHE[svc] = False LAST_STATUS_CHECK = now return STATUS_CACHE def pack_frame(pixels): bits = 0 for row in range(ROWS): for col in range(COLS): if pixels[row][col]: bits |= 1 << (row * COLS + col) w0 = bits & 0xFFFFFFFF w1 = (bits >> 32) & 0xFFFFFFFF w2 = (bits >> 64) & 0xFFFFFFFF w3 = (bits >> 96) & 0xFFFFFFFF return [str(w0), str(w1), str(w2), str(w3)] def render(active_notes): pixels = [[0] * COLS for _ in range(ROWS)] note_min, note_max = 36, 96 for note in active_notes: row = 7 - int((note - note_min) / (note_max - note_min) * 7) row = max(0, min(6, row)) col = note % 12 pixels[row][col] = 1 status = _check_services() ok = sum(1 for s in SERVICES if status.get(s)) if len(active_notes) > 0: ok += 1 for c in range(ok): pixels[7][c] = 1 return pack_frame(pixels) def _diag_frame(active_notes): pixels = [[0] * COLS for _ in range(ROWS)] status = _check_services() ok = sum(1 for s in SERVICES if status.get(s)) if len(active_notes) > 0: ok += 1 for r in range(ROWS): for c in range(ok): pixels[r][c] = 1 return pack_frame(pixels) def connect_router(): router = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) router.settimeout(0.01) try: router.connect(ROUTER_SOCK) except OSError: pass return router def try_send(router, data): try: router.sendall(data) return router, True except (BlockingIOError, OSError): return router, False def _send_frame(router, packer, params, msg_id): data = packer.pack([0, msg_id, "draw_frame", params]) router, ok = try_send(router, data) if ok: router.setblocking(False) try: while True: chunk = router.recv(4096) if not chunk: break except (BlockingIOError, OSError): pass finally: router.setblocking(True) router.settimeout(0.01) return router, ok def main(): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind(("0.0.0.0", VIZ_PORT)) sock.setblocking(False) router = connect_router() packer = msgpack.Packer() _poll.register(sock, POLLIN) active_notes = set() last_sent_params = None pending_bytes = None buf = b"" msg_id = 0 frames_sent = 0 frames_dropped = 0 bytes_received = 0 last_refresh = time.monotonic() last_diag = time.monotonic() diag_active = False diag_start = 0.0 pre_diag_params = None stats_last_write = time.monotonic() fps_sample_start = time.monotonic() fps_sample_frames = 0 events_processed = 0 raw_packets = 0 while True: now = time.monotonic() refresh_due = (now - last_refresh) >= REFRESH_INTERVAL diag_due = (now - last_diag) >= DIAG_INTERVAL and not diag_active and len(active_notes) == 0 if diag_due: diag_active = True diag_start = now pre_diag_params = last_sent_params msg_id = (msg_id + 1) % 65535 router, ok = _send_frame(router, packer, _diag_frame(active_notes), msg_id) if ok: frames_sent += 1 if diag_active and (now - diag_start) >= DIAG_DURATION: diag_active = False last_diag = now if pre_diag_params is not None: msg_id = (msg_id + 1) % 65535 params = render(active_notes) router, ok = _send_frame(router, packer, params, msg_id) if ok: frames_sent += 1 last_sent_params = params try: events = _poll.poll(5) except OSError: events = [] for fd, flags in events: if fd == sock.fileno() and (flags & POLLIN): try: while True: data, addr = sock.recvfrom(SOCK_BUFFER) buf += data bytes_received += len(data) raw_packets += 1 except BlockingIOError: pass msg = data.decode().strip() parts = msg.split() if len(parts) >= 3 and parts[0] == "midi": try: status = int(parts[1]) note = int(parts[2]) vel = int(parts[3]) if len(parts) >= 4 else 0 except ValueError: continue msg_type = status & 0xF0 changed = False if msg_type == 0x90 and vel > 0: if note not in active_notes: active_notes.add(note) changed = True elif msg_type == 0x80 or (msg_type == 0x90 and vel == 0): if note in active_notes: active_notes.discard(note) changed = True if changed: if diag_active: diag_active = False last_diag = now events_processed += 1 params = render(active_notes) if params != last_sent_params: msg_id = (msg_id + 1) % 65535 pending_bytes = packer.pack([0, msg_id, "draw_frame", params]) last_sent_params = params if pending_bytes is not None: router, ok = try_send(router, pending_bytes) if ok: pending_bytes = None frames_sent += 1 router.setblocking(False) try: while True: chunk = router.recv(4096) if not chunk: break except (BlockingIOError, OSError): pass finally: router.setblocking(True) router.settimeout(0.01) else: frames_dropped += 1 if refresh_due: last_refresh = now fps_sample_frames += 1 if not diag_active and last_sent_params is not None and pending_bytes is None: msg_id = (msg_id + 1) % 65535 refresh_bytes = packer.pack([0, msg_id, "draw_frame", last_sent_params]) router, ok = try_send(router, refresh_bytes) if ok: frames_sent += 1 if now - stats_last_write >= 2.0: fps = fps_sample_frames / (now - fps_sample_start) if (now - fps_sample_start) > 0 else 0 _write_stats(frames_sent, frames_dropped, fps, len(active_notes), events_processed, bytes_received, raw_packets) stats_last_write = now fps_sample_start = now fps_sample_frames = 0 events_processed = 0 def _write_stats(sent, dropped, fps, active_count, events, bytes_rcv, packets): try: with open(STATS_FILE, "w") as f: json.dump({ "frames_sent": sent, "frames_dropped": dropped, "fps": round(fps, 1), "active_notes": active_count, "midi_events": events, "bytes_received": bytes_rcv, "raw_packets": packets, "ts": time.time() }, f) except OSError: pass if __name__ == "__main__": main()