physiology_service.hpp

// physiology_service.hpp
// Created by Greg on 5/17/24.
// Copyright (C) 2024 Presage Security, Inc.
//
// SPDX-License-Identifier: LGPL-3.0-or-later
 
#pragma once
// === standard library includes (if any) ===
#include <chrono>
#include <thread>
#include <atomic>
#include <string>
#include <functional>
// === third-party includes (if any) ===
#include <reproc++/reproc.hpp>
#include <physiology/graph/physiology_core_service.grpc.pb.h>
#include <physiology/graph/physiology_core_service.pb.h>
// === local includes (if any) ===
#include <smartspectra/cpp/physiology_service.grpc.pb.h>
#include "log/log_buffer.hpp"
#include "log/absl_log_sink.hpp"
#include "log/glog_sink.hpp"
#include "log/core_log_sink.hpp"
//TODO: remove relative includes from public headers -- might require moving on_prem into <smartspectra/root>/cpp/smartspectra/
// the one below doesn't resolve w/ <> most likely due to the current header (physiology_service.hpp) being in a
// header file set with BASE_DIR at parent of smartspectra/cpp (see CMake file)
#include "smartspectra/cpp/physiology_service.pb.h"
#include "reactors/core_metrics_provider.hpp"
#include "reactors/edge_metrics_provider.hpp"
#include "reactors/status_code_provider.hpp"
#include "reactors/frame_with_timestamp_recorder.hpp"
#include "ipc_stream_writer.hpp"
#include <smartspectra/container/background_container.hpp>
#include <smartspectra/container/settings.hpp>
#include <smartspectra/video_source/video_source.hpp>
#include <smartspectra/video_source/camera_tuner.hpp>
 
 
namespace presage::smartspectra::grpc_bindings {
 
struct LogSettings {
    int verbosity_level = 0;
    bool log_dropped_frames = false;
    bool log_throughput_and_latency = false;
    bool log_edge_core_transfer_times = false;
};
 
template<container::settings::IntegrationMode TIntegrationMode>
class PhysiologyServiceImpl final : public Physiology::CallbackService {
public:
    explicit PhysiologyServiceImpl(
        video_source::InputTransformMode input_transform_mode = video_source::InputTransformMode::None,
        double preprocessed_data_buffer_duration_s = 0.2,
        std::optional<bool> enable_phasic_bp = std::nullopt,
        std::optional<bool> enable_eda = std::nullopt,
        std::optional<bool> use_full_range_face_detection = std::nullopt,
        std::optional<bool> use_full_pose_landmarks = std::nullopt,
        std::optional<bool> enable_pose_landmark_segmentation = std::nullopt,
        std::optional<bool> enable_micromotion = std::nullopt,
        bool use_camera = false,
        int camera_device_index = 0,
        const std::string& input_video_path = "",
        const std::string& input_video_time_path = "",
        int interframe_delay_ms = 20,
        bool start_with_recording_on = false,
        bool exit_on_input_video_completion = false,
        uint16_t physiology_core_port_number = 50052,
        bool physiology_core_manual_control = false,
        int physiology_core_connection_timeout_ms = 50000,
        const std::string& venv_directory = "",
        const LogSettings& log_settings = {},
        const int log_buffer_size = 1000,
        int64_t test_physiology_core_seed = -1,
        std::unique_ptr<ipc::IpcStreamWriter<TIntegrationMode>> ipc_stream_writer = nullptr,
        std::optional<std::string> video_output_directory = std::nullopt,
        bool enable_camera_tuning = true,
        video_source::CameraTunerSettings tuner_settings = {},
        std::chrono::seconds camera_tuning_cooldown_seconds = std::chrono::seconds(10),
        bool camera_tuning_recheck_on_recovery = false
    );
 
    ~PhysiologyServiceImpl() override;
 
    using BackgroundContainer =
        smartspectra::container::BackgroundContainer<
            presage::platform_independence::DeviceType::Cpu,
            smartspectra::container::settings::OperationMode::Continuous,
            TIntegrationMode
        >;
 
    using TFrameWithTimestampRecorder = FrameWithTimestampRecorder<TIntegrationMode>;
    using TStatusCodeProvider = StatusCodeProvider<TIntegrationMode>;
    using TMetricsProvider = CoreMetricsProvider<TIntegrationMode>;
    using TEdgeMetricsProvider = EdgeMetricsProvider<TIntegrationMode>;
 
