physiology_service_impl.hpp

//
// Created by greg on 5/17/24.
// Copyright (c) 2024 Presage Technologies
//
 
// === standard library includes (if any) ===
#include <filesystem>
#include <iomanip>
#include <mutex>
#include <future>
#include <optional>
// === third-party includes (if any) ===
#include <grpc/grpc.h>
#include <grpcpp/grpcpp.h>
#include <mediapipe/framework/port/status_macros.h>
#include <mediapipe/framework/port/opencv_core_inc.h>
#include <physiology/graph/physiology_core_service.grpc.pb.h>
#include <physiology/graph/physiology_core_service.pb.h>
#include <physiology/modules/rpc.h>
#include <physiology/modules/messages/status.h>
#include <physiology/modules/filesystem.h>
#include <physiology/modules/configuration.h>
#include <reproc++/drain.hpp>
#include <absl/log/log_sink_registry.h>
// === local includes (if any) ===
#include <smartspectra/container/settings.hpp>
#include <smartspectra/video_source/factory.hpp>
#include <mediapipe/framework/port/opencv_imgproc_inc.h>
#include "physiology_service.hpp"
#include "smartspectra_on_prem_config.hpp"
#include "reactors/log_stream_reactor.hpp"
#include "redis_ipc/redis_ipc_stream_writer.hpp"
 
namespace spectra = presage::physiology;
namespace cs = presage::smartspectra::container::settings;
namespace rpc = presage::rpc;
namespace vs = presage::smartspectra::video_source;
 
