#!/usr/bin/env python3 """Prometheus exporter for uno-q-audio-synth pipeline. Exposes MIDI, audio, viz, and service health metrics on :9090/metrics. Reads stats from /tmp/midi-stats.json (midi-bridge) and /tmp/viz-stats.json (viz). No external dependencies — uses only stdlib. """ import http.server, time, os, re, subprocess, json PORT = int(os.environ.get("METRICS_PORT", "9090")) COLLECT_INTERVAL = 5 metrics = { "synth_alsa_running": 0, "synth_hw_ptr_bytes_per_sec": 0.0, "synth_alsa_hw_ptr": 0, "synth_pd_running": 0, "synth_midi_bridge_running": 0, "synth_viz_running": 0, "synth_hphl_switch": 0, "synth_hphr_switch": 0, "synth_hphl_volume": 0, "synth_hphr_volume": 0, "synth_up": 1, "midi_messages_total": 0, "midi_notes_on_total": 0, "midi_notes_off_total": 0, "midi_errors_total": 0, "midi_last_status": 0, "midi_last_note": 0, "midi_last_velocity": 0, "midi_last_time_unix": 0.0, "viz_frames_sent": 0, "viz_frames_dropped": 0, "viz_pending_queue": 0, "viz_queue_full": 0, } prev_hw_ptr = 0 prev_hw_ptr_time = 0 MIDI_STATS_FILE = "/tmp/midi-stats.json" VIZ_STATS_FILE = "/tmp/viz-stats.json" def read_json(path): try: with open(path) as f: return json.load(f) except (OSError, ValueError): return {} def read_local_file(path): try: with open(path) as f: return f.read() except (OSError, PermissionError): return "" def get_alsa_status(): global prev_hw_ptr, prev_hw_ptr_time status = read_local_file("/proc/asound/card0/pcm0p/sub0/status") metrics["synth_alsa_running"] = 1 if "RUNNING" in status else 0 m = re.search(r"hw_ptr\s*:\s*(\d+)", status) if m: hw_ptr = int(m.group(1)) now = time.monotonic() metrics["synth_alsa_hw_ptr"] = hw_ptr if prev_hw_ptr_time > 0: dt = now - prev_hw_ptr_time if dt > 0: rate = (hw_ptr - prev_hw_ptr) / dt metrics["synth_hw_ptr_bytes_per_sec"] = max(0, rate) prev_hw_ptr = hw_ptr prev_hw_ptr_time = now def check_service(name): try: r = subprocess.run( ["systemctl", "is-active", name], capture_output=True, text=True, timeout=5 ) return 1 if r.stdout.strip() == "active" else 0 except Exception: return 0 def get_mixer_value(name): try: r = subprocess.run( ["amixer", "-c", "0", "cget", f"name={name}"], capture_output=True, text=True, timeout=5 ) for line in r.stdout.splitlines(): if "values=" in line and line.strip().startswith(":"): val = line.split("values=")[-1].strip() try: return int(val) except ValueError: return 1 if val in ("on", "LOHIFI") else 0 except Exception: return 0 return 0 def collect(): get_alsa_status() metrics["synth_pd_running"] = check_service("pd-synth-onboard") metrics["synth_midi_bridge_running"] = check_service("midi-bridge") metrics["synth_viz_running"] = check_service("led-matrix-viz") metrics["synth_hphl_switch"] = get_mixer_value("HPHL Switch") metrics["synth_hphr_switch"] = get_mixer_value("HPHR Switch") metrics["synth_hphl_volume"] = get_mixer_value("HPHL Volume") metrics["synth_hphr_volume"] = get_mixer_value("HPHR Volume") midi = read_json(MIDI_STATS_FILE) metrics["midi_messages_total"] = midi.get("midi_messages_total", 0) metrics["midi_notes_on_total"] = midi.get("midi_notes_on_total", 0) metrics["midi_notes_off_total"] = midi.get("midi_notes_off_total", 0) metrics["midi_errors_total"] = midi.get("midi_errors_total", 0) metrics["midi_last_status"] = midi.get("midi_last_status", 0) metrics["midi_last_note"] = midi.get("midi_last_note", 0) metrics["midi_last_velocity"] = midi.get("midi_last_velocity", 0) metrics["midi_last_time_unix"] = midi.get("midi_last_time_unix", 0.0) viz = read_json(VIZ_STATS_FILE) metrics["viz_frames_sent"] = viz.get("frames_sent", 0) metrics["viz_frames_dropped"] = viz.get("frames_dropped", 0) metrics["viz_pending_queue"] = viz.get("pending_queue", 0) metrics["viz_queue_full"] = 1 if viz.get("queue_full", False) else 0 def format_metrics(): lines = [ "# HELP synth_alsa_running Whether ALSA device is RUNNING", "# TYPE synth_alsa_running gauge", f"synth_alsa_running {metrics['synth_alsa_running']}", "# HELP synth_hw_ptr_bytes_per_sec Audio throughput in bytes/sec", "# TYPE synth_hw_ptr_bytes_per_sec gauge", f"synth_hw_ptr_bytes_per_sec {metrics['synth_hw_ptr_bytes_per_sec']:.0f}", "# HELP synth_alsa_hw_ptr Current ALSA hardware pointer", "# TYPE synth_alsa_hw_ptr gauge", f"synth_alsa_hw_ptr {metrics['synth_alsa_hw_ptr']}", "# HELP synth_pd_running Whether Pd service is active", "# TYPE synth_pd_running gauge", f"synth_pd_running {metrics['synth_pd_running']}", "# HELP synth_midi_bridge_running Whether midi-bridge service is active", "# TYPE synth_midi_bridge_running gauge", f"synth_midi_bridge_running {metrics['synth_midi_bridge_running']}", "# HELP synth_viz_running Whether led-matrix-viz service is active", "# TYPE synth_viz_running gauge", f"synth_viz_running {metrics['synth_viz_running']}", "# HELP synth_hphl_switch Headphone left switch state", "# TYPE synth_hphl_switch gauge", f"synth_hphl_switch {metrics['synth_hphl_switch']}", "# HELP synth_hphr_switch Headphone right switch state", "# TYPE synth_hphr_switch gauge", f"synth_hphr_switch {metrics['synth_hphr_switch']}", "# HELP synth_hphl_volume Headphone left volume (0-20)", "# TYPE synth_hphl_volume gauge", f"synth_hphl_volume {metrics['synth_hphl_volume']}", "# HELP synth_hphr_volume Headphone right volume (0-20)", "# TYPE synth_hphr_volume gauge", f"synth_hphr_volume {metrics['synth_hphr_volume']}", "# HELP synth_up Exporter is running", "# TYPE synth_up gauge", f"synth_up {metrics['synth_up']}", "# HELP midi_messages_total Total MIDI messages received from hardware", "# TYPE midi_messages_total counter", f"midi_messages_total {metrics['midi_messages_total']}", "# HELP midi_notes_on_total Total MIDI note-on messages", "# TYPE midi_notes_on_total counter", f"midi_notes_on_total {metrics['midi_notes_on_total']}", "# HELP midi_notes_off_total Total MIDI note-off messages", "# TYPE midi_notes_off_total counter", f"midi_notes_off_total {metrics['midi_notes_off_total']}", "# HELP midi_errors_total Total MIDI parse errors", "# TYPE midi_errors_total counter", f"midi_errors_total {metrics['midi_errors_total']}", "# HELP midi_last_status Status byte of last message", "# TYPE midi_last_status gauge", f"midi_last_status {metrics['midi_last_status']}", "# HELP midi_last_note Note number of last message", "# TYPE midi_last_note gauge", f"midi_last_note {metrics['midi_last_note']}", "# HELP midi_last_velocity Velocity of last message", "# TYPE midi_last_velocity gauge", f"midi_last_velocity {metrics['midi_last_velocity']}", "# HELP midi_last_time_unix Epoch of last MIDI message", "# TYPE midi_last_time_unix gauge", f"midi_last_time_unix {metrics['midi_last_time_unix']:.3f}", "# HELP viz_frames_sent_total Total frames sent to router", "# TYPE viz_frames_sent_total counter", f"viz_frames_sent_total {metrics['viz_frames_sent']}", "# HELP viz_frames_dropped_total Total frames dropped (queue full)", "# TYPE viz_frames_dropped_total counter", f"viz_frames_dropped_total {metrics['viz_frames_dropped']}", "# HELP viz_pending_queue Current pending frame queue depth", "# TYPE viz_pending_queue gauge", f"viz_pending_queue {metrics['viz_pending_queue']}", "# HELP viz_queue_full Router write buffer was full", "# TYPE viz_queue_full gauge", f"viz_queue_full {metrics['viz_queue_full']}", ] return "\n".join(lines) + "\n" DASHBOARD_PATH = os.path.join(os.path.dirname(__file__), "dashboard.html") class Handler(http.server.BaseHTTPRequestHandler): def do_GET(self): if self.path == "/metrics": collect() 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) elif self.path == "/health": body = b'{"status":"ok"}\n' self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(body) elif self.path in ("/dashboard", "/dashboard.html"): try: with open(DASHBOARD_PATH) as f: body = f.read().encode() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) except OSError: self.send_response(404) self.end_headers() else: self.send_response(404) self.end_headers() def log_message(self, format, *args): pass def main(): import threading t = threading.Thread(target=lambda: ( setInterval(COLLECT_INTERVAL, collect), ), daemon=True) t.start() server = http.server.HTTPServer(("0.0.0.0", PORT), Handler) print(f"[metrics] Listening on :{PORT}/metrics") print(f"[metrics] Dashboard: http://0.0.0.0:{PORT}/dashboard") server.serve_forever() def setInterval(secs, func): while True: func() time.sleep(secs) if __name__ == "__main__": main()