    using BackgroundContainerSettings = typename BackgroundContainer::SettingsType;
 
    grpc::ServerUnaryReactor* StartCore(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::Empty* request,
        ::google::protobuf::Empty* response
    ) override;
 
    grpc::ServerUnaryReactor* StartEdgeGraph(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::Empty* request,
        ::google::protobuf::Empty* response
    ) override;
 
    grpc::ServerUnaryReactor* StartPreprocessing(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::Empty* request,
        ::google::protobuf::Empty* response
    ) override;
 
    grpc::ServerUnaryReactor* WaitUntilGraphIsIdle(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::Empty* request,
        ::google::protobuf::Empty* response
    ) override;
 
    grpc::ServerUnaryReactor* SetRecording(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::BoolValue* request,
        ::google::protobuf::Empty* response
    ) override;
 
    grpc::ServerReadReactor<presage::smartspectra::FrameWithTimestamp>* AddFrameWithTimestamp(
        grpc::CallbackServerContext* context,
        google::protobuf::Empty* response
    ) override;
 
    grpc::ServerWriteReactor<physiology::StatusValue>* GetStatusCode(
        grpc::CallbackServerContext* context,
        const google::protobuf::Empty* request
    ) override;
 
    grpc::ServerUnaryReactor* GetStatusDescription(
        grpc::CallbackServerContext* context,
        const physiology::StatusValue* request,
        google::protobuf::StringValue* response
    ) override;
 
    grpc::ServerUnaryReactor* GetStatusHint(
        grpc::CallbackServerContext* context,
        const physiology::StatusValue* request,
        google::protobuf::StringValue* response
    ) override;
 
    grpc::ServerWriteReactor<presage::physiology::MetricsBuffer>* GetCoreMetrics(
        grpc::CallbackServerContext* context,
        const google::protobuf::Empty* request
    ) override;
 
    grpc::ServerWriteReactor<presage::physiology::MetricsBuffer>* GetMetrics(
        grpc::CallbackServerContext* context,
        const google::protobuf::Empty* request
    ) override;
 
    grpc::ServerWriteReactor<presage::physiology::Metrics>* GetEdgeMetrics(
        grpc::CallbackServerContext* context,
        const google::protobuf::Empty* request
    ) override;
 
    grpc::ServerUnaryReactor* StopCore(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::Empty* request,
        ::google::protobuf::Empty* response
    ) override;
 
    grpc::ServerUnaryReactor* StopEdgeGraph(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::Empty* request,
        ::google::protobuf::Empty* response
    ) override;
 
    grpc::ServerUnaryReactor* StopPreprocessing(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::Empty* request,
        ::google::protobuf::Empty* response
    ) override;
 
    grpc::ServerUnaryReactor* ResetProcessing(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::Empty* request,
        ::google::protobuf::Empty* response
    ) override;
 
    grpc::ServerUnaryReactor* CheckHealth(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::Empty* request,
        ::google::protobuf::StringValue* response
    ) override;
 
    grpc::ServerUnaryReactor* IsPhysiologyCoreUp(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::Empty* request,
        ::google::protobuf::BoolValue* response
    ) override;
 
    grpc::ServerUnaryReactor* IsEdgeGraphRunning(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::Empty* request,
        ::google::protobuf::BoolValue* response
    ) override;
 
    grpc::ServerUnaryReactor* IsPreprocessingRunning(
        grpc::CallbackServerContext* context,
        const ::google::protobuf::Empty* request,
        ::google::protobuf::BoolValue* response
    ) override;
 
    grpc::ServerWriteReactor<::presage::smartspectra::LogEntry>* StreamLogs(
        grpc::CallbackServerContext* context,
        const google::protobuf::Empty* request
    ) override;
 
    absl::Status AutoStartIfRequested();
    void SetOnInputVideoCompletionCallback(std::function<void()> callback);
 
private:
    template<typename TProviderWithContainer>
    TProviderWithContainer* GenericHookToProvider(std::shared_ptr<TProviderWithContainer>& provider);
 
    absl::Status StartCoreServer();
 
    absl::Status StartCoreClient();
 
    absl::Status StopCoreServer();
 
    absl::Status BuildAndStartContainer();
 
