From cbc508349067ef479375683146151fef2de6f9b4 Mon Sep 17 00:00:00 2001
From: joren <joren@ai-c-s.my.domain>
Date: Thu, 2 Apr 2026 19:48:02 +0200
Subject: [PATCH] feat: real-time dB level meters per node
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Backend:
- One pw_stream (INPUT, F32) per audio node; for sinks use
  stream.capture.sink=true to read the monitor port
- RT process callback computes instantaneous linear peak with no
  allocations; stored in std::atomic<float>
- Meter streams are filtered from the registry so they never appear
  in the graph or trigger recursive meter creation
- Meter streams are created on first node-ready event, destroyed on
  node removal, and cleaned up on engine close
- GET /api/peaks → {node_id: linear_peak} (polled by frontend)

Frontend:
- peaks store polled at 100 ms via setInterval; starts/stops with
  initGraph/destroyGraph
- Each node card grows 8 px and shows a 3 px meter bar at the bottom
  (green below -12 dB, yellow -12 to -3 dB, red above -3 dB)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 frontend/src/components/GraphCanvas.svelte |  21 ++-
 frontend/src/lib/stores.ts                 |  22 +++
 src/graph_engine.cpp                       | 148 ++++++++++++++++++++-
 src/graph_engine.h                         |  30 +++++
 src/web_server.cpp                         |  15 +++
 5 files changed, 232 insertions(+), 4 deletions(-)

diff --git a/frontend/src/components/GraphCanvas.svelte b/frontend/src/components/GraphCanvas.svelte
index e450afa..b6436df 100644
--- a/frontend/src/components/GraphCanvas.svelte
+++ b/frontend/src/components/GraphCanvas.svelte
@@ -1,7 +1,7 @@
 <script lang="ts">
   import { onMount, onDestroy } from 'svelte';
   import {
-    nodes, ports, links, connected, patchbay,
+    nodes, ports, links, connected, patchbay, peaks,
     portById,
     initGraph, destroyGraph,
     connectPorts, disconnectPorts,
@@ -139,6 +139,19 @@
     return $patchbay.aliases?.[nd.name] || nd.nick || nd.name;
   }
 
