The Singing Pager: Sonifying Telemetry¶

import os
import json
import asyncio
from fastapi import FastAPI, Request
from pythonosc.udp_client import SimpleUDPClient
from pythonosc import osc_server
from paho.mqtt import client as mqtt_client
from prometheus_client import start_http_server, Counter, Gauge, Histogram
import threading

app = FastAPI()

# OSC client for SuperCollider
osc_client = SimpleUDPClient(
    os.getenv("OSC_HOST", "localhost"), 
    int(os.getenv("OSC_PORT", "57120"))
)

# MQTT client
mqtt_client_instance = None

# Prometheus metrics
alerts_total = Counter("alerts_total", "Total alerts received", ["severity"])
mqtt_messages = Counter("mqtt_messages_total", "Total MQTT messages", ["topic"])
cpu_sound = Gauge("cpu_sound_frequency", "CPU sound frequency")
memory_sound = Gauge("memory_sound_volume", "Memory sound volume")
latency_histogram = Histogram("alert_processing_seconds", "Alert processing time")

# Sound parameters
sound_params = {
    "cpu_freq": 440.0,  # Base frequency for CPU
    "memory_vol": 0.5,   # Base volume for memory
    "alert_scale": 1.0   # Alert scale factor
}

def on_mqtt_connect(client, userdata, flags, rc):
    print(f"Connected to MQTT with result code {rc}")
    client.subscribe("alerts/#")
    client.subscribe("metrics/#")

def on_mqtt_message(client, userdata, msg):
    topic = msg.topic
    payload = msg.payload.decode()

    mqtt_messages.labels(topic=topic).inc()

    try:
        data = json.loads(payload)

        if topic.startswith("alerts/"):
            severity = topic.split("/")[-1]
            send_alert_sound(severity, data)
        elif topic.startswith("metrics/"):
            metric_type = topic.split("/")[-1]
            send_metric_sound(metric_type, data)

    except json.JSONDecodeError:
        print(f"Invalid JSON in MQTT message: {payload}")

def send_alert_sound(severity, data):
    """Send alert sound to SuperCollider based on severity."""
    severity_map = {
        "critical": {"freq": 880, "vol": 0.9, "dur": 2.0},
        "warning": {"freq": 660, "vol": 0.7, "dur": 1.5},
        "info": {"freq": 440, "vol": 0.5, "dur": 1.0}
    }

    params = severity_map.get(severity, severity_map["info"])

    # Send OSC message to SuperCollider
    osc_client.send_message("/alert", [
        params["freq"],
        params["vol"],
        params["dur"],
        severity
    ])

    alerts_total.labels(severity=severity).inc()

def send_metric_sound(metric_type, data):
    """Send metric sound to SuperCollider."""
    if metric_type == "cpu":
        freq = 440 + (data.get("value", 0) * 440)  # Scale frequency
        osc_client.send_message("/cpu", [freq])
        cpu_sound.set(freq)

    elif metric_type == "memory":
        vol = 0.1 + (data.get("value", 0) * 0.9)  # Scale volume
        osc_client.send_message("/memory", [vol])
        memory_sound.set(vol)

@app.post("/alert")
async def alert_webhook(request: Request):
    """Handle Prometheus Alertmanager webhooks."""
    payload = await request.json()

    for alert in payload.get("alerts", []):
        severity = alert["labels"].get("severity", "info")
        value = float(alert["annotations"].get("value", "1") or 1)

        # Send to SuperCollider
        send_alert_sound(severity, {"value": value})

        # Also publish to MQTT
        if mqtt_client_instance:
            mqtt_client_instance.publish(
                f"alerts/{severity}",
                json.dumps({"value": value, "alert": alert})
            )

    return {"ok": True}

@app.post("/metric")
async def metric_webhook(request: Request):
    """Handle metric updates."""
    data = await request.json()
    metric_type = data.get("type")
    value = data.get("value", 0)

    send_metric_sound(metric_type, {"value": value})

    # Publish to MQTT
    if mqtt_client_instance:
        mqtt_client_instance.publish(
            f"metrics/{metric_type}",
            json.dumps({"value": value})
        )

    return {"ok": True}

def setup_mqtt():
    """Setup MQTT client."""
    global mqtt_client_instance

    mqtt_client_instance = mqtt_client.Client()
    mqtt_client_instance.on_connect = on_mqtt_connect
    mqtt_client_instance.on_message = on_mqtt_message

    mqtt_host = os.getenv("MQTT_HOST", "localhost")
    mqtt_port = int(os.getenv("MQTT_PORT", "1883"))

    mqtt_client_instance.connect(mqtt_host, mqtt_port, 60)
    mqtt_client_instance.loop_start()

def start_prometheus_server():
    """Start Prometheus metrics server."""
    start_http_server(9109)

if __name__ == "__main__":
    # Start Prometheus metrics server
    start_prometheus_server()

    # Setup MQTT
    setup_mqtt()

    # Start FastAPI server
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)