auto_prefetch_service.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2021-2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include "prefetch_config.h"
#include "kcenon/pacs/storage/worklist_record.h"
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <queue>
#include <set>
#include <string>
#include <thread>
#include <vector>
 
// Forward declarations for kcenon ecosystem
namespace kcenon::thread {
class thread_pool;
}  // namespace kcenon::thread
 
namespace kcenon::logger {
class logger;
}  // namespace kcenon::logger
 
namespace kcenon::common::interfaces {
class IExecutor;
}  // namespace kcenon::common::interfaces
 
// Forward declarations for PACS modules
namespace kcenon::pacs::network {
class association;
}  // namespace kcenon::pacs::network
 
namespace kcenon::pacs::storage {
class index_database;
}  // namespace kcenon::pacs::storage
 
namespace kcenon::pacs::core {
class dicom_dataset;
}  // namespace kcenon::pacs::core
 
namespace kcenon::pacs::workflow {
 
struct prefetch_request {
    std::string patient_id;
 
    std::string patient_name;
 
    std::string scheduled_modality;
 
    std::string scheduled_body_part;
 
    std::string scheduled_study_uid;
 
    std::chrono::system_clock::time_point request_time;
 
    std::size_t retry_count{0};
};
 
class auto_prefetch_service {
public:
    // =========================================================================
    // Construction
    // =========================================================================
 
    explicit auto_prefetch_service(
        storage::index_database& database,
        const prefetch_service_config& config = {});
 
    auto_prefetch_service(
        storage::index_database& database,
        std::shared_ptr<kcenon::thread::thread_pool> thread_pool,
        const prefetch_service_config& config = {});
 
    auto_prefetch_service(
        storage::index_database& database,
        std::shared_ptr<kcenon::common::interfaces::IExecutor> executor,
        const prefetch_service_config& config = {});
 
    ~auto_prefetch_service();
 
    auto_prefetch_service(const auto_prefetch_service&) = delete;
    auto_prefetch_service& operator=(const auto_prefetch_service&) = delete;
 
    auto_prefetch_service(auto_prefetch_service&&) = delete;
    auto_prefetch_service& operator=(auto_prefetch_service&&) = delete;
 
    // =========================================================================
    // Lifecycle Management
    // =========================================================================
 
    void enable();
 
    void start();
 
    void disable(bool wait_for_completion = true);
 
    void stop(bool wait_for_completion = true);
 
    [[nodiscard]] auto is_enabled() const noexcept -> bool;
 
    [[nodiscard]] auto is_running() const noexcept -> bool;
 
    // =========================================================================
    // Manual Operations
    // =========================================================================
 
    [[nodiscard]] auto prefetch_priors(
        const std::string& patient_id,
        std::chrono::days lookback = std::chrono::days{365})
        -> prefetch_result;
 
    void trigger_for_worklist(
        const std::vector<storage::worklist_item>& worklist_items);
 
    void trigger_cycle();
 
    [[nodiscard]] auto run_prefetch_cycle() -> prefetch_result;
 
    // =========================================================================
    // Worklist Event Handler
    // =========================================================================
 
    void on_worklist_query(
        const std::vector<storage::worklist_item>& worklist_items);
 
    // =========================================================================
    // Statistics and Monitoring
    // =========================================================================
 
    [[nodiscard]] auto get_last_result() const
        -> std::optional<prefetch_result>;
 
    [[nodiscard]] auto get_cumulative_stats() const -> prefetch_result;
 
    [[nodiscard]] auto time_until_next_cycle() const
        -> std::optional<std::chrono::seconds>;
 
    [[nodiscard]] auto cycles_completed() const noexcept -> std::size_t;
 
    [[nodiscard]] auto pending_requests() const noexcept -> std::size_t;
 
    // =========================================================================
    // Configuration
    // =========================================================================
 
    void set_prefetch_interval(std::chrono::seconds interval);
 
    [[nodiscard]] auto get_prefetch_interval() const noexcept
        -> std::chrono::seconds;
 
    void set_prefetch_criteria(const prefetch_criteria& criteria);
 
    [[nodiscard]] auto get_prefetch_criteria() const noexcept
        -> const prefetch_criteria&;
 
    void set_cycle_complete_callback(
        prefetch_service_config::cycle_complete_callback callback);
 
    void set_error_callback(
        prefetch_service_config::error_callback callback);
 
private:
    // =========================================================================
    // Internal Methods
    // =========================================================================
 
    void run_loop();
 
    [[nodiscard]] auto execute_cycle() -> prefetch_result;
 
    [[nodiscard]] auto process_request(const prefetch_request& request)
        -> prefetch_result;
 
    [[nodiscard]] auto query_prior_studies(
        const remote_pacs_config& pacs_config,
        const std::string& patient_id,
        std::chrono::days lookback) -> std::vector<prior_study_info>;
 
    [[nodiscard]] auto filter_studies(
        const std::vector<prior_study_info>& studies,
        const prefetch_request& request) -> std::vector<prior_study_info>;
 
    [[nodiscard]] auto study_exists_locally(const std::string& study_uid)
        -> bool;
 
    [[nodiscard]] auto prefetch_study(
        const remote_pacs_config& pacs_config,
        const prior_study_info& study) -> bool;
 
    void update_stats(const prefetch_result& result);
 
    void queue_request(const prefetch_request& request);
 
    [[nodiscard]] auto dequeue_request()
        -> std::optional<prefetch_request>;
 
    // =========================================================================
    // Member Variables
    // =========================================================================
 
    storage::index_database& database_;
 
    std::shared_ptr<kcenon::thread::thread_pool> thread_pool_;
 
    std::shared_ptr<kcenon::common::interfaces::IExecutor> executor_;
 
    prefetch_service_config config_;
 
    std::thread worker_thread_;
 
    mutable std::mutex mutex_;
 
    std::condition_variable cv_;
 
    std::atomic<bool> stop_requested_{false};
 
    std::atomic<bool> enabled_{false};
 
    std::atomic<bool> cycle_in_progress_{false};
 
    std::queue<prefetch_request> request_queue_;
 
    std::set<std::string> queued_patients_;
 
    mutable std::mutex queue_mutex_;
 
    std::optional<prefetch_result> last_result_;
 
    prefetch_result cumulative_stats_;
 
    std::atomic<std::size_t> cycles_count_{0};
 
    std::chrono::steady_clock::time_point next_cycle_time_;
};
 
}  // namespace kcenon::pacs::workflow