Class SensorPipeline

capstone/sensor_pipeline.hpp:89–168

── Sensor Pipeline (Modules 07, 10, 12: RAII + Classes + Threads) ──

// ── Sensor Pipeline (Modules 07, 10, 12: RAII + Classes + Threads) ──
class SensorPipeline {
public:
    explicit SensorPipeline(std::string_view name)
        : name_(name) {}

    // RAII: destructor guarantees clean shutdown (Module 07)
    ~SensorPipeline() { stop(); }

    // Non-copyable, non-movable (owns threads — Module 10)
    SensorPipeline(const SensorPipeline&) = delete;
    SensorPipeline& operator=(const SensorPipeline&) = delete;

    void start() {
        running_ = true;
        producer_ = std::thread(&SensorPipeline::producerLoop, this);
        consumer_ = std::thread(&SensorPipeline::consumerLoop, this);
    }

    void stop() {
        if (!running_.exchange(false)) return; // Already stopped
        queue_.shutdown();
        if (producer_.joinable()) producer_.join();
        if (consumer_.joinable()) consumer_.join();
    }

    [[nodiscard]] std::size_t processedCount() const noexcept {
        return processed_count_.load(std::memory_order_relaxed);
    }

private:
    void producerLoop() {
        // Thread-local RNG — each thread gets its own engine (no data race)
        std::mt19937 rng(std::random_device{}());
        std::uniform_real_distribution<double> noise(0.0, 0.1);

        while (running_) {
            IMUReading reading{};
            reading.timestamp = std::chrono::steady_clock::now();
            reading.accel_z = -9.81 + noise(rng);

            if (!queue_.push(reading)) break;
            std::this_thread::sleep_for(config::IMU_PERIOD);
        }
    }

    void consumerLoop() {
        // Rely solely on pop()'s internal lock + shutdown signal,
        // NOT on an unguarded queue_.size() check which would be a data race.
        while (true) {
            auto maybe_reading = queue_.pop();
            if (!maybe_reading) break; // shutdown() was called and queue is drained

            // Simple magnitude filter (Module 03: arithmetic)
            const auto& r = *maybe_reading;
            [[maybe_unused]] double magnitude = std::sqrt(
                r.accel_x * r.accel_x +
                r.accel_y * r.accel_y +
                r.accel_z * r.accel_z);