feat: Create datawriter class

include/DataWriter.hpp

@@ -0,0 +1,42 @@
+#ifndef DATAWRITER_HPP
+#define DATAWRITER_HPP
+
+#include <blockingconcurrentqueue.h>
+
+#include <EEGData.hpp>
+#include <Marker.hpp>
+#include <atomic>
+#include <fstream>
+#include <memory>
+#include <string>
+#include <thread>
+
+class DataWriter {
+   public:
+    DataWriter() = default;
+    ~DataWriter();
+
+    DataWriter(const DataWriter&)            = delete;
+    DataWriter& operator=(const DataWriter&) = delete;
+    DataWriter(DataWriter&&)                 = delete;
+    DataWriter& operator=(DataWriter&&)      = delete;
+
+    void start(const std::string&                                    filePath,
+               std::shared_ptr<moodycamel::ConcurrentQueue<EEGData>> eegQueue,
+               std::shared_ptr<moodycamel::ConcurrentQueue<Marker>>  markerQueue);
+    void stop();
+
+   private:
+    void writeLoop();
+    void writeEEGData(const EEGData& data);
+    void writeMarker(const Marker& marker);
+
+    std::ofstream                                         outputFile;
+    std::shared_ptr<moodycamel::ConcurrentQueue<EEGData>> eegQueue;
+    std::shared_ptr<moodycamel::ConcurrentQueue<Marker>>  markerQueue;
+    std::thread                                           writerThread;
+
+    std::atomic<bool> stopRequested{false};
+};
+
+#endif  // DATAWRITER_HPP

include/EEGData.hpp

@@ -0,0 +1,20 @@
+#ifndef EEGDATA_HPP
+#define EEGDATA_HPP
+
+#include <utility>
+#include <vector>
+
+struct EEGData {
+    std::vector<float> channels;
+    double             timestamp = 0.0;
+
+    EEGData() = default;
+
+    EEGData(const std::vector<float>& channelValues, double sampleTimestamp)
+        : channels(channelValues), timestamp(sampleTimestamp) {}
+
+    EEGData(std::vector<float>&& channelValues, double sampleTimestamp)
+        : channels(std::move(channelValues)), timestamp(sampleTimestamp) {}
+};
+
+#endif  // EEGDATA_HPP

include/Marker.hpp

@@ -0,0 +1,20 @@
+#ifndef MARKER_HPP
+#define MARKER_HPP
+
+#include <string>
+#include <utility>
+
+struct Marker {
+    std::string eventName;
+    double      timestamp = 0.0;
+
+    Marker() = default;
+
+    Marker(const std::string& name, double markerTimestamp)
+        : eventName(name), timestamp(markerTimestamp) {}
+
+    Marker(std::string&& name, double markerTimestamp)
+        : eventName(std::move(name)), timestamp(markerTimestamp) {}
+};
+
+#endif  // MARKER_HPP

src/CMakeLists.txt

