13 files changed, 1094 insertions, 0 deletions
diff --git a/plugin/src/Caelestia/Services/CMakeLists.txt b/plugin/src/Caelestia/Services/CMakeLists.txt
new file mode 100644
index 0000000..8ce868b
--- /dev/null
+++ b/plugin/src/Caelestia/Services/CMakeLists.txt
@@ -0,0 +1,14 @@
+qml_module(caelestia-services
+    URI Caelestia.Services
+    SOURCES
+        service.hpp service.cpp
+        serviceref.hpp serviceref.cpp
+        beattracker.hpp beattracker.cpp
+        audiocollector.hpp audiocollector.cpp
+        audioprovider.hpp audioprovider.cpp
+        cavaprovider.hpp cavaprovider.cpp
+    LIBRARIES
+        PkgConfig::Pipewire
+        PkgConfig::Aubio
+        PkgConfig::Cava
+)
diff --git a/plugin/src/Caelestia/Services/audiocollector.cpp b/plugin/src/Caelestia/Services/audiocollector.cpp
new file mode 100644
index 0000000..9dc3871
--- /dev/null
+++ b/plugin/src/Caelestia/Services/audiocollector.cpp
@@ -0,0 +1,275 @@
+#include "audiocollector.hpp"
+
+#include "service.hpp"
+#include <algorithm>
+#include <pipewire/pipewire.h>
+#include <qdebug.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_idle(true)
+    , 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";
+        pw_deinit();
+        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<quint32>(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.state_changed = [](void* data, pw_stream_state, pw_stream_state state, const char*) {
+        auto* self = static_cast<PipeWireWorker*>(data);
+        self->streamStateChanged(state);
+    };
+    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);
+
+    const int success = pw_stream_connect(m_stream, PW_DIRECTION_INPUT, collector->nodeId(),
+        static_cast<pw_stream_flags>(
+            PW_STREAM_FLAG_AUTOCONNECT | PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_RT_PROCESS),
+        params, 1);
+    if (success < 0) {
+        qWarning() << "PipeWireWorker::init: failed to connect stream";
+        pw_stream_destroy(m_stream);
+        pw_main_loop_destroy(m_loop);
+        pw_deinit();
+        return;
+    }
+
+    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) {
+    auto* self = static_cast<PipeWireWorker*>(data);
+
+    if (self->m_token.stop_requested()) {
+        pw_main_loop_quit(self->m_loop);
+        return;
+    }
+
+    if (!self->m_idle) {
+        if (expirations < 10) {
+            self->m_collector->clearBuffer();
+        } else {
+            self->m_idle = true;
+            timespec timeout = { 0, 500 * SPA_NSEC_PER_MSEC };
+            pw_loop_update_timer(pw_main_loop_get_loop(self->m_loop), self->m_timer, &timeout, &timeout, false);
+        }
+    }
+}
+
+void PipeWireWorker::streamStateChanged(pw_stream_state state) {
+    m_idle = false;
+    switch (state) {
+    case PW_STREAM_STATE_PAUSED: {
+        timespec timeout = { 0, 10 * SPA_NSEC_PER_MSEC };
+        pw_loop_update_timer(pw_main_loop_get_loop(m_loop), m_timer, &timeout, &timeout, false);
+        break;
+    }
+    case PW_STREAM_STATE_STREAMING:
+        pw_loop_update_timer(pw_main_loop_get_loop(m_loop), m_timer, nullptr, nullptr, false);
+        break;
+    case PW_STREAM_STATE_ERROR:
+        pw_main_loop_quit(m_loop);
+        break;
+    default:
+        break;
+    }
+}
+
+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 qint16* samples = reinterpret_cast<const qint16*>(buf->datas[0].data);
+    if (samples == nullptr) {
+        return;
+    }
+
+    const quint32 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(QObject* parent)
+    : Service(parent)
+    , m_sampleRate(44100)
+    , m_chunkSize(512)
+    , m_nodeId(PW_ID_ANY)
+    , m_buffer1(m_chunkSize)
+    , m_buffer2(m_chunkSize)
+    , m_readBuffer(&m_buffer1)
+    , m_writeBuffer(&m_buffer2) {}
+
+AudioCollector::~AudioCollector() {
+    stop();
+}
+
+quint32 AudioCollector::sampleRate() const {
+    return m_sampleRate;
+}
+
+quint32 AudioCollector::chunkSize() const {
+    return m_chunkSize;
+}
+
+quint32 AudioCollector::nodeId() {
+    QMutexLocker locker(&m_nodeIdMutex);
+    return m_nodeId;
+}
+
+void AudioCollector::setNodeId(quint32 nodeId) {
+    {
+        QMutexLocker locker(&m_nodeIdMutex);
+
+        if (nodeId == m_nodeId) {
+            return;
+        }
+
+        m_nodeId = nodeId;
+    }
+    emit nodeIdChanged();
+
+    if (m_thread.joinable()) {
+        stop();
+        start();
+    }
+}
+
+void AudioCollector::clearBuffer() {
+    auto* writeBuffer = m_writeBuffer.load(std::memory_order_relaxed);
+    std::fill(writeBuffer->begin(), writeBuffer->end(), 0.0f);
+
+    auto* oldRead = m_readBuffer.exchange(writeBuffer, std::memory_order_acq_rel);
+    m_writeBuffer.store(oldRead, std::memory_order_release);
+}
+
+void AudioCollector::loadChunk(const qint16* samples, quint32 count) {
+    if (count > m_chunkSize) {
+        count = m_chunkSize;
+    }
+
+    auto* writeBuffer = m_writeBuffer.load(std::memory_order_relaxed);
+    std::transform(samples, samples + count, writeBuffer->begin(), [](qint16 sample) {
+        return sample / 32768.0f;
+    });
+
+    auto* oldRead = m_readBuffer.exchange(writeBuffer, std::memory_order_acq_rel);
+    m_writeBuffer.store(oldRead, std::memory_order_release);
+}
+
+quint32 AudioCollector::readChunk(float* out, quint32 count) {
+    if (count == 0 || count > m_chunkSize) {
+        count = m_chunkSize;
+    }
+
+    auto* readBuffer = m_readBuffer.load(std::memory_order_acquire);
+    std::memcpy(out, readBuffer->data(), count * sizeof(float));
+
+    return count;
+}
+
+quint32 AudioCollector::readChunk(double* out, quint32 count) {
+    if (count == 0 || count > m_chunkSize) {
+        count = m_chunkSize;
+    }
+
+    auto* readBuffer = m_readBuffer.load(std::memory_order_acquire);
+    std::transform(readBuffer->begin(), readBuffer->begin() + count, out, [](float sample) {
+        return static_cast<double>(sample);
+    });
+
+    return count;
+}
+
+void AudioCollector::start() {
+    if (m_thread.joinable()) {
+        return;
+    }
+
+    clearBuffer();
+
+    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
diff --git a/plugin/src/Caelestia/Services/audiocollector.hpp b/plugin/src/Caelestia/Services/audiocollector.hpp
new file mode 100644
index 0000000..74b0877
--- /dev/null
+++ b/plugin/src/Caelestia/Services/audiocollector.hpp
@@ -0,0 +1,86 @@
+#pragma once
+
+#include "service.hpp"
+#include <atomic>
+#include <pipewire/pipewire.h>
+#include <qmutex.h>
+#include <qqmlintegration.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;
+    bool m_idle;
+
+    std::stop_token m_token;
+    AudioCollector* m_collector;
+
+    void cleanup();
+
+    static void handleTimeout(void* data, uint64_t expirations);
+    void streamStateChanged(pw_stream_state state);
+    void processStream();
+    void processSamples(const qint16* samples, quint32 count);
+
+    [[nodiscard]] unsigned int nextPowerOf2(unsigned int n);
+};
+
+class AudioCollector : public Service {
+    Q_OBJECT
+    QML_ELEMENT
+
+    Q_PROPERTY(quint32 nodeId READ nodeId WRITE setNodeId NOTIFY nodeIdChanged)
+
+public:
+    explicit AudioCollector(QObject* parent = nullptr);
+    ~AudioCollector();
+
+    [[nodiscard]] quint32 sampleRate() const;
+    [[nodiscard]] quint32 chunkSize() const;
+
+    [[nodiscard]] quint32 nodeId();
+    void setNodeId(quint32 nodeId);
+
+    void clearBuffer();
+    void loadChunk(const qint16* samples, quint32 count);
+    quint32 readChunk(float* out, quint32 count = 0);
+    quint32 readChunk(double* out, quint32 count = 0);
+
+signals:
+    void sampleRateChanged();
+    void chunkSizeChanged();
+    void nodeIdChanged();
+
+private:
+    const quint32 m_sampleRate;
+    const quint32 m_chunkSize;
+    quint32 m_nodeId;
+    QMutex m_nodeIdMutex;
+
+    std::jthread m_thread;
+    std::vector<float> m_buffer1;
+    std::vector<float> m_buffer2;
+    std::atomic<std::vector<float>*> m_readBuffer;
+    std::atomic<std::vector<float>*> m_writeBuffer;
+    quint32 m_sampleCount;
+
+    void reload();
+    void start() override;
+    void stop() override;
+};
+
+} // namespace caelestia
diff --git a/plugin/src/Caelestia/Services/audioprovider.cpp b/plugin/src/Caelestia/Services/audioprovider.cpp
new file mode 100644
index 0000000..f31480e
--- /dev/null
+++ b/plugin/src/Caelestia/Services/audioprovider.cpp
@@ -0,0 +1,122 @@
+#include "audioprovider.hpp"
+
+#include "audiocollector.hpp"
+#include "service.hpp"
+#include <qdebug.h>
+#include <qthread.h>
+
+namespace caelestia {
+
+AudioProcessor::AudioProcessor(AudioCollector* collector, QObject* parent)
+    : QObject(parent)
+    , m_collector(collector) {}
+
+AudioProcessor::~AudioProcessor() {
+    stop();
+}
+
+void AudioProcessor::init() {
+    m_timer = new QTimer(this);
+    if (m_collector) {
+        m_timer->setInterval(static_cast<int>(m_collector->chunkSize() * 1000.0 / m_collector->sampleRate()));
+    }
+    connect(m_timer, &QTimer::timeout, this, &AudioProcessor::process);
+}
+
+void AudioProcessor::setCollector(AudioCollector* collector) {
+    if (m_collector == collector) {
+        return;
+    }
+
+    if (m_timer) {
+        if (m_timer->isActive()) {
+            if (m_collector) {
+                m_collector->unref();
+            }
+            if (collector) {
+                collector->ref();
+            }
+        }
+        if (collector) {
+            m_timer->setInterval(static_cast<int>(collector->chunkSize() * 1000.0 / collector->sampleRate()));
+        } else {
+            m_timer->stop();
+        }
+    }
+
+    m_collector = collector;
+}
+
+void AudioProcessor::start() {
+    if (m_timer && m_collector) {
+        m_collector->ref();
+        m_timer->start();
+    }
+}
+
+void AudioProcessor::stop() {
+    if (m_timer && m_collector) {
+        m_timer->stop();
+        m_collector->unref();
+    }
+}
+
+AudioProvider::AudioProvider(QObject* parent)
+    : Service(parent)
+    , m_collector(nullptr)
+    , m_processor(nullptr)
+    , m_thread(nullptr) {}
+
+AudioProvider::~AudioProvider() {
+    if (m_thread) {
+        m_thread->quit();
+        m_thread->wait();
+    }
+}
+
+AudioCollector* AudioProvider::collector() const {
+    return m_collector;
+}
+
+void AudioProvider::setCollector(AudioCollector* collector) {
+    if (m_collector == collector) {
+        return;
+    }
+
+    m_collector = collector;
+    emit collectorChanged();
+
+    if (m_processor) {
+        QMetaObject::invokeMethod(m_processor, "setCollector", Qt::QueuedConnection, Q_ARG(AudioCollector*, collector));
+    }
+}
+
+void AudioProvider::init() {
+    if (!m_processor) {
+        qWarning() << "AudioProvider::init: attempted to init with no processor set";
+        return;
+    }
+
+    m_thread = new QThread(this);
+    m_processor->moveToThread(m_thread);
+
+    connect(m_thread, &QThread::started, m_processor, &AudioProcessor::init);
+    connect(m_thread, &QThread::finished, m_processor, &AudioProcessor::deleteLater);
+    connect(m_thread, &QThread::finished, m_thread, &QThread::deleteLater);
+
+    m_thread->start();
+}
+
+void AudioProvider::start() {
+    if (m_processor) {
+        QMetaObject::invokeMethod(m_processor, "start", Qt::QueuedConnection);
+    }
+}
+
+void AudioProvider::stop() {
+    if (m_processor) {
+        QMetaObject::invokeMethod(m_processor, "stop", Qt::QueuedConnection);
+    }
+}
+
+} // namespace caelestia
diff --git a/plugin/src/Caelestia/Services/audioprovider.hpp b/plugin/src/Caelestia/Services/audioprovider.hpp
new file mode 100644
index 0000000..c92b965
--- /dev/null
+++ b/plugin/src/Caelestia/Services/audioprovider.hpp
@@ -0,0 +1,61 @@
+#pragma once
+
+#include "audiocollector.hpp"
+#include "service.hpp"
+#include <qqmlintegration.h>
+#include <qtimer.h>
+
+namespace caelestia {
+
+class AudioProcessor : public QObject {
+    Q_OBJECT
+
+public:
+    explicit AudioProcessor(AudioCollector* collector, QObject* parent = nullptr);
+    ~AudioProcessor();
+
+    void init();
+
+protected:
+    AudioCollector* m_collector;
+
+    Q_INVOKABLE virtual void setCollector(AudioCollector* collector);
+
+private:
+    QTimer* m_timer;
+
+    Q_INVOKABLE void start();
+    Q_INVOKABLE void stop();
+
+    virtual void process() = 0;
+};
+
+class AudioProvider : public Service {
+    Q_OBJECT
+
+    Q_PROPERTY(AudioCollector* collector READ collector WRITE setCollector NOTIFY collectorChanged)
+
+public:
+    explicit AudioProvider(QObject* parent = nullptr);
+    ~AudioProvider();
+
+    AudioCollector* collector() const;
+    void setCollector(AudioCollector* collector);
+
+signals:
+    void collectorChanged();
+
+protected:
+    AudioCollector* m_collector;
+    AudioProcessor* m_processor;
+
+    void init();
+
+private:
+    QThread* m_thread;
+
+    void start() override;
+    void stop() override;
+};
+
+} // namespace caelestia
diff --git a/plugin/src/Caelestia/Services/beattracker.cpp b/plugin/src/Caelestia/Services/beattracker.cpp
new file mode 100644
index 0000000..462abb8
--- /dev/null
+++ b/plugin/src/Caelestia/Services/beattracker.cpp
@@ -0,0 +1,82 @@
+#include "beattracker.hpp"
+
+#include "audiocollector.hpp"
+#include "audioprovider.hpp"
+#include <aubio/aubio.h>
+
+namespace caelestia {
+
+BeatProcessor::BeatProcessor(AudioCollector* collector, QObject* parent)
+    : AudioProcessor(collector, parent)
+    , m_tempo(nullptr)
+    , m_in(nullptr)
+    , m_out(new_fvec(2)) {
+    if (collector) {
+        m_tempo = new_aubio_tempo("default", 1024, collector->chunkSize(), collector->sampleRate());
+        m_in = new_fvec(collector->chunkSize());
+    }
+};
+
+BeatProcessor::~BeatProcessor() {
+    if (m_tempo) {
+        del_aubio_tempo(m_tempo);
+    }
+    if (m_in) {
+        del_fvec(m_in);
+    }
+    del_fvec(m_out);
+}
+
+void BeatProcessor::setCollector(AudioCollector* collector) {
+    AudioProcessor::setCollector(collector);
+
+    if (m_tempo) {
+        del_aubio_tempo(m_tempo);
+    }
+    if (m_in) {
+        del_fvec(m_in);
+    }
+
+    if (collector) {
+        m_tempo = new_aubio_tempo("default", 1024, collector->chunkSize(), collector->sampleRate());
+        m_in = new_fvec(collector->chunkSize());
+    } else {
+        m_tempo = nullptr;
+        m_in = nullptr;
+    }
+}
+
+void BeatProcessor::process() {
+    if (!m_collector || !m_tempo || !m_in) {
+        return;
+    }
+
+    m_collector->readChunk(m_in->data);
+
+    aubio_tempo_do(m_tempo, m_in, m_out);
+    if (!qFuzzyIsNull(m_out->data[0])) {
+        emit beat(aubio_tempo_get_bpm(m_tempo));
+    }
+}
+
+BeatTracker::BeatTracker(QObject* parent)
+    : AudioProvider(parent)
+    , m_bpm(120) {
+    m_processor = new BeatProcessor(m_collector);
+    init();
+
+    connect(static_cast<BeatProcessor*>(m_processor), &BeatProcessor::beat, this, &BeatTracker::updateBpm);
+}
+
+smpl_t BeatTracker::bpm() const {
+    return m_bpm;
+}
+
+void BeatTracker::updateBpm(smpl_t bpm) {
+    if (!qFuzzyCompare(bpm + 1.0f, m_bpm + 1.0f)) {
+        m_bpm = bpm;
+        emit bpmChanged();
+    }
+}
+
+} // namespace caelestia
diff --git a/plugin/src/Caelestia/Services/beattracker.hpp b/plugin/src/Caelestia/Services/beattracker.hpp
new file mode 100644
index 0000000..ab18373
--- /dev/null
+++ b/plugin/src/Caelestia/Services/beattracker.hpp
@@ -0,0 +1,52 @@
+#pragma once
+
+#include "audiocollector.hpp"
+#include "audioprovider.hpp"
+#include <aubio/aubio.h>
+#include <qqmlintegration.h>
+
+namespace caelestia {
+
+class BeatProcessor : public AudioProcessor {
+    Q_OBJECT
+
+public:
+    explicit BeatProcessor(AudioCollector* collector, QObject* parent = nullptr);
+    ~BeatProcessor();
+
+signals:
+    void beat(smpl_t bpm);
+
+protected:
+    void setCollector(AudioCollector* collector) override;
+
+private:
+    aubio_tempo_t* m_tempo;
+    fvec_t* m_in;
+    fvec_t* m_out;
+
+    void process() override;
+};
+
+class BeatTracker : public AudioProvider {
+    Q_OBJECT
+    QML_ELEMENT
+
+    Q_PROPERTY(smpl_t bpm READ bpm NOTIFY bpmChanged)
+
+public:
+    explicit BeatTracker(QObject* parent = nullptr);
+
+    [[nodiscard]] smpl_t bpm() const;
+
+signals:
+    void bpmChanged();
+    void beat(smpl_t bpm);
+
+private:
+    smpl_t m_bpm;
+
+    void updateBpm(smpl_t bpm);
+};
+
+} // namespace caelestia
diff --git a/plugin/src/Caelestia/Services/cavaprovider.cpp b/plugin/src/Caelestia/Services/cavaprovider.cpp
new file mode 100644
index 0000000..76a1a10
--- /dev/null
+++ b/plugin/src/Caelestia/Services/cavaprovider.cpp
@@ -0,0 +1,163 @@
+#include "cavaprovider.hpp"
+
+#include "audiocollector.hpp"
+#include "audioprovider.hpp"
+#include <cava/cavacore.h>
+#include <cmath>
+#include <cstddef>
+#include <qdebug.h>
+
+namespace caelestia {
+
+CavaProcessor::CavaProcessor(AudioCollector* collector, QObject* parent)
+    : AudioProcessor(collector, parent)
+    , m_plan(nullptr)
+    , m_in(nullptr)
+    , m_out(nullptr)
+    , m_bars(0) {
+    if (collector) {
+        m_in = new double[collector->chunkSize()];
+    }
+};
+
+CavaProcessor::~CavaProcessor() {
+    cleanup();
+    if (m_in) {
+        delete[] m_in;
+    }
+}
+
+void CavaProcessor::setCollector(AudioCollector* collector) {
+    AudioProcessor::setCollector(collector);
+
+    if (m_in) {
+        delete[] m_in;
+    }
+
+    if (collector) {
+        m_in = new double[collector->chunkSize()];
+    } else {
+        m_in = nullptr;
+    }
+
+    reload();
+}
+
+void CavaProcessor::setBars(int bars) {
+    if (bars < 0) {
+        qWarning() << "CavaProcessor::setBars: bars must be greater than 0. Setting to 0.";
+        bars = 0;
+    }
+
+    if (m_bars != bars) {
+        m_bars = bars;
+        reload();
+    }
+}
+
+void CavaProcessor::reload() {
+    cleanup();
+    initCava();
+}
+
+void CavaProcessor::cleanup() {
+    if (m_plan) {
+        cava_destroy(m_plan);
+        m_plan = nullptr;
+    }
+
+    if (m_out) {
+        delete[] m_out;
+        m_out = nullptr;
+    }
+}
+
+void CavaProcessor::initCava() {
+    if (m_plan || m_bars == 0 || !m_collector) {
+        return;
+    }
+
+    m_plan = cava_init(m_bars, m_collector->sampleRate(), 1, 1, 0.85, 50, 10000);
+
+    if (m_plan->status == -1) {
+        qWarning() << "CavaProcessor::initCava: failed to initialise cava plan";
+        cleanup();
+        return;
+    }
+
+    m_out = new double[static_cast<size_t>(m_bars)];
+}
+
+void CavaProcessor::process() {
+    if (!m_plan || m_bars == 0 || !m_collector || !m_in || !m_out) {
+        return;
+    }
+
+    const int count = static_cast<int>(m_collector->readChunk(m_in));
+
+    // Process in data via cava
+    cava_execute(m_in, count, m_out, m_plan);
+
+    // Apply monstercat filter
+    for (int i = 0; i < m_bars; i++) {
+        for (int j = i - 1; j >= 0; j--) {
+            m_out[j] = std::max(m_out[i] / std::pow(1.5, i - j), m_out[j]);
+        }
+        for (int j = i + 1; j < m_bars; j++) {
+            m_out[j] = std::max(m_out[i] / std::pow(1.5, j - i), m_out[j]);
+        }
+    }
+
+    // Update values
+    QVector<double> values(m_bars);
+    std::copy(m_out, m_out + m_bars, values.begin());
+    if (values != m_values) {
+        m_values = std::move(values);
+        emit valuesChanged(m_values);
+    }
+}
+
+CavaProvider::CavaProvider(QObject* parent)
+    : AudioProvider(parent)
+    , m_bars(0)
+    , m_values(m_bars, 0.0) {
+    m_processor = new CavaProcessor(m_collector);
+    init();
+
+    connect(static_cast<CavaProcessor*>(m_processor), &CavaProcessor::valuesChanged, this, &CavaProvider::updateValues);
+}
+
+int CavaProvider::bars() const {
+    return m_bars;
+}
+
+void CavaProvider::setBars(int bars) {
+    if (bars < 0) {
+        qWarning() << "CavaProvider::setBars: bars must be greater than 0. Setting to 0.";
+        bars = 0;
+    }
+
+    if (m_bars == bars) {
+        return;
+    }
+
+    m_values.resize(bars, 0.0);
+    m_bars = bars;
+    emit barsChanged();
+    emit valuesChanged();
+
+    QMetaObject::invokeMethod(m_processor, "setBars", Qt::QueuedConnection, Q_ARG(int, bars));
+}
+
+QVector<double> CavaProvider::values() const {
+    return m_values;
+}
+
+void CavaProvider::updateValues(QVector<double> values) {
+    if (values != m_values) {
+        m_values = values;
+        emit valuesChanged();
+    }
+}
+
+} // namespace caelestia
diff --git a/plugin/src/Caelestia/Services/cavaprovider.hpp b/plugin/src/Caelestia/Services/cavaprovider.hpp
new file mode 100644
index 0000000..6dab635
--- /dev/null
+++ b/plugin/src/Caelestia/Services/cavaprovider.hpp
@@ -0,0 +1,67 @@
+#pragma once
+
+#include "audiocollector.hpp"
+#include "audioprovider.hpp"
+#include <cava/cavacore.h>
+#include <qqmlintegration.h>
+
+namespace caelestia {
+
+class CavaProcessor : public AudioProcessor {
+    Q_OBJECT
+
+public:
+    explicit CavaProcessor(AudioCollector* collector, QObject* parent = nullptr);
+    ~CavaProcessor();
+
+signals:
+    void valuesChanged(QVector<double> values);
+
+protected:
+    void setCollector(AudioCollector* collector) override;
+
+private:
+    struct cava_plan* m_plan;
+    double* m_in;
+    double* m_out;
+
+    int m_bars;
+    QVector<double> m_values;
+
+    Q_INVOKABLE void setBars(int bars);
+
+    void reload();
+    void initCava();
+    void cleanup();
+
+    void process() override;
+};
+
+class CavaProvider : public AudioProvider {
+    Q_OBJECT
+    QML_ELEMENT
+
+    Q_PROPERTY(int bars READ bars WRITE setBars NOTIFY barsChanged)
+
+    Q_PROPERTY(QVector<double> values READ values NOTIFY valuesChanged)
+
+public:
+    explicit CavaProvider(QObject* parent = nullptr);
+
+    [[nodiscard]] int bars() const;
+    void setBars(int bars);
+
+    [[nodiscard]] QVector<double> values() const;
+
+signals:
+    void barsChanged();
+    void valuesChanged();
+
+private:
+    int m_bars;
+    QVector<double> m_values;
+
+    void updateValues(QVector<double> values);
+};
+
+} // namespace caelestia
diff --git a/plugin/src/Caelestia/Services/service.cpp b/plugin/src/Caelestia/Services/service.cpp
new file mode 100644
index 0000000..bc919c9
--- /dev/null
+++ b/plugin/src/Caelestia/Services/service.cpp
@@ -0,0 +1,70 @@
+#include "service.hpp"
+
+#include <qdebug.h>
+#include <qpointer.h>
+
+namespace caelestia {
+
+Service::Service(QObject* parent)
+    : QObject(parent)
+    , m_refCount(0) {}
+
+int Service::refCount() const {
+    QMutexLocker locker(&m_mutex);
+    return m_refCount;
+}
+
+void Service::ref() {
+    bool needsStart = false;
+
+    {
+        QMutexLocker locker(&m_mutex);
+        if (m_refCount == 0) {
+            needsStart = true;
+        }
+        m_refCount++;
+    }
+    emit refCountChanged();
+
+    if (needsStart) {
+        const QPointer<Service> self(this);
+        QMetaObject::invokeMethod(
+            this,
+            [self]() {
+                if (self) {
+                    self->start();
+                }
+            },
+            Qt::QueuedConnection);
+    }
+}
+
+void Service::unref() {
+    bool needsStop = false;
+
+    {
+        QMutexLocker locker(&m_mutex);
+        if (m_refCount == 0) {
+            return;
+        }
+        m_refCount--;
+        if (m_refCount == 0) {
+            needsStop = true;
+        }
+    }
+    emit refCountChanged();
+
+    if (needsStop) {
+        const QPointer<Service> self(this);
+        QMetaObject::invokeMethod(
+            this,
+            [self]() {
+                if (self) {
+                    self->stop();
+                }
+            },
+            Qt::QueuedConnection);
+    }
+}
+
+} // namespace caelestia
diff --git a/plugin/src/Caelestia/Services/service.hpp b/plugin/src/Caelestia/Services/service.hpp
new file mode 100644
index 0000000..787818b
--- /dev/null
+++ b/plugin/src/Caelestia/Services/service.hpp
@@ -0,0 +1,32 @@
+#pragma once
+
+#include <qmutex.h>
+#include <qobject.h>
+
+namespace caelestia {
+
+class Service : public QObject {
+    Q_OBJECT
+
+    Q_PROPERTY(int refCount READ refCount NOTIFY refCountChanged)
+
+public:
+    explicit Service(QObject* parent = nullptr);
+
+    [[nodiscard]] int refCount() const;
+
+    void ref();
+    void unref();
+
+signals:
+    void refCountChanged();
+
+private:
+    int m_refCount;
+    mutable QMutex m_mutex;
+
+    virtual void start() = 0;
+    virtual void stop() = 0;
+};
+
+} // namespace caelestia
diff --git a/plugin/src/Caelestia/Services/serviceref.cpp b/plugin/src/Caelestia/Services/serviceref.cpp
new file mode 100644
index 0000000..dc82811
--- /dev/null
+++ b/plugin/src/Caelestia/Services/serviceref.cpp
@@ -0,0 +1,42 @@
+#include "serviceref.hpp"
+
+#include "service.hpp"
+
+namespace caelestia {
+
+ServiceRef::ServiceRef(Service* service, QObject* parent)
+    : QObject(parent)
+    , m_service(service) {
+    if (m_service) {
+        m_service->ref();
+    }
+}
+
+ServiceRef::~ServiceRef() {
+    if (m_service) {
+        m_service->unref();
+    }
+}
+
+Service* ServiceRef::service() const {
+    return m_service;
+}
+
+void ServiceRef::setService(Service* service) {
+    if (m_service == service) {
+        return;
+    }
+
+    if (m_service) {
+        m_service->unref();
+    }
+
+    m_service = service;
+    emit serviceChanged();
+
+    if (m_service) {
+        m_service->ref();
+    }
+}
+
+} // namespace caelestia
diff --git a/plugin/src/Caelestia/Services/serviceref.hpp b/plugin/src/Caelestia/Services/serviceref.hpp
new file mode 100644
index 0000000..072419e
--- /dev/null
+++ b/plugin/src/Caelestia/Services/serviceref.hpp
@@ -0,0 +1,28 @@
+#pragma once
+
+#include "service.hpp"
+#include <qqmlintegration.h>
+
+namespace caelestia {
+
+class ServiceRef : public QObject {
+    Q_OBJECT
+    QML_ELEMENT
+
+    Q_PROPERTY(Service* service READ service WRITE setService NOTIFY serviceChanged)
+
+public:
+    explicit ServiceRef(Service* service = nullptr, QObject* parent = nullptr);
+    ~ServiceRef();
+
+    [[nodiscard]] Service* service() const;
+    void setService(Service* service);
+
+signals:
+    void serviceChanged();
+
+private:
+    Service* m_service;
+};
+
+} // namespace caelestia