SensorFramePipeline.hpp

Go to the documentation of this file.

#ifndef DAQ_SENSOR_FRAME_PIPELINE_HPP
#define DAQ_SENSOR_FRAME_PIPELINE_HPP
 
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <vector>
 
#include "protocol/EncryptedFrame.hpp"
#include "transport/NetworkSocket.hpp"
 
namespace fsw {
namespace streams {
 
/**
 * @brief UDP pipeline: DAQv2-Comms board packets → internal SensorBatch.
 */
class SensorFramePipeline {
public:
    explicit SensorFramePipeline(const std::string& bind_address, uint16_t bind_port);
    ~SensorFramePipeline() = default;
 
    std::optional<daq_comms::protocol::SensorBatch> poll();
 
    const std::string& last_source_ip() const {
        return last_source_ip_;
    }
 
    bool is_ready() const;
    std::string last_error() const;
 
    ssize_t send_to(const std::string& dest_ip, uint16_t dest_port, const uint8_t* data,
                    size_t size);
 
    bool set_broadcast(bool enable);
 
    struct LastHeartbeat {
        std::vector<uint8_t> data;
        std::string source_ip;
    };
    std::optional<LastHeartbeat> get_last_heartbeat();
 
    // Raw type-15 LOGS packet, surfaced like heartbeats (poll() returns no
    // SensorBatch for it). daq_bridge parses + forwards it to the backend.
    struct LastLog {
        std::vector<uint8_t> data;
        std::string source_ip;
    };
    std::optional<LastLog> get_last_log();
 
    struct EnvironmentalSample {
        float temperature_c;
        uint32_t pressure_pa;
        float humidity_rh;
        uint32_t sample_timestamp_ms;
        std::string source_ip;
    };
    std::optional<EnvironmentalSample> get_last_environmental();
 
private:
    std::unique_ptr<daq_comms::transport::UDPSocket> socket_;
    std::vector<uint8_t> receive_buffer_;
    std::string last_error_;
    std::string last_source_ip_;
 
    std::vector<uint8_t> last_heartbeat_buffer_;
    std::string last_heartbeat_source_ip_;
 
    std::vector<uint8_t> last_log_buffer_;
    std::string last_log_source_ip_;
    std::optional<EnvironmentalSample> last_environmental_;
 
    static constexpr size_t RECEIVE_BUFFER_SIZE = 8192;
};
 
}  // namespace streams
}  // namespace fsw
 
#endif