ipc_stream_writer.hpp

// ipc_stream_writer.hpp
// Created by greg on 1/26/25.
// Copyright (C) 2025 Presage Security, Inc.
//
// This program is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 3 of the License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
// Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with this program; if not, write to the Free Software Foundation,
// Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 
#pragma once
 
#include <memory>
#include <chrono>
#include <absl/status/status.h>
#include <mediapipe/framework/port/status_macros.h>
#include <physiology/modules/messages/status.pb.h>
#include <physiology/modules/messages/metrics.pb.h>
#include <opencv2/core/mat.hpp>
#include <smartspectra/container/background_container.hpp>
#include <glog/logging.h>
 
namespace presage::smartspectra::ipc {

template<container::settings::IntegrationMode TIntegrationMode>
class IpcStreamWriter {
public:
    using BackgroundContainer =
        smartspectra::container::BackgroundContainer<
            presage::platform_independence::DeviceType::Cpu,
            smartspectra::container::settings::OperationMode::Continuous,
            TIntegrationMode
        >;
 
    virtual ~IpcStreamWriter() = default;

    virtual absl::Status SwapContainer(std::shared_ptr<BackgroundContainer> new_container) {
        container_ = new_container;
        if (container_) {
            MP_RETURN_IF_ERROR(RegisterCallbacks());
            MP_RETURN_IF_ERROR(SendInitialStatus());
        }
        return absl::OkStatus();
    }

    const BackgroundContainer* GetContainer() const {
        return container_.get();
    }

    virtual absl::Status WriteStatus(const presage::physiology::StatusValue& status) = 0;

    virtual absl::Status WriteCoreMetrics(
        const presage::physiology::MetricsBuffer& metrics,
        int64_t timestamp_us
    ) = 0;

    virtual absl::Status WriteEdgeMetrics(const presage::physiology::Metrics& metrics, int64_t timestamp_us) = 0;

    virtual absl::Status WriteFrame(const cv::Mat& frame, int64_t timestamp_us) = 0;

    virtual void Close() = 0;

    virtual bool IsOperational() const = 0;

    virtual absl::Status Flush() = 0;
 
protected:
    virtual absl::Status RegisterCallbacks() {
        if (!container_ || !IsOperational()) {
            return absl::OkStatus();
        }
 
        // Register status callback
        MP_RETURN_IF_ERROR(container_->SetOnStatusChange(
            [this](presage::physiology::StatusValue status_value) {
                return WriteStatus(status_value);
            }
        ));
 
 
        // Register core metrics callback
        MP_RETURN_IF_ERROR(container_->SetOnCoreMetricsOutput(
            [this](presage::physiology::MetricsBuffer metrics, int64_t timestamp_us) {
                return WriteCoreMetrics(metrics, timestamp_us);
            }
        ));
 
        // Register edge metrics callback
        MP_RETURN_IF_ERROR(container_->SetOnEdgeMetricsOutput(
            [this](presage::physiology::Metrics metrics, int64_t timestamp_us) {
                return WriteEdgeMetrics(metrics, timestamp_us);
            }
        ));
 
        // Register video output callback for camera capture
        MP_RETURN_IF_ERROR(container_->SetOnVideoOutput(
            [this](cv::Mat& frame, int64_t timestamp_us) {
                return WriteFrame(frame, timestamp_us);
            }
        ));
 
        return absl::OkStatus();
    }

    virtual absl::Status SendInitialStatus() {
        if (!container_ || !IsOperational()) {
            return absl::OkStatus();
        }
 
        auto status_code = container_->GetStatusCode();
        auto current_time = std::chrono::system_clock::now().time_since_epoch();
        int64_t timestamp = std::chrono::duration_cast<std::chrono::microseconds>(current_time).count();
 
        physiology::StatusValue status_value;
        status_value.set_value(status_code);
        status_value.set_timestamp(timestamp);
 
        return WriteStatus(status_value);
    }
 
    // Protected member accessible to derived classes
    std::shared_ptr<BackgroundContainer> container_;
 
    int64_t GetCurrentTimestampUs() {
        using namespace std::chrono;
        return duration_cast<microseconds>(system_clock::now().time_since_epoch()).count();
    }
 
};

// template<container::settings::IntegrationMode TIntegrationMode>
// std::unique_ptr<IpcStreamWriter<TIntegrationMode>> CreateIpcStreamWriter(
//     StreamBackend backend,
//     const FileIpcConfig& file_config,
//     const RedisIpcConfig& redis_config
// );
 
} // namespace presage::smartspectra::ipc