+  // Level meter helpers
+  function meterDb(nodeId: number): number {
+    const p = $peaks[nodeId] ?? 0;
+    return p > 0 ? Math.max(-60, 20 * Math.log10(p)) : -60;
+  }
+  function meterWidth(nodeId: number, barW: number): number {
+    return barW * Math.max(0, (meterDb(nodeId) + 60) / 60);
+  }
+  function meterColor(nodeId: number): string {
+    const db = meterDb(nodeId);
+    return db >= -3 ? '#c44' : db >= -12 ? '#ca4' : '#4a9';
+  }
+
   // Build computed layout
   let graphNodes = $derived.by(() => {
     const n = $nodes;
@@ -194,7 +207,7 @@
       const headerH = 22;
       const portH = 16;
       const maxPorts = Math.max(allInPorts.length, allOutPorts.length, 1);
-      const height = headerH + maxPorts * portH + 20; // extra 20 for volume slider
+      const height = headerH + maxPorts * portH + 28; // 20 for volume slider + 8 for meter bar
       const width = 220;
 
       const portPositions = new Map<number, { x: number; y: number }>();
@@ -677,6 +690,10 @@
           {Math.round(Math.max(0, Math.min(1, nd.volume)) * 100)}%
         </text>
 
+        <!-- Level meter -->
+        <rect x={nd.x + 8} y={nd.y + nd.height - 6} width={nd.width - 16} height="3" rx="1" fill="#1a1a1a" />
+        <rect x={nd.x + 8} y={nd.y + nd.height - 6} width={meterWidth(nd.id, nd.width - 16)} height="3" rx="1" fill={meterColor(nd.id)} />
+
       </g>
     {/each}
   </svg>
diff --git a/frontend/src/lib/stores.ts b/frontend/src/lib/stores.ts
index d521587..4791575 100644
--- a/frontend/src/lib/stores.ts
+++ b/frontend/src/lib/stores.ts
@@ -8,6 +8,21 @@ export const ports = writable<Port[]>([]);
 export const links = writable<Link[]>([]);
 export const connected = writable(false);
 
+// Level meter peaks: node_id → linear peak (0–1)
+export const peaks = writable<Record<number, number>>({});
+let peakInterval: ReturnType<typeof setInterval> | null = null;
+
+async function pollPeaks() {
+  try {
+    const res = await fetch('/api/peaks');
+    if (!res.ok) return;
+    const raw: Record<string, number> = await res.json();
+    const out: Record<number, number> = {};
+    for (const [k, v] of Object.entries(raw)) out[Number(k)] = v;
+    peaks.set(out);
+  } catch {}
+}
+
 // Patchbay state
 export const patchbay = writable<PatchbayState>({
   profiles: {},
@@ -102,6 +117,8 @@ export async function initGraph() {
   unsubscribe = subscribe((graph: GraphMessage) => {
     applyGraph(graph);
   });
+
+  peakInterval = setInterval(pollPeaks, 100);
 }
 
 export function destroyGraph() {
@@ -110,6 +127,11 @@ export function destroyGraph() {
     unsubscribe = null;
     connected.set(false);
   }
+  if (peakInterval) {
+    clearInterval(peakInterval);
+    peakInterval = null;
+  }
+  peaks.set({});
 }
 
 // Patchbay operations
diff --git a/src/graph_engine.cpp b/src/graph_engine.cpp
index 2b78c7c..cb0cf95 100644
--- a/src/graph_engine.cpp
+++ b/src/graph_engine.cpp
@@ -27,6 +27,138 @@
 
 using namespace pwgraph;
 
+// ============================================================================
+// Level meter: pw_stream per audio node, RT process callback stores peak
+// ============================================================================
+
+GraphEngine::MeterStream::~MeterStream() {
+    if (stream) {
+        pw_stream_destroy(stream);
+        stream = nullptr;
+    }
+}
+
+static void on_meter_process(void *data) {
+    auto *ms = static_cast<GraphEngine::MeterStream*>(data);
+    struct pw_buffer *buf = pw_stream_dequeue_buffer(ms->stream);
+    if (!buf) return;
+
+    float peak = 0.0f;
+    struct spa_buffer *sbuf = buf->buffer;
+    for (uint32_t d = 0; d < sbuf->n_datas; d++) {
+        if (!sbuf->datas[d].data) continue;
+        const float *s = static_cast<const float*>(sbuf->datas[d].data);
+        uint32_t n = sbuf->datas[d].chunk->size / sizeof(float);
+        for (uint32_t i = 0; i < n; i++) {
+            float v = s[i] < 0 ? -s[i] : s[i];
+            if (v > peak) peak = v;
+        }
+    }
+    ms->peak.store(peak, std::memory_order_relaxed);
+    pw_stream_queue_buffer(ms->stream, buf);
+}
+
+static const struct pw_stream_events meter_stream_events = {
+    .version = PW_VERSION_STREAM_EVENTS,
+    .process = on_meter_process,
+};
+
+void GraphEngine::createMeterIfNeeded(uint32_t node_id, PortMode mode,
+                                      NodeType ntype, const std::string& name)
+{
+    // Only meter audio nodes
+    if ((ntype & NodeType::Audio) == NodeType::None) return;
+    // Skip nodes with no meaningful direction
+    if (mode == PortMode::None) return;
+
+    {
+        std::lock_guard<std::mutex> lk(m_meter_mutex);
+        if (m_meters.count(node_id)) return;  // already created
+    }
+
+    // Sinks (speakers/headphones) are metered via their monitor port
+    bool capture_sink = (mode == PortMode::Input || mode == PortMode::Duplex);
+
+    std::string meter_node_name = "pwweb.meter." + std::to_string(node_id);
+
+    struct pw_properties *props = pw_properties_new(
+        PW_KEY_MEDIA_TYPE,       "Audio",
+        PW_KEY_MEDIA_CATEGORY,   "Capture",
+        PW_KEY_MEDIA_ROLE,       "DSP",
+        PW_KEY_APP_NAME,         "pwweb-meter",
+        PW_KEY_NODE_NAME,        meter_node_name.c_str(),
+        PW_KEY_TARGET_OBJECT,    std::to_string(node_id).c_str(),
+        "stream.capture.sink",   capture_sink ? "true" : "false",
+        nullptr);
+
+    struct pw_stream *stream = pw_stream_new(m_pw.core, "pwweb-meter", props);
+    if (!stream) {
+        fprintf(stderr, "pwweb: meter stream alloc failed for node %u\n", node_id);
+        return;
+    }
+
+    auto *ms       = new MeterStream();
+    ms->node_id    = node_id;
+    ms->stream     = stream;
+    ms->engine_ref = this;
+
+    pw_stream_add_listener(stream, &ms->listener, &meter_stream_events, ms);
+
+    uint8_t pod_buf[1024];
+    struct spa_pod_builder b = SPA_POD_BUILDER_INIT(pod_buf, sizeof(pod_buf));
+    const struct spa_pod *params[1];
+    struct spa_audio_info_raw info = {};
+    info.format   = SPA_AUDIO_FORMAT_F32;
+    info.rate     = 48000;
+    info.channels = 2;
+    params[0] = spa_format_audio_raw_build(&b, SPA_PARAM_EnumFormat, &info);
+
+    int ret = pw_stream_connect(stream,
+        PW_DIRECTION_INPUT,
+        SPA_ID_INVALID,
+        static_cast<pw_stream_flags>(PW_STREAM_FLAG_AUTOCONNECT |
+                                     PW_STREAM_FLAG_MAP_BUFFERS),
+        params, 1);
+
+    if (ret < 0) {
+        fprintf(stderr, "pwweb: meter connect failed node %u (%s): %s\n",
+                node_id, name.c_str(), spa_strerror(ret));
+        delete ms;
+        return;
+    }
+
+    {
+        std::lock_guard<std::mutex> lk(m_meter_mutex);
+        m_meters[node_id] = ms;
+    }
+    fprintf(stderr, "pwweb: meter created for node %u (%s%s)\n",
+            node_id, name.c_str(), capture_sink ? ", monitor" : "");
+}
+
+void GraphEngine::destroyMeter(uint32_t node_id) {
+    std::lock_guard<std::mutex> lk(m_meter_mutex);
+    auto it = m_meters.find(node_id);
+    if (it == m_meters.end()) return;
+    delete it->second;
+    m_meters.erase(it);
+}
+
+void GraphEngine::clearMeters() {
+    std::lock_guard<std::mutex> lk(m_meter_mutex);
+    for (auto &[id, ms] : m_meters)
+        delete ms;
+    m_meters.clear();
+}
+
+std::unordered_map<uint32_t, float> GraphEngine::getPeaks() const {
+    std::lock_guard<std::mutex> lk(m_meter_mutex);
+    std::unordered_map<uint32_t, float> out;
+    out.reserve(m_meters.size());
+    for (auto &[id, ms] : m_meters)
+        out[id] = ms->peak.load(std::memory_order_relaxed);
+    return out;
+}
+
 // ============================================================================
 // Pending/sync helpers (ported from qpwgraph_pipewire.cpp)
 // ============================================================================
@@ -163,6 +295,11 @@ static void on_node_info(void *data, const struct pw_node_info *info) {
         }
     }
 
+    // On first ready: spin up a meter stream for this node
+    if (!nobj->node.ready && obj->engine_ref)
+        obj->engine_ref->createMeterIfNeeded(
+            nobj->node.id, nobj->node.mode, nobj->node.node_type, nobj->node.name);
+
     nobj->node.changed = true;
     nobj->node.ready = true;
 
@@ -435,6 +572,10 @@ static void on_registry_global(void *data,
     auto *engine = static_cast<GraphEngine*>(data);
 
     if (strcmp(type, PW_TYPE_INTERFACE_Node) == 0) {
+        // Skip our own meter streams (avoid graph clutter and infinite recursion)
+        const char *app = spa_dict_lookup(props, PW_KEY_APP_NAME);
+        if (app && strcmp(app, "pwweb-meter") == 0) return;
+
         // Parse node properties (ported from qpwgraph lines 444-489)
         const char *str = spa_dict_lookup(props, PW_KEY_NODE_DESCRIPTION);
         const char *nick = spa_dict_lookup(props, PW_KEY_NODE_NICK);
@@ -443,7 +584,6 @@ static void on_registry_global(void *data,
         if (!str || strlen(str) < 1) str = "node";
 
         std::string node_name;
-        const char *app = spa_dict_lookup(props, PW_KEY_APP_NAME);
         if (app && strlen(app) > 0 && strcmp(app, str) != 0) {
             node_name += app;
             node_name += '/';
@@ -708,6 +848,9 @@ void GraphEngine::close() {
     if (m_pw.loop)
         pw_thread_loop_stop(m_pw.loop);
 
+    // PW thread is stopped; safe to destroy meter streams now
+    clearMeters();
+
     if (m_pw.registry) {
         spa_hook_remove(&m_pw.registry_listener);
         pw_proxy_destroy((pw_proxy*)m_pw.registry);
@@ -796,13 +939,14 @@ void GraphEngine::removeObject(uint32_t id) {
         }
     }
 
-    // If it's a node, remove all its ports
+    // If it's a node, remove all its ports and tear down its meter
     if (obj->type == Object::ObjNode) {
         auto *nobj = static_cast<NodeObj*>(obj);
         auto port_ids_copy = nobj->node.port_ids;
         for (uint32_t pid : port_ids_copy) {
             removeObject(pid);
         }
+        destroyMeter(id);
     }
 
     // If it's a link, remove from output port's link list
diff --git a/src/graph_engine.h b/src/graph_engine.h
index aaf2bcc..bec5295 100644
--- a/src/graph_engine.h
+++ b/src/graph_engine.h
@@ -16,6 +16,7 @@ struct pw_thread_loop;
 struct pw_context;
 struct pw_core;
 struct pw_registry;
+struct pw_stream;
 
 namespace pwgraph {
 
@@ -135,6 +136,35 @@ public:
     uint32_t midiPortType()  const { return m_midi_type; }
     uint32_t videoPortType() const { return m_video_type; }
     uint32_t otherPortType() const { return m_other_type; }
+
+    // -----------------------------------------------------------------------
+    // Level metering
+    // -----------------------------------------------------------------------
+    struct MeterStream {
+        uint32_t            node_id   = 0;
+        struct pw_stream   *stream    = nullptr;
+        struct spa_hook     listener  = {};
+        std::atomic<float>  peak      {0.0f};   // instantaneous linear peak
+        GraphEngine        *engine_ref = nullptr;
+
+        MeterStream() = default;
+        ~MeterStream();
+        MeterStream(const MeterStream&) = delete;
+        MeterStream& operator=(const MeterStream&) = delete;
+    };
+
+    // Called from on_node_info when a node first becomes ready
+    void createMeterIfNeeded(uint32_t node_id, PortMode mode, NodeType ntype,
+                             const std::string& name);
+    void destroyMeter(uint32_t node_id);
+    void clearMeters();
+
+    // Returns linear peak (0–1) per node_id; called from HTTP thread
+    std::unordered_map<uint32_t, float> getPeaks() const;
+
+private:
+    std::unordered_map<uint32_t, MeterStream*> m_meters;
+    mutable std::mutex m_meter_mutex;
 };
 
 } // namespace pwgraph
diff --git a/src/web_server.cpp b/src/web_server.cpp
index 6a13341..34ff55f 100644
--- a/src/web_server.cpp
+++ b/src/web_server.cpp
@@ -302,6 +302,21 @@ void WebServer::setupRoutes() {
         res.set_header("Access-Control-Allow-Origin", "*");
     });
 
+    // Level meters: GET /api/peaks → {"<node_id>": <linear_peak_0_1>, ...}
+    m_http.Get("/api/peaks", [this](const httplib::Request &, httplib::Response &res) {
+        auto peaks = m_engine.getPeaks();
+        std::string json = "{";
+        bool first = true;
+        for (auto &[id, peak] : peaks) {
+            if (!first) json += ",";
+            json += "\"" + std::to_string(id) + "\":" + std::to_string(peak);
+            first = false;
+        }
+        json += "}";
+        res.set_content(json, "application/json");
+        res.set_header("Access-Control-Allow-Origin", "*");
+    });
+
     // REST API: POST /api/connect
     m_http.Post("/api/connect", [this](const httplib::Request &req, httplib::Response &res) {
         uint32_t out_id = 0, in_id = 0;