server.hpp

Go to the documentation of this file.

 
#ifndef CAMCOM_TESTBENCH_SERVER_HPP
#define CAMCOM_TESTBENCH_SERVER_HPP
 
#include "camcom/testbench/config.hpp"
#include "camcom/testbench/http_server.hpp"
#include "camcom/testbench/data_frame.hpp"
#include "camcom/testbench/fits_recorder.hpp"
#include "camcom/common/frameInfo.hpp"
#include "camcom/common/queue.hpp"
#include <string>
#include <atomic>
#include <memory>
#include <mutex>
#include <condition_variable>
#include <vector>
#include <thread>
#include <set>
#include <cmath>
#include <chrono>
 
 
// forward declare to avoid leaking httplib.h to the public API
namespace httplib::ws {
 
class WebSocket;
 
}; // namespace httplib::ws
 
 
namespace camcom::server {
 
class CameraController;

class CameraServer {
public:
    explicit CameraServer(const ServerConfig& config);
    ~CameraServer();
 
    CameraServer(const CameraServer&) = delete;
    CameraServer& operator=(const CameraServer&) = delete;
 
    bool start();
    void stop();
    bool isRunning() const { return running_; }
    const std::string& getServerId() const { return config_.camera_name; }
 
private:
    ServerConfig config_;
    std::atomic<bool> running_{false};
    std::unique_ptr<CameraController> camera_controller_;
    std::unique_ptr<HttpServer> http_server_;
 
    // --- Frame queue (capture -> publisher) ---
    common::Queue<std::shared_ptr<DataFrame>> frame_queue_;
    std::mutex queue_cv_mutex_;
    std::condition_variable queue_cv_;   // signalled when a frame is pushed
 
    // --- Capture thread (runs during acquisition) ---
    std::atomic<bool> capture_running_{false};
    std::thread capture_thread_;
    void captureFrameLoop();
    void startCapture();
    void stopCapture();
 
    // --- Publisher thread (runs while server is running) ---
    std::atomic<bool> publisher_running_{false};
    std::thread publisher_thread_;
    void publishFrameLoop();
 
    // --- Cached JPEG for HTTP /image (written by publisher) ---
    std::mutex image_cache_mutex_;
    std::vector<uint8_t> cached_image_data_;
    std::shared_ptr<DataFrame> cached_frame_;
    int cached_display_width_{0};    // display dimensions (may differ from frame
    int cached_display_height_{0};   //   dimensions when JPEG mode uses downscaling)
 
    // --- WebSocket streaming ---
    std::mutex ws_mutex_;
    std::set<httplib::ws::WebSocket*> websocket_clients_;
    std::chrono::steady_clock::time_point last_ws_frame_time_{std::chrono::steady_clock::now()};
 
    // --- Recording thread ---
    std::atomic<bool> recording_active_{false};
    std::atomic<bool> recorder_running_{false};
    std::atomic<bool> recording_waiting_{false};     // true during delay/absolute countdown
    std::thread recorder_thread_;
    common::Queue<std::shared_ptr<DataFrame>> recording_queue_;
    std::mutex recording_cv_mutex_;
    std::condition_variable recording_cv_;
    std::unique_ptr<FitsRecorder> fits_recorder_;
 
    // Recording runtime config (may be overridden per-session via /recording/start)
    RecordingConfig active_recording_config_;
 
    // Recording progress (read by /recording/status)
    std::atomic<int> recording_frames_target_{0};
    std::atomic<int> recording_frames_done_{0};
    std::atomic<size_t> recording_bytes_done_{0};
    std::chrono::steady_clock::time_point recording_start_time_;
    std::atomic<double> recording_wait_remaining_{0}; // seconds left in delay/absolute countdown
 
    // Recording drop stats
    std::atomic<uint32_t> recording_drop_count_{0};
    std::chrono::steady_clock::time_point recording_last_drop_log_{std::chrono::steady_clock::now()};
 
    // Written files (persisted after session ends for GUI listing)
    std::mutex recording_files_mutex_;
    std::vector<std::string> recording_written_files_;
 
    void recorderLoop();
    void startRecording(const RecordingConfig& cfg);
    void stopRecording();
 
    // --- Telemetry / monitoring thread ---
    std::atomic<bool> telemetry_running_{false};
    std::thread telemetry_thread_;
    void telemetryLoop();
 
    // Telemetry data (written by telemetry thread, read by HTTP handler)
    struct TelemetrySample {
        std::string value;
        std::string type;
        std::string camera_name;  // Camera-native parameter name
        std::string unit;         // Unit (from parameter_registry, empty if not defined)
        double timestamp{0};      // seconds since epoch
    };
    std::mutex telemetry_mutex_;
    std::map<std::string, TelemetrySample> telemetry_current_;
    // Parameters to sample (resolved at start from config + adapter)
    std::vector<std::string> telemetry_params_;
 
    // --- Acquisition statistics ---
    struct SlidingWindowStats {
        std::vector<double> samples;
        int pos{0};       // write cursor
        int count{0};     // number of valid samples (<= capacity)
 
        void init(int capacity) { samples.resize(capacity, 0.0); pos = 0; count = 0; }
        void push(double v) {
            samples[pos] = v;
            pos = (pos + 1) % static_cast<int>(samples.size());
            if (count < static_cast<int>(samples.size())) count++;
        }
        struct Summary { double mean{0}; double stddev{0}; double min_val{0}; double max_val{0}; };
        Summary compute() const {
            if (count == 0) return {};
            double sum = 0, mn = samples[0], mx = samples[0];
            for (int i = 0; i < count; i++) {
                double v = samples[i];
                sum += v;
                if (v < mn) mn = v;
                if (v > mx) mx = v;
            }
            double mean = sum / count;
            double sq_sum = 0;
            for (int i = 0; i < count; i++) {
                double d = samples[i] - mean;
                sq_sum += d * d;
            }
            return {mean, (count > 1 ? std::sqrt(sq_sum / (count - 1)) : 0.0), mn, mx};
        }
    };
 
    struct FrameStatistics {
        SlidingWindowStats inter_frame;       // inter-frame interval (ms)
        SlidingWindowStats processing_latency; // time from receive to queue push (ms)
        uint64_t frames_total{0};
        uint64_t bytes_total{0};
        uint64_t frames_dropped_capture{0};
        uint64_t frames_dropped_recording{0};
        std::chrono::steady_clock::time_point acquisition_start;
        std::chrono::steady_clock::time_point last_frame_time;
        bool has_last_frame{false};
 
        // Publishing stats (display pipeline: JPEG encode + HTTP cache update)
        SlidingWindowStats publish_interval;   // inter-publish interval (ms)
        SlidingWindowStats encode_time;        // JPEG encoding duration (ms)
        uint64_t frames_published{0};
        uint64_t publish_bytes_total{0};       // cumulative JPEG bytes published
        std::chrono::steady_clock::time_point last_publish_time;
        bool has_last_publish{false};
    };
 
    std::mutex stats_mutex_;
    FrameStatistics frame_stats_;
    void resetStatistics();
 
    // --- Helpers ---
    static void jpegWriteCallback(void* context, void* data, int size);
    std::vector<uint8_t> encodeFrameToJpeg(const DataFrame& frame, int quality,
                                            int max_width, int max_height,
                                            int& out_width, int& out_height);
    std::vector<uint8_t> normalizeToUint8(const DataFrame& frame);
    static std::vector<uint8_t> downscaleGray(const uint8_t* src, int src_w, int src_h,
                                               int dst_w, int dst_h);
};
 
} // namespace camcom::server
 
#endif