Nodes

src/pipewire/pipewirecontroller.cpp

@@ -6,11 +6,15 @@
 #include <QThread>
 #include <cstring>
 #include <cstdlib>
+#include <cmath>
 
 #include <pipewire/pipewire.h>
 #include <pipewire/keys.h>
 #include <pipewire/properties.h>
+#include <pipewire/stream.h>
 #include <spa/param/props.h>
+#include <spa/param/audio/format-utils.h>
+#include <spa/param/audio/raw.h>
 #include <spa/utils/dict.h>
 #include <spa/utils/type-info.h>
 
@@ -110,6 +114,51 @@ static const struct pw_core_events core_events = []() {
     return events;
 }();
 
+void meterProcess(void *data)
+{
+    auto *self = static_cast<PipeWireController*>(data);
+    if (!self || !self->m_meterStream) {
+        return;
+    }
+
+    struct pw_buffer *buf = pw_stream_dequeue_buffer(self->m_meterStream);
+    if (!buf || !buf->buffer || buf->buffer->n_datas == 0) {
+        if (buf) {
+            pw_stream_queue_buffer(self->m_meterStream, buf);
+        }
+        return;
+    }
+
+    struct spa_buffer *spaBuf = buf->buffer;
+    struct spa_data *data0 = &spaBuf->datas[0];
+    if (!data0->data || !data0->chunk) {
+        pw_stream_queue_buffer(self->m_meterStream, buf);
+        return;
+    }
+
+    const uint32_t size = data0->chunk->size;
+    const float *samples = static_cast<const float*>(data0->data);
+    const uint32_t count = size / sizeof(float);
+
+    float peak = 0.0f;
+    for (uint32_t i = 0; i < count; ++i) {
+        const float value = std::fabs(samples[i]);
+        if (value > peak) {
+            peak = value;
+        }
+    }
+
+    self->m_meterPeak.store(peak, std::memory_order_relaxed);
+    pw_stream_queue_buffer(self->m_meterStream, buf);
+}
+
+static const struct pw_stream_events meter_events = []() {
+    struct pw_stream_events events{};
+    events.version = PW_VERSION_STREAM_EVENTS;
+    events.process = meterProcess;
+    return events;
+}();
+
 PipeWireController::PipeWireController(QObject *parent)
     : QObject(parent)
 {
@@ -169,6 +218,10 @@ bool PipeWireController::initialize()
     }
     
     pw_registry_add_listener(m_registry, m_registryListener, &registry_events, this);
+
+    if (!setupMeterStream()) {
+        qWarning() << "Failed to set up meter stream";
+    }
     
     unlock();
     
@@ -201,6 +254,8 @@ void PipeWireController::shutdown()
         pw_proxy_destroy(reinterpret_cast<struct pw_proxy*>(m_registry));
         m_registry = nullptr;
     }
+
+    teardownMeterStream();
     
     if (m_core) {
         pw_core_disconnect(m_core);
@@ -250,6 +305,11 @@ QVector<LinkInfo> PipeWireController::links() const
     return m_links.values().toVector();
 }
 
+float PipeWireController::meterPeak() const
+{
+    return m_meterPeak.load(std::memory_order_relaxed);
+}
+
 uint32_t PipeWireController::createLink(uint32_t outputNodeId, uint32_t outputPortId,
                                         uint32_t inputNodeId, uint32_t inputPortId)
 {
@@ -421,6 +481,22 @@ void PipeWireController::handleNodeInfo(uint32_t id, const struct spa_dict *prop
     
     node.mediaClass = NodeInfo::mediaClassFromString(mediaClassStr);
     node.type = NodeInfo::typeFromProperties(mediaClassStr, appNameStr);
+
+    {
+        QMutexLocker lock(&m_nodesMutex);
+        for (auto it = m_ports.cbegin(); it != m_ports.cend(); ++it) {
+            const PortInfo &port = it.value();
+            if (port.nodeId != id) {
+                continue;
+            }
+
+            if (port.direction == PW_DIRECTION_INPUT) {
+                node.inputPorts.append(port);
+            } else if (port.direction == PW_DIRECTION_OUTPUT) {
+                node.outputPorts.append(port);
+            }
+        }
+    }
     
     {
         QMutexLocker lock(&m_nodesMutex);
@@ -460,22 +536,23 @@ void PipeWireController::handlePortInfo(uint32_t id, const struct spa_dict *prop
     QString portName = name ? toQString(name)
         : QString("port_") + QString::number(id);
     
-    PortInfo port(id, portName, direction);
+    uint32_t nodeId = nodeIdStr ? static_cast<uint32_t>(atoi(nodeIdStr)) : 0;
+    PortInfo port(id, nodeId, portName, direction);
     
     {
         QMutexLocker lock(&m_nodesMutex);
         m_ports.insert(id, port);
         
-        if (nodeIdStr) {
-            uint32_t nodeId = static_cast<uint32_t>(atoi(nodeIdStr));
-            if (m_nodes.contains(nodeId)) {
-                NodeInfo &node = m_nodes[nodeId];
-                if (direction == PW_DIRECTION_INPUT) {
-                    node.inputPorts.append(port);
-                } else if (direction == PW_DIRECTION_OUTPUT) {
-                    node.outputPorts.append(port);
+        if (nodeId != 0 && m_nodes.contains(nodeId)) {
+            NodeInfo &node = m_nodes[nodeId];
+            auto &ports = (direction == PW_DIRECTION_INPUT) ? node.inputPorts : node.outputPorts;
+            for (int i = 0; i < ports.size(); ++i) {
+                if (ports.at(i).id == id) {
+                    ports.removeAt(i);
+                    break;
                 }
             }
+            ports.append(port);
         }
     }
     
@@ -525,4 +602,62 @@ void PipeWireController::unlock()
     }
 }
 
+bool PipeWireController::setupMeterStream()
+{
+    if (!m_threadLoop || !m_core) {
+        return false;
+    }
+
+    struct pw_properties *props = pw_properties_new(
+        PW_KEY_MEDIA_TYPE, "Audio",
+        PW_KEY_MEDIA_CATEGORY, "Capture",
+        PW_KEY_MEDIA_CLASS, "Audio/Source",
+        nullptr);
+
+    m_meterStream = pw_stream_new_simple(
+        pw_thread_loop_get_loop(m_threadLoop),
+        "Potato-Meter",
+        props,
+        &meter_events,
+        this);
+
+    if (!m_meterStream) {
+        pw_properties_free(props);
+        return false;
+    }
+
+    uint8_t buffer[512];
+    spa_pod_builder builder = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer));
+
+    spa_audio_info_raw info{};
+    info.format = SPA_AUDIO_FORMAT_F32;
+    info.rate = 48000;
+    info.channels = 2;
+    info.position[0] = SPA_AUDIO_CHANNEL_FL;
+    info.position[1] = SPA_AUDIO_CHANNEL_FR;
+
+    const struct spa_pod *params[1];
+    params[0] = spa_format_audio_raw_build(&builder, SPA_PARAM_EnumFormat, &info);
+
+    const int res = pw_stream_connect(
+        m_meterStream,
+        PW_DIRECTION_INPUT,
+        PW_ID_ANY,
+        static_cast<pw_stream_flags>(PW_STREAM_FLAG_AUTOCONNECT | PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_RT_PROCESS),
+        params,
+        1);
+
+    return res == 0;
+}
+
+void PipeWireController::teardownMeterStream()
+{
+    if (!m_meterStream) {
+        return;
+    }
+
+    pw_stream_destroy(m_meterStream);
+    m_meterStream = nullptr;
+}
+
 } // namespace Potato