plugin/ap: properly buffer data

Run at a stable fps instead of in large chunks
Use one thread for collecting and another for processing

4 files changed, 224 insertions, 76 deletions

diff --git a/plugin/src/Caelestia/audioprovider.cpp b/plugin/src/Caelestia/audioprovider.cpp
index 2d76a5d..0219ca2 100644
--- a/plugin/src/Caelestia/audioprovider.cpp
+++ b/plugin/src/Caelestia/audioprovider.cpp
@@ -5,66 +5,84 @@
 #include <QDebug>
 #include <QIODevice>
 #include <QMediaDevices>
+#include <QMutexLocker>
 #include <QObject>
 #include <QThread>
+#include <QVector>
+#include <algorithm>
 #include <cstddef>
 #include <cstdint>
 
 namespace caelestia {
 
-AudioWorker::AudioWorker(int sampleRate, int hopSize, QObject* parent)
+AudioCollector::AudioCollector(AudioProvider* provider, QObject* parent)
     : QObject(parent)
-    , m_sampleRate(sampleRate)
-    , m_hopSize(hopSize)
     , m_source(nullptr)
-    , m_device(nullptr) {}
+    , m_device(nullptr)
+    , m_provider(provider)
+    , m_sampleRate(provider->sampleRate())
+    , m_chunkSize(provider->chunkSize())
+    , m_chunk(m_chunkSize)
+    , m_chunkOffset(0) {}
+
+AudioCollector::~AudioCollector() {
+    m_source->stop();
+}
 
-void AudioWorker::init() {
+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, &AudioWorker::handleStateChanged);
+    connect(m_source, &QAudioSource::stateChanged, this, &AudioCollector::handleStateChanged);
 };
 