namespace presage::smartspectra::grpc_bindings {
 
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
PhysiologyServiceImpl<TIntegrationMode>::PhysiologyServiceImpl(
    vs::InputTransformMode input_transform_mode,
    double preprocessed_data_buffer_duration_s,
    std::optional<bool> enable_phasic_bp,
    std::optional<bool> enable_eda,
    std::optional<bool> use_full_range_face_detection,
    std::optional<bool> use_full_pose_landmarks,
    std::optional<bool> enable_pose_landmark_segmentation,
    std::optional<bool> enable_micromotion,
    bool use_camera,
    int camera_device_index,
    const std::string& input_video_path,
    const std::string& input_video_time_path,
    int interframe_delay_ms,
    bool start_with_recording_on,
    bool exit_on_input_video_completion,
    uint16_t physiology_core_port_number,
    bool physiology_core_manual_control,
    int physiology_core_connection_timeout_ms,
    const std::string& venv_directory,
    const LogSettings& log_settings,
    const int log_buffer_size,
    int64_t test_physiology_core_seed,
    std::unique_ptr<ipc::IpcStreamWriter<TIntegrationMode>> ipc_stream_writer,
    std::optional<std::string> video_output_directory,
    bool enable_camera_tuning,
    video_source::CameraTunerSettings tuner_settings,
    std::chrono::seconds camera_tuning_cooldown_seconds,
    bool camera_tuning_recheck_on_recovery
) : physiology_core_port_number(physiology_core_port_number),
    physiology_core_manual_control(physiology_core_manual_control),
    test_physiology_core_seed(test_physiology_core_seed),
    physiology_core_connection_timeout_ms(physiology_core_connection_timeout_ms),
    venv_directory(venv_directory),
    log_settings(log_settings),
    log_buffer(log_buffer_size),
    absl_log_sink(this->log_buffer),
    glog_sink(this->log_buffer),
    core_stdout_log_sink(this->log_buffer, this->core_log_thread_shutdown),
    core_stderr_log_sink(this->log_buffer, this->core_log_thread_shutdown),
    container(nullptr),
    container_settings(
        {
            video_source::VideoSourceSettings{},
            cs::VideoSinkSettings{},
            /*headless=*/ false,
            /*interframe_delay_ms=*/ interframe_delay_ms,
            /*start_with_recording_on=*/ start_with_recording_on,
            /*start_time_offset_ms=*/ 0,
            /*scale_input=*/ true,
            /*binary_graph=*/ true,
            enable_phasic_bp,
            enable_eda,
            /*enable_dense_facemesh_points=*/false,
            use_full_range_face_detection,
            use_full_pose_landmarks,
            enable_pose_landmark_segmentation,
            enable_micromotion,
            /*enable_edge_metrics=*/true,
            /*print_graph_contents=*/false,
            /*log_transfer_timing_info=*/log_settings.log_edge_core_transfer_times,
            video_output_directory,
            /*verbosity_level=*/ 0,
            cs::ContinuousSettings{
                preprocessed_data_buffer_duration_s
            },
            cs::GrpcSettings{
                physiology_core_port_number
            }
        }
    ),
    input_transform_mode(input_transform_mode),
    use_camera(use_camera),
    input_video_path(input_video_path),
    input_video_time_path(input_video_time_path),
    start_with_recording_on(start_with_recording_on),
    exit_on_input_video_completion(exit_on_input_video_completion),
    local_input_enabled(use_camera || !input_video_path.empty()),
    camera_device_index(camera_device_index),
    ipc_stream_writer(std::move(ipc_stream_writer)),
    tuner_settings_(tuner_settings),
    camera_tuning_cooldown_duration(camera_tuning_cooldown_seconds),
    camera_tuning_recheck_on_recovery_(camera_tuning_recheck_on_recovery) {
    // Enable overlay rendering by default for on-prem service
    this->tuner_settings_.render_calibrating_overlay = enable_camera_tuning;
 
    absl::AddLogSink(&this->absl_log_sink);
    google::AddLogSink(&this->glog_sink);
}
 
 
template<container::settings::IntegrationMode TIntegrationMode>
PhysiologyServiceImpl<TIntegrationMode>::~PhysiologyServiceImpl() {
    // Attempt graceful shutdown of container/core to avoid dangling threads.
    if (this->container != nullptr) {
        absl::Status container_stop_status = this->StopAndDestroyContainer();
        if (!container_stop_status.ok()) {
            LOG(WARNING) << "Failed to stop preprocessing container during destruction: "
                         << container_stop_status.message();
        }
    }
    if (this->physiology_core_started && !this->physiology_core_manual_control) {
        absl::Status core_stop_status = this->StopCoreServer();
        if (!core_stop_status.ok()) {
            LOG(WARNING) << "Failed to stop Physiology Core during destruction: "
                         << core_stop_status.message();
        }
        this->physiology_core_started = false;
    }
    if (this->core_log_thread.joinable()) {
        this->core_log_thread.join();
    }
 
    // Stop local input thread (camera or video file) before tearing down
    if (this->local_input_enabled) {
        {
            std::lock_guard<std::mutex> lk(this->local_input_thread_mutex);
            this->local_input_thread_shutdown = true;
        }
        if (this->local_input_thread.joinable()) {
            this->local_input_thread.join();
        }
        this->local_input_source.reset();
    }
    // Shutdown the log buffer to release any waiting threads
    this->log_buffer.Shutdown();
 
    absl::RemoveLogSink(&this->absl_log_sink);
    google::RemoveLogSink(&this->glog_sink);
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status PhysiologyServiceImpl<TIntegrationMode>::AutoStartIfRequested() {
    if (!this->start_with_recording_on) {
        return absl::OkStatus();
    }
    if (this->physiology_core_manual_control) {
        LOG(INFO) << "Skipping auto-start: manual Physiology Core control enabled.";
        return absl::OkStatus();
    }
    if (!this->physiology_core_started) {
        LOG(INFO) << "Auto-start requested; launching Physiology Core.";
        MP_RETURN_IF_ERROR(this->StartCoreServer());
        MP_RETURN_IF_ERROR(this->StartCoreClient());
        this->physiology_core_started = true;
        LOG(INFO) << "Physiology Core auto-start complete.";
    }
    MP_RETURN_IF_ERROR(this->BuildAndStartContainer());
    LOG(INFO) << "Edge graph initialized; recording set to "
              << (this->recording_auto_started ? "on." : "off.");
    return absl::OkStatus();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
void PhysiologyServiceImpl<TIntegrationMode>::SetOnInputVideoCompletionCallback(std::function<void()> callback) {
    std::lock_guard<std::mutex> lock(this->input_video_completion_callback_mutex);
    this->input_video_completion_callback = std::move(callback);
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status PhysiologyServiceImpl<TIntegrationMode>::PublishRecordingStateToRedis(bool recording) {
    if (this->ipc_stream_writer == nullptr) {
        return absl::OkStatus();
    }
    auto* redis_writer =
        dynamic_cast<redis_ipc::RedisIpcStreamWriter<TIntegrationMode>*>(this->ipc_stream_writer.get());
    if (redis_writer == nullptr) {
        return absl::OkStatus();
    }
    return redis_writer->PublishRecordingState(recording);
}
 
template<typename TFunction>
inline grpc::ServerUnaryReactor* ExecuteAsyncWithDefaultReactor(
    grpc::CallbackServerContext* context,
    TFunction&& function
) {
    auto reactor = context->DefaultReactor();
    grpc::Status status = std::forward<TFunction>(function)();
    reactor->Finish(status);
    return reactor;
}
 
// NOTE: needs explicit template specialization w/ GCC, can't rely on template-from-function-argument inference
template<container::settings::IntegrationMode TIntegrationMode>
template<typename TProviderWithContainer>
TProviderWithContainer* PhysiologyServiceImpl<TIntegrationMode>::GenericHookToProvider(
    std::shared_ptr<TProviderWithContainer>& provider
) {
    // Disable gRPC streams when using IPC writer
    if (this->ipc_stream_writer != nullptr) {
        // Create a dummy provider that immediately finishes with an error
        auto error_provider = std::make_shared<TProviderWithContainer>(nullptr);
        error_provider->Finish(grpc::Status(grpc::StatusCode::UNAVAILABLE,
            "gRPC streaming disabled: using alternative IPC backend"));
        return error_provider.get();
    }
 
    std::unique_lock<std::mutex> lock(this->preprocessing_construction_mutex);
    this->preprocessing_construction_condition.wait(lock, [this] { return this->container != nullptr; });
    if (provider != nullptr) {
        if (provider->GetContainer() != this->container.get()) {
            // Assuming new gRPC connection, discard the old reactor safely and build new one.
            provider->Finish(grpc::Status::OK);
            provider->WaitUntilDone();
            provider.reset();
            provider = std::make_shared<TProviderWithContainer>(this->container);
        }
    } else {
        provider = std::make_shared<TProviderWithContainer>(this->container);
    }
    return provider.get();
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::StartCore(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request,
    google::protobuf::Empty* response
) {
    return ExecuteAsyncWithDefaultReactor(context, [this]() -> grpc::Status {
        if (this->physiology_core_manual_control) {
            const char* message = "Skipping start of Physiology Core Service: manual control enabled.";
            LOG(INFO) << message;
        } else {
            if (this->physiology_core_started) {
                return {::grpc::StatusCode::ALREADY_EXISTS, "Physiology core is already running."};
            }
            PHYSIOLOGY_GRPC_RETURN_IF_ERROR(StartCoreServer());
        }
        PHYSIOLOGY_GRPC_RETURN_IF_ERROR(StartCoreClient());
        this->physiology_core_started = true;
        return ::grpc::Status::OK;
    });
}
 
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
absl::Status PhysiologyServiceImpl<TIntegrationMode>::BuildAndStartContainer() {
    std::unique_lock<std::mutex> lock(this->preprocessing_construction_mutex);
    if (this->container == nullptr) {
        this->container = std::make_shared<BackgroundContainer>(this->container_settings);
        if (this->ipc_stream_writer == nullptr) {
            // Swap container for gRPC stream reactors
            if (this->frame_with_timestamp_recorder != nullptr) {
                this->frame_with_timestamp_recorder->SwapContainer(this->container);
            }
            if (this->status_code_provider != nullptr) {
                this->status_code_provider->SwapContainer(this->container);
            }
            if (this->metrics_provider != nullptr) {
                this->metrics_provider->SwapContainer(this->container);
            }
            if (this->edge_metrics_provider != nullptr) {
                this->edge_metrics_provider->SwapContainer(this->container);
            }
 
        } else {
            // Swap container for alternative IPC stream writer
            auto swap_status = this->ipc_stream_writer->SwapContainer(this->container);
            if (!swap_status.ok()) {
                return swap_status;
            }
 
            // Prepare camera tuner before registering status callbacks (camera only)
            if (this->use_camera) {
                if (!this->local_input_source) {
                    vs::VideoSourceSettings vs_settings{};
                    vs_settings.device_index = this->camera_device_index;
                    vs_settings.capture_height_px = 720;
                    vs_settings.capture_width_px = 1280;
                    vs_settings.input_transform_mode = this->input_transform_mode;
                    MP_ASSIGN_OR_RETURN(auto source, vs::BuildVideoSource(vs_settings));
                    this->local_input_source = std::move(source);
                }
 
                if (this->tuner_settings_.render_calibrating_overlay && this->camera_tuner == nullptr) {
                    this->camera_tuner = std::make_shared<video_source::CameraTuner>(this->tuner_settings_);
 
                    auto status = this->camera_tuner->SetVideoSource(this->local_input_source);
                    if (status.ok()) {
                        this->local_input_source = this->camera_tuner->GetVideoSource();
                        this->camera_tuning_enabled = true;
                        LOG(INFO) << "Camera tuner created and video source attached for IPC mode.";
                    } else {
                        LOG(WARNING) << "Camera tuning disabled: " << status.message();
                        this->camera_tuner.reset();
                        this->camera_tuning_enabled = false;
                    }
                }
            }
 
            auto* ipc_writer_ptr = this->ipc_stream_writer.get();
 
            // Set up status callbacks for camera tuning and IPC writer
            if (this->use_camera && this->camera_tuning_enabled.load()) {
                auto* tuner_ptr = this->camera_tuner.get();
 
                // OnStatusCode: called on EVERY status update for camera tuning
                PHYSIOLOGY_GRPC_RETURN_IF_ERROR(this->container->SetOnStatusCode(
                    [this, tuner_ptr, ipc_writer_ptr](presage::physiology::StatusValue status_value) {
                        if (tuner_ptr && tuner_ptr->IsTuning()) {
                            auto status_code = static_cast<presage::physiology::StatusCode>(status_value.value());
                            auto tuning_status = tuner_ptr->ProcessFrame(status_code, status_value.timestamp());
                            if (!tuning_status.ok()) {
                                LOG(WARNING) << "Camera tuner ProcessFrame failed: " << tuning_status.message();
                            }
                        } else if (ipc_writer_ptr != nullptr) {
                            auto flush_status = this->FlushPendingStatusAfterTuning(ipc_writer_ptr);
                            if (!flush_status.ok()) {
                                return flush_status;
                            }
                        }
                        return absl::OkStatus();
                    }
                ));
                LOG(INFO) << "Camera tuning status monitoring enabled.";
            }
 
            // OnStatusChange: called only when status changes, for IPC writer
            PHYSIOLOGY_GRPC_RETURN_IF_ERROR(this->container->SetOnStatusChange(
                [this, ipc_writer_ptr](presage::physiology::StatusValue status_value) {
                    auto new_status = static_cast<presage::physiology::StatusCode>(status_value.value());
                    auto previous_status = this->latest_status_code_.exchange(new_status);
                    this->latest_status_change_timestamp_us_.store(status_value.timestamp());
 
                    // Handle tuning logic (may trigger tuning based on status)
                    auto tuning_status = this->HandleStatusForTuning(status_value, previous_status);
                    if (!tuning_status.ok()) {
                        return tuning_status;
                    }
 
                    if (!this->camera_tuner || !this->camera_tuner->IsTuning()) {
                        auto flush_status = this->FlushPendingStatusAfterTuning(ipc_writer_ptr);
                        if (!flush_status.ok()) {
                            return flush_status;
                        }
                        return ipc_writer_ptr->WriteStatus(status_value);
                    }
 
                    if (new_status == presage::physiology::StatusCode::OK) {
                        presage::physiology::StatusValue masked_status(status_value);
                        {
                            std::lock_guard<std::mutex> lock(this->pending_status_mutex_);
                            this->pending_status_after_tuning_ = status_value;
                        }
                        masked_status.set_value(presage::physiology::StatusCode::CAMERA_TUNING_IN_PROGRESS);
                        return ipc_writer_ptr->WriteStatus(masked_status);
                    }
 
                    return ipc_writer_ptr->WriteStatus(status_value);
                }
            ));
        }
        if (this->log_settings.log_dropped_frames) {
            PHYSIOLOGY_GRPC_RETURN_IF_ERROR(this->container->SetOnFrameSentThrough(
                [](bool frame_sent_through, int64_t input_timestamp) {
                    if (!frame_sent_through) {
                        LOG(INFO) << "Dropped frame with capture timestamp " << input_timestamp << " microseconds.";
                    }
                    return absl::OkStatus();
                }
            ));
        }
        if (this->log_settings.log_throughput_and_latency) {
            PHYSIOLOGY_GRPC_RETURN_IF_ERROR(this->container->SetOnCorePerformanceTelemetry([](
                double effective_core_fps,
                double effective_core_latency,
                int64_t input_timestamp
            ) {
                LOG(INFO) << "Effective system throughput: " << std::fixed << std::setprecision(3)
                          << effective_core_fps
                          << " FPS/HZ, system latency: " << std::fixed << std::setprecision(3)
                          << effective_core_latency << " s [window: 3s for both]";
                return absl::OkStatus();
            }));
        }
    }
    PHYSIOLOGY_GRPC_RETURN_IF_ERROR(container->Initialize());
    if (!this->container->GraphIsRunning()) {
        PHYSIOLOGY_GRPC_RETURN_IF_ERROR(container->StartGraph());
    }
    if (this->start_with_recording_on && !this->recording_auto_started) {
        auto recording_status = this->container->SetRecording(true);
        if (!recording_status.ok()) {
            LOG(WARNING) << "Failed to auto-start recording: " << recording_status.message();
        } else {
            auto publish_status = PublishRecordingStateToRedis(/*recording=*/true);
            if (!publish_status.ok()) {
                LOG(WARNING) << "Failed to publish auto-start recording state: " << publish_status.message();
            }
            this->recording_auto_started = true;
            this->is_recording.store(true);
        }
    }
    // notify any awaiting threads in async calls that we have a container available
    this->preprocessing_construction_condition.notify_all();
 
    // If configured, start local ingestion thread (camera or video file) after container is ready
    if (this->local_input_enabled) {
        std::lock_guard<std::mutex> local_lock(this->local_input_thread_mutex);
        if (!this->local_input_thread.joinable()) {
            if (!this->local_input_source) {
                vs::VideoSourceSettings vs_settings{};
                vs_settings.input_transform_mode = this->input_transform_mode;
                if (!this->input_video_path.empty()) {
                    vs_settings.input_video_path = this->input_video_path;
                    vs_settings.input_video_time_path = this->input_video_time_path;
                } else {
                    vs_settings.device_index = this->camera_device_index;
                    vs_settings.capture_height_px = 720;
                    vs_settings.capture_width_px = 1280;
                }
                MP_ASSIGN_OR_RETURN(auto source, vs::BuildVideoSource(vs_settings));
                this->local_input_source = std::move(source);
            }
            this->local_input_thread_shutdown = false;
            this->local_input_thread = std::thread([this]() {
                if (this->use_camera && this->camera_tuning_enabled.load() && this->camera_tuner != nullptr) {
                    LOG(INFO) << "Starting initial camera tuning...";
                    auto start_status = this->camera_tuner->StartTuning();
                    if (start_status.ok()) {
                        this->is_tuning_camera.store(true);
                        LOG(INFO) << "Initial camera tuning started";
                        {
                            std::lock_guard<std::mutex> lock(this->camera_tuning_mutex);
                            this->last_camera_tuning_time = std::chrono::steady_clock::now();
                        }
                    } else {
                        LOG(WARNING) << "Failed to start initial camera tuning: " << start_status.message();
                        {
                            std::lock_guard<std::mutex> lock(this->camera_error_mutex_);
                            this->last_camera_error_ = start_status;
                        }
                    }
                }
 
                int64_t previous_frame_ts_us = 0;
                const bool fixed_delay_enabled = !this->use_camera && this->container_settings.interframe_delay_ms > 0;
 
                while (!this->local_input_thread_shutdown.load()) {
                    // Ensure container is valid and running
                    std::shared_ptr<BackgroundContainer> local_container;
                    {
                        std::unique_lock<std::mutex> c_lock(this->preprocessing_construction_mutex);
                        if (this->container == nullptr || !this->container->GraphIsRunning()) {
                            // Avoid tight spin if container not ready
                            c_lock.unlock();
                            std::this_thread::sleep_for(std::chrono::milliseconds(10));
                            continue;
                        }
                        local_container = this->container;
                    }
 
                    cv::Mat frame_bgr;
                    (*this->local_input_source) >> frame_bgr;
                    if (frame_bgr.empty()) {
                        this->recording_auto_started = false;
                        if (!this->input_video_path.empty()) {
                            LOG(INFO) << "Input video source exhausted; stopping recording.";
                            absl::Status stop_status = local_container->SetRecording(false);
                        if (!stop_status.ok()) {
                            LOG(WARNING) << "Failed to stop recording after replay finished: "
                                         << stop_status.message();
                        } else {
                            auto publish_status = PublishRecordingStateToRedis(/*recording=*/false);
                            if (!publish_status.ok()) {
                                LOG(WARNING) << "Failed to publish recording stop after replay: "
                                             << publish_status.message();
                            }
                            this->is_recording.store(false);
                        }
                            if (this->exit_on_input_video_completion) {
                                std::function<void()> callback_copy;
                                {
                                    std::lock_guard<std::mutex> cb_lock(this->input_video_completion_callback_mutex);
                                    callback_copy = this->input_video_completion_callback;
                                }
                                if (callback_copy) {
                                    LOG(INFO) << "Input video playback complete; signaling shutdown.";
                                    callback_copy();
                                }
                            }
                            this->local_input_thread_shutdown = true;
                            break;
                        }
 
                        std::this_thread::sleep_for(std::chrono::milliseconds(5));
                        continue;
                    }
 
                    if (fixed_delay_enabled) {
                        std::this_thread::sleep_for(std::chrono::milliseconds(this->container_settings.interframe_delay_ms));
                    }
 
                    int64_t ts_us = this->local_input_source->GetFrameTimestamp();
 
                    cv::Mat frame_rgb;
                    cv::cvtColor(frame_bgr, frame_rgb, cv::COLOR_BGR2RGB);
 
                    // Check if tuning completed
                    if (this->use_camera && this->camera_tuner != nullptr) {
                        if (this->is_tuning_camera.load() && !this->camera_tuner->IsTuning()) {
                            this->is_tuning_camera.store(false);
                            auto stage = this->camera_tuner->GetCurrentStage();
                            if (stage == vs::TuningStage::COMPLETE) {
                                LOG(INFO) << "Camera tuning completed successfully";
                            } else if (stage == vs::TuningStage::FAILED) {
                                LOG(WARNING) << "Camera tuning failed";
                            } else {
                                LOG(INFO) << "Camera tuning stopped";
                            }
                        }
 
                        // Apply tuning overlay if active
                        frame_rgb = this->camera_tuner->ProcessFrameForDisplay(frame_rgb);
                    }
 
                    absl::Status s = local_container->AddFrameWithTimestamp(frame_rgb, ts_us);
                    if (!s.ok()) {
                        // Minor backoff to avoid hot loop on errors
                        std::this_thread::sleep_for(std::chrono::milliseconds(2));
                    }
                }
            });
        }
    }
    return absl::OkStatus();
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::StartEdgeGraph(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request,
    ::google::protobuf::Empty* response
) {
    return ExecuteAsyncWithDefaultReactor(context, [this]() -> grpc::Status {
        if (this->physiology_core_started || this->physiology_core_manual_control) {
            return ::rpc::abslStatusToGrpc(BuildAndStartContainer());
        }
        return {::grpc::StatusCode::FAILED_PRECONDITION, "Physiology Core is not running, unable to start graph."};
    });
}

template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::StartPreprocessing(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request,
    google::protobuf::Empty* response
) {
    return this->StartEdgeGraph(context, request, response);
}
 
/*
 * @deprecated
 */
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::IsPreprocessingRunning(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request,
    ::google::protobuf::BoolValue* response
) {
    return this->IsEdgeGraphRunning(context, request, response);
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::WaitUntilGraphIsIdle(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request,
    google::protobuf::Empty* response
) {
    return ExecuteAsyncWithDefaultReactor(context, [this]() -> grpc::Status {
        return rpc::abslStatusToGrpc(container->WaitUntilGraphIsIdle());
    });
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::SetRecording(
    grpc::CallbackServerContext* context,
    const google::protobuf::BoolValue* request,
    google::protobuf::Empty* response
) {
    return ExecuteAsyncWithDefaultReactor(context, [this, request]() -> grpc::Status {
        if (this->container == nullptr) {
            return {
                grpc::StatusCode::FAILED_PRECONDITION,
                "Physiology Preprocessing is not running, unable to start or stop recording."
            };
        }
 
        bool recording = request->value();
        bool was_recording = this->is_recording.load();
 
        // If tuning is active and trying to start recording, defer the request
        if (recording && this->is_tuning_camera.load()) {
            LOG(INFO) << "Ignoring recording start request - camera tuning in progress";
            // Return success but don't actually start recording
            return grpc::Status::OK;
        }
 
        // Update recording state
        this->is_recording.store(recording);
 
        // Handle tuning state based on recording transition
        if (this->camera_tuning_enabled.load() && this->camera_tuner != nullptr) {
            if (recording && !was_recording) {
                // Started recording - stop any ongoing tuning (shouldn't happen due to check above, but be safe)
                if (this->is_tuning_camera.load()) {
                    LOG(INFO) << "Recording started - interrupting tuning";
                    // CameraTuner lacks an explicit stop API; clearing the flag prevents further processing.
                    this->is_tuning_camera.store(false);
                }
            } else if (!recording && was_recording) {
                // Stopped recording - enable re-tuning after cooldown
                LOG(INFO) << "Recording stopped - tuning will check status after cooldown";
                std::lock_guard<std::mutex> lock(this->camera_tuning_mutex);
                this->last_camera_tuning_time = std::chrono::steady_clock::now();
            }
        }
 
        auto status = container->SetRecording(recording);
 
        // Publish recording state to Redis IPC if using Redis backend
        if (status.ok()) {
            auto publish_status = PublishRecordingStateToRedis(recording);
            if (!publish_status.ok()) {
                LOG(WARNING) << "Failed to publish recording state: " << publish_status.message();
            }
        }
 
        return rpc::abslStatusToGrpc(status);
    });
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerReadReactor<presage::smartspectra::FrameWithTimestamp>*
PhysiologyServiceImpl<TIntegrationMode>::AddFrameWithTimestamp(
    grpc::CallbackServerContext* context,
    google::protobuf::Empty* response
) {
    // Disable gRPC frame streaming when using IPC writer
    if (this->ipc_stream_writer != nullptr) {
        auto error_recorder = std::make_shared<TFrameWithTimestampRecorder>(nullptr, this->input_transform_mode);
        error_recorder->Finish(grpc::Status(grpc::StatusCode::UNAVAILABLE,
            "gRPC frame streaming disabled: using alternative IPC backend"));
        return error_recorder.get();
    }
 
    std::unique_lock<std::mutex> lock(this->preprocessing_construction_mutex);
    this->preprocessing_construction_condition.wait(lock, [this] { return this->container != nullptr; });
    if (this->frame_with_timestamp_recorder != nullptr) {
        if (this->frame_with_timestamp_recorder->GetContainer() != this->container.get()) {
            // Assuming new gRPC connection, discard the old reactor safely and build new one.
            this->frame_with_timestamp_recorder->Finish(grpc::Status::OK);
            this->frame_with_timestamp_recorder->WaitUntilDone();
            this->frame_with_timestamp_recorder.reset();
            this->frame_with_timestamp_recorder = std::make_shared<TFrameWithTimestampRecorder>(this->container,
                                                                                                this->input_transform_mode);
        }
    } else {
        this->frame_with_timestamp_recorder = std::make_shared<TFrameWithTimestampRecorder>(this->container,
                                                                                            this->input_transform_mode);
    }
    return this->frame_with_timestamp_recorder.get();
}
 
inline cv::Mat BuildOpenCVMatFromProtoBuf(const smartspectra::Mat& mat_protobuf) {
    cv::Mat mat_cv(mat_protobuf.rows(), mat_protobuf.cols(), mat_protobuf.element_type());
    size_t data_bytesize = mat_cv.rows * mat_cv.cols * mat_cv.elemSize();
    auto data_protobuf = const_cast<char*>(mat_protobuf.data().data());
    std::copy(reinterpret_cast<unsigned char*>(data_protobuf),
              reinterpret_cast<unsigned char*>(data_protobuf + data_bytesize),
              mat_cv.data);
    return mat_cv;
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerWriteReactor<physiology::StatusValue>* PhysiologyServiceImpl<TIntegrationMode>::GetStatusCode(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request
) {
    return this->GenericHookToProvider<TStatusCodeProvider>(this->status_code_provider);
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::GetStatusDescription(
    grpc::CallbackServerContext* context,
    const physiology::StatusValue* request,
    google::protobuf::StringValue* response
) {
    return ExecuteAsyncWithDefaultReactor(
        context,
        [this, &response, &request]() -> grpc::Status {
            response->set_value(spectra::GetStatusDescription(request->value()));
            return ::grpc::Status::OK;
        }
    );
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::GetStatusHint(
    grpc::CallbackServerContext* context,
    const physiology::StatusValue* request,
    google::protobuf::StringValue* response
) {
    return ExecuteAsyncWithDefaultReactor(
        context,
        [this, &response, &request]() -> grpc::Status {
            response->set_value(spectra::GetStatusHint(request->value()));
            return ::grpc::Status::OK;
        }
    );
}
 
template<container::settings::IntegrationMode TIntegrationMode>
grpc::ServerWriteReactor<presage::physiology::MetricsBuffer>*
PhysiologyServiceImpl<TIntegrationMode>::GetCoreMetrics(
    grpc::CallbackServerContext* context,
    const google::protobuf::Empty* request
) {
    return this->GenericHookToProvider<TMetricsProvider>(this->metrics_provider);
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerWriteReactor<presage::physiology::MetricsBuffer>*
PhysiologyServiceImpl<TIntegrationMode>::GetMetrics(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request
) {
    return this->GetCoreMetrics(context, request);
}
 
template<container::settings::IntegrationMode TIntegrationMode>
grpc::ServerWriteReactor<presage::physiology::Metrics>* PhysiologyServiceImpl<TIntegrationMode>::GetEdgeMetrics(
    grpc::CallbackServerContext* context,
    const google::protobuf::Empty* request
) {
    return this->GenericHookToProvider<TEdgeMetricsProvider>(this->edge_metrics_provider);
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::StopCore(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request,
    google::protobuf::Empty* response
) {
    return ExecuteAsyncWithDefaultReactor(context, [this]() -> grpc::Status {
        if (this->container != nullptr) {
            PHYSIOLOGY_GRPC_RETURN_IF_ERROR(this->StopAndDestroyContainer());
        }
        absl::Status status = absl::OkStatus();
        if (this->physiology_core_manual_control) {
            const char* message = "Skipping shutdown of Physiology Core Service: manual control enabled.";
            LOG(INFO) << message;
        } else {
            if (this->physiology_core_started) {
                status = StopCoreServer();
                if (status.ok()) {
                    this->physiology_core_started = false;
                }
            } else {
                LOG(INFO) << "No need to shut down Physiology Core Service: process not started.";
                status = absl::OkStatus();
            }
            if (this->core_stub != nullptr) {
                this->core_stub.reset();
            }
        }
        return rpc::abslStatusToGrpc(status);
    });
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
absl::Status PhysiologyServiceImpl<TIntegrationMode>::StopAndDestroyContainer() {
    // Stop local input thread before shutting down the container (if enabled)
    if (this->local_input_enabled) {
        std::lock_guard<std::mutex> local_lock(this->local_input_thread_mutex);
        this->local_input_thread_shutdown = true;
        if (this->local_input_thread.joinable()) {
            this->local_input_thread.join();
        }
        // keep local_input_source around to allow restart without re-probing device or re-opening file
    }
    if (this->container != nullptr) {
        const bool graph_was_running = this->container->GraphIsRunning();
        if (graph_was_running) {
            absl::Status idle_status = container->WaitUntilGraphIsIdle();
            if (!idle_status.ok() && !absl::IsFailedPrecondition(idle_status)) {
                LOG(WARNING) << "WaitUntilGraphIsIdle failed prior to shutdown: " << idle_status.message();
            }
        }
        absl::Status stop_status = container->StopGraph();
        if (!stop_status.ok()) {
            LOG(WARNING) << "StopGraph failed during shutdown: " << stop_status.message();
        }
    }
    this->container.reset();
    this->recording_auto_started = false;
    return absl::OkStatus();
}

template<container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::StopEdgeGraph(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request,
    ::google::protobuf::Empty* response
) {
    return ExecuteAsyncWithDefaultReactor(
        context,
        [this]() -> grpc::Status {
            return rpc::abslStatusToGrpc(this->StopAndDestroyContainer());
        }
    );
}

template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::StopPreprocessing(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request,
    google::protobuf::Empty* response
) {
    return this->StopEdgeGraph(context, request, response);
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::ResetProcessing(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request,
    google::protobuf::Empty* response
) {
    return ExecuteAsyncWithDefaultReactor(
        context,
        [this, &request, &response]() -> grpc::Status {
            if (!physiology_core_manual_control && !this->physiology_core_started) {
                return {::grpc::StatusCode::FAILED_PRECONDITION,
                        "Physiology Core has not been started, unable to clean test buffer."};
            }
 
            if (this->container == nullptr) {
                return {::grpc::StatusCode::FAILED_PRECONDITION,
                        "Physiology Preprocessing has not been started, unable to clean test buffer."};
            }
 
            // stop the container if it is running
            PHYSIOLOGY_GRPC_RETURN_IF_ERROR(StopAndDestroyContainer());
 
            grpc::ClientContext core_context;
            // forward request to Physiology Core
            PHYSIOLOGY_GRPC_RETURN_IF_ERROR(core_stub->ResetProcessing(&core_context, *request, response));
 
            // restart the container
            PHYSIOLOGY_GRPC_RETURN_IF_ERROR(BuildAndStartContainer());
            return ::grpc::Status::OK;
        }
    );
 
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::CheckHealth(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request,
    ::google::protobuf::StringValue* response
) {
    return ExecuteAsyncWithDefaultReactor(context, [response]() {
        response->set_value("Service is healthy");
        return ::grpc::Status::OK;
    });
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::IsPhysiologyCoreUp(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request,
    ::google::protobuf::BoolValue* response
) {
    return ExecuteAsyncWithDefaultReactor(context, [this, response]() {
        response->set_value(this->physiology_core_started);
        return ::grpc::Status::OK;
    });
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerUnaryReactor* PhysiologyServiceImpl<TIntegrationMode>::IsEdgeGraphRunning(
    grpc::CallbackServerContext* context,
    const ::google::protobuf::Empty* request,
    ::google::protobuf::BoolValue* response
) {
    return ExecuteAsyncWithDefaultReactor(context, [this, response]() {
        response->set_value(this->container != nullptr && this->container->GraphIsRunning());
        return ::grpc::Status::OK;
    });
}
 
// region ======================= Physiology Core Start/Stop ===========================================================
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
absl::Status PhysiologyServiceImpl<TIntegrationMode>::StartCoreClient() {
    channel = grpc::CreateChannel("0.0.0.0:" + std::to_string(this->physiology_core_port_number),
                                  grpc::InsecureChannelCredentials());
    int retry_interval_ms = 50;
    int remaining_time = this->physiology_core_connection_timeout_ms;
    while (channel->GetState(true) != GRPC_CHANNEL_READY && remaining_time > 0) {
        channel->WaitForConnected(gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
                                               gpr_time_from_millis(retry_interval_ms, GPR_TIMESPAN)));
        remaining_time -= retry_interval_ms;
    }
    if (channel->GetState(true) != GRPC_CHANNEL_READY) {
        channel.reset();
        return absl::InternalError(
            "Failed to connect to Physiology Core gRPC Server within " +
            std::to_string(this->physiology_core_connection_timeout_ms) + " milliseconds, aborting."
        );
    }
    this->core_stub = presage::physiology::PhysiologyCore::NewStub(channel);
    return absl::OkStatus();
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
absl::Status PhysiologyServiceImpl<TIntegrationMode>::StartCoreServer() {
    if (!this->venv_verified) {
        PHYSIOLOGY_GRPC_RETURN_IF_ERROR(this->VerifyVenv());
        this->venv_verified = true;
    }
    const std::filesystem::path core_server_script = "grpc_core_server.py";
    const std::filesystem::path local_core_path = PHYSIOLOGY_CORE_DIRECTORY;
    const std::filesystem::path local_core_server_absolute_path = local_core_path / core_server_script;
 
    bool mock_setting = false;
 
    if (!std::filesystem::is_regular_file(local_core_server_absolute_path)) {
        LOG(INFO) << "Local Physiology Core gRPC Server not found at \""
                     + local_core_server_absolute_path.string()
                     + "\". Assuming development setting and checking for the example physio core server instead...";
 
        const std::filesystem::path mock_core_server_script = "mock_grpc_core_server.py";
        const std::filesystem::path local_tests_path = SMARTSPECTRA_ON_PREM_TESTS_DIRECTORY;
        const std::filesystem::path local_mock_core_server_absolute_path = local_tests_path / mock_core_server_script;
        if (!std::filesystem::is_regular_file(local_mock_core_server_absolute_path)) {
            return absl::NotFoundError("Mock Physiology Core gRPC Server not found at \""
                                       + local_mock_core_server_absolute_path.string()
                                       + "\".");
        }
        mock_setting = true;
 
    }
 
    std::string port_string = std::to_string(this->physiology_core_port_number);
    std::string random_seed_str = std::to_string(this->test_physiology_core_seed);
 
    std::vector<std::string> string_arguments;
    std::vector<const char*> arguments;
    if (this->venv_directory.empty()) {
        // try running locally w/o Core install
        string_arguments = {
            "python",
            core_server_script,
            "--port",
            port_string
        };
        if (mock_setting) {
            string_arguments.emplace_back("--seed");
            string_arguments.push_back(random_seed_str);
        } else {
            string_arguments.emplace_back("--headless");
            if (this->log_settings.log_edge_core_transfer_times) {
                string_arguments.emplace_back("--log_transfer_timing_info");
            }
        }
        std::map<std::string, std::string> variables;
        if (!std::string(PHYSIOLOGY_CORE_DIRECTORY).empty()) {
            this->physiology_core_process_options.working_directory = PHYSIOLOGY_CORE_DIRECTORY;
            variables["PYTHONPATH"] = PHYSIOLOGY_CORE_DIRECTORY;
        } else {
            variables["PYTHONPATH"] = ".";
        }
        this->physiology_core_process_options.env.extra = reproc::env(variables);
    } else {
        std::filesystem::path venv_activate_script = this->venv_directory / "bin" / "activate";
        std::string command = "source " + venv_activate_script.string()
                              + " && python " + local_core_server_absolute_path.string()
                              + " --port " + port_string + " --headless";
        if (this->log_settings.log_edge_core_transfer_times) {
            command += " --log_transfer_timing_info";
        }
        string_arguments = {
            "bash",
            "-c",
            command,
        };
    }
    for (const auto& argument: string_arguments) {
        arguments.push_back(argument.c_str());
    }
    // Ensure the arguments end with a nullptr
    arguments.push_back(nullptr);
 
    reproc::stop_actions stop = {
        {reproc::stop::noop,      reproc::milliseconds(0)},
        {reproc::stop::terminate, reproc::milliseconds(5000)},
        {reproc::stop::kill,      reproc::milliseconds(2000)}
    };
 
    this->physiology_core_process_options.stop = stop;
    this->physiology_core_process_options.redirect.err.type = reproc::redirect::pipe;
    this->physiology_core_process_options.redirect.out.type = reproc::redirect::pipe;
    this->physiology_core_process_options.redirect.parent = true;
    this->physiology_core_process_options.redirect.discard = false;
 
    std::error_code ec = this->physiology_core_process.start(arguments.data(), this->physiology_core_process_options);
    if (ec) {
        std::string additional_debugging_info;
        if (ec.message().find("No such file") != std::string::npos) {
            std::string ps_working_directory = std::filesystem::current_path().string();
            std::string files_in_ps_working_directory =
                presage::filesystem::GetListOfFilesInDirectoryAsString(ps_working_directory);
            std::string pc_working_directory_debugging_info;
            std::string physiology_core_directory(PHYSIOLOGY_CORE_DIRECTORY);
            if (this->venv_directory.empty() && !physiology_core_directory.empty()) {
                pc_working_directory_debugging_info =
                    "Physiology Core working directory: " + physiology_core_directory + ".\n";
                std::string files_in_pc_working_directory =
                    presage::filesystem::GetListOfFilesInDirectoryAsString(PHYSIOLOGY_CORE_DIRECTORY);
                if (files_in_pc_working_directory.empty()) {
                    pc_working_directory_debugging_info += "Nothing found in Physiology Core working directory.";
                } else {
                    pc_working_directory_debugging_info += "Files in Physiology Core working directory:\n"
                                                           + files_in_pc_working_directory;
                }
            }
            additional_debugging_info = +"\nPhysiology Service working directory: " + ps_working_directory +
                                        ".\nFiles in working directory:\n" + files_in_ps_working_directory +
                                        pc_working_directory_debugging_info;
        }
        return absl::InternalError(
            "Failed to start Physiology Core gRPC server: " +
            ec.message() + "." + additional_debugging_info
        );
    }
 
    this->core_log_thread_shutdown = false;
    this->core_log_thread = std::thread([this]() {
        std::error_code ec =
            reproc::drain(
                /*process=*/this->physiology_core_process,
                /*out=*/this->core_stdout_log_sink,
                /*err=*/this->core_stderr_log_sink
            );
        if (ec) {
            if (ec == std::errc::operation_canceled) {
                // everything is fine.
                return;
            }
            LOG(ERROR) << "Error draining process output: " << ec.message();
        }
    });
 
    return absl::OkStatus();
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
absl::Status PhysiologyServiceImpl<TIntegrationMode>::StopCoreServer() {
    this->core_log_thread_shutdown = true;
 
    // make core server try to log something, so it knows to stop the logging thread
 
    bool have_connection_to_core_server = true;
    if (this->core_stub != nullptr) {
        grpc::ClientContext core_context;
        physiology::BlueTooth request;
        google::protobuf::Empty response;
        this->core_stub->IssueBlueTooth(&core_context, request, &response);
        if (this->core_log_thread.joinable()) {
            this->core_log_thread.join();
        }
    } else {
        have_connection_to_core_server = false;
    }
 
    std::error_code ec;
    int status_code;
 
    // Try to stop the core server regardless of whether streams were shut down. A shutdown w/ a non-zero exit code
    // is still better than none at all.
    std::tie(status_code, ec) = this->physiology_core_process.stop(this->physiology_core_process_options.stop);
 
    this->physiology_core_process = reproc::process();
    if (ec) {
        if (!have_connection_to_core_server) {
            return absl::InternalError("Physiology Core gRPC server connection could not be established, "
                                       "was not able to stop the Core server 100% gracefully, "
                                       "but it should be stopped.");
        } else {
            return absl::InternalError(
                "Failed to gracefully shut down Physiology Core server. Error: " + ec.message() +
                ". Status code: " + std::to_string(status_code) + "."
            );
        }
    } else {
        return absl::OkStatus();
    }
}
 
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status PhysiologyServiceImpl<TIntegrationMode>::VerifyVenv() {
    if (this->venv_directory.empty()) {
        LOG(INFO) << "No venv directory specified, skipping venv activation.";
        return absl::OkStatus();
    }
    std::string usage_message =
        "Please ensure that you have a Python 3.11 virtual environment configured inside the root of the "
        "physiology repository, that Physiology-Core package installed on that, and that you are running "
        "physiology_server from the Physiology repository's root directory.";
    std::filesystem::path venv_activate_script = this->venv_directory / "bin" / "activate";
    if (!std::filesystem::exists(this->venv_directory)) {
        return absl::NotFoundError("Venv directory not found at \"" + this->venv_directory.string() +
                                   "\". " + usage_message);
    } else if (!std::filesystem::is_directory(this->venv_directory)) {
        return absl::NotFoundError("\"" + this->venv_directory.string() + "\" is not a directory. " + usage_message);
    } else if (!std::filesystem::is_regular_file(venv_activate_script)) {
        return absl::NotFoundError("Venv activate script not found at \"" + venv_activate_script.string() + "\". " +
                                   usage_message);
    }
    return absl::OkStatus();
}
 
// endregion ===========================================================================================================
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
grpc::ServerWriteReactor<::presage::smartspectra::LogEntry>* PhysiologyServiceImpl<TIntegrationMode>::StreamLogs(
    grpc::CallbackServerContext* context,
    const google::protobuf::Empty* request
) {
    return new LogStreamReactor(log_buffer);
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
bool PhysiologyServiceImpl<TIntegrationMode>::StatusRequiresImmediateRetune(
    presage::physiology::StatusCode status
) {
    using presage::physiology::StatusCode;
    switch (status) {
        case StatusCode::IMAGE_TOO_DARK:
        case StatusCode::IMAGE_TOO_BRIGHT:
            return true;
        default:
            return false;
    }
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
bool PhysiologyServiceImpl<TIntegrationMode>::StatusShouldRetuneAfterRecovery(
    presage::physiology::StatusCode status
) {
    using presage::physiology::StatusCode;
    switch (status) {
        case StatusCode::IMAGE_TOO_DARK:
        case StatusCode::IMAGE_TOO_BRIGHT:
        case StatusCode::NO_FACES_FOUND:
            return true;
        default:
            return false;
    }
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
absl::Status PhysiologyServiceImpl<TIntegrationMode>::HandleStatusForTuning(
    const presage::physiology::StatusValue& status_output,
    presage::physiology::StatusCode previous_status_code
) {
    // Only process if tuning is enabled
    if (!this->camera_tuning_enabled.load() || this->camera_tuner == nullptr) {
        return absl::OkStatus();
    }
 
    // Don't trigger tuning if already tuning
    if (this->is_tuning_camera.load()) {
        return absl::OkStatus();
    }
 
    // Don't trigger tuning if recording is active
    if (this->is_recording.load()) {
        return absl::OkStatus();
    }
 
    const auto current_status = static_cast<presage::physiology::StatusCode>(status_output.value());
    const bool status_requires_immediate_retune = StatusRequiresImmediateRetune(current_status);
    const bool recovered_from_issue =
        this->camera_tuning_recheck_on_recovery_ &&
        current_status == physiology::StatusCode::OK &&
        StatusShouldRetuneAfterRecovery(previous_status_code);
 
    // Nothing to do if status is OK and no recovery-trigger requested.
    if (!status_requires_immediate_retune && !recovered_from_issue) {
        return absl::OkStatus();
    }
 
    // Check cooldown timer
    {
        std::lock_guard<std::mutex> lock(this->camera_tuning_mutex);
        auto now = std::chrono::steady_clock::now();
        auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(
            now - this->last_camera_tuning_time
        );
 
        if (elapsed < this->camera_tuning_cooldown_duration) {
            // Still in cooldown period
            return absl::OkStatus();
        }
 
        // Cooldown expired, trigger tuning
        this->last_camera_tuning_time = now;
    }
 
    // Trigger tuning by starting the tuning process
    auto start_status = this->camera_tuner->StartTuning();
    if (!start_status.ok()) {
        LOG(WARNING) << "Failed to start camera tuning: " << start_status.message();
        // Store error for potential inspection
        {
            std::lock_guard<std::mutex> lock(this->camera_error_mutex_);
            this->last_camera_error_ = start_status;
        }
        return start_status;
    }
 
    this->is_tuning_camera.store(true);
    if (status_requires_immediate_retune) {
        LOG(INFO) << "Status code " << spectra::GetStatusDescription(current_status)
                  << " triggered camera tuning";
    } else {
        LOG(INFO) << "Status recovered (" << spectra::GetStatusDescription(previous_status_code)
                  << " -> " << spectra::GetStatusDescription(current_status)
                  << "); re-triggering camera tuning to verify conditions";
    }
 
    return absl::OkStatus();
}
 
template<smartspectra::container::settings::IntegrationMode TIntegrationMode>
absl::Status PhysiologyServiceImpl<TIntegrationMode>::FlushPendingStatusAfterTuning(
    ipc::IpcStreamWriter<TIntegrationMode>* writer
) {
    if (writer == nullptr) {
        return absl::OkStatus();
    }
 
    std::optional<presage::physiology::StatusValue> pending_status;
    {
        std::lock_guard<std::mutex> lock(this->pending_status_mutex_);
        if (!this->pending_status_after_tuning_.has_value()) {
            return absl::OkStatus();
        }
        pending_status = this->pending_status_after_tuning_;
        this->pending_status_after_tuning_.reset();
    }
 
    return writer->WriteStatus(*pending_status);
}
 
} // namespace presage::smartspectra::grpc_bindings