From 054bc9bfe3b29d00434d1484e3d79f2fc7f81901 Mon Sep 17 00:00:00 2001 From: Michal Date: Wed, 1 Jul 2026 18:16:56 +0200 Subject: [PATCH 1/5] feat: create main loop --- .gitignore | 3 + README.md | 13 +- include/Runtime.hpp | 87 +++++- include/config/DeviceConfig.hpp | 3 +- include/config/OutputConfig.hpp | 18 ++ include/datawriter/CSVFormatStrategy.hpp | 2 + .../datawriter/DataFormatStrategyFactory.hpp | 18 ++ include/datawriter/IDataFormatStrategy.hpp | 2 + src/Runtime.cpp | 147 ++++++++++- src/config/ConfigParser.cpp | 9 + src/config/ConfigValidation.cpp | 4 + src/datawriter/CMakeLists.txt | 1 + src/datawriter/DataFormatStrategyFactory.cpp | 34 +++ src/main.cpp | 42 ++- tests/unit_tests/config_parser_test.cpp | 30 +++ tests/unit_tests/config_validation_test.cpp | 17 ++ .../data_format_strategy_factory_test.cpp | 25 ++ tests/unit_tests/runtime_test.cpp | 249 ++++++++++++++++++ 18 files changed, 688 insertions(+), 16 deletions(-) create mode 100644 include/config/OutputConfig.hpp create mode 100644 include/datawriter/DataFormatStrategyFactory.hpp create mode 100644 src/datawriter/DataFormatStrategyFactory.cpp create mode 100644 tests/unit_tests/data_format_strategy_factory_test.cpp create mode 100644 tests/unit_tests/runtime_test.cpp diff --git a/.gitignore b/.gitignore index 9017fac..3a06a20 100644 --- a/.gitignore +++ b/.gitignore @@ -60,3 +60,6 @@ _deps/ .Trashes ehthumbs.db Thumbs.db + +# config files +*.json diff --git a/README.md b/README.md index 4c74db1..6254121 100644 --- a/README.md +++ b/README.md @@ -192,7 +192,7 @@ otherwise the same experiment stops being portable between labs. | Stimulus timing, trial structure, marker/event names | Expected stream shape: channel count, sample rate | | Anything the editor authors and versions with the study | Channel table: index, label, enabled, unit | | | Reference / ground electrodes, impedance check thresholds | -| | *(planned)* output file format for `DataWriter` | +| | Output format for `DataWriter` (`output.format`) | Consequences of the split: @@ -227,12 +227,16 @@ way — `channels` is a top-level key, so it is a top-level `DeviceConfig` field { "index": 1, "label": "Oz", "enabled": false, "unit": "microvolts" } // ... one entry per expected_channel_count, indices unique and in range ], - "impedance_check": { "supported": true, "threshold_kohm": 5.0 } + "impedance_check": { "supported": true, "threshold_kohm": 5.0 }, + "output": { "format": "csv" } // -> DataFormatStrategyFactory, defaults to csv } ``` `config_version`, `device_name`, `montage_standard`, `lsl_stream` and `channels` -are required; `reference`, `ground` and `impedance_check` default when absent. +are required; `reference`, `ground`, `impedance_check` and `output` default when +absent. If `output` is present it must carry a non-empty `format`; whether that +format is *known* is decided by `DataFormatStrategyFactory` when the writer is +built, not by config validation. Channels with `"enabled": false` stay in the config (they document the cap) but are **not** acquired: `LSLReader` drops them from every sample. @@ -320,7 +324,8 @@ stay fatal — they are logged and the worker exits instead of retrying forever. | `ConfigParser` | Implemented | `config.json` → `DeviceConfig` (1:1 mapping, major-version checked, see §5), nlohmann/json | | Config `validate()` | Implemented | semantic rules on the config types themselves, independent of JSON (see §5) | | `DataWriter` | Implemented | strategy-based; `CSVFormatStrategy` | -| `Runtime` orchestration | **Stub** | currently does nothing | +| `DataFormatStrategyFactory` | Implemented | `output.format` → format strategy; unknown formats rejected | +| `Runtime` orchestration | Implemented | owns SDL session, parses both config files, wires the queues, drives the render loop, stops workers | The class diagram in older docs is partly aspirational; the table above reflects the actual code. diff --git a/include/Runtime.hpp b/include/Runtime.hpp index 838b4a6..f577e0c 100644 --- a/include/Runtime.hpp +++ b/include/Runtime.hpp @@ -1,18 +1,93 @@ #ifndef RUNTIME_HPP #define RUNTIME_HPP -#include +#include + +#include +#include +#include +#include +#include + +class Scene; +class LSLReader; +class DataWriter; +class Renderer; +struct EEGData; +struct Marker; + +// Tag matches the other forward declarations in include/ (Scene, SceneObject, +// Component). SDL declares it as a struct, so all four are technically +// mismatched; changing one alone trips -Wmismatched-tags. +class SDL_Renderer; + +struct RuntimePaths { + std::string config; // device config.json + std::string experiment; // serialized experiment scene (protobuf) + std::string outputDir = "."; // directory the recorded CSV is written to +}; class Runtime { public: - Runtime() = default; - ~Runtime() = default; + using RenderTargetFactory = + std::function(const std::string& windowTitle)>; + + explicit Runtime(const RuntimePaths& paths); + Runtime(const RuntimePaths& paths, const RenderTargetFactory& renderTargetFactory); + ~Runtime(); Runtime(const Runtime&) = delete; Runtime& operator=(const Runtime&) = delete; Runtime(Runtime&&) = delete; - Runtime& operator=(Runtime&&) = delete; - static void start(); + Runtime& operator=(Runtime&&) = delete; + + // Drives the experiment on the calling thread until the render loop ends + // (SDL_QUIT or requestStop), then stops the workers. Single-shot: the stop + // source is never re-armed, so a second run() returns immediately. + void run(); + + // Safe to call from any thread, including while run() is in progress. + void requestStop(); + + // Empty until run() has computed it. Not synchronized: read it before run() + // starts or after it has returned. + const std::string& outputPath() const noexcept { return outputFilePath; } + + static RenderTargetFactory defaultRenderTargetFactory(); + + private: + struct SdlSession { + SdlSession(); + ~SdlSession(); + + SdlSession(const SdlSession&) = delete; + SdlSession& operator=(const SdlSession&) = delete; + SdlSession(SdlSession&&) = delete; + SdlSession& operator=(SdlSession&&) = delete; + }; + + void startWorkers(const std::string& outputPath); + void shutdown(); + std::string makeOutputPath() const; + + RuntimePaths paths; + DeviceConfig config; + + SdlSession sdlSession; + + std::shared_ptr scene; + std::shared_ptr> eegQueue; + std::shared_ptr> markerQueue; + + std::shared_ptr sdlRenderer; + + std::unique_ptr lslReader; + std::unique_ptr dataWriter; + std::unique_ptr renderer; + + std::string outputExtension; + std::string outputFilePath; + std::stop_source stopSource; }; -#endif // RUNTIME_HPP \ No newline at end of file +#endif // RUNTIME_HPP diff --git a/include/config/DeviceConfig.hpp b/include/config/DeviceConfig.hpp index 256c5d0..ca7e741 100644 --- a/include/config/DeviceConfig.hpp +++ b/include/config/DeviceConfig.hpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include @@ -36,7 +37,7 @@ struct DeviceConfig { GroundConfig ground; // ground std::vector channels; // channels ImpedanceConfig impedance; // impedance_check - // TODO: DataWriterConfig writer; // EEG output file format strategy + OutputConfig output; // output // Checks every rule a device config must satisfy, including the cross-field // ones no single member can check (channel count matching the stream, unique diff --git a/include/config/OutputConfig.hpp b/include/config/OutputConfig.hpp new file mode 100644 index 0000000..3fdb527 --- /dev/null +++ b/include/config/OutputConfig.hpp @@ -0,0 +1,18 @@ +#ifndef OUTPUTCONFIG_HPP +#define OUTPUTCONFIG_HPP + +#include + +// How recorded data should be persisted. The `format` selects the DataWriter +// format strategy (see DataFormatStrategyFactory) and, through it, the output +// file extension. +struct OutputConfig { + std::string format = "csv"; + + // Throws std::invalid_argument on an empty format. Whether a non-empty + // format is *known* is DataFormatStrategyFactory's decision, so that check + // stays out of the config layer. + void validate() const; +}; + +#endif // OUTPUTCONFIG_HPP diff --git a/include/datawriter/CSVFormatStrategy.hpp b/include/datawriter/CSVFormatStrategy.hpp index fb76dbe..4a9efca 100644 --- a/include/datawriter/CSVFormatStrategy.hpp +++ b/include/datawriter/CSVFormatStrategy.hpp @@ -15,6 +15,8 @@ class CSVFormatStrategy : public IDataFormatStrategy { CSVFormatStrategy(CSVFormatStrategy&&) = delete; CSVFormatStrategy& operator=(CSVFormatStrategy&&) = delete; + std::string fileExtension() const override { return "csv"; } + void open(const std::string& filepath) override; void close() override; diff --git a/include/datawriter/DataFormatStrategyFactory.hpp b/include/datawriter/DataFormatStrategyFactory.hpp new file mode 100644 index 0000000..9699e62 --- /dev/null +++ b/include/datawriter/DataFormatStrategyFactory.hpp @@ -0,0 +1,18 @@ +#ifndef DATAFORMATSTRATEGYFACTORY_HPP +#define DATAFORMATSTRATEGYFACTORY_HPP + +#include +#include +#include + +// Builds the DataWriter format strategy selected by the config's output format. +// Adding a new output format means registering it here; nothing else in the +// runtime needs to change. +class DataFormatStrategyFactory { + public: + // Returns the strategy for the given format (case-insensitive), e.g. "csv". + // Throws std::invalid_argument if the format is unknown. + static std::unique_ptr create(const std::string& format); +}; + +#endif // DATAFORMATSTRATEGYFACTORY_HPP diff --git a/include/datawriter/IDataFormatStrategy.hpp b/include/datawriter/IDataFormatStrategy.hpp index c178e43..28992a7 100644 --- a/include/datawriter/IDataFormatStrategy.hpp +++ b/include/datawriter/IDataFormatStrategy.hpp @@ -16,6 +16,8 @@ class IDataFormatStrategy { IDataFormatStrategy(IDataFormatStrategy&&) = delete; IDataFormatStrategy& operator=(IDataFormatStrategy&&) = delete; + virtual std::string fileExtension() const = 0; + virtual void open(const std::string& filepath) = 0; virtual void close() = 0; diff --git a/src/Runtime.cpp b/src/Runtime.cpp index 200a3d5..2082d6b 100644 --- a/src/Runtime.cpp +++ b/src/Runtime.cpp @@ -1,3 +1,148 @@ +#include + #include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { +constexpr int kWindowWidth = 1280; +constexpr int kWindowHeight = 720; +constexpr const char* kDefaultTitle = "NeuronIDE"; +constexpr const char* kFallbackName = "experiment"; + +std::string sdlError(const char* what) { return std::string(what) + ": " + SDL_GetError(); } + +// The experiment name comes from an authored protobuf file and ends up in a file +// name, so anything that is not plainly safe becomes '_'. Left unfiltered, a name +// like "block 1/run" would resolve to a missing subdirectory (and "../x" would +// escape the output directory entirely). +std::string sanitizeForFileName(std::string name) { + if (name.empty()) { + return kFallbackName; + } + + std::replace_if( + name.begin(), name.end(), + [](unsigned char character) { + return std::isalnum(character) == 0 && character != '-' && character != '_'; + }, + '_'); + return name; +} +} // namespace + +Runtime::SdlSession::SdlSession() { + if (SDL_Init(SDL_INIT_VIDEO | SDL_INIT_EVENTS) != 0) { + throw std::runtime_error(sdlError("Runtime: SDL_Init failed")); + } +} + +Runtime::SdlSession::~SdlSession() { SDL_Quit(); } + +Runtime::RenderTargetFactory Runtime::defaultRenderTargetFactory() { + return [](const std::string& windowTitle) -> std::shared_ptr { + const std::string title = windowTitle.empty() ? kDefaultTitle : windowTitle; + + SDL_Window* window = + SDL_CreateWindow(title.c_str(), SDL_WINDOWPOS_CENTERED, SDL_WINDOWPOS_CENTERED, + kWindowWidth, kWindowHeight, SDL_WINDOW_SHOWN); + if (window == nullptr) { + throw std::runtime_error(sdlError("Runtime: SDL_CreateWindow failed")); + } + + SDL_Renderer* renderer = + SDL_CreateRenderer(window, -1, SDL_RENDERER_ACCELERATED | SDL_RENDERER_PRESENTVSYNC); + if (renderer == nullptr) { + SDL_DestroyWindow(window); + throw std::runtime_error(sdlError("Runtime: SDL_CreateRenderer failed")); + } + + return {renderer, [window](SDL_Renderer* target) { + if (target != nullptr) { + SDL_DestroyRenderer(target); + } + SDL_DestroyWindow(window); + }}; + }; +} + +Runtime::Runtime(const RuntimePaths& paths) : Runtime(paths, defaultRenderTargetFactory()) {} + +Runtime::Runtime(const RuntimePaths& paths, const RenderTargetFactory& renderTargetFactory) + : paths(paths), + config(ConfigParser::parse(paths.config)), + scene(Parser::parse(paths.experiment)), + eegQueue(std::make_shared>()), + markerQueue(std::make_shared>()) { + if (!renderTargetFactory) { + throw std::invalid_argument("Runtime: render target factory must not be null"); + } + + sdlRenderer = renderTargetFactory(scene->getExperimentName()); + if (!sdlRenderer) { + throw std::runtime_error("Runtime: render target factory returned no renderer"); + } + + auto formatStrategy = DataFormatStrategyFactory::create(config.output.format); + outputExtension = formatStrategy->fileExtension(); + + lslReader = std::make_unique(config); + dataWriter = std::make_unique(std::move(formatStrategy)); + renderer = std::make_unique(scene, sdlRenderer, markerQueue); +} + +Runtime::~Runtime() { shutdown(); } + +std::string Runtime::makeOutputPath() const { + const std::string name = sanitizeForFileName(scene->getExperimentName()); + + const auto now = std::chrono::system_clock::now().time_since_epoch(); + const auto epoch = std::chrono::duration_cast(now).count(); + + const std::filesystem::path fileName = + name + "_" + std::to_string(epoch) + "." + outputExtension; + return (std::filesystem::path(paths.outputDir) / fileName).string(); +} + +void Runtime::startWorkers(const std::string& outputPath) { + dataWriter->start(outputPath, eegQueue, markerQueue); + lslReader->start(eegQueue); +} + +void Runtime::run() { + outputFilePath = makeOutputPath(); + startWorkers(outputFilePath); + + renderer->render(stopSource.get_token()); + + shutdown(); +} + +void Runtime::requestStop() { stopSource.request_stop(); } + +void Runtime::shutdown() { + stopSource.request_stop(); -void Runtime::start() { std::cout << "Runtime started." << "\n"; } \ No newline at end of file + if (lslReader) { + lslReader->stop(); + } + if (dataWriter) { + dataWriter->stop(); + } +} diff --git a/src/config/ConfigParser.cpp b/src/config/ConfigParser.cpp index 11f93ef..5c43f5b 100644 --- a/src/config/ConfigParser.cpp +++ b/src/config/ConfigParser.cpp @@ -129,6 +129,14 @@ GroundConfig buildGround(const json& root) { return ground; } +OutputConfig buildOutput(const json& root) { + OutputConfig output; + if (root.contains("output")) { + output.format = requireField(root.at("output"), "format", "output"); + } + return output; +} + ImpedanceConfig buildImpedance(const json& root) { ImpedanceConfig impedance; if (root.contains("impedance_check")) { @@ -176,6 +184,7 @@ DeviceConfig ConfigParser::parseStream(std::istream& stream) { config.ground = buildGround(root); config.channels = buildChannels(root); config.impedance = buildImpedance(root); + config.output = buildOutput(root); // Mapping is done; the semantic rules belong to the types themselves. config.validate(); diff --git a/src/config/ConfigValidation.cpp b/src/config/ConfigValidation.cpp index ff44b99..1656af7 100644 --- a/src/config/ConfigValidation.cpp +++ b/src/config/ConfigValidation.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -59,11 +60,14 @@ void ImpedanceConfig::validate() const { } } +void OutputConfig::validate() const { requireNonEmpty(format, "format", "OutputConfig"); } + void DeviceConfig::validate() const { configVersion.validate(); requireNonEmpty(deviceName, "device_name", "DeviceConfig"); lsl.validate(); impedance.validate(); + output.validate(); if (static_cast(channels.size()) != lsl.expectedChannelCount) { throw std::invalid_argument("DeviceConfig: channel count mismatch: 'channels' has " + diff --git a/src/datawriter/CMakeLists.txt b/src/datawriter/CMakeLists.txt index c0b9322..b9a63e9 100644 --- a/src/datawriter/CMakeLists.txt +++ b/src/datawriter/CMakeLists.txt @@ -1,5 +1,6 @@ add_library(datawriter OBJECT CSVFormatStrategy.cpp + DataFormatStrategyFactory.cpp DataWriter.cpp ) diff --git a/src/datawriter/DataFormatStrategyFactory.cpp b/src/datawriter/DataFormatStrategyFactory.cpp new file mode 100644 index 0000000..d9560ea --- /dev/null +++ b/src/datawriter/DataFormatStrategyFactory.cpp @@ -0,0 +1,34 @@ +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { +std::string toLower(std::string value) { + std::transform(value.begin(), value.end(), value.begin(), + [](unsigned char character) { return std::tolower(character); }); + return value; +} + +using StrategyCreator = std::function()>; + +const std::unordered_map& creators() { + static const std::unordered_map registry = { + {"csv", [] { return std::make_unique(); }}, + }; + return registry; +} +} // namespace + +std::unique_ptr DataFormatStrategyFactory::create(const std::string& format) { + const auto entry = creators().find(toLower(format)); + if (entry == creators().end()) { + throw std::invalid_argument("DataFormatStrategyFactory: unknown output format '" + format + + "'"); + } + return entry->second(); +} diff --git a/src/main.cpp b/src/main.cpp index 824bdc3..3dcb40c 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -1,8 +1,42 @@ #include +#include +#include +#include +#include +#include + +namespace { +constexpr const char* kDefaultConfigPath = "config.json"; + +void printUsage(const char* program) { + std::cerr << "Usage: " << program << " [config.json] \n" + << " config.json defaults to '" << kDefaultConfigPath + << "' in the working directory.\n"; +} +} // namespace int main(int argc, char* argv[]) { - (void)argc; - (void)argv; - Runtime::start(); + const std::span args(argv, static_cast(argc)); + + RuntimePaths paths{.config = kDefaultConfigPath, .experiment = ""}; + + if (args.size() == 2) { + paths.experiment = args[1]; + } else if (args.size() >= 3) { + paths.config = args[1]; + paths.experiment = args[2]; + } else { + printUsage(args.empty() ? "NeuronIDE" : args[0]); + return 1; + } + + try { + Runtime runtime(paths); + runtime.run(); + } catch (const std::exception& e) { + std::cerr << "NeuronIDE: fatal: " << e.what() << "\n"; + return 1; + } + return 0; -} \ No newline at end of file +} diff --git a/tests/unit_tests/config_parser_test.cpp b/tests/unit_tests/config_parser_test.cpp index 279b5ff..a4db027 100644 --- a/tests/unit_tests/config_parser_test.cpp +++ b/tests/unit_tests/config_parser_test.cpp @@ -161,6 +161,36 @@ TEST(ConfigParserTest, ParsesReferenceGroundAndImpedance) { EXPECT_DOUBLE_EQ(config.impedance.thresholdKohm, kImpedanceThreshold); } +TEST(ConfigParserTest, ParsesOutputFormat) { + const std::string jsonText = R"json({ + "config_version": "1.0", "device_name": "Dev", "montage_standard": "10-20", + "lsl_stream": { + "name": "s", "type": "EEG", "source_id": "x", + "expected_channel_count": 1, "expected_sample_rate_hz": 250 + }, + "channels": [ { "index": 0, "label": "Fz", "enabled": true, "unit": "uV" } ], + "output": { "format": "csv" } + })json"; + EXPECT_EQ(parseString(jsonText).output.format, "csv"); +} + +TEST(ConfigParserTest, OutputFormatDefaultsToCsvWhenSectionAbsent) { + EXPECT_EQ(parseString(kMinimalConfig).output.format, "csv"); +} + +TEST(ConfigParserTest, MissingOutputFormatFieldThrows) { + const std::string jsonText = R"json({ + "config_version": "1.0", "device_name": "Dev", "montage_standard": "10-20", + "lsl_stream": { + "name": "s", "type": "EEG", "source_id": "x", + "expected_channel_count": 1, "expected_sample_rate_hz": 250 + }, + "channels": [ { "index": 0, "label": "Fz", "enabled": true, "unit": "uV" } ], + "output": { } + })json"; + EXPECT_THROW(parseString(jsonText), std::invalid_argument); +} + TEST(ConfigParserTest, MissingLslStreamThrows) { const std::string jsonText = R"json({ "config_version": "1.0", diff --git a/tests/unit_tests/config_validation_test.cpp b/tests/unit_tests/config_validation_test.cpp index ec6167f..a84bc86 100644 --- a/tests/unit_tests/config_validation_test.cpp +++ b/tests/unit_tests/config_validation_test.cpp @@ -125,6 +125,23 @@ TEST(ConfigValidationTest, NegativeImpedanceThresholdThrows) { EXPECT_THROW(config.validate(), std::invalid_argument); } +TEST(ConfigValidationTest, EmptyOutputFormatThrows) { + DeviceConfig config = makeValidConfig(); + config.output.format.clear(); + + EXPECT_THROW(config.validate(), std::invalid_argument); +} + +TEST(ConfigValidationTest, UnknownOutputFormatIsNotAConfigRule) { + // Which formats exist is DataFormatStrategyFactory's knowledge, not the + // config layer's; an unknown-but-present format passes validation and is + // rejected when the strategy is built. + DeviceConfig config = makeValidConfig(); + config.output.format = "parquet"; + + EXPECT_NO_THROW(config.validate()); +} + TEST(ConfigValidationTest, NegativeVersionComponentThrows) { DeviceConfig config = makeValidConfig(); config.configVersion.minor = -1; diff --git a/tests/unit_tests/data_format_strategy_factory_test.cpp b/tests/unit_tests/data_format_strategy_factory_test.cpp new file mode 100644 index 0000000..040bb55 --- /dev/null +++ b/tests/unit_tests/data_format_strategy_factory_test.cpp @@ -0,0 +1,25 @@ +#include + +#include +#include +#include + +TEST(DataFormatStrategyFactoryTest, CreatesCsvStrategy) { + const auto strategy = DataFormatStrategyFactory::create("csv"); + ASSERT_NE(strategy, nullptr); + EXPECT_EQ(strategy->fileExtension(), "csv"); +} + +TEST(DataFormatStrategyFactoryTest, FormatMatchingIsCaseInsensitive) { + const auto strategy = DataFormatStrategyFactory::create("CSV"); + ASSERT_NE(strategy, nullptr); + EXPECT_EQ(strategy->fileExtension(), "csv"); +} + +TEST(DataFormatStrategyFactoryTest, UnknownFormatThrows) { + EXPECT_THROW(DataFormatStrategyFactory::create("parquet"), std::invalid_argument); +} + +TEST(DataFormatStrategyFactoryTest, EmptyFormatThrows) { + EXPECT_THROW(DataFormatStrategyFactory::create(""), std::invalid_argument); +} diff --git a/tests/unit_tests/runtime_test.cpp b/tests/unit_tests/runtime_test.cpp new file mode 100644 index 0000000..616184d --- /dev/null +++ b/tests/unit_tests/runtime_test.cpp @@ -0,0 +1,249 @@ +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "utils/ParserTestUtils.hpp" + +namespace { +namespace fs = std::filesystem; + +constexpr int kSurfaceSize = 10; +constexpr int kSurfaceDepth = 32; +constexpr auto kRenderSpinWait = std::chrono::milliseconds(50); + +constexpr const char* kValidConfig = R"json({ + "config_version": "1.0", + "device_name": "Dev", + "montage_standard": "10-20", + "lsl_stream": { + "name": "runtime_test_stream", "type": "EEG", "source_id": "runtime-test-src", + "expected_channel_count": 1, "expected_sample_rate_hz": 250 + }, + "channels": [ { "index": 0, "label": "Fz", "enabled": true, "unit": "uV" } ] +})json"; + +// A software renderer backed by an in-memory surface: no window, GPU, or +// display required, so it runs anywhere (including headless CI). Injected in +// place of Runtime's production SDL window/vsync-renderer factory. +Runtime::RenderTargetFactory softwareRenderTargetFactory() { + return [](const std::string&) -> std::shared_ptr { + SDL_Surface* surface = SDL_CreateRGBSurfaceWithFormat( + 0, kSurfaceSize, kSurfaceSize, kSurfaceDepth, SDL_PIXELFORMAT_RGBA32); + SDL_Renderer* renderer = SDL_CreateSoftwareRenderer(surface); + return {renderer, [surface](SDL_Renderer* target) { + if (target != nullptr) { + SDL_DestroyRenderer(target); + } + if (surface != nullptr) { + SDL_FreeSurface(surface); + } + }}; + }; +} + +void writeText(const fs::path& path, const std::string& content) { + std::ofstream out(path, std::ios::binary); + out << content; +} + +void writeExperimentFile(const fs::path& path, const NeuronIDE::Scene& scene) { + std::ofstream out(path, std::ios::binary); + scene.SerializeToOstream(&out); +} + +std::vector readAllLines(const fs::path& path) { + std::ifstream input(path); + std::vector lines; + std::string line; + while (std::getline(input, line)) { + lines.push_back(line); + } + return lines; +} + +// Owns the temp files that back a Runtime so a test never leaves artifacts on +// disk, and provides a valid config + experiment scene by default. +class RuntimeFixture : public ::testing::Test { + protected: + void SetUp() override { + // SdlSession initializes SDL_INIT_VIDEO; the dummy driver makes that + // succeed without a display. The software renderer is unaffected by it. + setenv("SDL_VIDEODRIVER", "dummy", 1); + + tempDir = fs::temp_directory_path() / + ("neuronide_runtime_" + + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count())); + fs::create_directories(tempDir); + + writeText(tempDir / "config.json", kValidConfig); + writeExperimentFile(tempDir / "experiment.pb", utils::buildSimpleScene()); + } + + void TearDown() override { + std::error_code errorCode; + fs::remove_all(tempDir, errorCode); + } + + const fs::path& dir() const { return tempDir; } + + RuntimePaths paths() const { + return RuntimePaths{.config = (tempDir / "config.json").string(), + .experiment = (tempDir / "experiment.pb").string(), + .outputDir = tempDir.string()}; + } + + private: + fs::path tempDir; +}; +} // namespace + +TEST_F(RuntimeFixture, ConstructsWithValidInputs) { + EXPECT_NO_THROW({ const Runtime runtime(paths(), softwareRenderTargetFactory()); }); +} + +TEST_F(RuntimeFixture, OutputPathEmptyBeforeRun) { + const Runtime runtime(paths(), softwareRenderTargetFactory()); + EXPECT_TRUE(runtime.outputPath().empty()); +} + +TEST_F(RuntimeFixture, MissingConfigThrows) { + RuntimePaths badPaths = paths(); + badPaths.config = (dir() / "does_not_exist.json").string(); + + EXPECT_THROW(Runtime(badPaths, softwareRenderTargetFactory()), std::exception); +} + +TEST_F(RuntimeFixture, MissingExperimentThrows) { + RuntimePaths badPaths = paths(); + badPaths.experiment = (dir() / "does_not_exist.pb").string(); + + EXPECT_THROW(Runtime(badPaths, softwareRenderTargetFactory()), std::exception); +} + +TEST_F(RuntimeFixture, UnknownOutputFormatThrows) { + const std::string badFormatConfig = R"json({ + "config_version": "1.0", "device_name": "Dev", "montage_standard": "10-20", + "lsl_stream": { + "name": "runtime_test_stream", "type": "EEG", "source_id": "runtime-test-src", + "expected_channel_count": 1, "expected_sample_rate_hz": 250 + }, + "channels": [ { "index": 0, "label": "Fz", "enabled": true, "unit": "uV" } ], + "output": { "format": "parquet" } + })json"; + writeText(dir() / "config.json", badFormatConfig); + + EXPECT_THROW(Runtime(paths(), softwareRenderTargetFactory()), std::invalid_argument); +} + +TEST_F(RuntimeFixture, NullRenderTargetFactoryThrows) { + EXPECT_THROW(Runtime(paths(), Runtime::RenderTargetFactory{}), std::invalid_argument); +} + +TEST_F(RuntimeFixture, FactoryReturningNullRendererThrows) { + const auto nullFactory = [](const std::string&) -> std::shared_ptr { + return nullptr; + }; + EXPECT_THROW(Runtime(paths(), nullFactory), std::runtime_error); +} + +TEST_F(RuntimeFixture, OutputFilenameDerivedFromExperimentNameAndDir) { + Runtime runtime(paths(), softwareRenderTargetFactory()); + + // Stop the render loop immediately so run() returns after a single pass. + runtime.requestStop(); + runtime.run(); + + const fs::path output = runtime.outputPath(); + EXPECT_EQ(output.parent_path(), dir()); + // Scene project name is "TestProject" (see ParserTestUtils::buildSimpleScene). + EXPECT_EQ(output.filename().string().rfind("TestProject_", 0), 0U); + EXPECT_EQ(output.extension(), ".csv"); +} + +TEST_F(RuntimeFixture, OutputFilenameSanitizesTheExperimentName) { + // An authored name is not a safe file name: unfiltered, "block 1/run" would + // point at a subdirectory that does not exist. + writeExperimentFile(dir() / "experiment.pb", + utils::buildSimpleScene({.projectName = "block 1/run"})); + + Runtime runtime(paths(), softwareRenderTargetFactory()); + runtime.requestStop(); + runtime.run(); + + const fs::path output = runtime.outputPath(); + EXPECT_EQ(output.parent_path(), dir()); + EXPECT_EQ(output.filename().string().rfind("block_1_run_", 0), 0U); + EXPECT_TRUE(fs::exists(output)); +} + +TEST_F(RuntimeFixture, RunWritesRecordingWithCsvHeader) { + Runtime runtime(paths(), softwareRenderTargetFactory()); + + runtime.requestStop(); + runtime.run(); + + const fs::path output = runtime.outputPath(); + ASSERT_TRUE(fs::exists(output)); + + const auto lines = readAllLines(output); + ASSERT_FALSE(lines.empty()); + EXPECT_EQ(lines.front(), "type,timestamp,payload"); +} + +TEST_F(RuntimeFixture, RunStopsOnQuitEvent) { + Runtime runtime(paths(), softwareRenderTargetFactory()); + + SDL_Event quit; + quit.type = SDL_QUIT; + ASSERT_EQ(SDL_PushEvent(&quit), 1); + + // Must return promptly on SDL_QUIT even though no stop was requested. + runtime.run(); + + EXPECT_TRUE(fs::exists(runtime.outputPath())); +} + +TEST_F(RuntimeFixture, RequestStopFromAnotherThreadEndsRun) { + Runtime runtime(paths(), softwareRenderTargetFactory()); + + // Ensure no stale SDL_QUIT from a previous test ends the loop early. + SDL_FlushEvent(SDL_QUIT); + + std::thread worker([&runtime] { runtime.run(); }); + std::this_thread::sleep_for(kRenderSpinWait); + runtime.requestStop(); + worker.join(); + + EXPECT_TRUE(fs::exists(runtime.outputPath())); +} + +// Covers the production SDL window + vsync renderer factory where the platform +// can provide an accelerated renderer; skipped on headless machines that can't. +TEST_F(RuntimeFixture, DefaultRenderTargetFactoryOnCapablePlatform) { + // Let SDL pick the best available driver instead of the forced dummy one, + // which cannot provide an accelerated renderer. + unsetenv("SDL_VIDEODRIVER"); + + try { + Runtime runtime(paths(), Runtime::defaultRenderTargetFactory()); + + SDL_Event quit; + quit.type = SDL_QUIT; + ASSERT_EQ(SDL_PushEvent(&quit), 1); + runtime.run(); + + EXPECT_TRUE(fs::exists(runtime.outputPath())); + } catch (const std::exception& e) { + GTEST_SKIP() << "No accelerated render target on this platform: " << e.what(); + } +} From 5b4d1519557d58179e82e56c0e13992a8c30d7a6 Mon Sep 17 00:00:00 2001 From: Michal Date: Fri, 9 Oct 2026 17:01:37 +0200 Subject: [PATCH 2/5] fix(runtime): gate experiment on live EEG and protect recordings - wait for first EEG sample before rendering; stop run if acquisition fails - make run() single-shot, never overwrite output files, readable file names - require vsync and add optional "display" section to config.json - validate config before opening the window, enforce run() thread - fix scene/renderer teardown order, update tests and README --- README.md | 57 ++- include/Runtime.hpp | 68 ++- include/config/DeviceConfig.hpp | 6 +- include/config/DisplayConfig.hpp | 22 + include/datawriter/IDataFormatStrategy.hpp | 2 + include/lslreader/LSLReader.hpp | 47 +- src/Runtime.cpp | 227 ++++++++-- src/config/ConfigParser.cpp | 24 + src/config/ConfigValidation.cpp | 13 + src/datawriter/CSVFormatStrategy.cpp | 19 +- src/lslreader/LSLReader.cpp | 76 +++- tests/unit_tests/config_parser_test.cpp | 49 ++ tests/unit_tests/config_validation_test.cpp | 17 + tests/unit_tests/data_writer_test.cpp | 18 + tests/unit_tests/lslreader_test.cpp | 95 ++++ tests/unit_tests/runtime_test.cpp | 468 ++++++++++++++++---- 16 files changed, 1057 insertions(+), 151 deletions(-) create mode 100644 include/config/DisplayConfig.hpp diff --git a/README.md b/README.md index 6254121..efa93db 100644 --- a/README.md +++ b/README.md @@ -192,6 +192,7 @@ otherwise the same experiment stops being portable between labs. | Stimulus timing, trial structure, marker/event names | Expected stream shape: channel count, sample rate | | Anything the editor authors and versions with the study | Channel table: index, label, enabled, unit | | | Reference / ground electrodes, impedance check thresholds | +| | Participant display: monitor, fullscreen, window size (`display`) | | | Output format for `DataWriter` (`output.format`) | Consequences of the split: @@ -210,7 +211,7 @@ way — `channels` is a top-level key, so it is a top-level `DeviceConfig` field ```jsonc { - "config_version": "1.0", // "MAJOR.MINOR", checked first (see below) + "config_version": "1.1", // "MAJOR.MINOR", checked first (see below) "device_name": "OpenBCI Cyton 8ch", "montage_standard": "10-20", "lsl_stream": { // -> DeviceConfig::lsl (LSLConfig) @@ -228,13 +229,20 @@ way — `channels` is a top-level key, so it is a top-level `DeviceConfig` field // ... one entry per expected_channel_count, indices unique and in range ], "impedance_check": { "supported": true, "threshold_kohm": 5.0 }, + "display": { // -> DeviceConfig::display; every key optional + "index": 1, // monitor facing the participant (default 0) + "fullscreen": true, // native-resolution fullscreen (default true) + "width": 1280, "height": 720 // window size, used only when not fullscreen + }, "output": { "format": "csv" } // -> DataFormatStrategyFactory, defaults to csv } ``` `config_version`, `device_name`, `montage_standard`, `lsl_stream` and `channels` -are required; `reference`, `ground`, `impedance_check` and `output` default when -absent. If `output` is present it must carry a non-empty `format`; whether that +are required; `reference`, `ground`, `impedance_check`, `display` and `output` +default when absent. Unlike the other optional sections, each key inside `display` +is optional on its own (`"display": { "index": 1 }` is complete). If `output` is +present it must carry a non-empty `format`; whether that format is *known* is decided by `DataFormatStrategyFactory` when the writer is built, not by config validation. Channels with `"enabled": false` stay in the config (they document the cap) but @@ -310,6 +318,39 @@ the re-resolved stream is re-validated (channel count, sample rate) before acquisition continues. Config errors (mismatched stream shape, no enabled channels) stay fatal — they are logged and the worker exits instead of retrying forever. +`LSLReader` reports where it is as a typed `AcquisitionState`: `Idle` → `Resolving` +→ `Streaming` (set on the first *queued sample*, not on resolve), back to +`Resolving` after a lost stream, or `Failed` for good. `waitWhileResolving(deadline, +stop_token)` blocks until the state changes; `failure()` holds the reason, and an +optional callback passed to `start()` fires once on the reader thread on failure. +A `Failed` state survives `stop()` so the reason can be read after joining. + +### `Runtime::run()` sequence + +`Runtime` builds everything that can reject the configuration (both config files, +output directory, format strategy, `LSLReader`) **before** it creates the window, so a +bad config never flashes a window at the participant. `run()` then: + +1. **Acquisition gate.** Starts `LSLReader` and waits for the first EEG sample, + handling window events meanwhile. **No stimulus is shown without EEG**: a stream + that fails validation, or delivers nothing within the acquisition timeout + (default 30 s), throws; closing the window or `requestStop()` returns without + recording. +2. **Recording.** Creates `_[_N].` in the output + directory and runs the render loop. Format strategies must create the file + exclusively — an existing recording is never overwritten. +3. **Shutdown.** Stops `LSLReader`, then `DataWriter` (producer before consumer, so + the writer drains everything). If acquisition failed for good mid-experiment, + the failure callback has already stopped the render loop, and `run()` throws + once the recording is closed. + +`run()` is single-shot and must be called on the thread that constructed the +`Runtime` (SDL handles window events only there); both are enforced with +`std::logic_error`. The default render target opens the window on +`display.index` and **refuses to run without vsync** — SDL silently drops +`PRESENTVSYNC` when the driver can't honour it, and marker timestamps are only +correct when `SDL_RenderPresent` waits for the refresh. + ## 7. Implementation status | Area / class | Status | Notes | @@ -320,12 +361,12 @@ stay fatal — they are logged and the worker exits instead of retrying forever. | `ComponentRegistry` | Implemented | proto-type → factory, macro-based self-registration | | `specifiic components` | **Planned** | defined in `neuronide.proto`, not yet implemented in C++ | | `Renderer` | Implemented | SDL + vsync, marker timestamping | -| `LSLReader` | Implemented | LSL inlet → `eegQueue`, clock-synced (see §4); driven by `DeviceConfig`, enabled channels only, re-resolves lost streams | +| `LSLReader` | Implemented | LSL inlet → `eegQueue`, clock-synced (see §4); driven by `DeviceConfig`, enabled channels only, re-resolves lost streams, reports `AcquisitionState` | | `ConfigParser` | Implemented | `config.json` → `DeviceConfig` (1:1 mapping, major-version checked, see §5), nlohmann/json | | Config `validate()` | Implemented | semantic rules on the config types themselves, independent of JSON (see §5) | | `DataWriter` | Implemented | strategy-based; `CSVFormatStrategy` | | `DataFormatStrategyFactory` | Implemented | `output.format` → format strategy; unknown formats rejected | -| `Runtime` orchestration | Implemented | owns SDL session, parses both config files, wires the queues, drives the render loop, stops workers | +| `Runtime` orchestration | Implemented | owns SDL session, parses both config files, gates the experiment on live EEG, records to a fresh file, drives the render loop, stops workers (see §6) | The class diagram in older docs is partly aspirational; the table above reflects the actual code. @@ -375,7 +416,7 @@ sudo apt install cmake clang-format clang-tidy libsdl2-dev protobuf-compiler gco ```bash cmake -B build cmake --build build -./build/src/NeuronIDE config.json experiment.neuroz # parses the device config, parses scene, starts LSLReader, DataWriter, Renderer. +./build/src/NeuronIDE config.json experiment.neuroz # waits for the EEG stream, then records and runs the experiment ``` `NeuronIDE` takes the path to a device `config.json` (defaults to `config.json` in @@ -390,8 +431,8 @@ ctest -L unit --output-on-failure # unit tests only ctest -L component --output-on-failure ``` -> Note: `LSLReader` unit tests open a local LSL stream and exercise a real -> outlet→inlet round-trip over loopback; they need loopback multicast to be +> Note: `LSLReader` and `Runtime` unit tests open local LSL streams and exercise a +> real outlet→inlet round-trip over loopback; they need loopback multicast to be > available. ### Formatting & static analysis diff --git a/include/Runtime.hpp b/include/Runtime.hpp index f577e0c..d181041 100644 --- a/include/Runtime.hpp +++ b/include/Runtime.hpp @@ -3,11 +3,13 @@ #include +#include #include #include #include #include #include +#include class Scene; class LSLReader; @@ -24,16 +26,24 @@ class SDL_Renderer; struct RuntimePaths { std::string config; // device config.json std::string experiment; // serialized experiment scene (protobuf) - std::string outputDir = "."; // directory the recorded CSV is written to + std::string outputDir = "."; // existing directory the recording is written to }; class Runtime { public: - using RenderTargetFactory = - std::function(const std::string& windowTitle)>; + using RenderTargetFactory = std::function( + const std::string& windowTitle, const DisplayConfig& display)>; + // How long run() waits for the first EEG sample before giving up. + static constexpr std::chrono::seconds kDefaultAcquisitionTimeout{30}; + + // Parses and validates both input files and builds every worker before the + // render target is created, so a bad configuration fails without opening a + // window. The constructing thread becomes the only thread allowed to call + // run(). explicit Runtime(const RuntimePaths& paths); - Runtime(const RuntimePaths& paths, const RenderTargetFactory& renderTargetFactory); + Runtime(const RuntimePaths& paths, const RenderTargetFactory& renderTargetFactory, + std::chrono::milliseconds acquisitionTimeout = kDefaultAcquisitionTimeout); ~Runtime(); Runtime(const Runtime&) = delete; @@ -41,18 +51,36 @@ class Runtime { Runtime(Runtime&&) = delete; Runtime& operator=(Runtime&&) = delete; - // Drives the experiment on the calling thread until the render loop ends - // (SDL_QUIT or requestStop), then stops the workers. Single-shot: the stop - // source is never re-armed, so a second run() returns immediately. + // Runs the experiment on the calling thread, which must be the thread that + // constructed this Runtime: SDL window events can only be handled there. + // + // 1. Starts EEG acquisition and waits for the first sample. The experiment + // never starts without EEG: a stream that fails validation, or delivers + // nothing within the acquisition timeout, throws std::runtime_error. + // Closing the window or requestStop() during the wait returns without + // recording anything. + // 2. Creates a new recording file (an existing file is never overwritten) + // and renders until SDL_QUIT or requestStop(). + // 3. Stops the workers. If acquisition failed for good during the + // experiment, rendering stops at once, the recording is closed, and + // run() throws std::runtime_error. + // + // Single-shot: throws std::logic_error when called a second time or from + // the wrong thread. void run(); // Safe to call from any thread, including while run() is in progress. void requestStop(); - // Empty until run() has computed it. Not synchronized: read it before run() - // starts or after it has returned. + // Path of the recording, or empty if run() has not started one (yet, or + // at all). Not synchronized: read it before run() starts or after it has + // returned. const std::string& outputPath() const noexcept { return outputFilePath; } + // Opens a window on display.index (fullscreen or display.width x height) + // with an accelerated renderer, and throws std::runtime_error unless that + // renderer actually presents in sync with the display refresh. Marker + // timestamps are only correct when it does. static RenderTargetFactory defaultRenderTargetFactory(); private: @@ -66,21 +94,29 @@ class Runtime { SdlSession& operator=(SdlSession&&) = delete; }; - void startWorkers(const std::string& outputPath); - void shutdown(); - std::string makeOutputPath() const; + bool awaitAcquisition(); + void startRecording(); + void shutdown(); + [[noreturn]] void throwAcquisitionFailure(const std::string& phase) const; + [[nodiscard]] std::string makeOutputPath() const; - RuntimePaths paths; - DeviceConfig config; + RuntimePaths paths; + DeviceConfig config; + std::chrono::milliseconds acquisitionTimeout; + std::thread::id ownerThread; + bool hasRun = false; SdlSession sdlSession; + // Declared before `scene` on purpose: members are destroyed in reverse + // order, so the scene - and any SDL_Texture its components hold - is + // released while the SDL_Renderer that owns those textures is still alive. + std::shared_ptr sdlRenderer; + std::shared_ptr scene; std::shared_ptr> eegQueue; std::shared_ptr> markerQueue; - std::shared_ptr sdlRenderer; - std::unique_ptr lslReader; std::unique_ptr dataWriter; std::unique_ptr renderer; diff --git a/include/config/DeviceConfig.hpp b/include/config/DeviceConfig.hpp index ca7e741..a7ea465 100644 --- a/include/config/DeviceConfig.hpp +++ b/include/config/DeviceConfig.hpp @@ -3,6 +3,7 @@ #include #include +#include #include #include #include @@ -25,8 +26,8 @@ struct ImpedanceConfig { void validate() const; }; -// Describes the acquisition hardware (the cap and its LSL stream), not the -// experiment - experiment content lives in the protobuf file. Mirrors +// Describes the lab hardware (the cap, its LSL stream, the participant's +// display), not the experiment - experiment content lives in the protobuf file. Mirrors // `config.json` 1:1: every JSON key maps onto exactly one field below. struct DeviceConfig { ConfigVersion configVersion; // config_version @@ -37,6 +38,7 @@ struct DeviceConfig { GroundConfig ground; // ground std::vector channels; // channels ImpedanceConfig impedance; // impedance_check + DisplayConfig display; // display OutputConfig output; // output // Checks every rule a device config must satisfy, including the cross-field diff --git a/include/config/DisplayConfig.hpp b/include/config/DisplayConfig.hpp new file mode 100644 index 0000000..80e8006 --- /dev/null +++ b/include/config/DisplayConfig.hpp @@ -0,0 +1,22 @@ +#ifndef DISPLAYCONFIG_HPP +#define DISPLAYCONFIG_HPP + +// The screen the participant looks at (`display` in `config.json`). It belongs +// to the lab setup, not the experiment: the same experiment file runs on +// whichever monitor faces the participant at this particular machine. +struct DisplayConfig { + static constexpr int kDefaultWindowWidth = 1280; + static constexpr int kDefaultWindowHeight = 720; + + int index = 0; // display.index: monitor the stimulus window opens on + bool fullscreen = true; // display.fullscreen: native-resolution fullscreen + int width = kDefaultWindowWidth; // display.width: window size, windowed mode only + int height = kDefaultWindowHeight; // display.height + + // Throws std::invalid_argument on a negative index or a non-positive window + // size. Whether `index` names a connected monitor can only be checked when + // the window is created. + void validate() const; +}; + +#endif // DISPLAYCONFIG_HPP diff --git a/include/datawriter/IDataFormatStrategy.hpp b/include/datawriter/IDataFormatStrategy.hpp index 28992a7..606d78e 100644 --- a/include/datawriter/IDataFormatStrategy.hpp +++ b/include/datawriter/IDataFormatStrategy.hpp @@ -18,6 +18,8 @@ class IDataFormatStrategy { virtual std::string fileExtension() const = 0; + // Creates `filepath` for writing. Must throw rather than overwrite an + // existing file: a truncated recording cannot be recovered. virtual void open(const std::string& filepath) = 0; virtual void close() = 0; diff --git a/include/lslreader/LSLReader.hpp b/include/lslreader/LSLReader.hpp index c289c78..70001d3 100644 --- a/include/lslreader/LSLReader.hpp +++ b/include/lslreader/LSLReader.hpp @@ -3,20 +3,37 @@ #include +#include +#include #include #include +#include +#include +#include #include +#include +#include #include #include struct EEGData; +enum class AcquisitionState : std::uint8_t { + Idle, // not started, or stopped + Resolving, // looking for the stream: right after start(), and again after it was lost + Streaming, // samples from the resolved stream are being queued + Failed, // gave up for good (see LSLReader::failure()); the worker has exited +}; + // Acquires the device's LSL stream on its own worker thread and pushes samples // onto eegQueue. Only the channels the device config marks as enabled are // forwarded; each pushed EEGData holds those channels in config declaration // order, so its values line up with the enabled entries of DeviceConfig::channels. class LSLReader { public: + // Invoked once, on the reader thread, when acquisition enters Failed. + using FailureCallback = std::function; + // Validates the config (see DeviceConfig::validate) and throws // std::invalid_argument if it is invalid or enables no channels. explicit LSLReader(DeviceConfig deviceConfig); @@ -27,18 +44,44 @@ class LSLReader { LSLReader(LSLReader&&) = delete; LSLReader& operator=(LSLReader&&) = delete; - void start(std::shared_ptr> eegQueue); + // Returns immediately; the stream is resolved on the worker thread. The + // state is Resolving by the time start() returns. + void start(std::shared_ptr> eegQueue, + FailureCallback onFailure = {}); + // Joins the worker. A Failed state and its failure() survive stop(), so the + // reason can still be read after the worker is gone; any other state + // becomes Idle. void stop(); + AcquisitionState state() const; + + // Blocks while the reader is Resolving, until `deadline` or until stopToken + // fires, and returns the state it ended on. + AcquisitionState waitWhileResolving(std::chrono::steady_clock::time_point deadline, + const std::stop_token& stopToken) const; + + // Why acquisition failed; null unless state() is Failed. + std::exception_ptr failure() const; + private: void readLoop(const std::stop_token& stopToken); + void setState(AcquisitionState next); + void fail(std::exception_ptr reason); DeviceConfig config; // Sample offsets to forward, in config declaration order. std::vector enabledChannelIndices; bool forwardsWholeSample = false; std::shared_ptr> eegQueue; - std::jthread readerThread; + FailureCallback onFailure; + + mutable std::mutex stateMutex; + mutable std::condition_variable_any stateChanged; + AcquisitionState currentState = AcquisitionState::Idle; + std::exception_ptr failureReason; + + // Last, so it is joined before the members the worker uses are destroyed. + std::jthread readerThread; }; #endif // LSLREADER_HPP diff --git a/src/Runtime.cpp b/src/Runtime.cpp index 2082d6b..bab8e83 100644 --- a/src/Runtime.cpp +++ b/src/Runtime.cpp @@ -2,15 +2,19 @@ #include #include +#include #include #include #include +#include #include #include #include #include #include +#include #include +#include #include #include #include @@ -18,13 +22,19 @@ #include #include #include +#include #include namespace { -constexpr int kWindowWidth = 1280; -constexpr int kWindowHeight = 720; -constexpr const char* kDefaultTitle = "NeuronIDE"; -constexpr const char* kFallbackName = "experiment"; +namespace fs = std::filesystem; + +constexpr const char* kDefaultTitle = "NeuronIDE"; +constexpr const char* kFallbackName = "experiment"; +constexpr const char* kTimestampFormat = "%Y%m%dT%H%M%S"; // e.g. 20261009T143012 + +constexpr std::size_t kTimestampBufferSize = 32; +// How often the window's events are handled while waiting for the EEG stream. +constexpr auto kEventPollInterval = std::chrono::milliseconds(50); std::string sdlError(const char* what) { return std::string(what) + ": " + SDL_GetError(); } @@ -45,6 +55,44 @@ std::string sanitizeForFileName(std::string name) { '_'); return name; } + +// Local wall-clock time, as a researcher reads it when looking for a session. +std::string localTimestamp() { + const std::time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now()); + std::tm local{}; + localtime_r(&now, &local); + + std::array buffer{}; + const std::size_t length = + std::strftime(buffer.data(), buffer.size(), kTimestampFormat, &local); + return {buffer.data(), length}; +} + +std::string describe(const std::exception_ptr& error) { + if (!error) { + return "acquisition stopped unexpectedly"; + } + try { + std::rethrow_exception(error); + } catch (const std::exception& e) { + return e.what(); + } catch (...) { + return "unknown error"; + } +} + +// Handles every pending window event, so the window stays responsive while +// nothing is being rendered yet, and reports whether the operator closed it. +bool quitRequested() { + bool quit = false; + SDL_Event event{}; + while (SDL_PollEvent(&event) == 1) { + if (event.type == SDL_QUIT) { + quit = true; + } + } + return quit; +} } // namespace Runtime::SdlSession::SdlSession() { @@ -56,24 +104,60 @@ Runtime::SdlSession::SdlSession() { Runtime::SdlSession::~SdlSession() { SDL_Quit(); } Runtime::RenderTargetFactory Runtime::defaultRenderTargetFactory() { - return [](const std::string& windowTitle) -> std::shared_ptr { + return [](const std::string& windowTitle, + const DisplayConfig& display) -> std::shared_ptr { const std::string title = windowTitle.empty() ? kDefaultTitle : windowTitle; - SDL_Window* window = - SDL_CreateWindow(title.c_str(), SDL_WINDOWPOS_CENTERED, SDL_WINDOWPOS_CENTERED, - kWindowWidth, kWindowHeight, SDL_WINDOW_SHOWN); - if (window == nullptr) { + const int displayCount = SDL_GetNumVideoDisplays(); + if (displayCount < 1) { + throw std::runtime_error(sdlError("Runtime: no display available")); + } + if (display.index >= displayCount) { + throw std::runtime_error("Runtime: display.index is " + std::to_string(display.index) + + " but only " + std::to_string(displayCount) + + " display(s) are connected"); + } + + // Desktop fullscreen keeps the monitor's native mode: no mode switch, + // and the compositor can hand the window the display directly. + Uint32 windowFlags = SDL_WINDOW_SHOWN; + if (display.fullscreen) { + windowFlags |= SDL_WINDOW_FULLSCREEN_DESKTOP; + } + const int position = static_cast(SDL_WINDOWPOS_CENTERED_DISPLAY(display.index)); + + std::unique_ptr window( + SDL_CreateWindow(title.c_str(), position, position, display.width, display.height, + windowFlags), + &SDL_DestroyWindow); + if (!window) { throw std::runtime_error(sdlError("Runtime: SDL_CreateWindow failed")); } - SDL_Renderer* renderer = - SDL_CreateRenderer(window, -1, SDL_RENDERER_ACCELERATED | SDL_RENDERER_PRESENTVSYNC); - if (renderer == nullptr) { - SDL_DestroyWindow(window); + std::unique_ptr renderer( + SDL_CreateRenderer(window.get(), -1, + SDL_RENDERER_ACCELERATED | SDL_RENDERER_PRESENTVSYNC), + &SDL_DestroyRenderer); + if (!renderer) { throw std::runtime_error(sdlError("Runtime: SDL_CreateRenderer failed")); } - return {renderer, [window](SDL_Renderer* target) { + // SDL quietly drops PRESENTVSYNC when the driver cannot honour it and + // still returns a renderer. Markers are stamped right after + // SDL_RenderPresent, so without vsync they would not match the frame + // that is actually on screen. + SDL_RendererInfo info{}; + if (SDL_GetRendererInfo(renderer.get(), &info) != 0) { + throw std::runtime_error(sdlError("Runtime: SDL_GetRendererInfo failed")); + } + if ((info.flags & SDL_RENDERER_PRESENTVSYNC) == 0U) { + throw std::runtime_error( + std::string("Runtime: renderer '") + info.name + + "' cannot present in sync with the display refresh; marker timestamps would " + "not match the frames on screen"); + } + + return {renderer.release(), [window = window.release()](SDL_Renderer* target) { if (target != nullptr) { SDL_DestroyRenderer(target); } @@ -84,58 +168,137 @@ Runtime::RenderTargetFactory Runtime::defaultRenderTargetFactory() { Runtime::Runtime(const RuntimePaths& paths) : Runtime(paths, defaultRenderTargetFactory()) {} -Runtime::Runtime(const RuntimePaths& paths, const RenderTargetFactory& renderTargetFactory) +Runtime::Runtime(const RuntimePaths& paths, const RenderTargetFactory& renderTargetFactory, + std::chrono::milliseconds acquisitionTimeout) : paths(paths), config(ConfigParser::parse(paths.config)), + acquisitionTimeout(acquisitionTimeout), + ownerThread(std::this_thread::get_id()), scene(Parser::parse(paths.experiment)), eegQueue(std::make_shared>()), markerQueue(std::make_shared>()) { if (!renderTargetFactory) { throw std::invalid_argument("Runtime: render target factory must not be null"); } - - sdlRenderer = renderTargetFactory(scene->getExperimentName()); - if (!sdlRenderer) { - throw std::runtime_error("Runtime: render target factory returned no renderer"); + if (acquisitionTimeout <= std::chrono::milliseconds::zero()) { + throw std::invalid_argument("Runtime: acquisition timeout must be positive"); + } + if (!fs::is_directory(paths.outputDir)) { + throw std::invalid_argument("Runtime: output directory does not exist: " + paths.outputDir); } + // Everything that can still reject the configuration runs before the window + // opens, so a bad config fails without flashing a window at the participant. auto formatStrategy = DataFormatStrategyFactory::create(config.output.format); outputExtension = formatStrategy->fileExtension(); lslReader = std::make_unique(config); dataWriter = std::make_unique(std::move(formatStrategy)); - renderer = std::make_unique(scene, sdlRenderer, markerQueue); + + sdlRenderer = renderTargetFactory(scene->getExperimentName(), config.display); + if (!sdlRenderer) { + throw std::runtime_error("Runtime: render target factory returned no renderer"); + } + + renderer = std::make_unique(scene, sdlRenderer, markerQueue); } Runtime::~Runtime() { shutdown(); } +// _., e.g. "ssvep_20261009T143012.csv". A numeric +// suffix keeps a second recording started within the same second from reusing +// the name; the format strategy refuses to overwrite regardless, so losing a +// race on the name fails loudly instead of destroying a recording. std::string Runtime::makeOutputPath() const { - const std::string name = sanitizeForFileName(scene->getExperimentName()); + const std::string stem = + sanitizeForFileName(scene->getExperimentName()) + "_" + localTimestamp(); + const fs::path directory(paths.outputDir); - const auto now = std::chrono::system_clock::now().time_since_epoch(); - const auto epoch = std::chrono::duration_cast(now).count(); + fs::path candidate = directory / (stem + "." + outputExtension); + for (int attempt = 2; fs::exists(candidate); ++attempt) { + candidate = directory / (stem + "_" + std::to_string(attempt) + "." + outputExtension); + } + return candidate.string(); +} + +// True once samples are flowing; false if the window was closed or +// requestStop() was called first. +bool Runtime::awaitAcquisition() { + std::cerr << "Runtime: waiting for EEG stream '" << config.lsl.name << "'...\n"; + + const std::stop_token token = stopSource.get_token(); + const auto deadline = std::chrono::steady_clock::now() + acquisitionTimeout; + + while (!token.stop_requested()) { + if (quitRequested()) { + return false; + } + + const auto now = std::chrono::steady_clock::now(); + if (now >= deadline) { + throw std::runtime_error( + "Runtime: EEG stream '" + config.lsl.name + "' delivered no samples within " + + std::to_string( + std::chrono::duration_cast(acquisitionTimeout).count()) + + " s; the experiment was not started"); + } - const std::filesystem::path fileName = - name + "_" + std::to_string(epoch) + "." + outputExtension; - return (std::filesystem::path(paths.outputDir) / fileName).string(); + switch ( + lslReader->waitWhileResolving(std::min(now + kEventPollInterval, deadline), token)) { + case AcquisitionState::Streaming: + return true; + case AcquisitionState::Resolving: + break; // not yet: handle window events and keep waiting + case AcquisitionState::Idle: + case AcquisitionState::Failed: + throwAcquisitionFailure("before the experiment started"); + } + } + return false; } -void Runtime::startWorkers(const std::string& outputPath) { - dataWriter->start(outputPath, eegQueue, markerQueue); - lslReader->start(eegQueue); +void Runtime::startRecording() { + outputFilePath = makeOutputPath(); + dataWriter->start(outputFilePath, eegQueue, markerQueue); } void Runtime::run() { - outputFilePath = makeOutputPath(); - startWorkers(outputFilePath); + if (std::this_thread::get_id() != ownerThread) { + throw std::logic_error( + "Runtime: run() must be called on the thread that constructed the Runtime"); + } + if (hasRun) { + throw std::logic_error("Runtime: run() was already called; a Runtime runs once"); + } + hasRun = true; - renderer->render(stopSource.get_token()); + try { + // A failure that ends acquisition for good also ends the experiment: + // stimuli without EEG are useless for analysis. + lslReader->start(eegQueue, [this] { stopSource.request_stop(); }); + if (awaitAcquisition()) { + startRecording(); + renderer->render(stopSource.get_token()); + } + } catch (...) { + shutdown(); + throw; + } shutdown(); + + if (lslReader->state() == AcquisitionState::Failed) { + throwAcquisitionFailure("during the experiment; the recording was stopped"); + } } void Runtime::requestStop() { stopSource.request_stop(); } +void Runtime::throwAcquisitionFailure(const std::string& phase) const { + throw std::runtime_error("Runtime: EEG acquisition failed " + phase + ": " + + describe(lslReader->failure())); +} + void Runtime::shutdown() { stopSource.request_stop(); diff --git a/src/config/ConfigParser.cpp b/src/config/ConfigParser.cpp index 5c43f5b..79c8e68 100644 --- a/src/config/ConfigParser.cpp +++ b/src/config/ConfigParser.cpp @@ -30,6 +30,12 @@ T requireField(const json& obj, const char* key, std::string_view ctx) { return requireMember(obj, key, ctx)->get(); } +template +T optionalField(const json& obj, const char* key, const T& fallback) { + const auto member = obj.find(key); + return member == obj.end() ? fallback : member->get(); +} + bool toUnsigned(std::string_view text, int& out) { const bool digitsOnly = !text.empty() && std::all_of(text.begin(), text.end(), [](unsigned char character) { @@ -137,6 +143,23 @@ OutputConfig buildOutput(const json& root) { return output; } +// Unlike the other optional sections, every key in `display` is optional on its +// own: `{ "index": 1 }` alone is a complete statement ("fullscreen on monitor 1"). +DisplayConfig buildDisplay(const json& root) { + DisplayConfig display; + if (root.contains("display")) { + const json& displayJson = root.at("display"); + if (!displayJson.is_object()) { + throw std::invalid_argument("ConfigParser: 'display' must be an object"); + } + display.index = optionalField(displayJson, "index", display.index); + display.fullscreen = optionalField(displayJson, "fullscreen", display.fullscreen); + display.width = optionalField(displayJson, "width", display.width); + display.height = optionalField(displayJson, "height", display.height); + } + return display; +} + ImpedanceConfig buildImpedance(const json& root) { ImpedanceConfig impedance; if (root.contains("impedance_check")) { @@ -184,6 +207,7 @@ DeviceConfig ConfigParser::parseStream(std::istream& stream) { config.ground = buildGround(root); config.channels = buildChannels(root); config.impedance = buildImpedance(root); + config.display = buildDisplay(root); config.output = buildOutput(root); // Mapping is done; the semantic rules belong to the types themselves. diff --git a/src/config/ConfigValidation.cpp b/src/config/ConfigValidation.cpp index 1656af7..e3f987e 100644 --- a/src/config/ConfigValidation.cpp +++ b/src/config/ConfigValidation.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -60,6 +61,17 @@ void ImpedanceConfig::validate() const { } } +void DisplayConfig::validate() const { + if (index < 0) { + throw std::invalid_argument("DisplayConfig: 'index' must not be negative, got " + + std::to_string(index)); + } + if (width <= 0 || height <= 0) { + throw std::invalid_argument("DisplayConfig: 'width' and 'height' must be positive, got " + + std::to_string(width) + "x" + std::to_string(height)); + } +} + void OutputConfig::validate() const { requireNonEmpty(format, "format", "OutputConfig"); } void DeviceConfig::validate() const { @@ -67,6 +79,7 @@ void DeviceConfig::validate() const { requireNonEmpty(deviceName, "device_name", "DeviceConfig"); lsl.validate(); impedance.validate(); + display.validate(); output.validate(); if (static_cast(channels.size()) != lsl.expectedChannelCount) { diff --git a/src/datawriter/CSVFormatStrategy.cpp b/src/datawriter/CSVFormatStrategy.cpp index 69e9be1..de84da2 100644 --- a/src/datawriter/CSVFormatStrategy.cpp +++ b/src/datawriter/CSVFormatStrategy.cpp @@ -1,12 +1,29 @@ +#include +#include #include #include #include #include +#include CSVFormatStrategy::~CSVFormatStrategy() { CSVFormatStrategy::close(); } void CSVFormatStrategy::open(const std::string& filepath) { - outputFile.open(filepath, std::ios::out | std::ios::trunc); + // "x" (C11) creates the file, or fails if it already exists, in one atomic + // step: an existing recording is never truncated, not even by a race on + // the file name. C++20 streams cannot express this (std::ios::noreplace is + // C++23), hence the short-lived C handle. + // NOLINTNEXTLINE(cppcoreguidelines-owning-memory) + std::FILE* created = std::fopen(filepath.c_str(), "wx"); + if (created == nullptr) { + const int error = errno; + throw std::system_error(error, std::generic_category(), + "Failed to create CSV output file " + filepath); + } + // NOLINTNEXTLINE(cppcoreguidelines-owning-memory) + std::fclose(created); + + outputFile.open(filepath, std::ios::out); if (!outputFile.is_open()) { throw std::runtime_error("Failed to open CSV output file: " + filepath); } diff --git a/src/lslreader/LSLReader.cpp b/src/lslreader/LSLReader.cpp index 8775397..7e2c31c 100644 --- a/src/lslreader/LSLReader.cpp +++ b/src/lslreader/LSLReader.cpp @@ -1,9 +1,12 @@ #include "lslreader/LSLReader.hpp" +#include #include #include #include +#include #include +#include #include #include #include @@ -108,15 +111,25 @@ LSLReader::LSLReader(DeviceConfig deviceConfig) LSLReader::~LSLReader() { stop(); } -void LSLReader::start(std::shared_ptr> eegQueue) { +void LSLReader::start(std::shared_ptr> eegQueue, + FailureCallback onFailure) { stop(); - this->eegQueue = std::move(eegQueue); - readerThread = std::jthread([this](const std::stop_token& stopToken) { + this->eegQueue = std::move(eegQueue); + this->onFailure = std::move(onFailure); + { + const std::scoped_lock lock(stateMutex); + failureReason = nullptr; + currentState = AcquisitionState::Resolving; + } + stateChanged.notify_all(); + + readerThread = std::jthread([this](const std::stop_token& stopToken) { try { readLoop(stopToken); } catch (const std::exception& e) { std::cerr << "LSLReader: fatal error, stopping acquisition: " << e.what() << "\n"; + fail(std::current_exception()); } }); } @@ -128,10 +141,59 @@ void LSLReader::stop() { if (readerThread.joinable()) { readerThread.join(); } + + { + const std::scoped_lock lock(stateMutex); + if (currentState != AcquisitionState::Failed) { + currentState = AcquisitionState::Idle; + } + } + stateChanged.notify_all(); +} + +AcquisitionState LSLReader::state() const { + const std::scoped_lock lock(stateMutex); + return currentState; +} + +AcquisitionState LSLReader::waitWhileResolving(std::chrono::steady_clock::time_point deadline, + const std::stop_token& stopToken) const { + std::unique_lock lock(stateMutex); + stateChanged.wait_until(lock, stopToken, deadline, + [this] { return currentState != AcquisitionState::Resolving; }); + return currentState; +} + +std::exception_ptr LSLReader::failure() const { + const std::scoped_lock lock(stateMutex); + return failureReason; +} + +void LSLReader::setState(AcquisitionState next) { + { + const std::scoped_lock lock(stateMutex); + currentState = next; + } + stateChanged.notify_all(); +} + +void LSLReader::fail(std::exception_ptr reason) { + { + const std::scoped_lock lock(stateMutex); + failureReason = std::move(reason); + currentState = AcquisitionState::Failed; + } + stateChanged.notify_all(); + + if (onFailure) { + onFailure(); + } } void LSLReader::readLoop(const std::stop_token& stopToken) { while (!stopToken.stop_requested()) { + setState(AcquisitionState::Resolving); + const std::optional info = resolveStream(config.lsl, stopToken); if (!info.has_value()) { return; // stop requested while waiting for the cap @@ -144,6 +206,7 @@ void LSLReader::readLoop(const std::stop_token& stopToken) { lsl::post_monotonize); std::vector sample; + bool streaming = false; while (!stopToken.stop_requested()) { const double timestamp = inlet.pull_sample(sample, kPullTimeout); if (timestamp == 0.0) { @@ -156,6 +219,13 @@ void LSLReader::readLoop(const std::stop_token& stopToken) { eegQueue->enqueue( EEGData{timestamp, pickChannels(sample, enabledChannelIndices)}); } + + // Reported on the first queued sample, not on resolve: a stream + // that resolves but never delivers is not acquiring anything. + if (!streaming) { + streaming = true; + setState(AcquisitionState::Streaming); + } } } catch (const lsl::lost_error& e) { std::cerr << "LSLReader: stream '" << config.lsl.name << "' lost (" << e.what() diff --git a/tests/unit_tests/config_parser_test.cpp b/tests/unit_tests/config_parser_test.cpp index a4db027..adf0012 100644 --- a/tests/unit_tests/config_parser_test.cpp +++ b/tests/unit_tests/config_parser_test.cpp @@ -191,6 +191,55 @@ TEST(ConfigParserTest, MissingOutputFormatFieldThrows) { EXPECT_THROW(parseString(jsonText), std::invalid_argument); } +// A minimal valid config with the given raw `display` section. +std::string withDisplay(const std::string& display) { + return R"json({ + "config_version": "1.1", "device_name": "Dev", "montage_standard": "10-20", + "lsl_stream": { + "name": "s", "type": "EEG", "source_id": "x", + "expected_channel_count": 1, "expected_sample_rate_hz": 250 + }, + "channels": [ { "index": 0, "label": "Fz", "enabled": true, "unit": "uV" } ], + "display": )json" + + display + "\n}"; +} + +TEST(ConfigParserTest, ParsesDisplaySection) { + constexpr int kIndex = 1; + constexpr int kWidth = 1920; + constexpr int kHeight = 1080; + + const DeviceConfig config = parseString(withDisplay( + R"json({ "index": 1, "fullscreen": false, "width": 1920, "height": 1080 })json")); + EXPECT_EQ(config.display.index, kIndex); + EXPECT_FALSE(config.display.fullscreen); + EXPECT_EQ(config.display.width, kWidth); + EXPECT_EQ(config.display.height, kHeight); +} + +TEST(ConfigParserTest, DisplayDefaultsToFullscreenOnFirstMonitorWhenSectionAbsent) { + const DeviceConfig config = parseString(kMinimalConfig); + EXPECT_EQ(config.display.index, 0); + EXPECT_TRUE(config.display.fullscreen); +} + +TEST(ConfigParserTest, DisplayKeysAreIndividuallyOptional) { + const DeviceConfig config = parseString(withDisplay(R"json({ "index": 2 })json")); + EXPECT_EQ(config.display.index, 2); + EXPECT_TRUE(config.display.fullscreen); + EXPECT_EQ(config.display.width, DisplayConfig::kDefaultWindowWidth); + EXPECT_EQ(config.display.height, DisplayConfig::kDefaultWindowHeight); +} + +TEST(ConfigParserTest, DisplayFieldWithWrongTypeThrows) { + EXPECT_THROW(parseString(withDisplay(R"json({ "fullscreen": "yes" })json")), + std::invalid_argument); +} + +TEST(ConfigParserTest, DisplayThatIsNotAnObjectThrows) { + EXPECT_THROW(parseString(withDisplay("1")), std::invalid_argument); +} + TEST(ConfigParserTest, MissingLslStreamThrows) { const std::string jsonText = R"json({ "config_version": "1.0", diff --git a/tests/unit_tests/config_validation_test.cpp b/tests/unit_tests/config_validation_test.cpp index a84bc86..70cc67b 100644 --- a/tests/unit_tests/config_validation_test.cpp +++ b/tests/unit_tests/config_validation_test.cpp @@ -125,6 +125,23 @@ TEST(ConfigValidationTest, NegativeImpedanceThresholdThrows) { EXPECT_THROW(config.validate(), std::invalid_argument); } +TEST(ConfigValidationTest, NegativeDisplayIndexThrows) { + DeviceConfig config = makeValidConfig(); + config.display.index = -1; + + EXPECT_THROW(config.validate(), std::invalid_argument); +} + +TEST(ConfigValidationTest, NonPositiveWindowSizeThrows) { + DeviceConfig config = makeValidConfig(); + config.display.width = 0; + EXPECT_THROW(config.validate(), std::invalid_argument); + + config = makeValidConfig(); + config.display.height = -1; + EXPECT_THROW(config.validate(), std::invalid_argument); +} + TEST(ConfigValidationTest, EmptyOutputFormatThrows) { DeviceConfig config = makeValidConfig(); config.output.format.clear(); diff --git a/tests/unit_tests/data_writer_test.cpp b/tests/unit_tests/data_writer_test.cpp index ef34c2d..e0c31a2 100644 --- a/tests/unit_tests/data_writer_test.cpp +++ b/tests/unit_tests/data_writer_test.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -84,3 +85,20 @@ TEST(DataWriterTest, FlushesEegAndMarkerRecords) { fs::remove(filePath); } + +TEST(CSVFormatStrategyTest, OpenRefusesToOverwriteAnExistingFile) { + const auto filePath = makeTempFilePath("csv_existing"); + { + std::ofstream existing(filePath); + existing << "earlier recording\n"; + } + + CSVFormatStrategy strategy; + EXPECT_THROW(strategy.open(filePath.string()), std::runtime_error); + + const auto lines = readAllLines(filePath); + ASSERT_EQ(lines.size(), 1U); + EXPECT_EQ(lines.front(), "earlier recording"); + + fs::remove(filePath); +} diff --git a/tests/unit_tests/lslreader_test.cpp b/tests/unit_tests/lslreader_test.cpp index bc029bd..f2a3c9c 100644 --- a/tests/unit_tests/lslreader_test.cpp +++ b/tests/unit_tests/lslreader_test.cpp @@ -1,6 +1,7 @@ #include #include +#include #include #include #include @@ -11,9 +12,11 @@ #include #include #include +#include #include #include #include +#include #include #include "lsl_cpp.h" @@ -32,6 +35,7 @@ constexpr auto kSubscribePoll = std::chrono::milliseconds(20); constexpr double kRecoveryValue = 42.0; constexpr auto kValidationWait = std::chrono::milliseconds(1500); +constexpr auto kStateWait = std::chrono::seconds(10); class ScopedStreamRedirect { public: @@ -136,6 +140,22 @@ std::string runAndCaptureDiagnostics(const DeviceConfig& config, int channelCoun return captured.str(); } +// Starts the reader with its diagnostics silenced, waits while it resolves, and +// stops it. Returns the state the wait ended on. +AcquisitionState startWaitAndStop( + LSLReader& reader, const std::shared_ptr>& eegQueue, + LSLReader::FailureCallback onFailure) { + std::ostringstream captured; + const ScopedStreamRedirect redirect(std::cerr, captured.rdbuf()); + + reader.start(eegQueue, std::move(onFailure)); + const std::stop_source neverStopped; + const AcquisitionState reached = reader.waitWhileResolving( + std::chrono::steady_clock::now() + kStateWait, neverStopped.get_token()); + reader.stop(); + return reached; +} + // Keeps pushing a marked sample until one of them comes back through the queue, // so the check does not depend on when exactly the inlet reconnects. Samples // from before the drop carry other values and are discarded. @@ -303,3 +323,78 @@ TEST(LSLReaderTest, StopBeforeStreamResolvedExitsCleanly) { TEST(LSLReaderTest, DestroyingUnstartedReaderIsSafe) { EXPECT_NO_THROW({ const LSLReader reader(makeConfig()); }); } + +TEST(LSLReaderTest, IsIdleUntilStartedAndResolvingRightAfterStart) { + DeviceConfig config = makeConfig(); + config.lsl.name = "neuronide_test_state_absent"; + config.lsl.sourceId = "neuronide-test-state-absent"; + + auto eegQueue = std::make_shared>(); + LSLReader reader(config); + EXPECT_EQ(reader.state(), AcquisitionState::Idle); + + reader.start(eegQueue); + EXPECT_EQ(reader.state(), AcquisitionState::Resolving); + + reader.stop(); + EXPECT_EQ(reader.state(), AcquisitionState::Idle); + EXPECT_EQ(reader.failure(), nullptr); +} + +TEST(LSLReaderTest, ReportsStreamingOnceSamplesArrive) { + DeviceConfig config = makeConfig(); + config.lsl.name = "neuronide_test_state_streaming"; + config.lsl.sourceId = "neuronide-test-state-streaming"; + + lsl::stream_outlet outlet = makeOutlet(config); + auto eegQueue = std::make_shared>(); + LSLReader reader(config); + reader.start(eegQueue); + + ASSERT_TRUE(waitForConsumer(outlet)) << "LSLReader did not subscribe (needs loopback)"; + EXPECT_EQ(reader.state(), AcquisitionState::Resolving) << "no sample has been queued yet"; + + const std::stop_source neverStopped; + pushSamples(outlet, makeSample(), kSamplesToPush); + EXPECT_EQ(reader.waitWhileResolving(std::chrono::steady_clock::now() + kStateWait, + neverStopped.get_token()), + AcquisitionState::Streaming); + + reader.stop(); +} + +TEST(LSLReaderTest, ReportsFailureAndNotifiesOnceForRejectedStream) { + DeviceConfig config = makeConfig(); + config.lsl.name = "neuronide_test_state_failed"; + config.lsl.sourceId = "neuronide-test-state-failed"; + + lsl::stream_outlet outlet = makeOutletWithShape(config, kMismatchedChannels, kSampleRate); + auto eegQueue = std::make_shared>(); + LSLReader reader(config); + + std::atomic notifications{0}; + EXPECT_EQ(startWaitAndStop(reader, eegQueue, [¬ifications] { ++notifications; }), + AcquisitionState::Failed); + + EXPECT_EQ(notifications.load(), 1); + EXPECT_EQ(reader.state(), AcquisitionState::Failed) << "a failure must survive stop()"; + EXPECT_NE(reader.failure(), nullptr); +} + +TEST(LSLReaderTest, WaitWhileResolvingReturnsWhenStopTokenFires) { + DeviceConfig config = makeConfig(); + config.lsl.name = "neuronide_test_state_wait_stop"; + config.lsl.sourceId = "neuronide-test-state-wait-stop"; + + auto eegQueue = std::make_shared>(); + LSLReader reader(config); + reader.start(eegQueue); + + std::stop_source stopSource; + stopSource.request_stop(); + EXPECT_EQ(reader.waitWhileResolving(std::chrono::steady_clock::now() + kStateWait, + stopSource.get_token()), + AcquisitionState::Resolving); + + reader.stop(); +} diff --git a/tests/unit_tests/runtime_test.cpp b/tests/unit_tests/runtime_test.cpp index 616184d..99a7c53 100644 --- a/tests/unit_tests/runtime_test.cpp +++ b/tests/unit_tests/runtime_test.cpp @@ -2,55 +2,138 @@ #include #include +#include #include +#include +#include #include #include #include +#include #include +#include #include +#include #include #include +#include #include +#include "lsl_cpp.h" #include "utils/ParserTestUtils.hpp" +// These tests drive a real LSLReader against a local LSL outlet over loopback, +// like the LSLReader tests do. + namespace { namespace fs = std::filesystem; -constexpr int kSurfaceSize = 10; -constexpr int kSurfaceDepth = 32; -constexpr auto kRenderSpinWait = std::chrono::milliseconds(50); - -constexpr const char* kValidConfig = R"json({ - "config_version": "1.0", - "device_name": "Dev", - "montage_standard": "10-20", - "lsl_stream": { - "name": "runtime_test_stream", "type": "EEG", "source_id": "runtime-test-src", - "expected_channel_count": 1, "expected_sample_rate_hz": 250 - }, - "channels": [ { "index": 0, "label": "Fz", "enabled": true, "unit": "uV" } ] -})json"; +constexpr int kSurfaceSize = 10; +constexpr int kSurfaceDepth = 32; +constexpr int kStreamChannels = 1; +constexpr int kWrongChannels = 2; +constexpr double kSampleRateHz = 250.0; +constexpr double kSampleValue = 7.5; + +constexpr auto kSamplePeriod = std::chrono::milliseconds(4); // 250 Hz +// Generous: resolving a stream over loopback usually takes well under a second. +constexpr auto kAcquisitionTimeout = std::chrono::seconds(10); +constexpr auto kNoStreamTimeout = std::chrono::milliseconds(500); +// Longer than kAcquisitionTimeout, so a recording that does start is always seen. +constexpr auto kRecordingWait = std::chrono::seconds(15); +constexpr auto kRecordingPoll = std::chrono::milliseconds(10); +constexpr auto kExperimentDuration = std::chrono::milliseconds(300); +// Upper bound for a lost stream to be noticed, re-resolved and rejected. +constexpr auto kFailureWait = std::chrono::seconds(15); + +// The stream name is unique per test, so tests running in parallel processes +// never resolve each other's outlets. +std::string currentStreamName() { + const auto* info = ::testing::UnitTest::GetInstance()->current_test_info(); + return std::string("runtime_test_") + info->name(); +} + +struct ConfigOptions { + std::string display = R"json({ "fullscreen": false })json"; + std::string outputFormat = "csv"; + bool channelEnabled = true; +}; + +std::string makeConfigJson(const std::string& streamName, const ConfigOptions& options = {}) { + return R"json({ + "config_version": "1.1", + "device_name": "Dev", + "montage_standard": "10-20", + "lsl_stream": { + "name": ")json" + + streamName + R"json(", "type": "EEG", "source_id": "runtime-test-src", + "expected_channel_count": 1, "expected_sample_rate_hz": 250 + }, + "channels": [ { "index": 0, "label": "Fz", "enabled": )json" + + (options.channelEnabled ? "true" : "false") + R"json(, "unit": "uV" } ], + "display": )json" + + options.display + R"json(, + "output": { "format": ")json" + + options.outputFormat + R"json(" } + })json"; +} + +std::shared_ptr createSoftwareRenderer() { + SDL_Surface* surface = SDL_CreateRGBSurfaceWithFormat(0, kSurfaceSize, kSurfaceSize, + kSurfaceDepth, SDL_PIXELFORMAT_RGBA32); + SDL_Renderer* renderer = SDL_CreateSoftwareRenderer(surface); + return {renderer, [surface](SDL_Renderer* target) { + if (target != nullptr) { + SDL_DestroyRenderer(target); + } + if (surface != nullptr) { + SDL_FreeSurface(surface); + } + }}; +} // A software renderer backed by an in-memory surface: no window, GPU, or // display required, so it runs anywhere (including headless CI). Injected in // place of Runtime's production SDL window/vsync-renderer factory. Runtime::RenderTargetFactory softwareRenderTargetFactory() { - return [](const std::string&) -> std::shared_ptr { - SDL_Surface* surface = SDL_CreateRGBSurfaceWithFormat( - 0, kSurfaceSize, kSurfaceSize, kSurfaceDepth, SDL_PIXELFORMAT_RGBA32); - SDL_Renderer* renderer = SDL_CreateSoftwareRenderer(surface); - return {renderer, [surface](SDL_Renderer* target) { - if (target != nullptr) { - SDL_DestroyRenderer(target); - } - if (surface != nullptr) { - SDL_FreeSurface(surface); - } - }}; + return [](const std::string&, const DisplayConfig&) { return createSoftwareRenderer(); }; +} + +// Records whether Runtime asked for a render target at all. +Runtime::RenderTargetFactory windowSpy(bool& windowOpened) { + return [&windowOpened](const std::string&, const DisplayConfig&) { + windowOpened = true; + return createSoftwareRenderer(); }; } +// Stands in for the EEG headset: an LSL outlet streaming a constant sample at +// the configured rate until it is destroyed. +class FakeCap { + public: + FakeCap(const std::string& streamName, int channelCount) + : outlet(lsl::stream_info(streamName, "EEG", channelCount, kSampleRateHz, lsl::cf_double64, + streamName + "-src")), + pusher([this, channelCount](const std::stop_token& stopToken) { + const std::vector sample(static_cast(channelCount), + kSampleValue); + while (!stopToken.stop_requested()) { + outlet.push_sample(sample); + std::this_thread::sleep_for(kSamplePeriod); + } + }) {} + ~FakeCap() = default; + + FakeCap(const FakeCap&) = delete; + FakeCap& operator=(const FakeCap&) = delete; + FakeCap(FakeCap&&) = delete; + FakeCap& operator=(FakeCap&&) = delete; + + private: + lsl::stream_outlet outlet; + std::jthread pusher; // last: stops pushing before the outlet goes away +}; + void writeText(const fs::path& path, const std::string& content) { std::ofstream out(path, std::ios::binary); out << content; @@ -71,6 +154,56 @@ std::vector readAllLines(const fs::path& path) { return lines; } +std::vector recordingsIn(const fs::path& dir) { + std::vector recordings; + for (const auto& entry : fs::directory_iterator(dir)) { + if (entry.path().extension() == ".csv") { + recordings.push_back(entry.path()); + } + } + return recordings; +} + +std::ptrdiff_t countEegRows(const fs::path& recording) { + const auto lines = readAllLines(recording); + return std::count_if(lines.begin(), lines.end(), + [](const std::string& line) { return line.rfind("eeg,", 0) == 0; }); +} + +// Sleeps for `duration`, waking early when `stopToken` fires. +void sleepUnlessStopped(std::chrono::milliseconds duration, const std::stop_token& stopToken) { + std::mutex mutex; + std::condition_variable_any wakeUp; + std::unique_lock lock(mutex); + wakeUp.wait_for(lock, stopToken, duration, [] { return false; }); +} + +using HelperAction = std::function; + +// Runs `action` on a helper thread once `count` recording files exist in `dir` +// - i.e. once run() is past the acquisition wait and rendering - while run() +// itself stays on the test's main thread, as SDL requires. After +// kRecordingWait the action runs anyway, so a broken run() cannot hang a test; +// destroying the returned thread before then skips the action. +std::jthread onceRecording(const fs::path& dir, HelperAction action, std::size_t count = 1) { + return std::jthread([dir, count, action = std::move(action)](const std::stop_token& stopToken) { + const auto deadline = std::chrono::steady_clock::now() + kRecordingWait; + while (recordingsIn(dir).size() < count && std::chrono::steady_clock::now() < deadline) { + if (stopToken.stop_requested()) { + return; + } + std::this_thread::sleep_for(kRecordingPoll); + } + action(stopToken); + }); +} + +void pushQuitEvent() { + SDL_Event quit{}; + quit.type = SDL_QUIT; + SDL_PushEvent(&quit); +} + // Owns the temp files that back a Runtime so a test never leaves artifacts on // disk, and provides a valid config + experiment scene by default. class RuntimeFixture : public ::testing::Test { @@ -85,7 +218,8 @@ class RuntimeFixture : public ::testing::Test { std::to_string(std::chrono::steady_clock::now().time_since_epoch().count())); fs::create_directories(tempDir); - writeText(tempDir / "config.json", kValidConfig); + streamName = currentStreamName(); + writeText(tempDir / "config.json", makeConfigJson(streamName)); writeExperimentFile(tempDir / "experiment.pb", utils::buildSimpleScene()); } @@ -94,7 +228,8 @@ class RuntimeFixture : public ::testing::Test { fs::remove_all(tempDir, errorCode); } - const fs::path& dir() const { return tempDir; } + const fs::path& dir() const { return tempDir; } + const std::string& stream() const { return streamName; } RuntimePaths paths() const { return RuntimePaths{.config = (tempDir / "config.json").string(), @@ -102,11 +237,33 @@ class RuntimeFixture : public ::testing::Test { .outputDir = tempDir.string()}; } + std::unique_ptr makeRuntime( + std::chrono::milliseconds acquisitionTimeout = kAcquisitionTimeout) const { + return std::make_unique(paths(), softwareRenderTargetFactory(), + acquisitionTimeout); + } + + // Records for kExperimentDuration, then stops the runtime from another + // thread. `recordingNumber` says which recording in dir() to wait for. + void runBriefly(Runtime& runtime, std::size_t recordingNumber = 1) const { + const std::jthread stopper = onceRecording( + dir(), + [&runtime](const std::stop_token& stopToken) { + sleepUnlessStopped(kExperimentDuration, stopToken); + runtime.requestStop(); + }, + recordingNumber); + runtime.run(); + } + private: - fs::path tempDir; + fs::path tempDir; + std::string streamName; }; } // namespace +// --- Construction ---------------------------------------------------------- + TEST_F(RuntimeFixture, ConstructsWithValidInputs) { EXPECT_NO_THROW({ const Runtime runtime(paths(), softwareRenderTargetFactory()); }); } @@ -130,17 +287,15 @@ TEST_F(RuntimeFixture, MissingExperimentThrows) { EXPECT_THROW(Runtime(badPaths, softwareRenderTargetFactory()), std::exception); } +TEST_F(RuntimeFixture, MissingOutputDirectoryThrows) { + RuntimePaths badPaths = paths(); + badPaths.outputDir = (dir() / "no_such_dir").string(); + + EXPECT_THROW(Runtime(badPaths, softwareRenderTargetFactory()), std::invalid_argument); +} + TEST_F(RuntimeFixture, UnknownOutputFormatThrows) { - const std::string badFormatConfig = R"json({ - "config_version": "1.0", "device_name": "Dev", "montage_standard": "10-20", - "lsl_stream": { - "name": "runtime_test_stream", "type": "EEG", "source_id": "runtime-test-src", - "expected_channel_count": 1, "expected_sample_rate_hz": 250 - }, - "channels": [ { "index": 0, "label": "Fz", "enabled": true, "unit": "uV" } ], - "output": { "format": "parquet" } - })json"; - writeText(dir() / "config.json", badFormatConfig); + writeText(dir() / "config.json", makeConfigJson(stream(), {.outputFormat = "parquet"})); EXPECT_THROW(Runtime(paths(), softwareRenderTargetFactory()), std::invalid_argument); } @@ -150,20 +305,149 @@ TEST_F(RuntimeFixture, NullRenderTargetFactoryThrows) { } TEST_F(RuntimeFixture, FactoryReturningNullRendererThrows) { - const auto nullFactory = [](const std::string&) -> std::shared_ptr { + const auto nullFactory = [](const std::string&, + const DisplayConfig&) -> std::shared_ptr { return nullptr; }; EXPECT_THROW(Runtime(paths(), nullFactory), std::runtime_error); } -TEST_F(RuntimeFixture, OutputFilenameDerivedFromExperimentNameAndDir) { - Runtime runtime(paths(), softwareRenderTargetFactory()); +TEST_F(RuntimeFixture, NonPositiveAcquisitionTimeoutThrows) { + EXPECT_THROW(Runtime(paths(), softwareRenderTargetFactory(), std::chrono::milliseconds(0)), + std::invalid_argument); +} + +TEST_F(RuntimeFixture, ConfigErrorsAreReportedBeforeTheWindowOpens) { + // A valid config, but LSLReader rejects a cap with every channel disabled. + writeText(dir() / "config.json", makeConfigJson(stream(), {.channelEnabled = false})); + + bool windowOpened = false; + EXPECT_THROW(Runtime(paths(), windowSpy(windowOpened)), std::invalid_argument); + EXPECT_FALSE(windowOpened); +} + +TEST_F(RuntimeFixture, PassesTheConfiguredDisplayToTheRenderTargetFactory) { + writeText(dir() / "config.json", + makeConfigJson(stream(), {.display = R"json({ "index": 0, "width": 800 })json"})); + + DisplayConfig received; + const auto factory = [&received](const std::string&, const DisplayConfig& display) { + received = display; + return createSoftwareRenderer(); + }; + const Runtime runtime(paths(), factory); + + constexpr int kConfiguredWidth = 800; + EXPECT_EQ(received.width, kConfiguredWidth); +} + +// --- Acquisition gate ------------------------------------------------------ + +TEST_F(RuntimeFixture, RunThrowsWithoutRecordingWhenNoStreamAppears) { + auto runtime = makeRuntime(kNoStreamTimeout); + + EXPECT_THROW(runtime->run(), std::runtime_error); + EXPECT_TRUE(runtime->outputPath().empty()); + EXPECT_TRUE(recordingsIn(dir()).empty()) << "no stimulus may run without EEG"; +} + +TEST_F(RuntimeFixture, RunThrowsWithoutRecordingWhenStreamShapeDoesNotMatchConfig) { + const FakeCap cap(stream(), kWrongChannels); + auto runtime = makeRuntime(); + + try { + runtime->run(); + FAIL() << "a stream with the wrong channel count must not start the experiment"; + } catch (const std::runtime_error& e) { + EXPECT_NE(std::string(e.what()).find("channels"), std::string::npos) << e.what(); + } + EXPECT_TRUE(recordingsIn(dir()).empty()); +} + +TEST_F(RuntimeFixture, ClosingTheWindowWhileWaitingForTheStreamAbortsWithoutRecording) { + auto runtime = makeRuntime(); + pushQuitEvent(); + + EXPECT_NO_THROW(runtime->run()); + EXPECT_TRUE(runtime->outputPath().empty()); + EXPECT_TRUE(recordingsIn(dir()).empty()); +} + +TEST_F(RuntimeFixture, RequestStopBeforeRunAbortsWithoutRecording) { + auto runtime = makeRuntime(); + runtime->requestStop(); + + EXPECT_NO_THROW(runtime->run()); + EXPECT_TRUE(recordingsIn(dir()).empty()); +} + +TEST_F(RuntimeFixture, RunStopsTheExperimentWhenAcquisitionFailsMidway) { + auto cap = std::make_unique(stream(), kStreamChannels); + auto runtime = makeRuntime(); + + // Once recording, swap the cap for one the config rejects: LSLReader loses + // the stream, re-resolves it, fails validation and gives up for good. + const std::jthread saboteur = + onceRecording(dir(), [&cap, &runtime, this](const std::stop_token& stopToken) { + cap.reset(); + cap = std::make_unique(stream(), kWrongChannels); + + // Safety net only: the failure itself must end run(). + sleepUnlessStopped(kFailureWait, stopToken); + runtime->requestStop(); + }); + + try { + runtime->run(); + FAIL() << "run() must report that acquisition failed during the experiment"; + } catch (const std::runtime_error& e) { + EXPECT_NE(std::string(e.what()).find("during the experiment"), std::string::npos) + << e.what(); + } + + const auto recordings = recordingsIn(dir()); + ASSERT_EQ(recordings.size(), 1U); + EXPECT_GT(countEegRows(recordings.front()), 0) << "data before the failure must be kept"; +} + +// --- Recording ------------------------------------------------------------- + +TEST_F(RuntimeFixture, RecordsEegUntilStopIsRequestedFromAnotherThread) { + const FakeCap cap(stream(), kStreamChannels); + auto runtime = makeRuntime(); + + const auto started = std::chrono::steady_clock::now(); + runBriefly(*runtime); + const auto elapsed = std::chrono::steady_clock::now() - started; - // Stop the render loop immediately so run() returns after a single pass. - runtime.requestStop(); - runtime.run(); + EXPECT_GE(elapsed, kExperimentDuration) << "run() returned before stop was requested"; - const fs::path output = runtime.outputPath(); + const fs::path output = runtime->outputPath(); + ASSERT_TRUE(fs::exists(output)); + const auto lines = readAllLines(output); + ASSERT_FALSE(lines.empty()); + EXPECT_EQ(lines.front(), "type,timestamp,payload"); + EXPECT_GT(countEegRows(output), 0) << "samples from the stream must reach the recording"; +} + +TEST_F(RuntimeFixture, RunStopsOnQuitEvent) { + const FakeCap cap(stream(), kStreamChannels); + auto runtime = makeRuntime(); + const std::jthread closer = + onceRecording(dir(), [](const std::stop_token&) { pushQuitEvent(); }); + + // Must return on SDL_QUIT even though no stop was requested. + runtime->run(); + + EXPECT_TRUE(fs::exists(runtime->outputPath())); +} + +TEST_F(RuntimeFixture, OutputFilenameDerivedFromExperimentNameAndDir) { + const FakeCap cap(stream(), kStreamChannels); + auto runtime = makeRuntime(); + runBriefly(*runtime); + + const fs::path output = runtime->outputPath(); EXPECT_EQ(output.parent_path(), dir()); // Scene project name is "TestProject" (see ParserTestUtils::buildSimpleScene). EXPECT_EQ(output.filename().string().rfind("TestProject_", 0), 0U); @@ -176,74 +460,84 @@ TEST_F(RuntimeFixture, OutputFilenameSanitizesTheExperimentName) { writeExperimentFile(dir() / "experiment.pb", utils::buildSimpleScene({.projectName = "block 1/run"})); - Runtime runtime(paths(), softwareRenderTargetFactory()); - runtime.requestStop(); - runtime.run(); + const FakeCap cap(stream(), kStreamChannels); + auto runtime = makeRuntime(); + runBriefly(*runtime); - const fs::path output = runtime.outputPath(); + const fs::path output = runtime->outputPath(); EXPECT_EQ(output.parent_path(), dir()); EXPECT_EQ(output.filename().string().rfind("block_1_run_", 0), 0U); EXPECT_TRUE(fs::exists(output)); } -TEST_F(RuntimeFixture, RunWritesRecordingWithCsvHeader) { - Runtime runtime(paths(), softwareRenderTargetFactory()); +TEST_F(RuntimeFixture, SecondRunThrowsAndLeavesTheRecordingIntact) { + const FakeCap cap(stream(), kStreamChannels); + auto runtime = makeRuntime(); + runBriefly(*runtime); - runtime.requestStop(); - runtime.run(); + const fs::path output = runtime->outputPath(); + const auto linesBefore = readAllLines(output); + const auto recordedRows = countEegRows(output); + ASSERT_GT(recordedRows, 0); - const fs::path output = runtime.outputPath(); - ASSERT_TRUE(fs::exists(output)); - - const auto lines = readAllLines(output); - ASSERT_FALSE(lines.empty()); - EXPECT_EQ(lines.front(), "type,timestamp,payload"); + EXPECT_THROW(runtime->run(), std::logic_error); + EXPECT_EQ(readAllLines(output), linesBefore); + EXPECT_EQ(recordingsIn(dir()).size(), 1U); } -TEST_F(RuntimeFixture, RunStopsOnQuitEvent) { - Runtime runtime(paths(), softwareRenderTargetFactory()); +TEST_F(RuntimeFixture, ConsecutiveRuntimesWriteSeparateRecordings) { + const FakeCap cap(stream(), kStreamChannels); - SDL_Event quit; - quit.type = SDL_QUIT; - ASSERT_EQ(SDL_PushEvent(&quit), 1); + auto first = makeRuntime(); + runBriefly(*first); + first.reset(); - // Must return promptly on SDL_QUIT even though no stop was requested. - runtime.run(); + auto second = makeRuntime(); + runBriefly(*second, 2); - EXPECT_TRUE(fs::exists(runtime.outputPath())); + const auto recordings = recordingsIn(dir()); + ASSERT_EQ(recordings.size(), 2U); + for (const auto& recording : recordings) { + EXPECT_GT(countEegRows(recording), 0) << recording; + } } -TEST_F(RuntimeFixture, RequestStopFromAnotherThreadEndsRun) { - Runtime runtime(paths(), softwareRenderTargetFactory()); +// --- Threading ------------------------------------------------------------- - // Ensure no stale SDL_QUIT from a previous test ends the loop early. - SDL_FlushEvent(SDL_QUIT); +TEST_F(RuntimeFixture, RunFromAnotherThreadThrows) { + auto runtime = makeRuntime(); - std::thread worker([&runtime] { runtime.run(); }); - std::this_thread::sleep_for(kRenderSpinWait); - runtime.requestStop(); - worker.join(); + bool rejected = false; + std::jthread([&runtime, &rejected] { + try { + runtime->run(); + } catch (const std::logic_error&) { + rejected = true; + } + }).join(); - EXPECT_TRUE(fs::exists(runtime.outputPath())); + EXPECT_TRUE(rejected) << "SDL events can only be handled on the thread that built the window"; } // Covers the production SDL window + vsync renderer factory where the platform -// can provide an accelerated renderer; skipped on headless machines that can't. +// can provide one; skipped on headless machines (or without vsync). TEST_F(RuntimeFixture, DefaultRenderTargetFactoryOnCapablePlatform) { // Let SDL pick the best available driver instead of the forced dummy one, // which cannot provide an accelerated renderer. unsetenv("SDL_VIDEODRIVER"); + std::unique_ptr runtime; try { - Runtime runtime(paths(), Runtime::defaultRenderTargetFactory()); - - SDL_Event quit; - quit.type = SDL_QUIT; - ASSERT_EQ(SDL_PushEvent(&quit), 1); - runtime.run(); - - EXPECT_TRUE(fs::exists(runtime.outputPath())); + runtime = std::make_unique(paths(), Runtime::defaultRenderTargetFactory(), + kAcquisitionTimeout); } catch (const std::exception& e) { - GTEST_SKIP() << "No accelerated render target on this platform: " << e.what(); + GTEST_SKIP() << "No vsynced accelerated render target on this platform: " << e.what(); } + + const FakeCap cap(stream(), kStreamChannels); + const std::jthread closer = + onceRecording(dir(), [](const std::stop_token&) { pushQuitEvent(); }); + runtime->run(); + + EXPECT_TRUE(fs::exists(runtime->outputPath())); } From 74ef2c7df3cad0383279177b297cfd3c04429856 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Szandarowski?= <59362160+MichalSzandar@users.noreply.github.com> Date: Fri, 9 Oct 2026 17:14:05 +0200 Subject: [PATCH 3/5] Change timeout duration from seconds to milliseconds Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/Runtime.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Runtime.cpp b/src/Runtime.cpp index bab8e83..f12d43f 100644 --- a/src/Runtime.cpp +++ b/src/Runtime.cpp @@ -239,8 +239,8 @@ bool Runtime::awaitAcquisition() { throw std::runtime_error( "Runtime: EEG stream '" + config.lsl.name + "' delivered no samples within " + std::to_string( - std::chrono::duration_cast(acquisitionTimeout).count()) + - " s; the experiment was not started"); + std::chrono::duration_cast(acquisitionTimeout).count()) + + " ms; the experiment was not started"); } switch ( From 2fcde238097ce19c4aa84fc3c2f5e15b7ff75d7e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Szandarowski?= <59362160+MichalSzandar@users.noreply.github.com> Date: Fri, 9 Oct 2026 17:14:38 +0200 Subject: [PATCH 4/5] Fix condition for setting experiment path Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/main.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main.cpp b/src/main.cpp index 3dcb40c..dc062db 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -22,7 +22,7 @@ int main(int argc, char* argv[]) { if (args.size() == 2) { paths.experiment = args[1]; - } else if (args.size() >= 3) { + } else if (args.size() == 3) { paths.config = args[1]; paths.experiment = args[2]; } else { From 9751e5ae6ca1e744aaee92f082040e443b6be223 Mon Sep 17 00:00:00 2001 From: Michal Date: Fri, 9 Oct 2026 17:27:08 +0200 Subject: [PATCH 5/5] fix(datawriter): close overwrite race and make output timestamps portable - write recordings through the handle created with fopen("wx") (new ExclusiveOutputFile) instead of reopening the path, which could truncate another file - format the file-name timestamp with std::chrono time zones instead of POSIX localtime_r --- include/datawriter/CSVFormatStrategy.hpp | 4 +- include/datawriter/ExclusiveOutputFile.hpp | 53 ++++++++++++++++++ include/datawriter/IDataFormatStrategy.hpp | 3 +- src/Runtime.cpp | 26 ++++----- src/datawriter/CMakeLists.txt | 1 + src/datawriter/CSVFormatStrategy.cpp | 48 ++++------------ src/datawriter/ExclusiveOutputFile.cpp | 64 ++++++++++++++++++++++ tests/unit_tests/data_writer_test.cpp | 32 +++++++++++ 8 files changed, 176 insertions(+), 55 deletions(-) create mode 100644 include/datawriter/ExclusiveOutputFile.hpp create mode 100644 src/datawriter/ExclusiveOutputFile.cpp diff --git a/include/datawriter/CSVFormatStrategy.hpp b/include/datawriter/CSVFormatStrategy.hpp index 4a9efca..88bac77 100644 --- a/include/datawriter/CSVFormatStrategy.hpp +++ b/include/datawriter/CSVFormatStrategy.hpp @@ -1,8 +1,8 @@ #ifndef CSVFORMATSTRATEGY_HPP #define CSVFORMATSTRATEGY_HPP +#include #include -#include #include class CSVFormatStrategy : public IDataFormatStrategy { @@ -25,7 +25,7 @@ class CSVFormatStrategy : public IDataFormatStrategy { void writeMarker(const Marker& marker) override; private: - std::ofstream outputFile; + ExclusiveOutputFile outputFile; }; #endif // CSVFORMATSTRATEGY_HPP diff --git a/include/datawriter/ExclusiveOutputFile.hpp b/include/datawriter/ExclusiveOutputFile.hpp new file mode 100644 index 0000000..ac1eb48 --- /dev/null +++ b/include/datawriter/ExclusiveOutputFile.hpp @@ -0,0 +1,53 @@ +#ifndef EXCLUSIVEOUTPUTFILE_HPP +#define EXCLUSIVEOUTPUTFILE_HPP + +#include +#include +#include +#include + +// An output file owned from creation to close. open() creates the path +// atomically and fails if it already exists, and every write goes through that +// same handle: the path is never reopened, so a file another process puts under +// the same name can be neither truncated nor written into. This is how format +// strategies keep IDataFormatStrategy's no-overwrite contract. +class ExclusiveOutputFile { + public: + ExclusiveOutputFile(); + ~ExclusiveOutputFile(); + + ExclusiveOutputFile(const ExclusiveOutputFile&) = delete; + ExclusiveOutputFile& operator=(const ExclusiveOutputFile&) = delete; + ExclusiveOutputFile(ExclusiveOutputFile&&) = delete; + ExclusiveOutputFile& operator=(ExclusiveOutputFile&&) = delete; + + // Throws std::system_error if `path` already exists or cannot be created. + void open(const std::string& path); + // Flushes and closes; does nothing when no file is open. + void close(); + [[nodiscard]] bool isOpen() const noexcept { return file != nullptr; } + + // Formatted output into the file while it is open. + std::ostream& stream() noexcept { return output; } + + private: + // Forwards every write to the C handle, which does its own buffering. + class HandleBuffer : public std::streambuf { + public: + void attach(std::FILE* target) noexcept { file = target; } + + protected: + int_type overflow(int_type character) override; + std::streamsize xsputn(const char_type* text, std::streamsize count) override; + int sync() override; + + private: + std::FILE* file = nullptr; + }; + + std::FILE* file = nullptr; + HandleBuffer buffer; + std::ostream output; // after `buffer`, which it writes through +}; + +#endif // EXCLUSIVEOUTPUTFILE_HPP diff --git a/include/datawriter/IDataFormatStrategy.hpp b/include/datawriter/IDataFormatStrategy.hpp index 606d78e..7ba2228 100644 --- a/include/datawriter/IDataFormatStrategy.hpp +++ b/include/datawriter/IDataFormatStrategy.hpp @@ -19,7 +19,8 @@ class IDataFormatStrategy { virtual std::string fileExtension() const = 0; // Creates `filepath` for writing. Must throw rather than overwrite an - // existing file: a truncated recording cannot be recovered. + // existing file - a truncated recording cannot be recovered - and must + // write through the handle that created it (see ExclusiveOutputFile). virtual void open(const std::string& filepath) = 0; virtual void close() = 0; diff --git a/src/Runtime.cpp b/src/Runtime.cpp index f12d43f..10e4936 100644 --- a/src/Runtime.cpp +++ b/src/Runtime.cpp @@ -2,11 +2,9 @@ #include #include -#include #include #include #include -#include #include #include #include @@ -14,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -28,11 +27,9 @@ namespace { namespace fs = std::filesystem; -constexpr const char* kDefaultTitle = "NeuronIDE"; -constexpr const char* kFallbackName = "experiment"; -constexpr const char* kTimestampFormat = "%Y%m%dT%H%M%S"; // e.g. 20261009T143012 +constexpr const char* kDefaultTitle = "NeuronIDE"; +constexpr const char* kFallbackName = "experiment"; -constexpr std::size_t kTimestampBufferSize = 32; // How often the window's events are handled while waiting for the EEG stream. constexpr auto kEventPollInterval = std::chrono::milliseconds(50); @@ -56,16 +53,12 @@ std::string sanitizeForFileName(std::string name) { return name; } -// Local wall-clock time, as a researcher reads it when looking for a session. +// Local wall-clock time, as a researcher reads it when looking for a session, +// e.g. "20261009T143012". std::string localTimestamp() { - const std::time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now()); - std::tm local{}; - localtime_r(&now, &local); - - std::array buffer{}; - const std::size_t length = - std::strftime(buffer.data(), buffer.size(), kTimestampFormat, &local); - return {buffer.data(), length}; + const auto now = std::chrono::floor(std::chrono::system_clock::now()); + return std::format("{:%Y%m%dT%H%M%S}", + std::chrono::zoned_time{std::chrono::current_zone(), now}); } std::string describe(const std::exception_ptr& error) { @@ -239,7 +232,8 @@ bool Runtime::awaitAcquisition() { throw std::runtime_error( "Runtime: EEG stream '" + config.lsl.name + "' delivered no samples within " + std::to_string( - std::chrono::duration_cast(acquisitionTimeout).count()) + + std::chrono::duration_cast(acquisitionTimeout) + .count()) + " ms; the experiment was not started"); } diff --git a/src/datawriter/CMakeLists.txt b/src/datawriter/CMakeLists.txt index b9a63e9..067c363 100644 --- a/src/datawriter/CMakeLists.txt +++ b/src/datawriter/CMakeLists.txt @@ -2,6 +2,7 @@ add_library(datawriter OBJECT CSVFormatStrategy.cpp DataFormatStrategyFactory.cpp DataWriter.cpp + ExclusiveOutputFile.cpp ) target_include_directories(datawriter PUBLIC diff --git a/src/datawriter/CSVFormatStrategy.cpp b/src/datawriter/CSVFormatStrategy.cpp index de84da2..53b60de 100644 --- a/src/datawriter/CSVFormatStrategy.cpp +++ b/src/datawriter/CSVFormatStrategy.cpp @@ -1,54 +1,30 @@ -#include -#include +#include #include #include #include -#include -#include +#include CSVFormatStrategy::~CSVFormatStrategy() { CSVFormatStrategy::close(); } -void CSVFormatStrategy::open(const std::string& filepath) { - // "x" (C11) creates the file, or fails if it already exists, in one atomic - // step: an existing recording is never truncated, not even by a race on - // the file name. C++20 streams cannot express this (std::ios::noreplace is - // C++23), hence the short-lived C handle. - // NOLINTNEXTLINE(cppcoreguidelines-owning-memory) - std::FILE* created = std::fopen(filepath.c_str(), "wx"); - if (created == nullptr) { - const int error = errno; - throw std::system_error(error, std::generic_category(), - "Failed to create CSV output file " + filepath); - } - // NOLINTNEXTLINE(cppcoreguidelines-owning-memory) - std::fclose(created); - - outputFile.open(filepath, std::ios::out); - if (!outputFile.is_open()) { - throw std::runtime_error("Failed to open CSV output file: " + filepath); - } -} +void CSVFormatStrategy::open(const std::string& filepath) { outputFile.open(filepath); } -void CSVFormatStrategy::close() { - if (outputFile.is_open()) { - outputFile.flush(); - outputFile.close(); - } -} +void CSVFormatStrategy::close() { outputFile.close(); } -void CSVFormatStrategy::writeHeader() { outputFile << "type,timestamp,payload\n"; } +void CSVFormatStrategy::writeHeader() { outputFile.stream() << "type,timestamp,payload\n"; } void CSVFormatStrategy::writeEEGData(const EEGData& data) { - outputFile << "eeg," << data.timestamp << ",\""; + std::ostream& out = outputFile.stream(); + out << "eeg," << data.timestamp << ",\""; for (std::size_t index = 0; index < data.channels.size(); ++index) { if (index > 0) { - outputFile << ','; + out << ','; } - outputFile << data.channels[index]; + out << data.channels[index]; } - outputFile << '"' << '\n'; + out << '"' << '\n'; } void CSVFormatStrategy::writeMarker(const Marker& marker) { - outputFile << "marker," << marker.timestamp << ",\"" << marker.eventName << '"' << '\n'; + outputFile.stream() << "marker," << marker.timestamp << ",\"" << marker.eventName << '"' + << '\n'; } diff --git a/src/datawriter/ExclusiveOutputFile.cpp b/src/datawriter/ExclusiveOutputFile.cpp new file mode 100644 index 0000000..cadbc90 --- /dev/null +++ b/src/datawriter/ExclusiveOutputFile.cpp @@ -0,0 +1,64 @@ +#include +#include +#include +#include +#include +#include + +ExclusiveOutputFile::ExclusiveOutputFile() : output(&buffer) {} + +ExclusiveOutputFile::~ExclusiveOutputFile() { close(); } + +void ExclusiveOutputFile::open(const std::string& path) { + close(); + + // "x" (C11) creates the file, or fails if it already exists, in one atomic + // step. C++20 streams cannot express this (std::ios::noreplace is C++23), + // hence the C handle, which then stays open for every write. + // NOLINTNEXTLINE(cppcoreguidelines-owning-memory) + file = std::fopen(path.c_str(), "wx"); + if (file == nullptr) { + const int error = errno; + throw std::system_error(error, std::generic_category(), + "Cannot create output file " + path); + } + + buffer.attach(file); + output.clear(); +} + +void ExclusiveOutputFile::close() { + if (file == nullptr) { + return; + } + + output.flush(); + buffer.attach(nullptr); + // NOLINTNEXTLINE(cppcoreguidelines-owning-memory) + std::fclose(file); + file = nullptr; +} + +ExclusiveOutputFile::HandleBuffer::int_type ExclusiveOutputFile::HandleBuffer::overflow( + int_type character) { + if (traits_type::eq_int_type(character, traits_type::eof())) { + return traits_type::not_eof(character); + } + if (file == nullptr || std::fputc(character, file) == EOF) { + return traits_type::eof(); + } + return character; +} + +std::streamsize ExclusiveOutputFile::HandleBuffer::xsputn(const char_type* text, + std::streamsize count) { + if (file == nullptr || count <= 0) { + return 0; + } + return static_cast( + std::fwrite(text, sizeof(char_type), static_cast(count), file)); +} + +int ExclusiveOutputFile::HandleBuffer::sync() { + return file == nullptr || std::fflush(file) == 0 ? 0 : -1; +} diff --git a/tests/unit_tests/data_writer_test.cpp b/tests/unit_tests/data_writer_test.cpp index e0c31a2..e841ce9 100644 --- a/tests/unit_tests/data_writer_test.cpp +++ b/tests/unit_tests/data_writer_test.cpp @@ -102,3 +102,35 @@ TEST(CSVFormatStrategyTest, OpenRefusesToOverwriteAnExistingFile) { fs::remove(filePath); } + +TEST(CSVFormatStrategyTest, WritesThroughTheCreatedFileEvenIfThePathIsReplaced) { +#ifdef _WIN32 + GTEST_SKIP() << "Windows does not allow renaming a file that is open"; +#endif + const auto filePath = makeTempFilePath("csv_swapped"); + const auto movedPath = fs::path(filePath).replace_extension(".moved.csv"); + + CSVFormatStrategy strategy; + strategy.open(filePath.string()); + + // Another process moves our file away and puts its own under the same name. + fs::rename(filePath, movedPath); + { + std::ofstream other(filePath); + other << "other recording\n"; + } + + strategy.writeHeader(); + strategy.close(); + + const auto otherLines = readAllLines(filePath); + ASSERT_EQ(otherLines.size(), 1U); + EXPECT_EQ(otherLines.front(), "other recording") << "the other file must not be touched"; + + const auto ownLines = readAllLines(movedPath); + ASSERT_EQ(ownLines.size(), 1U); + EXPECT_EQ(ownLines.front(), "type,timestamp,payload"); + + fs::remove(filePath); + fs::remove(movedPath); +}