plugin/ap: fix collector

Actually read from speakers not mic

modules/background/Visualiser.qml

@@ -4,6 +4,7 @@ import qs.components
 import qs.components.misc
 import qs.services
 import qs.config
+import Caelestia
 import Quickshell
 import Quickshell.Widgets
 import QtQuick
@@ -15,8 +16,8 @@ Item {
     required property ShellScreen screen
     required property Wallpaper wallpaper
 
-    Ref {
+    ServiceRef {
-        service: Cava
+        service: Cava.provider
     }
 
     MultiEffect {
@@ -65,7 +66,7 @@ Item {
             id: bar
 
             required property int modelData
-            property real value: Math.max(1, Math.min(100, Cava.values[side.isRight ? modelData : side.count - modelData - 1])) / 100
+            property real value: Math.max(0, Math.min(1, Cava.values[side.isRight ? modelData : side.count - modelData - 1]))
 
             clip: true
 

modules/dashboard/Media.qml

@@ -55,8 +55,8 @@ Item {
         onTriggered: Players.active?.positionChanged()
     }
 
-    Ref {
+    ServiceRef {
-        service: Cava
+        service: Cava.provider
     }
 
     ServiceRef {
@@ -91,10 +91,10 @@ Item {
             id: visualiserBar
 
             required property int modelData
-            readonly property int value: Math.max(1, Math.min(100, Cava.values[modelData]))
+            readonly property real value: Math.max(0, Math.min(1, Cava.values[modelData]))
 
             readonly property real angle: modelData * 2 * Math.PI / Config.services.visualiserBars
-            readonly property real magnitude: value / 100 * Config.dashboard.sizes.mediaVisualiserSize
+            readonly property real magnitude: value * Config.dashboard.sizes.mediaVisualiserSize
             readonly property real cos: Math.cos(angle)
             readonly property real sin: Math.sin(angle)
 

nix/default.nix

@@ -23,6 +23,7 @@
   quickshell,
   aubio,
   libcava,
+  libpipewire,
   xkeyboard-config,
   cmake,
   ninja,
@@ -87,7 +88,7 @@
     };
 
     nativeBuildInputs = [cmake ninja pkg-config];
-    buildInputs = [qt6.qtbase qt6.qtdeclarative qt6.qtmultimedia libqalculate aubio libcava];
+    buildInputs = [qt6.qtbase qt6.qtdeclarative qt6.qtmultimedia libqalculate libpipewire aubio libcava];
 
     dontWrapQtApps = true;
     cmakeFlags =

plugin/src/Caelestia/CMakeLists.txt

@@ -1,5 +1,6 @@
 find_package(PkgConfig REQUIRED)
 pkg_check_modules(Qalculate IMPORTED_TARGET libqalculate REQUIRED)
+pkg_check_modules(Pipewire IMPORTED_TARGET libpipewire-0.3 REQUIRED)
 pkg_check_modules(Aubio IMPORTED_TARGET aubio REQUIRED)
 pkg_check_modules(Cava IMPORTED_TARGET cava REQUIRED)
 
@@ -14,6 +15,7 @@ qt_add_qml_module(caelestia
         beattracker.hpp beattracker.cpp
         service.hpp service.cpp
         serviceref.hpp serviceref.cpp
+        audiocollector.hpp audiocollector.cpp
         audioprovider.hpp audioprovider.cpp
         cavaprovider.hpp cavaprovider.cpp
 )
@@ -37,5 +39,5 @@ install(FILES "${module_typeinfo}" DESTINATION "${module_dir}")
 
 target_link_libraries(caelestia PRIVATE
     Qt::Core Qt::Qml Qt::Gui Qt::Concurrent Qt::Multimedia
-    PkgConfig::Qalculate PkgConfig::Aubio PkgConfig::Cava
+    PkgConfig::Qalculate PkgConfig::Pipewire PkgConfig::Aubio PkgConfig::Cava
 )

plugin/src/Caelestia/audiocollector.cpp

@@ -0,0 +1,248 @@
+#include "audiocollector.hpp"
+#include "service.hpp"
+
+#include <QDebug>
+#include <QVector>
+#include <algorithm>
+#include <cstdint>
+#include <mutex>
+#include <pipewire/pipewire.h>
+#include <qmutex.h>
+#include <spa/param/audio/format-utils.h>
+#include <spa/param/latency-utils.h>
+#include <stop_token>
+#include <vector>
+
+namespace caelestia {
+
+PipeWireWorker::PipeWireWorker(std::stop_token token, AudioCollector* collector)
+    : m_loop(nullptr)
+    , m_stream(nullptr)
+    , m_timer(nullptr)
+    , m_token(token)
+    , m_collector(collector) {
+    pw_init(nullptr, nullptr);
+
+    m_loop = pw_main_loop_new(nullptr);
+    if (!m_loop) {
+        qWarning() << "PipeWireWorker::init: failed to create PipeWire main loop";
+        return;
+    }
+
+    timespec timeout = { 0, 10 * SPA_NSEC_PER_MSEC };
+    m_timer = pw_loop_add_timer(pw_main_loop_get_loop(m_loop), handleTimeout, this);
+    pw_loop_update_timer(pw_main_loop_get_loop(m_loop), m_timer, &timeout, &timeout, false);
+
+    auto props = pw_properties_new(
+        PW_KEY_MEDIA_TYPE, "Audio", PW_KEY_MEDIA_CATEGORY, "Capture", PW_KEY_MEDIA_ROLE, "Music", nullptr);
+    pw_properties_set(props, PW_KEY_STREAM_CAPTURE_SINK, "true");
+    pw_properties_setf(props, PW_KEY_NODE_LATENCY, "%u/%u", nextPowerOf2(512 * collector->sampleRate() / 48000),
+        collector->sampleRate());
+    pw_properties_set(props, PW_KEY_NODE_PASSIVE, "true");
+    pw_properties_set(props, PW_KEY_NODE_VIRTUAL, "true");
+    pw_properties_set(props, PW_KEY_STREAM_DONT_REMIX, "false");
+    pw_properties_set(props, "channelmix.upmix", "true");
+
+    std::vector<uint8_t> buffer(collector->chunkSize());
+    spa_pod_builder b;
+    spa_pod_builder_init(&b, buffer.data(), static_cast<uint32_t>(buffer.size()));
+
+    spa_audio_info_raw info{};
+    info.format = SPA_AUDIO_FORMAT_S16;
+    info.rate = collector->sampleRate();
+    info.channels = 1;
+
+    const spa_pod* params[1];
+    params[0] = spa_format_audio_raw_build(&b, SPA_PARAM_EnumFormat, &info);
+
+    pw_stream_events events{};
+    events.process = [](void* data) {
+        auto* self = static_cast<PipeWireWorker*>(data);
+        self->processStream();
+    };
+
+    m_stream = pw_stream_new_simple(pw_main_loop_get_loop(m_loop), "caelestia-shell", props, &events, this);
+
+    pw_stream_connect(m_stream, PW_DIRECTION_INPUT, PW_ID_ANY,
+        static_cast<pw_stream_flags>(PW_STREAM_FLAG_AUTOCONNECT | PW_STREAM_FLAG_MAP_BUFFERS), params, 1);
+
+    pw_main_loop_run(m_loop);
+
+    pw_stream_destroy(m_stream);
+    pw_main_loop_destroy(m_loop);
+    pw_deinit();
+}
+
+void PipeWireWorker::handleTimeout(void* data, uint64_t expirations) {
+    Q_UNUSED(expirations);
+    auto* self = static_cast<PipeWireWorker*>(data);
+
+    if (self->m_token.stop_requested()) {
+        pw_main_loop_quit(self->m_loop);
+    }
+}
+
+void PipeWireWorker::processStream() {
+    if (m_token.stop_requested()) {
+        pw_main_loop_quit(m_loop);
+        return;
+    }
+
+    pw_buffer* buffer = pw_stream_dequeue_buffer(m_stream);
+    if (buffer == nullptr) {
+        return;
+    }
+
+    const spa_buffer* buf = buffer->buffer;
+    const int16_t* samples = reinterpret_cast<const int16_t*>(buf->datas[0].data);
+    if (samples == nullptr) {
+        return;
+    }
+
+    const uint32_t count = buf->datas[0].chunk->size / 2;
+    m_collector->loadChunk(samples, count);
+
+    pw_stream_queue_buffer(m_stream, buffer);
+}
+
+unsigned int PipeWireWorker::nextPowerOf2(unsigned int n) {
+    if (n == 0) {
+        return 1;
+    }
+
+    n--;
+    n |= n >> 1;
+    n |= n >> 2;
+    n |= n >> 4;
+    n |= n >> 8;
+    n |= n >> 16;
+    n++;
+
+    return n;
+}
+
+AudioCollector::AudioCollector(uint32_t sampleRate, uint32_t chunkSize, uint32_t bufferSize, QObject* parent)
+    : Service(parent)
+    , m_buffer(bufferSize)
+    , m_bufferIndex(0)
+    , m_bufferFull(false)
+    , m_sampleRate(sampleRate)
+    , m_chunkSize(chunkSize)
+    , m_bufferSize(bufferSize) {}
+
+AudioCollector::~AudioCollector() {
+    stop();
+}
+
+AudioCollector* AudioCollector::instance() {
+    std::lock_guard<std::mutex> lock(s_mutex);
+    if (s_instance == nullptr) {
+        s_instance = new AudioCollector();
+    }
+    return s_instance;
+}
+
+uint32_t AudioCollector::sampleRate() const {
+    return m_sampleRate;
+}
+
+uint32_t AudioCollector::chunkSize() const {
+    return m_chunkSize;
+}
+
+uint32_t AudioCollector::bufferSize() const {
+    return m_bufferSize;
+}
+
+void AudioCollector::loadChunk(const int16_t* samples, uint32_t count) {
+    std::lock_guard<std::mutex> lock(m_bufferMutex);
+
+    while (count > 0) {
+        const auto spaceToEnd = m_bufferSize - m_bufferIndex;
+        const auto toCopy = (count < spaceToEnd) ? count : spaceToEnd;
+
+        std::transform(samples, samples + toCopy, m_buffer.begin() + m_bufferIndex, [](int16_t sample) {
+            return sample / 32768.0f;
+        });
+
+        m_bufferIndex = (m_bufferIndex + toCopy) % m_bufferSize;
+        if (m_bufferIndex == 0) {
+            m_bufferFull = true;
+        }
+
+        samples += toCopy;
+        count -= toCopy;
+    }
+}
+
+uint32_t AudioCollector::readChunk(float* out, uint32_t count) {
+    std::lock_guard<std::mutex> lock(m_bufferMutex);
+
+    const auto available = m_bufferFull ? m_bufferSize : m_bufferIndex;
+    if (count == 0 || count > available) {
+        count = available;
+    }
+
+    const auto start = m_bufferFull ? (m_bufferIndex + m_bufferSize - count) % m_bufferSize
+                                    : (m_bufferIndex >= count ? m_bufferIndex - count : 0);
+    const auto firstChunk = std::min(count, m_bufferSize - start);
+
+    std::copy(m_buffer.begin() + start, m_buffer.begin() + start + firstChunk, out);
+
+    if (firstChunk < count) {
+        std::copy(m_buffer.begin(), m_buffer.begin() + (count - firstChunk), out + firstChunk);
+    }
+
+    return count;
+}
+
+uint32_t AudioCollector::readChunk(double* out, uint32_t count) {
+    std::lock_guard<std::mutex> lock(m_bufferMutex);
+
+    const auto available = m_bufferFull ? m_bufferSize : m_bufferIndex;
+    if (count == 0 || count > available) {
+        count = available;
+    }
+
+    const auto start = m_bufferFull ? (m_bufferIndex + m_bufferSize - count) % m_bufferSize
+                                    : (m_bufferIndex >= count ? m_bufferIndex - count : 0);
+    const auto firstChunk = std::min(count, m_bufferSize - start);
+
+    std::transform(m_buffer.begin() + start, m_buffer.begin() + start + firstChunk, out, [](float sample) {
+        return static_cast<double>(sample);
+    });
+
+    if (firstChunk < count) {
+        std::transform(m_buffer.begin(), m_buffer.begin() + (count - firstChunk), out + firstChunk, [](float sample) {
+            return static_cast<double>(sample);
+        });
+    }
+
+    return count;
+}
+
+void AudioCollector::start() {
+    if (m_thread.joinable()) {
+        return;
+    }
+
+    {
+        std::lock_guard<std::mutex> lock(m_bufferMutex);
+        m_buffer.clear();
+        m_bufferIndex = 0;
+        m_bufferFull = false;
+    }
+
+    m_thread = std::jthread([this](std::stop_token token) {
+        PipeWireWorker worker(token, this);
+    });
+}
+
+void AudioCollector::stop() {
+    if (m_thread.joinable()) {
+        m_thread.request_stop();
+        m_thread.join();
+    }
+}
+
+} // namespace caelestia

plugin/src/Caelestia/audiocollector.hpp

@@ -0,0 +1,76 @@
+#pragma once
+
+#include "service.hpp"
+#include <QObject>
+#include <cstdint>
+#include <mutex>
+#include <pipewire/pipewire.h>
+#include <spa/param/audio/format-utils.h>
+#include <stop_token>
+#include <thread>
+#include <vector>
+
+namespace caelestia {
+
+class AudioCollector;
+
+class PipeWireWorker {
+public:
+    explicit PipeWireWorker(std::stop_token token, AudioCollector* collector);
+
+    void run();
+
+private:
+    pw_main_loop* m_loop;
+    pw_stream* m_stream;
+    spa_source* m_timer;
+
+    std::stop_token m_token;
+    AudioCollector* m_collector;
+
+    void cleanup();
+
+    static void handleTimeout(void* data, uint64_t expirations);
+    void processStream();
+    void processSamples(const int16_t* samples, uint32_t count);
+
+    [[nodiscard]] unsigned int nextPowerOf2(unsigned int n);
+};
+
+class AudioCollector : public Service {
+    Q_OBJECT
+
+public:
+    explicit AudioCollector(
+        uint32_t sampleRate = 44100, uint32_t chunkSize = 512, uint32_t bufferSize = 1024, QObject* parent = nullptr);
+    ~AudioCollector();
+
+    static AudioCollector* instance();
+
+    [[nodiscard]] uint32_t sampleRate() const;
+    [[nodiscard]] uint32_t chunkSize() const;
+    [[nodiscard]] uint32_t bufferSize() const;
+
+    void loadChunk(const int16_t* samples, uint32_t count);
+    uint32_t readChunk(float* out, uint32_t count = 0);
+    uint32_t readChunk(double* out, uint32_t count = 0);
+
+private:
+    inline static AudioCollector* s_instance = nullptr;
+    inline static std::mutex s_mutex;
+
+    std::jthread m_thread;
+    std::vector<float> m_buffer;
+    uint32_t m_bufferIndex;
+    bool m_bufferFull;
+    std::mutex m_bufferMutex;
+
+    const uint32_t m_sampleRate;
+    const uint32_t m_chunkSize;
+    const uint32_t m_bufferSize;
+
+    void start() override;
+    void stop() override;
+};
+
+} // namespace caelestia

plugin/src/Caelestia/audioprovider.cpp

@@ -1,5 +1,6 @@
 #include "audioprovider.hpp"
 
+#include "audiocollector.hpp"
 #include "service.hpp"
 #include <QAudioSource>
 #include <QDebug>
@@ -9,121 +10,27 @@
 #include <QObject>
 #include <QThread>
 #include <QVector>
-#include <algorithm>
-#include <cstddef>
-#include <cstdint>
-#include <cstdlib>
 
 namespace caelestia {
 
-AudioCollector::AudioCollector(AudioProvider* provider, QObject* parent)
+AudioProcessor::AudioProcessor(QObject* parent)
     : QObject(parent)
-    , m_source(nullptr)
+    , m_sampleRate(AudioCollector::instance()->sampleRate())
-    , m_device(nullptr)
+    , m_chunkSize(AudioCollector::instance()->chunkSize())
-    , m_provider(provider)
+    , m_bufferSize(AudioCollector::instance()->bufferSize()) {}
-    , m_sampleRate(provider->sampleRate())
-    , m_chunkSize(provider->chunkSize())
-    , m_chunk(m_chunkSize)
-    , m_chunkOffset(0) {}
-
-AudioCollector::~AudioCollector() {
-    m_source->stop();
-}
-
-void AudioCollector::init() {
-    QAudioFormat format;
-    format.setSampleRate(m_sampleRate);
-    format.setChannelCount(1);
-    format.setSampleFormat(QAudioFormat::Int16);
-
-    m_source = new QAudioSource(QMediaDevices::defaultAudioInput(), format, this);
-    connect(m_source, &QAudioSource::stateChanged, this, &AudioCollector::handleStateChanged);
-};
-
-void AudioCollector::start() {
-    if (!m_source) {
-        return;
-    }
-
-    m_device = m_source->start();
-    connect(m_device, &QIODevice::readyRead, this, &AudioCollector::loadChunk);
-}
-
-void AudioCollector::stop() {
-    if (m_source) {
-        m_source->stop();
-        m_device = nullptr;
-    }
-}
-
-void AudioCollector::loadChunk() {
-    const QByteArray data = m_device->readAll();
-    const int16_t* samples = reinterpret_cast<const int16_t*>(data.constData());
-    const size_t count = static_cast<size_t>(data.size()) / sizeof(int16_t);
-
-    size_t i = 0;
-    while (i < count) {
-        const int spaceLeft = m_chunkSize - m_chunkOffset;
-        const auto toCopy = std::min<size_t>(static_cast<size_t>(spaceLeft), count - i);
-
-        std::transform(samples + i, samples + i + toCopy, m_chunk.begin() + m_chunkOffset, [](int16_t sample) {
-            return sample / 32768.0;
-        });
-
-        m_chunkOffset += toCopy;
-        i += toCopy;
-
-        if (m_chunkOffset == m_chunkSize) {
-            if (std::any_of(m_chunk.constBegin(), m_chunk.constEnd(), [](double d) {
-                    return std::abs(d) > 1e-6;
-                })) {
-                m_provider->loadChunk(m_chunk);
-            }
-            m_chunkOffset = 0;
-        }
-    }
-}
-
-void AudioCollector::handleStateChanged(QtAudio::State state) const {
-    if (state == QtAudio::StoppedState && m_source->error() != QtAudio::NoError) {
-        switch (m_source->error()) {
-        case QtAudio::OpenError:
-            qWarning() << "AudioWorker: failed to open audio device";
-            break;
-        case QtAudio::IOError:
-            qWarning() << "AudioWorker: an error occurred during read/write of audio device";
-            break;
-        case QtAudio::UnderrunError:
-            qWarning() << "AudioWorker: audio data is not being fed to audio device fast enough";
-            break;
-        case QtAudio::FatalError:
-            qCritical() << "AudioWorker: fatal error in audio device";
-            break;
-        default:
-            break;
-        }
-    }
-}
-
-AudioProcessor::AudioProcessor(AudioProvider* provider, QObject* parent)
-    : QObject(parent)
-    , m_sampleRate(provider->sampleRate())
-    , m_chunkSize(provider->chunkSize())
-    , m_provider(provider) {}
 
 AudioProcessor::~AudioProcessor() {
-    if (m_timer) {
+    stop();
-        m_timer->stop();
-    }
 }
 
 void AudioProcessor::init() {
     m_timer = new QTimer(this);
     m_timer->setInterval(static_cast<int>(m_chunkSize * 1000.0 / m_sampleRate));
-    connect(m_timer, &QTimer::timeout, this, &AudioProcessor::handleTimeout);
+    connect(m_timer, &QTimer::timeout, this, &AudioProcessor::process);
 }
 
 void AudioProcessor::start() {
+    AudioCollector::instance()->ref();
     if (m_timer) {
         m_timer->start();
     }
@@ -133,60 +40,19 @@ void AudioProcessor::stop() {
     if (m_timer) {
         m_timer->stop();
     }
+    AudioCollector::instance()->unref();
 }
 
-void AudioProcessor::handleTimeout() {
+AudioProvider::AudioProvider(QObject* parent)
-    const QVector<double> chunk = m_provider->nextChunk();
-    if (!chunk.isEmpty()) {
-        processChunk(chunk);
-    }
-}
-
-AudioProvider::AudioProvider(int sampleRate, int chunkSize, QObject* parent)
     : Service(parent)
-    , m_sampleRate(sampleRate)
-    , m_chunkSize(chunkSize)
-    , m_collector(new AudioCollector(this))
     , m_processor(nullptr)
-    , m_collectorThread(new QThread(this))
+    , m_thread(nullptr) {}
-    , m_processorThread(nullptr) {
-    m_collector->moveToThread(m_collectorThread);
-
-    connect(m_collectorThread, &QThread::started, m_collector, &AudioCollector::init);
-    connect(m_collectorThread, &QThread::finished, m_collector, &AudioCollector::deleteLater);
-    connect(m_collectorThread, &QThread::finished, m_collectorThread, &QThread::deleteLater);
-
-    m_collectorThread->start();
-}
 
 AudioProvider::~AudioProvider() {
-    m_collectorThread->quit();
+    if (m_thread) {
-    if (m_processorThread) {
+        m_thread->quit();
-        m_processorThread->quit();
+        m_thread->wait();
-        m_processorThread->wait();
     }
-    m_collectorThread->wait();
-}
-
-int AudioProvider::sampleRate() const {
-    return m_sampleRate;
-}
-
-int AudioProvider::chunkSize() const {
-    return m_chunkSize;
-}
-
-QVector<double> AudioProvider::nextChunk() {
-    QMutexLocker lock(&m_mutex);
-    if (m_chunks.isEmpty()) {
-        return {};
-    }
-    return m_chunks.dequeue();
-}
-
-void AudioProvider::loadChunk(const QVector<double>& chunk) {
-    QMutexLocker lock(&m_mutex);
-    m_chunks.enqueue(chunk);
 }
 
 void AudioProvider::init() {
@@ -195,25 +61,23 @@ void AudioProvider::init() {
         return;
     }
 
-    m_processorThread = new QThread(this);
+    m_thread = new QThread(this);
-    m_processor->moveToThread(m_processorThread);
+    m_processor->moveToThread(m_thread);
 
-    connect(m_processorThread, &QThread::started, m_processor, &AudioProcessor::init);
+    connect(m_thread, &QThread::started, m_processor, &AudioProcessor::init);
-    connect(m_processorThread, &QThread::finished, m_processor, &AudioProcessor::deleteLater);
+    connect(m_thread, &QThread::finished, m_processor, &AudioProcessor::deleteLater);
-    connect(m_processorThread, &QThread::finished, m_processorThread, &QThread::deleteLater);
+    connect(m_thread, &QThread::finished, m_thread, &QThread::deleteLater);
 
-    m_processorThread->start();
+    m_thread->start();
 }
 
 void AudioProvider::start() {
-    QMetaObject::invokeMethod(m_collector, "start", Qt::QueuedConnection);
     if (m_processor) {
         QMetaObject::invokeMethod(m_processor, "start", Qt::QueuedConnection);
     }
 }
 
 void AudioProvider::stop() {
-    QMetaObject::invokeMethod(m_collector, "stop", Qt::QueuedConnection);
     if (m_processor) {
         QMetaObject::invokeMethod(m_processor, "stop", Qt::QueuedConnection);
     }

plugin/src/Caelestia/audioprovider.hpp

@@ -9,91 +9,48 @@
 #include <QThread>
 #include <QTimer>
 #include <QVector>
+#include <cstdint>
 #include <qqmlintegration.h>
 
 namespace caelestia {
 
-class AudioProvider;
-
-class AudioCollector : public QObject {
-    Q_OBJECT
-
-public:
-    explicit AudioCollector(AudioProvider* provider, QObject* parent = nullptr);
-    ~AudioCollector();
-
-    void init();
-
-private:
-    QAudioSource* m_source;
-    QIODevice* m_device;
-
-    AudioProvider* m_provider;
-    int m_sampleRate;
-    int m_chunkSize;
-
-    QVector<double> m_chunk;
-    int m_chunkOffset;
-
-    Q_INVOKABLE void start();
-    Q_INVOKABLE void stop();
-
-    void handleStateChanged(QtAudio::State state) const;
-    void loadChunk();
-};
-
 class AudioProcessor : public QObject {
     Q_OBJECT
 
 public:
-    explicit AudioProcessor(AudioProvider* provider, QObject* parent = nullptr);
+    explicit AudioProcessor(QObject* parent = nullptr);
     ~AudioProcessor();
 
     void init();
 
 protected:
-    int m_sampleRate;
+    uint32_t m_sampleRate;
-    int m_chunkSize;
+    uint32_t m_chunkSize;
+    uint32_t m_bufferSize;
 
 private:
-    AudioProvider* m_provider;
     QTimer* m_timer;
 
     Q_INVOKABLE void start();
     Q_INVOKABLE void stop();
 
-    void handleTimeout();
+    virtual void process() = 0;
-    virtual void processChunk(const QVector<double>& chunk) = 0;
 };
 
 class AudioProvider : public Service {
     Q_OBJECT
 
 public:
-    explicit AudioProvider(int sampleRate = 44100, int chunkSize = 512, QObject* parent = nullptr);
+    explicit AudioProvider(QObject* parent = nullptr);
     ~AudioProvider();
 
-    [[nodiscard]] int sampleRate() const;
-    [[nodiscard]] int chunkSize() const;
-
-    [[nodiscard]] QVector<double> nextChunk();
-    void loadChunk(const QVector<double>& chunk);
-
 protected:
-    int m_sampleRate;
-    int m_chunkSize;
-
-    QMutex m_mutex;
-    QQueue<QVector<double>> m_chunks;
-
-    AudioCollector* m_collector;
     AudioProcessor* m_processor;
 
     void init();
 
 private:
-    QThread* m_collectorThread;
+    QThread* m_thread;
-    QThread* m_processorThread;
 
     void start() override;
     void stop() override;

plugin/src/Caelestia/beattracker.cpp

@@ -1,15 +1,16 @@
 #include "beattracker.hpp"
 
+#include "audiocollector.hpp"
 #include "audioprovider.hpp"
 #include <QObject>
 #include <aubio/aubio.h>
 
 namespace caelestia {
 
-BeatProcessor::BeatProcessor(AudioProvider* provider, QObject* parent)
+BeatProcessor::BeatProcessor(QObject* parent)
-    : AudioProcessor(provider, parent)
+    : AudioProcessor(parent)
-    , m_tempo(new_aubio_tempo("default", 1024, static_cast<uint_t>(m_chunkSize), static_cast<uint_t>(m_sampleRate)))
+    , m_tempo(new_aubio_tempo("default", 1024, m_chunkSize, m_sampleRate))
-    , m_in(new_fvec(static_cast<uint_t>(m_chunkSize)))
+    , m_in(new_fvec(m_chunkSize))
     , m_out(new_fvec(2)) {};
 
 BeatProcessor::~BeatProcessor() {
@@ -18,10 +19,8 @@ BeatProcessor::~BeatProcessor() {
     del_fvec(m_out);
 }
 
-void BeatProcessor::processChunk(const QVector<double>& chunk) {
+void BeatProcessor::process() {
-    std::transform(chunk.constBegin(), chunk.constEnd(), m_in->data, [](double d) {
+    AudioCollector::instance()->readChunk(m_in->data, m_chunkSize);
-        return static_cast<float>(d);
-    });
 
     aubio_tempo_do(m_tempo, m_in, m_out);
     if (m_out->data[0] != 0.0f) {
@@ -29,10 +28,10 @@ void BeatProcessor::processChunk(const QVector<double>& chunk) {
     }
 }
 
-BeatTracker::BeatTracker(int sampleRate, int chunkSize, QObject* parent)
+BeatTracker::BeatTracker(QObject* parent)
-    : AudioProvider(sampleRate, chunkSize, parent)
+    : AudioProvider(parent)
     , m_bpm(120) {
-    m_processor = new BeatProcessor(this);
+    m_processor = new BeatProcessor();
     init();
 
     connect(static_cast<BeatProcessor*>(m_processor), &BeatProcessor::beat, this, &BeatTracker::updateBpm);

plugin/src/Caelestia/beattracker.hpp

@@ -11,7 +11,7 @@ class BeatProcessor : public AudioProcessor {
     Q_OBJECT
 
 public:
-    explicit BeatProcessor(AudioProvider* provider, QObject* parent = nullptr);
+    explicit BeatProcessor(QObject* parent = nullptr);
     ~BeatProcessor();
 
 signals:
@@ -22,7 +22,7 @@ private:
     fvec_t* m_in;
     fvec_t* m_out;
 
-    void processChunk(const QVector<double>& chunk) override;
+    void process() override;
 };
 
 class BeatTracker : public AudioProvider {
@@ -33,7 +33,7 @@ class BeatTracker : public AudioProvider {
     Q_PROPERTY(smpl_t bpm READ bpm NOTIFY bpmChanged)
 
 public:
-    explicit BeatTracker(int sampleRate = 44100, int chunkSize = 512, QObject* parent = nullptr);
+    explicit BeatTracker(QObject* parent = nullptr);
 
     [[nodiscard]] smpl_t bpm() const;
 

plugin/src/Caelestia/cavaprovider.cpp

@@ -1,5 +1,6 @@
 #include "cavaprovider.hpp"
 
+#include "audiocollector.hpp"
 #include "audioprovider.hpp"
 #include <QDebug>
 #include <QObject>
@@ -9,8 +10,8 @@
 
 namespace caelestia {
 
-CavaProcessor::CavaProcessor(AudioProvider* provider, QObject* parent)
+CavaProcessor::CavaProcessor(QObject* parent)
-    : AudioProcessor(provider, parent)
+    : AudioProcessor(parent)
     , m_plan(nullptr)
     , m_in(new double[static_cast<size_t>(m_chunkSize)])
     , m_out(nullptr)
@@ -68,15 +69,15 @@ void CavaProcessor::initCava() {
     m_out = new double[static_cast<size_t>(m_bars)];
 }
 
-void CavaProcessor::processChunk(const QVector<double>& chunk) {
+void CavaProcessor::process() {
     if (!m_plan || m_bars == 0) {
         return;
     }
 
-    std::copy(chunk.constBegin(), chunk.constEnd(), m_in);
+    const int count = static_cast<int>(AudioCollector::instance()->readChunk(m_in));
 
     // Process in data via cava
-    cava_execute(m_in, m_chunkSize, m_out, m_plan);
+    cava_execute(m_in, count, m_out, m_plan);
 
     // Apply monstercat filter
     for (int i = 0; i < m_bars; i++) {
@@ -97,11 +98,11 @@ void CavaProcessor::processChunk(const QVector<double>& chunk) {
     }
 }
 
-CavaProvider::CavaProvider(int sampleRate, int chunkSize, QObject* parent)
+CavaProvider::CavaProvider(QObject* parent)
-    : AudioProvider(sampleRate, chunkSize, parent)
+    : AudioProvider(parent)
     , m_bars(0)
     , m_values(m_bars) {
-    m_processor = new CavaProcessor(this);
+    m_processor = new CavaProcessor();
     init();
 
     connect(static_cast<CavaProcessor*>(m_processor), &CavaProcessor::valuesChanged, this, &CavaProvider::updateValues);

plugin/src/Caelestia/cavaprovider.hpp

@@ -11,7 +11,7 @@ class CavaProcessor : public AudioProcessor {
     Q_OBJECT
 
 public:
-    explicit CavaProcessor(AudioProvider* provider, QObject* parent = nullptr);
+    explicit CavaProcessor(QObject* parent = nullptr);
     ~CavaProcessor();
 
 signals:
@@ -31,7 +31,7 @@ private:
     void initCava();
     void cleanup();
 
-    void processChunk(const QVector<double>& chunk) override;
+    void process() override;
 };
 
 class CavaProvider : public AudioProvider {
@@ -43,7 +43,7 @@ class CavaProvider : public AudioProvider {
     Q_PROPERTY(QVector<double> values READ values NOTIFY valuesChanged)
 
 public:
-    explicit CavaProvider(int sampleRate = 48000, int chunkSize = 512, QObject* parent = nullptr);
+    explicit CavaProvider(QObject* parent = nullptr);
 
     [[nodiscard]] int bars() const;
     void setBars(int bars);

plugin/src/Caelestia/service.cpp

@@ -1,6 +1,7 @@
 #include "service.hpp"
 
 #include <QDebug>
+#include <QMutexLocker>
 #include <QObject>
 
 namespace caelestia {
@@ -9,30 +10,46 @@ Service::Service(QObject* parent)
     : QObject(parent)
     , m_refCount(0) {}
 
-int Service::refCount() const {
+int Service::refCount() {
+    QMutexLocker locker(&m_mutex);
     return m_refCount;
 }
 
 void Service::ref() {
-    if (m_refCount == 0) {
+    bool needsStart = false;
-        start();
-    }
 
-    m_refCount++;
+    {
+        QMutexLocker locker(&m_mutex);
+        if (m_refCount == 0) {
+            needsStart = true;
+        }
+        m_refCount++;
+    }
     emit refCountChanged();
+
+    if (needsStart) {
+        QMetaObject::invokeMethod(this, &Service::start, Qt::QueuedConnection);
+    }
 }
 
 void Service::unref() {
-    if (m_refCount == 0) {
+    bool needsStop = false;
-        qWarning() << "ServiceRef::unref: attempted to unref service with no active refs";
-        return;
-    }
 
-    m_refCount--;
+    {
+        QMutexLocker locker(&m_mutex);
+        if (m_refCount == 0) {
+            qWarning() << "ServiceRef::unref: attempted to unref service with no active refs";
+            return;
+        }
+        m_refCount--;
+        if (m_refCount == 0) {
+            needsStop = true;
+        }
+    }
     emit refCountChanged();
 
-    if (m_refCount == 0) {
+    if (needsStop) {
-        stop();
+        QMetaObject::invokeMethod(this, &Service::stop, Qt::QueuedConnection);
     }
 }
 

plugin/src/Caelestia/service.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <QMutex>
 #include <QObject>
 
 namespace caelestia {
@@ -12,7 +13,7 @@ class Service : public QObject {
 public:
     explicit Service(QObject* parent = nullptr);
 
-    [[nodiscard]] int refCount() const;
+    [[nodiscard]] int refCount();
 
     void ref();
     void unref();
@@ -22,6 +23,7 @@ signals:
 
 private:
     int m_refCount;
+    QMutex m_mutex;
 
     virtual void start() = 0;
     virtual void stop() = 0;

services/Cava.qml

@@ -1,36 +1,18 @@
 pragma Singleton
 
 import qs.config
+import Caelestia
 import Quickshell
-import Quickshell.Io
-import QtQuick
 
 Singleton {
     id: root
 
-    property list<int> values: Array(Config.services.visualiserBars)
+    readonly property alias provider: provider
-    property int refCount
+    readonly property alias values: provider.values
 
-    Connections {
+    CavaProvider {
-        target: Config.services
+        id: provider
 
-        function onVisualiserBarsChanged() {
+        bars: Config.services.visualiserBars
-            root.values = Array(Config.services.visualiserBars);
-            cavaProc.running = false;
-            cavaProc.running = true;
-        }
-    }
-
-    Process {
-        id: cavaProc
-
-        running: true
-        command: ["sh", "-c", `printf '[general]\nframerate=60\nbars=${Config.services.visualiserBars}\nsleep_timer=3\n[output]\nchannels=mono\nmethod=raw\nraw_target=/dev/stdout\ndata_format=ascii\nascii_max_range=100\n[smoothing]\nnoise_reduction=85\nmonstercat=1\ngravity=120\n[eq]\n1=0.8\n2=0.9\n3=1\n4=1.1\n5=1.2' | cava -p /dev/stdin`]
-        stdout: SplitParser {
-            onRead: data => {
-                if (root.refCount)
-                    root.values = data.slice(0, -1).split(";").map(v => parseInt(v, 10));
-            }
-        }
     }
 }