-AudioWorker::~AudioWorker() {
-    m_source->stop();
-    delete m_source;
-}
-
-void AudioWorker::start() {
+void AudioCollector::start() {
     if (!m_source) {
         return;
     }
 
     m_device = m_source->start();
-    connect(m_device, &QIODevice::readyRead, this, &AudioWorker::processData);
+    connect(m_device, &QIODevice::readyRead, this, &AudioCollector::loadChunk);
 }
 
-void AudioWorker::stop() {
-    m_source->stop();
-    m_device = nullptr;
+void AudioCollector::stop() {
+    if (m_source) {
+        m_source->stop();
+        m_device = nullptr;
+    }
 }
 
-template <typename T> void AudioWorker::process(T* outBuf) {
+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);
-    const size_t hopSize = static_cast<size_t>(m_hopSize);
 
-    for (size_t i = 0; i < count; ++i) {
-        outBuf[i % hopSize] = static_cast<T>(samples[i] / 32768.0);
-        if ((i + 1) % hopSize == 0) {
-            consumeData();
+    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) {
+            m_provider->withLock([&] {
+                m_provider->loadChunk(m_chunk);
+            });
+
+            m_chunkOffset = 0;
         }
     }
 }
-template void AudioWorker::process(float* outBuf);
-template void AudioWorker::process(double* outBuf);
 
-void AudioWorker::handleStateChanged(QtAudio::State state) const {
+void AudioCollector::handleStateChanged(QtAudio::State state) const {
     if (state == QtAudio::StoppedState && m_source->error() != QtAudio::NoError) {
         switch (m_source->error()) {
         case QtAudio::OpenError:
@@ -85,43 +103,128 @@ void AudioWorker::handleStateChanged(QtAudio::State state) const {
     }
 }
 
-AudioProvider::AudioProvider(QObject* parent)
+AudioProcessor::AudioProcessor(AudioProvider* provider, QObject* parent)
+    : QObject(parent)
+    , m_sampleRate(provider->sampleRate())
+    , m_chunkSize(provider->chunkSize())
+    , m_provider(provider) {}
+
+AudioProcessor::~AudioProcessor() {
+    if (m_timer) {
+        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);
+}
+
+void AudioProcessor::start() {
+    if (m_timer) {
+        m_timer->start();
+    }
+}
+
+void AudioProcessor::stop() {
+    if (m_timer) {
+        m_timer->stop();
+    }
+}
+
+void AudioProcessor::handleTimeout() {
+    QVector<double> chunk;
+
+    m_provider->withLock([&] {
+        if (m_provider->hasChunks()) {
+            chunk = m_provider->nextChunk();
+        }
+    });
+
+    if (!chunk.isEmpty()) {
+        processChunk(chunk);
+    }
+}
+
+AudioProvider::AudioProvider(int sampleRate, int chunkSize, QObject* parent)
     : Service(parent)
-    , m_worker(nullptr)
-    , m_thread(nullptr) {}
+    , m_sampleRate(sampleRate)
+    , m_chunkSize(chunkSize)
+    , m_collector(new AudioCollector(this))
+    , m_processor(nullptr)
+    , m_collectorThread(new QThread(this))
+    , 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() {
-    if (m_thread) {
-        m_thread->quit();
-        m_thread->wait();
+    m_collectorThread->quit();
+    if (m_processorThread) {
+        m_processorThread->quit();
+        m_processorThread->wait();
     }
+    m_collectorThread->wait();
+}
+
+int AudioProvider::sampleRate() const {
+    return m_sampleRate;
+}
+
+int AudioProvider::chunkSize() const {
+    return m_chunkSize;
+}
+
+void AudioProvider::withLock(std::function<void()> fn) {
+    QMutexLocker locker(&m_mutex);
+    fn();
+}
+
+bool AudioProvider::hasChunks() const {
+    return !m_chunks.isEmpty();
+}
+
+QVector<double> AudioProvider::nextChunk() {
+    return m_chunks.dequeue();
+}
+
+void AudioProvider::loadChunk(QVector<double> chunk) {
+    m_chunks.enqueue(std::move(chunk));
 }
 
 void AudioProvider::init() {
-    if (!m_worker) {
-        qWarning() << "AudioProvider::init: attempted to init with no worker set";
+    if (!m_processor) {
+        qWarning() << "AudioProvider::init: attempted to init with no processor set";
         return;
     }
 
-    m_thread = new QThread(this);
-    m_worker->moveToThread(m_thread);
+    m_processorThread = new QThread(this);
+    m_processor->moveToThread(m_processorThread);
 
-    connect(m_thread, &QThread::started, m_worker, &AudioWorker::init);
-    connect(m_thread, &QThread::finished, m_worker, &AudioWorker::deleteLater);
-    connect(m_thread, &QThread::finished, m_thread, &QThread::deleteLater);
+    connect(m_processorThread, &QThread::started, m_processor, &AudioProcessor::init);
+    connect(m_processorThread, &QThread::finished, m_processor, &AudioProcessor::deleteLater);
+    connect(m_processorThread, &QThread::finished, m_processorThread, &QThread::deleteLater);
 
-    m_thread->start();
+    m_processorThread->start();
 }
 
 void AudioProvider::start() {
-    if (m_worker) {
-        QMetaObject::invokeMethod(m_worker, "start", Qt::QueuedConnection);
+    QMetaObject::invokeMethod(m_collector, "start", Qt::QueuedConnection);
+    if (m_processor) {
+        QMetaObject::invokeMethod(m_processor, "start", Qt::QueuedConnection);
     }
 }
 
 void AudioProvider::stop() {
-    if (m_worker) {
-        QMetaObject::invokeMethod(m_worker, "stop", Qt::QueuedConnection);
+    QMetaObject::invokeMethod(m_collector, "stop", Qt::QueuedConnection);
+    if (m_processor) {
+        QMetaObject::invokeMethod(m_processor, "stop", Qt::QueuedConnection);
     }
 }
 
diff --git a/plugin/src/Caelestia/audioprovider.hpp b/plugin/src/Caelestia/audioprovider.hpp
index 205be2c..2acdb2d 100644
--- a/plugin/src/Caelestia/audioprovider.hpp
+++ b/plugin/src/Caelestia/audioprovider.hpp
@@ -3,54 +3,100 @@
 #include "service.hpp"
 #include <QAudioSource>
 #include <QIODevice>
+#include <QMutex>
 #include <QObject>
+#include <QQueue>
 #include <QThread>
+#include <QTimer>
+#include <QVector>
 #include <qqmlintegration.h>
 
 namespace caelestia {
 
-class AudioWorker : public QObject {
+class AudioProvider;
+
+class AudioCollector : public QObject {
     Q_OBJECT
 
 public:
-    explicit AudioWorker(int sampleRate = 44100, int hopSize = 512, QObject* parent = nullptr);
-    ~AudioWorker();
+    explicit AudioCollector(AudioProvider* provider, QObject* parent = nullptr);
+    ~AudioCollector();
 
     void init();
 
-protected:
-    int m_sampleRate;
-    int m_hopSize;
-
-    template <typename T> void process(T* outBuf);
-
 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);
+    ~AudioProcessor();
+
+    void init();
 
-    virtual void processData() = 0;
-    virtual void consumeData() = 0;
+protected:
+    int m_sampleRate;
+    int m_chunkSize;
+
+private:
+    AudioProvider* m_provider;
+    QTimer* m_timer;
+
+    Q_INVOKABLE void start();
+    Q_INVOKABLE void stop();
+
+    void handleTimeout();
+    virtual void processChunk(const QVector<double>& chunk) = 0;
 };
 
 class AudioProvider : public Service {
     Q_OBJECT
 
 public:
-    explicit AudioProvider(QObject* parent = nullptr);
+    explicit AudioProvider(int sampleRate = 44100, int chunkSize = 512, QObject* parent = nullptr);
     ~AudioProvider();
 
+    [[nodiscard]] int sampleRate() const;
+    [[nodiscard]] int chunkSize() const;
+
+    void withLock(std::function<void()> fn);
+
+    [[nodiscard]] bool hasChunks() const;
+    [[nodiscard]] QVector<double> nextChunk();
+    void loadChunk(QVector<double> chunk);
+
 protected:
-    AudioWorker* m_worker;
+    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_thread;
+    QThread* m_collectorThread;
+    QThread* m_processorThread;
 
     void start() override;
     void stop() override;
diff --git a/plugin/src/Caelestia/beattracker.cpp b/plugin/src/Caelestia/beattracker.cpp
index b59abd7..11938d0 100644
--- a/plugin/src/Caelestia/beattracker.cpp
+++ b/plugin/src/Caelestia/beattracker.cpp
@@ -6,36 +6,36 @@
 
 namespace caelestia {
 
-BeatWorker::BeatWorker(uint_t sampleRate, uint_t hopSize, QObject* parent)
-    : AudioWorker(static_cast<int>(sampleRate), static_cast<int>(hopSize), parent)
-    , m_tempo(new_aubio_tempo("default", 1024, hopSize, sampleRate))
-    , m_in(new_fvec(hopSize))
+BeatProcessor::BeatProcessor(AudioProvider* provider, QObject* parent)
+    : AudioProcessor(provider, parent)
+    , m_tempo(new_aubio_tempo("default", 1024, static_cast<uint_t>(m_chunkSize), static_cast<uint_t>(m_sampleRate)))
+    , m_in(new_fvec(static_cast<uint_t>(m_chunkSize)))
     , m_out(new_fvec(2)) {};
 
-BeatWorker::~BeatWorker() {
+BeatProcessor::~BeatProcessor() {
     del_aubio_tempo(m_tempo);
     del_fvec(m_in);
     del_fvec(m_out);
 }
 
-void BeatWorker::processData() {
-    process(m_in->data);
-}
+void BeatProcessor::processChunk(const QVector<double>& chunk) {
+    std::transform(chunk.constBegin(), chunk.constEnd(), m_in->data, [](double d) {
+        return static_cast<float>(d);
+    });
 
-void BeatWorker::consumeData() {
     aubio_tempo_do(m_tempo, m_in, m_out);
     if (m_out->data[0] != 0.0f) {
         emit beat(aubio_tempo_get_bpm(m_tempo));
     }
 }
 
-BeatTracker::BeatTracker(uint_t sampleRate, uint_t hopSize, QObject* parent)
-    : AudioProvider(parent)
+BeatTracker::BeatTracker(int sampleRate, int chunkSize, QObject* parent)
+    : AudioProvider(sampleRate, chunkSize, parent)
     , m_bpm(120) {
-    m_worker = new BeatWorker(sampleRate, hopSize, this);
+    m_processor = new BeatProcessor(this);
     init();
 
-    connect(static_cast<BeatWorker*>(m_worker), &BeatWorker::beat, this, &BeatTracker::updateBpm);
+    connect(static_cast<BeatProcessor*>(m_processor), &BeatProcessor::beat, this, &BeatTracker::updateBpm);
 }
 
 smpl_t BeatTracker::bpm() const {
diff --git a/plugin/src/Caelestia/beattracker.hpp b/plugin/src/Caelestia/beattracker.hpp
index edb9c0c..a490f9d 100644
--- a/plugin/src/Caelestia/beattracker.hpp
+++ b/plugin/src/Caelestia/beattracker.hpp
@@ -7,12 +7,12 @@
 
 namespace caelestia {
 
-class BeatWorker : public AudioWorker {
+class BeatProcessor : public AudioProcessor {
     Q_OBJECT
 
 public:
-    explicit BeatWorker(uint_t sampleRate = 44100, uint_t hopSize = 512, QObject* parent = nullptr);
-    ~BeatWorker();
+    explicit BeatProcessor(AudioProvider* provider, QObject* parent = nullptr);
+    ~BeatProcessor();
 
 signals:
     void beat(smpl_t bpm);
@@ -22,8 +22,7 @@ private:
     fvec_t* m_in;
     fvec_t* m_out;
 
-    void processData() override;
-    void consumeData() override;
+    void processChunk(const QVector<double>& chunk) override;
 };
 
 class BeatTracker : public AudioProvider {
@@ -34,7 +33,7 @@ class BeatTracker : public AudioProvider {
     Q_PROPERTY(smpl_t bpm READ bpm NOTIFY bpmChanged)
 
 public:
-    explicit BeatTracker(uint_t sampleRate = 44100, uint_t hopSize = 512, QObject* parent = nullptr);
+    explicit BeatTracker(int sampleRate = 44100, int chunkSize = 512, QObject* parent = nullptr);
 
     [[nodiscard]] smpl_t bpm() const;