@@ -2,6 +2,7 @@ protobuf_generate_cpp(PROTO_SRCS PROTO_HDRS ${CMAKE_CURRENT_SOURCE_DIR}/../proto
 
 add_library(runtime_core OBJECT
     Runtime.cpp
+    DataWriter.cpp
     parser/Parser.cpp
     scene/components/ComponentRegistry.cpp
     scene/components/BlinkComponent.cpp

src/DataWriter.cpp

@@ -0,0 +1,111 @@
+#include <DataWriter.hpp>
+#include <chrono>
+#include <sstream>
+#include <stdexcept>
+#include <thread>
+#include <utility>
+
+// anonymous namespace here for private utility-style definitions
+namespace {
+constexpr auto kWriteLoopSleep = std::chrono::milliseconds(10);
+
+std::string formatChannels(const std::vector<float>& channels) {
+    std::ostringstream stream;
+    for (std::size_t index = 0; index < channels.size(); ++index) {
+        if (index > 0) {
+            stream << ',';
+        }
+        stream << channels[index];
+    }
+    return stream.str();
+}
+}  // namespace
+
+DataWriter::~DataWriter() { stop(); }
+
+void DataWriter::start(const std::string&                                    filePath,
+                       std::shared_ptr<moodycamel::ConcurrentQueue<EEGData>> eegQueue,
+                       std::shared_ptr<moodycamel::ConcurrentQueue<Marker>>  markerQueue) {
+    stop();
+
+    outputFile.open(filePath, std::ios::out | std::ios::trunc);
+    if (!outputFile.is_open()) {
+        throw std::runtime_error("Failed to open data writer output file: " + filePath);
+    }
+
+    this->eegQueue    = std::move(eegQueue);
+    this->markerQueue = std::move(markerQueue);
+    stopRequested.store(false);
+
+    outputFile << "type,timestamp,payload\n";
+    writerThread = std::thread(&DataWriter::writeLoop, this);
+}
+
+void DataWriter::stop() {
+    stopRequested.store(true);
+
+    if (writerThread.joinable()) {
+        writerThread.join();
+    }
+
+    if (outputFile.is_open()) {
+        outputFile.flush();
+        outputFile.close();
+    }
+}
+
+void DataWriter::writeLoop() {
+    while (!stopRequested.load()) {
+        bool wroteData = false;
+
+        if (eegQueue) {
+            EEGData eegData;
+            while (eegQueue->try_dequeue(eegData)) {
+                writeEEGData(eegData);
+                wroteData = true;
+            }
+        }
+
+        if (markerQueue) {
+            Marker marker;
+            while (markerQueue->try_dequeue(marker)) {
+                writeMarker(marker);
+                wroteData = true;
+            }
+        }
+
+        if (!wroteData) {
+            std::this_thread::sleep_for(kWriteLoopSleep);
+        }
+    }
+
+    if (eegQueue) {
+        EEGData eegData;
+        while (eegQueue->try_dequeue(eegData)) {
+            writeEEGData(eegData);
+        }
+    }
+
+    if (markerQueue) {
+        Marker marker;
+        while (markerQueue->try_dequeue(marker)) {
+            writeMarker(marker);
+        }
+    }
+}
+
+void DataWriter::writeEEGData(const EEGData& data) {
+    if (!outputFile.is_open()) {
+        return;
+    }
+
+    outputFile << "eeg," << data.timestamp << ",\"" << formatChannels(data.channels) << '"' << '\n';
+}
+
+void DataWriter::writeMarker(const Marker& marker) {
+    if (!outputFile.is_open()) {
+        return;
+    }
+
+    outputFile << "marker," << marker.timestamp << ",\"" << marker.eventName << '"' << '\n';
+}

tests/unit_tests/data_writer_test.cpp

@@ -0,0 +1,83 @@
+#include <gtest/gtest.h>
+
+#include <DataWriter.hpp>
+#include <chrono>
+#include <filesystem>
+#include <fstream>
+#include <memory>
+#include <string>
+#include <thread>
+#include <vector>
+
+namespace {
+namespace fs = std::filesystem;
+
+constexpr float  kChannelOne      = 1.25F;
+constexpr float  kChannelTwo      = 2.5F;
+constexpr float  kChannelThree    = 3.75F;
+constexpr double kEegTimestamp    = 12.5;
+constexpr double kMarkerTimestamp = 13.25;
+constexpr auto   kWriteWait       = std::chrono::milliseconds(50);
+
+fs::path makeTempFilePath(const std::string& nameStem) {
+    const auto uniqueSuffix =
+        std::to_string(std::chrono::steady_clock::now().time_since_epoch().count());
+    return fs::temp_directory_path() / (nameStem + "_" + uniqueSuffix + ".csv");
+}
+
+std::vector<std::string> readAllLines(const fs::path& filePath) {
+    std::ifstream            input(filePath);
+    std::vector<std::string> lines;
+    std::string              line;
+
+    while (std::getline(input, line)) {
+        lines.push_back(line);
+    }
+
+    return lines;
+}
+}  // namespace
+
+TEST(DataWriterTest, WritesCsvHeaderOnStart) {
+    const auto filePath = makeTempFilePath("datawriter_header");
+
+    auto eegQueue    = std::make_shared<moodycamel::ConcurrentQueue<EEGData>>();
+    auto markerQueue = std::make_shared<moodycamel::ConcurrentQueue<Marker>>();
+
+    {
+        DataWriter writer;
+        writer.start(filePath.string(), eegQueue, markerQueue);
+        writer.stop();
+    }
+
+    const auto lines = readAllLines(filePath);
+    ASSERT_FALSE(lines.empty());
+    EXPECT_EQ(lines.front(), "type,timestamp,payload");
+
+    fs::remove(filePath);
+}
+
+TEST(DataWriterTest, FlushesEegAndMarkerRecords) {
+    const auto filePath = makeTempFilePath("datawriter_records");
+
+    auto eegQueue    = std::make_shared<moodycamel::ConcurrentQueue<EEGData>>();
+    auto markerQueue = std::make_shared<moodycamel::ConcurrentQueue<Marker>>();
+
+    eegQueue->enqueue(EEGData{{kChannelOne, kChannelTwo, kChannelThree}, kEegTimestamp});
+    markerQueue->enqueue(Marker{"stimulus_on", kMarkerTimestamp});
+
+    {
+        DataWriter writer;
+        writer.start(filePath.string(), eegQueue, markerQueue);
+        std::this_thread::sleep_for(kWriteWait);
+        writer.stop();
+    }
+
+    const auto lines = readAllLines(filePath);
+    ASSERT_GE(lines.size(), 3U);
+    EXPECT_EQ(lines[0], "type,timestamp,payload");
+    EXPECT_EQ(lines[1], "eeg,12.5,\"1.25,2.5,3.75\"");
+    EXPECT_EQ(lines[2], "marker,13.25,\"stimulus_on\"");
+
+    fs::remove(filePath);
+}