uno-q-audio-synth/scripts/prometheus-exporter.py

#!/usr/bin/env python3
"""Prometheus exporter for uno-q-audio-synth pipeline.

Exposes MIDI, audio, and service health metrics on :9090/metrics.
No external dependencies — uses only stdlib.
"""
import http.server, time, os, re, subprocess, socket, threading

PORT = int(os.environ.get("METRICS_PORT", "9090"))
SSH_KEY = os.environ.get("SSH_KEY", "/home/david/.ssh/unoq_deploy_key")
BOARD = os.environ.get("BOARD_HOST", "127.0.0.1")
COLLECT_INTERVAL = 5  # seconds between metric collections

# ── Gauges (updated by collector) ──
metrics = {
    "synth_alsa_running": 0,
    "synth_hw_ptr_bytes_per_sec": 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,
}

prev_hw_ptr = 0
prev_hw_ptr_time = 0


def read_local_file(path):
    try:
        with open(path) as f:
            return f.read()
    except (OSError, PermissionError):
        return ""


def get_alsa_status():
    """Read ALSA PCM status from /proc."""
    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):
    """Check if a systemd service is active (local only)."""
    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):
    """Read an ALSA mixer value via amixer."""
    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_metrics():
    """Background thread: poll system state periodically."""
    while True:
        try:
            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")
        except Exception:
            pass

        time.sleep(COLLECT_INTERVAL)


def format_metrics():
    lines = []
    lines.append("# HELP synth_alsa_running Whether ALSA device is RUNNING")
    lines.append("# TYPE synth_alsa_running gauge")
    lines.append(f"synth_alsa_running {metrics['synth_alsa_running']}")

    lines.append("# HELP synth_hw_ptr_bytes_per_sec Audio throughput in bytes/sec")
    lines.append("# TYPE synth_hw_ptr_bytes_per_sec gauge")
    lines.append(f"synth_hw_ptr_bytes_per_sec {metrics['synth_hw_ptr_bytes_per_sec']:.0f}")

    lines.append("# HELP synth_alsa_hw_ptr Current ALSA hardware pointer")
    lines.append("# TYPE synth_alsa_hw_ptr gauge")
    lines.append(f"synth_alsa_hw_ptr {metrics['synth_alsa_hw_ptr']}")

    lines.append("# HELP synth_pd_running Whether Pd service is active")
    lines.append("# TYPE synth_pd_running gauge")
    lines.append(f"synth_pd_running {metrics['synth_pd_running']}")

    lines.append("# HELP synth_midi_bridge_running Whether midi-bridge service is active")
    lines.append("# TYPE synth_midi_bridge_running gauge")
    lines.append(f"synth_midi_bridge_running {metrics['synth_midi_bridge_running']}")

    lines.append("# HELP synth_viz_running Whether led-matrix-viz service is active")
    lines.append("# TYPE synth_viz_running gauge")
    lines.append(f"synth_viz_running {metrics['synth_viz_running']}")

    lines.append("# HELP synth_hphl_switch Headphone left switch state")
    lines.append("# TYPE synth_hphl_switch gauge")
    lines.append(f"synth_hphl_switch {metrics['synth_hphl_switch']}")

    lines.append("# HELP synth_hphr_switch Headphone right switch state")
    lines.append("# TYPE synth_hphr_switch gauge")
    lines.append(f"synth_hphr_switch {metrics['synth_hphr_switch']}")

    lines.append("# HELP synth_hphl_volume Headphone left volume (0-20)")
    lines.append("# TYPE synth_hphl_volume gauge")
    lines.append(f"synth_hphl_volume {metrics['synth_hphl_volume']}")

    lines.append("# HELP synth_hphr_volume Headphone right volume (0-20)")
    lines.append("# TYPE synth_hphr_volume gauge")
    lines.append(f"synth_hphr_volume {metrics['synth_hphr_volume']}")

    lines.append("# HELP synth_up Exporter is running")
    lines.append("# TYPE synth_up gauge")
    lines.append(f"synth_up {metrics['synth_up']}")

    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)
        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)
        else:
            self.send_response(404)
            self.end_headers()

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


def main():
    # Start background collector
    t = threading.Thread(target=collect_metrics, daemon=True)
    t.start()

    server = http.server.HTTPServer(("0.0.0.0", PORT), MetricsHandler)
    print(f"[metrics] Listening on :{PORT}/metrics")
    server.serve_forever()


if __name__ == "__main__":
    main()