    absl::Status StopAndDestroyContainer();
 
    absl::Status VerifyVenv();
    absl::Status PublishRecordingStateToRedis(bool recording);
 
    absl::Status HandleStatusForTuning(
        const presage::physiology::StatusValue& status_output,
        presage::physiology::StatusCode previous_status_code
    );
 
    absl::Status FlushPendingStatusAfterTuning(ipc::IpcStreamWriter<TIntegrationMode>* writer);
 
    // Restrict re-tuning to the subset of status codes that camera tuning can address.
    static bool StatusRequiresImmediateRetune(presage::physiology::StatusCode status);
    static bool StatusShouldRetuneAfterRecovery(presage::physiology::StatusCode status);
 
 
    // parameters
    const uint16_t physiology_core_port_number;
    const bool physiology_core_manual_control;
    const int64_t test_physiology_core_seed;
    const int physiology_core_connection_timeout_ms;
 
    // state
    std::shared_ptr<BackgroundContainer> container;
    std::shared_ptr<TFrameWithTimestampRecorder> frame_with_timestamp_recorder;
    std::shared_ptr<TStatusCodeProvider> status_code_provider;
    std::shared_ptr<TMetricsProvider> metrics_provider;
    std::shared_ptr<TEdgeMetricsProvider> edge_metrics_provider;
    std::mutex preprocessing_construction_mutex;
    std::condition_variable preprocessing_construction_condition;
    video_source::InputTransformMode input_transform_mode;
 
    const BackgroundContainerSettings container_settings;
    reproc::options physiology_core_process_options;
    reproc::process physiology_core_process;
    bool physiology_core_started = false;
    std::shared_ptr<grpc::Channel> channel;
    std::unique_ptr<presage::physiology::PhysiologyCore::Stub> core_stub = nullptr;
    std::filesystem::path venv_directory;
    bool venv_verified = false;
 
    LogSettings log_settings;
    LogBuffer log_buffer;
    AbslLogSink absl_log_sink;
    GlogSink glog_sink;
 
    CoreLogSink core_stdout_log_sink;
    CoreLogSink core_stderr_log_sink;
    std::thread core_log_thread;
    std::atomic<bool> core_log_thread_shutdown = false;
 
    // Camera/video ingestion thread (optional)
    const bool use_camera;
    const std::string input_video_path;
    const std::string input_video_time_path;
    const bool local_input_enabled;
    const bool start_with_recording_on;
    const bool exit_on_input_video_completion;
    bool recording_auto_started = false;
    const int camera_device_index;
    std::mutex local_input_thread_mutex;
    std::thread local_input_thread;
    std::atomic<bool> local_input_thread_shutdown = false;
    std::shared_ptr<video_source::VideoSource> local_input_source;
    std::function<void()> input_video_completion_callback;
    std::mutex input_video_completion_callback_mutex;
 
    // IPC stream writer (optional, alternative to gRPC streams)
    std::unique_ptr<ipc::IpcStreamWriter<TIntegrationMode>> ipc_stream_writer;
 
    // Camera tuning (optional, for IPC modes with camera)
    std::shared_ptr<video_source::CameraTuner> camera_tuner;
    video_source::CameraTunerSettings tuner_settings_;
    std::atomic<bool> camera_tuning_enabled = false;
    std::atomic<bool> is_recording = false;
    std::atomic<bool> is_tuning_camera = false;
    std::chrono::steady_clock::time_point last_camera_tuning_time;
    std::chrono::seconds camera_tuning_cooldown_duration;
    const bool camera_tuning_recheck_on_recovery_;
    std::mutex camera_tuning_mutex;
    std::atomic<int64_t> latest_status_change_timestamp_us_{0};
    std::atomic<presage::physiology::StatusCode> latest_status_code_{presage::physiology::StatusCode::PROCESSING_NOT_STARTED};
    std::mutex pending_status_mutex_;
    std::optional<presage::physiology::StatusValue> pending_status_after_tuning_;
 
    // Camera thread error tracking
    std::optional<absl::Status> last_camera_error_;
    std::mutex camera_error_mutex_;
};
 
typedef PhysiologyServiceImpl<smartspectra::container::settings::IntegrationMode::Grpc> PhysiologyServiceImpl_GrpcBackend;
} // namespace presage::smartspectra::grpc_bindings