kcenon::pacs::workflow::auto_prefetch_service Class Reference

#include <auto_prefetch_service.h>

Collaboration diagram for kcenon::pacs::workflow::auto_prefetch_service:

Public Member Functions
	auto_prefetch_service (storage::index_database &database, const prefetch_service_config &config={})
	Construct auto prefetch service.
	auto_prefetch_service (storage::index_database &database, std::shared_ptr< kcenon::thread::thread_pool > thread_pool, const prefetch_service_config &config={})
	Construct auto prefetch service with thread pool.
	auto_prefetch_service (storage::index_database &database, std::shared_ptr< kcenon::common::interfaces::IExecutor > executor, const prefetch_service_config &config={})
	Construct auto prefetch service with IExecutor (recommended)
	~auto_prefetch_service ()
	Destructor - ensures graceful shutdown.
	auto_prefetch_service (const auto_prefetch_service &)=delete
	Non-copyable.
auto_prefetch_service &	operator= (const auto_prefetch_service &)=delete
	auto_prefetch_service (auto_prefetch_service &&)=delete
	Non-movable.
auto_prefetch_service &	operator= (auto_prefetch_service &&)=delete
void	enable ()
	Enable the prefetch service.
void	start ()
	Start the prefetch service (alias for enable)
void	disable (bool wait_for_completion=true)
	Disable/stop the prefetch service.
void	stop (bool wait_for_completion=true)
	Stop the prefetch service (alias for disable)
auto	is_enabled () const noexcept -> bool
	Check if the service is enabled/running.
auto	is_running () const noexcept -> bool
	Check if the service is running (alias for is_enabled)
auto	prefetch_priors (const std::string &patient_id, std::chrono::days lookback=std::chrono::days{365}) -> prefetch_result
	Manually prefetch prior studies for a patient.
void	trigger_for_worklist (const std::vector< storage::worklist_item > &worklist_items)
	Trigger prefetch for worklist items.
void	trigger_cycle ()
	Trigger next cycle immediately.
auto	run_prefetch_cycle () -> prefetch_result
	Run a prefetch cycle manually.
void	on_worklist_query (const std::vector< storage::worklist_item > &worklist_items)
	Handle worklist query event.
auto	get_last_result () const -> std::optional< prefetch_result >
	Get the result of the last prefetch cycle.
auto	get_cumulative_stats () const -> prefetch_result
	Get cumulative statistics since service started.
auto	time_until_next_cycle () const -> std::optional< std::chrono::seconds >
	Get the time until the next scheduled prefetch cycle.
auto	cycles_completed () const noexcept -> std::size_t
	Get the number of cycles completed.
auto	pending_requests () const noexcept -> std::size_t
	Get the number of pending prefetch requests.
void	set_prefetch_interval (std::chrono::seconds interval)
	Update the prefetch interval.
auto	get_prefetch_interval () const noexcept -> std::chrono::seconds
	Get the current prefetch interval.
void	set_prefetch_criteria (const prefetch_criteria &criteria)
	Update the prefetch criteria.
auto	get_prefetch_criteria () const noexcept -> const prefetch_criteria &
	Get the current prefetch criteria.
void	set_cycle_complete_callback (prefetch_service_config::cycle_complete_callback callback)
	Set the cycle complete callback.
void	set_error_callback (prefetch_service_config::error_callback callback)
	Set the error callback.

Private Member Functions
void	run_loop ()
	Background thread main loop.
auto	execute_cycle () -> prefetch_result
	Execute a single prefetch cycle.
auto	process_request (const prefetch_request &request) -> prefetch_result
	Process a single prefetch request.
auto	query_prior_studies (const remote_pacs_config &pacs_config, const std::string &patient_id, std::chrono::days lookback) -> std::vector< prior_study_info >
	Query remote PACS for prior studies.
auto	filter_studies (const std::vector< prior_study_info > &studies, const prefetch_request &request) -> std::vector< prior_study_info >
	Filter prior studies based on criteria.
auto	study_exists_locally (const std::string &study_uid) -> bool
	Check if study already exists locally.
auto	prefetch_study (const remote_pacs_config &pacs_config, const prior_study_info &study) -> bool
	Prefetch a single study via C-MOVE.
void	update_stats (const prefetch_result &result)
	Update cumulative statistics.
void	queue_request (const prefetch_request &request)
	Add request to queue (deduplicated)
auto	dequeue_request () -> std::optional< prefetch_request >
	Get next request from queue.

Private Attributes
storage::index_database &	database_
	Reference to PACS index database.
std::shared_ptr< kcenon::thread::thread_pool >	thread_pool_
	Thread pool for parallel prefetches (optional, legacy)
std::shared_ptr< kcenon::common::interfaces::IExecutor >	executor_
	IExecutor for task execution (recommended, Issue #487)
prefetch_service_config	config_
	Service configuration.
std::thread	worker_thread_
	Background worker thread.
std::mutex	mutex_
	Mutex for thread synchronization.
std::condition_variable	cv_
	Condition variable for sleep/wake.
std::atomic< bool >	stop_requested_ {false}
	Flag to signal shutdown.
std::atomic< bool >	enabled_ {false}
	Flag indicating service is enabled.
std::atomic< bool >	cycle_in_progress_ {false}
	Flag indicating a cycle is in progress.
std::queue< prefetch_request >	request_queue_
	Queue of pending prefetch requests.
std::set< std::string >	queued_patients_
	Set of patient IDs currently in queue (for deduplication)
std::mutex	queue_mutex_
	Mutex for request queue.
std::optional< prefetch_result >	last_result_
	Last prefetch result.
prefetch_result	cumulative_stats_
	Cumulative statistics.
std::atomic< std::size_t >	cycles_count_ {0}
	Number of completed cycles.
std::chrono::steady_clock::time_point	next_cycle_time_
	Time of next scheduled cycle.

Detailed Description

Definition at line 203 of file auto_prefetch_service.h.

Constructor & Destructor Documentation

auto_prefetch_service() [1/5]

kcenon::pacs::workflow::auto_prefetch_service::auto_prefetch_service ( storage::index_database & database, const prefetch_service_config & config = {} )

explicit

Construct auto prefetch service.

Parameters

database	Reference to the PACS index database for checking existing studies and receiving worklist events
config	Service configuration

Definition at line 33 of file auto_prefetch_service.cpp.

    : database_(database)
    , config_(config) {
    if (config_.auto_start && config_.enabled) {
        start();
    }
}

References kcenon::pacs::workflow::prefetch_service_config::auto_start, config_, and kcenon::pacs::workflow::prefetch_service_config::enabled.

auto_prefetch_service() [2/5]

kcenon::pacs::workflow::auto_prefetch_service::auto_prefetch_service	(	storage::index_database &	database,
		std::shared_ptr< kcenon::thread::thread_pool >	thread_pool,
		const prefetch_service_config &	config = {} )

Construct auto prefetch service with thread pool.

Parameters

database	Reference to the PACS index database
thread_pool	Thread pool for parallel prefetch operations
config	Service configuration

Definition at line 43 of file auto_prefetch_service.cpp.

    : database_(database)
    , thread_pool_(std::move(thread_pool))
    , config_(config) {
    if (config_.auto_start && config_.enabled) {
        start();
    }
}

References kcenon::pacs::workflow::prefetch_service_config::auto_start, config_, and kcenon::pacs::workflow::prefetch_service_config::enabled.

auto_prefetch_service() [3/5]

kcenon::pacs::workflow::auto_prefetch_service::auto_prefetch_service	(	storage::index_database &	database,
		std::shared_ptr< kcenon::common::interfaces::IExecutor >	executor,
		const prefetch_service_config &	config = {} )

Construct auto prefetch service with IExecutor (recommended)

This constructor accepts the standardized IExecutor interface from common_system, enabling better testability and decoupling from specific thread pool implementations.

Parameters

database	Reference to the PACS index database
executor	IExecutor for task execution (from common_system)
config	Service configuration

See also

Issue #487 - IExecutor integration

Definition at line 55 of file auto_prefetch_service.cpp.

    : database_(database)
    , executor_(std::move(executor))
    , config_(config) {
    if (config_.auto_start && config_.enabled) {
        start();
    }
}

References kcenon::pacs::workflow::prefetch_service_config::auto_start, config_, and kcenon::pacs::workflow::prefetch_service_config::enabled.

~auto_prefetch_service()

kcenon::pacs::workflow::auto_prefetch_service::~auto_prefetch_service

(

)

Destructor - ensures graceful shutdown.

Definition at line 67 of file auto_prefetch_service.cpp.

                                              {
    stop(true);
}

References stop().

Here is the call graph for this function:

auto_prefetch_service() [4/5]

kcenon::pacs::workflow::auto_prefetch_service::auto_prefetch_service ( const auto_prefetch_service & )

delete

Non-copyable.

auto_prefetch_service() [5/5]

kcenon::pacs::workflow::auto_prefetch_service::auto_prefetch_service ( auto_prefetch_service && )

delete

Non-movable.

Member Function Documentation

cycles_completed()

auto kcenon::pacs::workflow::auto_prefetch_service::cycles_completed ( ) const -> std::size_t

nodiscardnoexcept

Get the number of cycles completed.

Returns

Total number of completed prefetch cycles

Definition at line 228 of file auto_prefetch_service.cpp.

                                                                         {
    return cycles_count_.load();
}

References cycles_count_.

dequeue_request()

auto kcenon::pacs::workflow::auto_prefetch_service::dequeue_request ( ) -> std::optional<prefetch_request>

nodiscardprivate

Get next request from queue.

Returns

Optional containing next request, or nullopt if empty

Definition at line 807 of file auto_prefetch_service.cpp.

                                     {
    std::lock_guard<std::mutex> lock(queue_mutex_);
 
    if (request_queue_.empty()) {
        return std::nullopt;
    }
 
    auto request = std::move(request_queue_.front());
    request_queue_.pop();
    queued_patients_.erase(request.patient_id);
 
    return request;
}

disable()

void kcenon::pacs::workflow::auto_prefetch_service::disable

(

bool

wait_for_completion = true

)

Disable/stop the prefetch service.

Gracefully stops the service, waiting for any in-progress prefetch operations to complete.

Parameters

wait_for_completion

If true, waits for current operations

Definition at line 97 of file auto_prefetch_service.cpp.

                                                            {
    stop(wait_for_completion);
}

References stop().

Here is the call graph for this function:

enable()

void kcenon::pacs::workflow::auto_prefetch_service::enable

(

)

Enable the prefetch service.

Starts the background prefetch thread and begins monitoring worklist queries for prefetch candidates.

Definition at line 75 of file auto_prefetch_service.cpp.

                                   {
    start();
}

References start().

Here is the call graph for this function:

execute_cycle()

auto kcenon::pacs::workflow::auto_prefetch_service::execute_cycle ( ) -> prefetch_result

nodiscardprivate

Execute a single prefetch cycle.

Returns

Prefetch result

Definition at line 334 of file auto_prefetch_service.cpp.

                                                             {
    prefetch_result cycle_result;
    cycle_result.timestamp = std::chrono::system_clock::now();
    auto cycle_start = std::chrono::steady_clock::now();
 
    // Process all pending requests
    while (auto request = dequeue_request()) {
        if (stop_requested_.load()) {
            break;
        }
 
        auto result = process_request(*request);
        cycle_result += result;
    }
 
    cycle_result.duration = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - cycle_start);
 
    integration::logger_adapter::info(
        "Prefetch cycle completed patients={} studies_prefetched={} studies_failed={} duration_ms={}",
        cycle_result.patients_processed,
        cycle_result.studies_prefetched,
        cycle_result.studies_failed,
        cycle_result.duration.count());
 
    // Record metrics
    integration::monitoring_adapter::record_histogram(
        "prefetch_cycle_duration_ms",
        static_cast<double>(cycle_result.duration.count()));
    integration::monitoring_adapter::increment_counter(
        "prefetch_studies_total",
        static_cast<int64_t>(cycle_result.studies_prefetched));
    integration::monitoring_adapter::increment_counter(
        "prefetch_failures_total",
        static_cast<int64_t>(cycle_result.studies_failed));
 
    return cycle_result;
}

References kcenon::pacs::workflow::prefetch_result::duration, kcenon::pacs::integration::monitoring_adapter::increment_counter(), kcenon::pacs::integration::logger_adapter::info(), kcenon::pacs::workflow::prefetch_result::patients_processed, kcenon::pacs::integration::monitoring_adapter::record_histogram(), kcenon::pacs::workflow::prefetch_result::studies_failed, kcenon::pacs::workflow::prefetch_result::studies_prefetched, and kcenon::pacs::workflow::prefetch_result::timestamp.

Referenced by run_loop().

Here is the call graph for this function:

Here is the caller graph for this function:

filter_studies()

auto kcenon::pacs::workflow::auto_prefetch_service::filter_studies ( const std::vector< prior_study_info > & studies, const prefetch_request & request ) -> std::vector<prior_study_info>

nodiscardprivate

Filter prior studies based on criteria.

Parameters

studies	List of candidate studies
request	Original prefetch request

Returns

Filtered list of studies to prefetch

Definition at line 606 of file auto_prefetch_service.cpp.

                                                                    {
 
    std::vector<prior_study_info> filtered;
 
    for (const auto& study : studies) {
        // Apply modality filters
        bool modality_match = true;
 
        // Check include list (if not empty, study must match)
        if (!config_.criteria.include_modalities.empty()) {
            modality_match = false;
            for (const auto& mod : study.modalities) {
                if (config_.criteria.include_modalities.count(mod) > 0) {
                    modality_match = true;
                    break;
                }
            }
        }
 
        // Check exclude list
        if (modality_match && !config_.criteria.exclude_modalities.empty()) {
            for (const auto& mod : study.modalities) {
                if (config_.criteria.exclude_modalities.count(mod) > 0) {
                    modality_match = false;
                    break;
                }
            }
        }
 
        if (!modality_match) {
            continue;
        }
 
        // Apply body part filter
        if (!config_.criteria.include_body_parts.empty()) {
            if (config_.criteria.include_body_parts.count(
                    study.body_part_examined) == 0) {
                continue;
            }
        }
 
        filtered.push_back(study);
    }
 
    // Sort by preference if enabled
    if (config_.criteria.prefer_same_modality ||
        config_.criteria.prefer_same_body_part) {
        std::ranges::sort(filtered, [&](const auto& a, const auto& b) {
            int score_a = 0;
            int score_b = 0;
 
            if (config_.criteria.prefer_same_modality) {
                if (a.modalities.count(request.scheduled_modality) > 0) {
                    score_a += 10;
                }
                if (b.modalities.count(request.scheduled_modality) > 0) {
                    score_b += 10;
                }
            }
 
            if (config_.criteria.prefer_same_body_part) {
                if (a.body_part_examined == request.scheduled_body_part) {
                    score_a += 5;
                }
                if (b.body_part_examined == request.scheduled_body_part) {
                    score_b += 5;
                }
            }
 
            // Higher score first, then by date (newer first)
            if (score_a != score_b) {
                return score_a > score_b;
            }
            return a.study_date > b.study_date;
        });
    }
 
    // Limit results
    if (config_.criteria.max_studies_per_patient > 0 &&
        filtered.size() > config_.criteria.max_studies_per_patient) {
        filtered.resize(config_.criteria.max_studies_per_patient);
    }
 
    return filtered;
}

get_cumulative_stats()

auto kcenon::pacs::workflow::auto_prefetch_service::get_cumulative_stats ( ) const -> prefetch_result

nodiscard

Get cumulative statistics since service started.

Returns

Cumulative prefetch statistics

Definition at line 207 of file auto_prefetch_service.cpp.

                                                                          {
    std::lock_guard<std::mutex> lock(mutex_);
    return cumulative_stats_;
}

References cumulative_stats_, and mutex_.

get_last_result()

auto kcenon::pacs::workflow::auto_prefetch_service::get_last_result ( ) const -> std::optional<prefetch_result>

nodiscard

Get the result of the last prefetch cycle.

Returns

Last prefetch result, or nullopt if no cycle has run

Definition at line 201 of file auto_prefetch_service.cpp.

                                    {
    std::lock_guard<std::mutex> lock(mutex_);
    return last_result_;
}

References last_result_, and mutex_.

get_prefetch_criteria()

auto kcenon::pacs::workflow::auto_prefetch_service::get_prefetch_criteria ( ) const -> const prefetch_criteria&

nodiscardnoexcept

Get the current prefetch criteria.

Returns

Current selection criteria

Definition at line 259 of file auto_prefetch_service.cpp.

                                {
    std::lock_guard<std::mutex> lock(mutex_);
    return config_.criteria;
}

References config_, kcenon::pacs::workflow::prefetch_service_config::criteria, and mutex_.

get_prefetch_interval()

auto kcenon::pacs::workflow::auto_prefetch_service::get_prefetch_interval ( ) const -> std::chrono::seconds

nodiscardnoexcept

Get the current prefetch interval.

Returns

Current interval between prefetch cycles

Definition at line 247 of file auto_prefetch_service.cpp.

                        {
    std::lock_guard<std::mutex> lock(mutex_);
    return config_.prefetch_interval;
}

References config_, mutex_, and kcenon::pacs::workflow::prefetch_service_config::prefetch_interval.

is_enabled()

auto kcenon::pacs::workflow::auto_prefetch_service::is_enabled ( ) const -> bool

nodiscardnoexcept

Check if the service is enabled/running.

Returns

true if the service is actively prefetching

Definition at line 120 of file auto_prefetch_service.cpp.

                                                              {
    return enabled_.load();
}

References enabled_.

Referenced by is_running().

Here is the caller graph for this function:

is_running()

auto kcenon::pacs::workflow::auto_prefetch_service::is_running ( ) const -> bool

nodiscardnoexcept

Check if the service is running (alias for is_enabled)

Returns

true if the service is actively prefetching

Definition at line 124 of file auto_prefetch_service.cpp.

                                                              {
    return is_enabled();
}

References is_enabled().

Here is the call graph for this function:

on_worklist_query()

void kcenon::pacs::workflow::auto_prefetch_service::on_worklist_query

(

const std::vector< storage::worklist_item > &

worklist_items

)

Handle worklist query event.

Called when a worklist query is processed. Extracts patient information and queues prefetch requests for each unique patient.

Parameters

worklist_items

The worklist items returned by the query

Definition at line 170 of file auto_prefetch_service.cpp.

                                                           {
 
    for (const auto& item : worklist_items) {
        if (item.patient_id.empty()) {
            continue;
        }
 
        prefetch_request request;
        request.patient_id = item.patient_id;
        request.patient_name = item.patient_name;
        request.scheduled_modality = item.modality;
        request.scheduled_study_uid = item.study_uid;
        request.request_time = std::chrono::system_clock::now();
 
        queue_request(request);
    }
 
    // Wake up worker if there are new requests
    cv_.notify_one();
 
    integration::logger_adapter::debug(
        "Queued prefetch requests from worklist worklist_items={} queue_size={}",
        worklist_items.size(),
        pending_requests());
}

References cv_, kcenon::pacs::integration::logger_adapter::debug(), kcenon::pacs::workflow::prefetch_request::patient_id, kcenon::pacs::workflow::prefetch_request::patient_name, pending_requests(), queue_request(), kcenon::pacs::workflow::prefetch_request::request_time, kcenon::pacs::workflow::prefetch_request::scheduled_modality, and kcenon::pacs::workflow::prefetch_request::scheduled_study_uid.

Referenced by trigger_for_worklist().

Here is the call graph for this function:

Here is the caller graph for this function:

operator=() [1/2]

auto_prefetch_service & kcenon::pacs::workflow::auto_prefetch_service::operator= ( auto_prefetch_service && )

delete

operator=() [2/2]

auto_prefetch_service & kcenon::pacs::workflow::auto_prefetch_service::operator= ( const auto_prefetch_service & )

delete

pending_requests()

auto kcenon::pacs::workflow::auto_prefetch_service::pending_requests ( ) const -> std::size_t

nodiscardnoexcept

Get the number of pending prefetch requests.

Returns

Number of patients waiting to be prefetched

Definition at line 232 of file auto_prefetch_service.cpp.

                                                                         {
    std::lock_guard<std::mutex> lock(queue_mutex_);
    return request_queue_.size();
}

References queue_mutex_, and request_queue_.

Referenced by on_worklist_query().

Here is the caller graph for this function:

prefetch_priors()

auto kcenon::pacs::workflow::auto_prefetch_service::prefetch_priors ( const std::string & patient_id, std::chrono::days lookback = std::chrono::days{365} ) -> prefetch_result

nodiscard

Manually prefetch prior studies for a patient.

Immediately triggers a prefetch operation for the specified patient, regardless of whether the service is enabled or the patient is in the worklist.

Parameters

patient_id	The patient ID to prefetch priors for
lookback	Lookback period (default: use config value)

Returns

Result with count of prefetched studies, or error

Definition at line 132 of file auto_prefetch_service.cpp.

                                               {
 
    prefetch_request request;
    request.patient_id = patient_id;
    request.request_time = std::chrono::system_clock::now();
 
    // Override criteria lookback if specified
    auto saved_lookback = config_.criteria.lookback_period;
    config_.criteria.lookback_period = lookback;
 
    auto result = process_request(request);
 
    config_.criteria.lookback_period = saved_lookback;
 
    return result;
}

References kcenon::pacs::workflow::prefetch_request::patient_id, and kcenon::pacs::workflow::prefetch_request::request_time.

prefetch_study()

auto kcenon::pacs::workflow::auto_prefetch_service::prefetch_study ( const remote_pacs_config & pacs_config, const prior_study_info & study ) -> bool

nodiscardprivate

Prefetch a single study via C-MOVE.

Parameters

pacs_config	Remote PACS configuration
study	Study information

Returns

true if prefetch was successful

Definition at line 701 of file auto_prefetch_service.cpp.

                                           {
 
    integration::logger_adapter::debug(
        "Prefetching study study_uid={} patient_id={} remote_pacs={}",
        study.study_instance_uid,
        study.patient_id,
        pacs_config.ae_title);
 
    try {
        // Configure association for C-MOVE
        network::association_config assoc_config;
        assoc_config.calling_ae_title = pacs_config.local_ae_title;
        assoc_config.called_ae_title = pacs_config.ae_title;
        assoc_config.proposed_contexts.push_back({
            1,
            std::string(services::study_root_move_sop_class_uid),
            {"1.2.840.10008.1.2.1", "1.2.840.10008.1.2"}
        });
 
        // Connect to remote PACS
        auto connect_result = network::association::connect(
            pacs_config.host,
            pacs_config.port,
            assoc_config,
            std::chrono::duration_cast<network::association::duration>(
                pacs_config.connection_timeout));
 
        if (connect_result.is_err()) {
            integration::logger_adapter::error(
                "Failed to connect to remote PACS for C-MOVE remote_pacs={} error={}",
                pacs_config.ae_title,
                connect_result.error().message);
            return false;
        }
 
        auto assoc = std::move(connect_result.value());
 
        // Configure retrieve SCU with our local AE as move destination
        services::retrieve_scu_config retrieve_config;
        retrieve_config.mode = services::retrieve_mode::c_move;
        retrieve_config.model = services::query_model::study_root;
        retrieve_config.level = services::query_level::study;
        retrieve_config.move_destination = pacs_config.local_ae_title;
        retrieve_config.timeout = std::chrono::duration_cast<std::chrono::milliseconds>(
            pacs_config.association_timeout);
 
        services::retrieve_scu scu(retrieve_config);
        auto move_result = scu.retrieve_study(
            assoc, study.study_instance_uid);
 
        // Release association
        (void)assoc.release();
 
        if (move_result.is_ok() && move_result.value().is_success()) {
            integration::logger_adapter::info(
                "Successfully prefetched study study_uid={} completed={} remote_pacs={}",
                study.study_instance_uid,
                move_result.value().completed,
                pacs_config.ae_title);
            return true;
        }
 
        if (move_result.is_err()) {
            integration::logger_adapter::error(
                "C-MOVE failed study_uid={} remote_pacs={} error={}",
                study.study_instance_uid,
                pacs_config.ae_title,
                move_result.error().message);
        } else if (move_result.value().has_failures()) {
            integration::logger_adapter::error(
                "C-MOVE completed with failures study_uid={} completed={} failed={} remote_pacs={}",
                study.study_instance_uid,
                move_result.value().completed,
                move_result.value().failed,
                pacs_config.ae_title);
        }
 
        return false;
    } catch (const std::exception& e) {
        integration::logger_adapter::error(
            "Exception during prefetch study_uid={} remote_pacs={} error={}",
            study.study_instance_uid,
            pacs_config.ae_title,
            e.what());
        return false;
    }
}

References kcenon::pacs::services::c_move, kcenon::pacs::network::association_config::called_ae_title, kcenon::pacs::network::association_config::calling_ae_title, kcenon::pacs::network::association::connect(), kcenon::pacs::integration::logger_adapter::debug(), kcenon::pacs::integration::logger_adapter::error(), kcenon::pacs::integration::logger_adapter::info(), kcenon::pacs::services::retrieve_scu_config::level, kcenon::pacs::services::retrieve_scu_config::mode, kcenon::pacs::services::retrieve_scu_config::model, kcenon::pacs::services::retrieve_scu_config::move_destination, kcenon::pacs::network::association_config::proposed_contexts, kcenon::pacs::services::retrieve_scu::retrieve_study(), kcenon::pacs::services::study, kcenon::pacs::services::study_root, kcenon::pacs::services::study_root_move_sop_class_uid, and kcenon::pacs::services::retrieve_scu_config::timeout.

Here is the call graph for this function:

process_request()

auto kcenon::pacs::workflow::auto_prefetch_service::process_request ( const prefetch_request & request ) -> prefetch_result

nodiscardprivate

Process a single prefetch request.

Parameters

request

The prefetch request to process

Returns

Prefetch result for this request

Definition at line 373 of file auto_prefetch_service.cpp.

                       {
 
    prefetch_result result;
    result.patients_processed = 1;
    result.timestamp = std::chrono::system_clock::now();
    auto start_time = std::chrono::steady_clock::now();
 
    integration::logger_adapter::debug(
        "Processing prefetch request patient_id={} scheduled_modality={}",
        request.patient_id,
        request.scheduled_modality);
 
    // Query each remote PACS for prior studies
    for (const auto& pacs : config_.remote_pacs) {
        if (!pacs.is_valid()) {
            continue;
        }
 
        // Query for prior studies
        auto prior_studies = query_prior_studies(
            pacs,
            request.patient_id,
            config_.criteria.lookback_period);
 
        // Filter based on criteria
        auto filtered_studies = filter_studies(prior_studies, request);
 
        // Prefetch each study
        for (const auto& study : filtered_studies) {
            // Skip if already present locally
            if (study_exists_locally(study.study_instance_uid)) {
                ++result.studies_already_present;
                continue;
            }
 
            // Skip the scheduled study itself
            if (study.study_instance_uid == request.scheduled_study_uid) {
                continue;
            }
 
            // Attempt prefetch
            bool success = prefetch_study(pacs, study);
 
            if (success) {
                ++result.studies_prefetched;
                result.series_prefetched += study.number_of_series;
                result.instances_prefetched += study.number_of_instances;
 
                if (config_.on_prefetch_complete) {
                    config_.on_prefetch_complete(
                        request.patient_id, study, true, "");
                }
            } else {
                ++result.studies_failed;
 
                if (config_.on_prefetch_error) {
                    config_.on_prefetch_error(
                        request.patient_id,
                        study.study_instance_uid,
                        "Failed to prefetch study");
                }
            }
 
            // Check rate limiting
            if (config_.rate_limit_per_minute > 0) {
                auto elapsed = std::chrono::steady_clock::now() - start_time;
                auto elapsed_minutes =
                    std::chrono::duration_cast<std::chrono::minutes>(elapsed)
                        .count();
                if (elapsed_minutes < 1) {
                    auto prefetched_this_minute =
                        result.studies_prefetched + result.studies_failed;
                    if (prefetched_this_minute >= config_.rate_limit_per_minute) {
                        // Wait until the next minute
                        std::this_thread::sleep_for(
                            std::chrono::seconds(60) - elapsed);
                    }
                }
            }
        }
    }
 
    result.duration = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - start_time);
 
    return result;
}

References kcenon::pacs::integration::logger_adapter::debug(), kcenon::pacs::workflow::prefetch_result::duration, kcenon::pacs::workflow::prefetch_result::instances_prefetched, kcenon::pacs::workflow::prefetch_result::patients_processed, kcenon::pacs::workflow::prefetch_result::series_prefetched, kcenon::pacs::workflow::prefetch_result::studies_already_present, kcenon::pacs::workflow::prefetch_result::studies_failed, kcenon::pacs::workflow::prefetch_result::studies_prefetched, and kcenon::pacs::workflow::prefetch_result::timestamp.

Here is the call graph for this function:

query_prior_studies()

auto kcenon::pacs::workflow::auto_prefetch_service::query_prior_studies ( const remote_pacs_config & pacs_config, const std::string & patient_id, std::chrono::days lookback ) -> std::vector<prior_study_info>

nodiscardprivate

Query remote PACS for prior studies.

Parameters

pacs_config	Remote PACS configuration
patient_id	Patient ID to query
lookback	Lookback period

Returns

Vector of prior study information

Definition at line 462 of file auto_prefetch_service.cpp.

                                                           {
 
    std::vector<prior_study_info> results;
 
    // Calculate date range
    auto now = std::chrono::system_clock::now();
    auto from_time = now - lookback;
 
    // Format dates as YYYYMMDD
    auto format_date = [](std::chrono::system_clock::time_point tp) {
        auto time_t = std::chrono::system_clock::to_time_t(tp);
        std::tm tm = *std::localtime(&time_t);
        char buffer[9];
        std::strftime(buffer, sizeof(buffer), "%Y%m%d", &tm);
        return std::string(buffer);
    };
 
    std::string from_date = format_date(from_time);
    std::string to_date = format_date(now);
 
    integration::logger_adapter::debug(
        "Querying prior studies remote_pacs={} patient_id={} from_date={} to_date={}",
        pacs_config.ae_title,
        patient_id,
        from_date,
        to_date);
 
    try {
        // Configure association for C-FIND
        network::association_config assoc_config;
        assoc_config.calling_ae_title = pacs_config.local_ae_title;
        assoc_config.called_ae_title = pacs_config.ae_title;
        assoc_config.proposed_contexts.push_back({
            1,
            std::string(services::study_root_find_sop_class_uid),
            {"1.2.840.10008.1.2.1", "1.2.840.10008.1.2"}
        });
 
        // Connect to remote PACS
        auto connect_result = network::association::connect(
            pacs_config.host,
            pacs_config.port,
            assoc_config,
            std::chrono::duration_cast<network::association::duration>(
                pacs_config.connection_timeout));
 
        if (connect_result.is_err()) {
            integration::logger_adapter::error(
                "Failed to connect to remote PACS for C-FIND remote_pacs={} error={}",
                pacs_config.ae_title,
                connect_result.error().message);
            return results;
        }
 
        auto assoc = std::move(connect_result.value());
 
        // Build query keys
        services::study_query_keys keys;
        keys.patient_id = patient_id;
        keys.study_date = from_date + "-" + to_date;
 
        // Execute C-FIND query
        services::query_scu_config query_config;
        query_config.model = services::query_model::study_root;
        query_config.level = services::query_level::study;
        query_config.timeout = std::chrono::duration_cast<std::chrono::milliseconds>(
            pacs_config.association_timeout);
 
        services::query_scu scu(query_config);
        auto find_result = scu.find_studies(assoc, keys);
 
        if (find_result.is_ok() && find_result.value().is_success()) {
            for (const auto& dataset : find_result.value().matches) {
                prior_study_info info;
                info.study_instance_uid =
                    dataset.get_string(core::tags::study_instance_uid);
                info.patient_id = patient_id;
                info.patient_name =
                    dataset.get_string(core::tags::patient_name);
                info.study_date =
                    dataset.get_string(core::tags::study_date);
                info.study_description =
                    dataset.get_string(core::tags::study_description);
 
                // Parse modalities in study (backslash-separated)
                auto modalities_str =
                    dataset.get_string(core::tags::modalities_in_study);
                if (!modalities_str.empty()) {
                    std::istringstream ss(modalities_str);
                    std::string mod;
                    while (std::getline(ss, mod, '\\')) {
                        if (!mod.empty()) {
                            info.modalities.insert(mod);
                        }
                    }
                }
 
                // Parse numeric fields
                auto num_series_str = dataset.get_string(
                    core::tags::number_of_study_related_series);
                if (!num_series_str.empty()) {
                    try {
                        info.number_of_series = std::stoull(num_series_str);
                    } catch (...) {}
                }
 
                auto num_instances_str = dataset.get_string(
                    core::tags::number_of_study_related_instances);
                if (!num_instances_str.empty()) {
                    try {
                        info.number_of_instances = std::stoull(num_instances_str);
                    } catch (...) {}
                }
 
                results.push_back(std::move(info));
            }
 
            integration::logger_adapter::debug(
                "C-FIND query returned matches={} remote_pacs={} patient_id={}",
                results.size(),
                pacs_config.ae_title,
                patient_id);
        } else if (find_result.is_err()) {
            integration::logger_adapter::error(
                "C-FIND query failed remote_pacs={} error={}",
                pacs_config.ae_title,
                find_result.error().message);
        }
 
        // Release association
        (void)assoc.release();
    } catch (const std::exception& e) {
        integration::logger_adapter::error(
            "Exception querying prior studies remote_pacs={} error={}",
            pacs_config.ae_title,
            e.what());
    }
 
    return results;
}

References kcenon::pacs::network::association_config::called_ae_title, kcenon::pacs::network::association_config::calling_ae_title, kcenon::pacs::network::association::connect(), kcenon::pacs::integration::logger_adapter::debug(), kcenon::pacs::integration::logger_adapter::error(), kcenon::pacs::services::query_scu::find_studies(), kcenon::pacs::services::query_scu_config::level, kcenon::pacs::core::tags::modalities_in_study, kcenon::pacs::services::query_scu_config::model, kcenon::pacs::core::tags::number_of_study_related_instances, kcenon::pacs::core::tags::number_of_study_related_series, kcenon::pacs::services::study_query_keys::patient_id, kcenon::pacs::core::tags::patient_name, kcenon::pacs::network::association_config::proposed_contexts, kcenon::pacs::services::study, kcenon::pacs::core::tags::study_date, kcenon::pacs::services::study_query_keys::study_date, kcenon::pacs::core::tags::study_description, kcenon::pacs::core::tags::study_instance_uid, kcenon::pacs::services::study_root, kcenon::pacs::services::study_root_find_sop_class_uid, and kcenon::pacs::services::query_scu_config::timeout.

Here is the call graph for this function:

queue_request()

void kcenon::pacs::workflow::auto_prefetch_service::queue_request ( const prefetch_request & request )

private

Add request to queue (deduplicated)

Parameters

request

The prefetch request to queue

Definition at line 795 of file auto_prefetch_service.cpp.

                                                                         {
    std::lock_guard<std::mutex> lock(queue_mutex_);
 
    // Deduplicate by patient ID
    if (queued_patients_.count(request.patient_id) > 0) {
        return;
    }
 
    request_queue_.push(request);
    queued_patients_.insert(request.patient_id);
}

References kcenon::pacs::workflow::prefetch_request::patient_id, queue_mutex_, queued_patients_, and request_queue_.

Referenced by on_worklist_query().

Here is the caller graph for this function:

run_loop()

void kcenon::pacs::workflow::auto_prefetch_service::run_loop ( )

private

Background thread main loop.

Definition at line 281 of file auto_prefetch_service.cpp.

                                     {
    integration::logger_adapter::debug("Prefetch service worker thread started");
 
    while (!stop_requested_.load()) {
        std::unique_lock<std::mutex> lock(mutex_);
 
        // Wait until next cycle time or until woken up
        auto wait_until = next_cycle_time_;
        cv_.wait_until(lock, wait_until, [this]() {
            return stop_requested_.load() ||
                   std::chrono::steady_clock::now() >= next_cycle_time_ ||
                   !request_queue_.empty();
        });
 
        if (stop_requested_.load()) {
            break;
        }
 
        // Check if we should run a cycle
        auto now = std::chrono::steady_clock::now();
        bool should_run_cycle = (now >= next_cycle_time_) ||
                                (!request_queue_.empty());
 
        if (should_run_cycle) {
            lock.unlock();
 
            cycle_in_progress_.store(true);
            auto result = execute_cycle();
            cycle_in_progress_.store(false);
 
            lock.lock();
 
            // Update last result
            last_result_ = result;
            update_stats(result);
            ++cycles_count_;
 
            // Schedule next cycle
            next_cycle_time_ = std::chrono::steady_clock::now() +
                               config_.prefetch_interval;
 
            // Invoke callback
            if (config_.on_cycle_complete) {
                lock.unlock();
                config_.on_cycle_complete(result);
                lock.lock();
            }
        }
    }
 
    integration::logger_adapter::debug("Prefetch service worker thread stopped");
}

References config_, cv_, cycle_in_progress_, cycles_count_, kcenon::pacs::integration::logger_adapter::debug(), execute_cycle(), last_result_, mutex_, next_cycle_time_, kcenon::pacs::workflow::prefetch_service_config::on_cycle_complete, kcenon::pacs::workflow::prefetch_service_config::prefetch_interval, request_queue_, stop_requested_, and update_stats().

Referenced by start().

Here is the call graph for this function:

Here is the caller graph for this function:

run_prefetch_cycle()

auto kcenon::pacs::workflow::auto_prefetch_service::run_prefetch_cycle ( ) -> prefetch_result

nodiscard

Run a prefetch cycle manually.

Executes a complete prefetch cycle for all queued patients. Can be called whether the service is enabled or not.

Returns

Prefetch result with statistics

Definition at line 162 of file auto_prefetch_service.cpp.

                                                                  {
    return execute_cycle();
}

set_cycle_complete_callback()

void kcenon::pacs::workflow::auto_prefetch_service::set_cycle_complete_callback

(

prefetch_service_config::cycle_complete_callback

callback

)

Set the cycle complete callback.

Parameters

callback

Function called after each prefetch cycle

Definition at line 265 of file auto_prefetch_service.cpp.

                                                             {
    std::lock_guard<std::mutex> lock(mutex_);
    config_.on_cycle_complete = std::move(callback);
}

References config_, mutex_, and kcenon::pacs::workflow::prefetch_service_config::on_cycle_complete.

set_error_callback()

void kcenon::pacs::workflow::auto_prefetch_service::set_error_callback

(

prefetch_service_config::error_callback

callback

)

Set the error callback.

Parameters

callback

Function called when a prefetch fails

Definition at line 271 of file auto_prefetch_service.cpp.

                                                    {
    std::lock_guard<std::mutex> lock(mutex_);
    config_.on_prefetch_error = std::move(callback);
}

References config_, mutex_, and kcenon::pacs::workflow::prefetch_service_config::on_prefetch_error.

set_prefetch_criteria()

void kcenon::pacs::workflow::auto_prefetch_service::set_prefetch_criteria

(

const prefetch_criteria &

criteria

)

Update the prefetch criteria.

Parameters

criteria

New selection criteria for prior studies

Definition at line 253 of file auto_prefetch_service.cpp.

                                       {
    std::lock_guard<std::mutex> lock(mutex_);
    config_.criteria = criteria;
}

References config_, kcenon::pacs::workflow::prefetch_service_config::criteria, and mutex_.

set_prefetch_interval()

void kcenon::pacs::workflow::auto_prefetch_service::set_prefetch_interval

(

std::chrono::seconds

interval

)

Update the prefetch interval.

Parameters

interval

New interval between prefetch cycles

Note

Takes effect at the next cycle

Definition at line 241 of file auto_prefetch_service.cpp.

                               {
    std::lock_guard<std::mutex> lock(mutex_);
    config_.prefetch_interval = interval;
}

References config_, mutex_, and kcenon::pacs::workflow::prefetch_service_config::prefetch_interval.

start()

void kcenon::pacs::workflow::auto_prefetch_service::start

(

)

Start the prefetch service (alias for enable)

Definition at line 79 of file auto_prefetch_service.cpp.

                                  {
    if (enabled_.exchange(true)) {
        return;  // Already enabled
    }
 
    stop_requested_.store(false);
    next_cycle_time_ = std::chrono::steady_clock::now();
 
    worker_thread_ = std::thread([this]() {
        run_loop();
    });
 
    integration::logger_adapter::info(
        "Auto prefetch service started interval_seconds={} max_concurrent={}",
        config_.prefetch_interval.count(),
        config_.max_concurrent_prefetches);
}

References config_, enabled_, kcenon::pacs::integration::logger_adapter::info(), kcenon::pacs::workflow::prefetch_service_config::max_concurrent_prefetches, next_cycle_time_, kcenon::pacs::workflow::prefetch_service_config::prefetch_interval, run_loop(), stop_requested_, and worker_thread_.

Referenced by enable().

Here is the call graph for this function:

Here is the caller graph for this function:

stop()

void kcenon::pacs::workflow::auto_prefetch_service::stop

(

bool

wait_for_completion = true

)

Stop the prefetch service (alias for disable)

Parameters

wait_for_completion

If true, waits for current operations

Definition at line 101 of file auto_prefetch_service.cpp.

                                                         {
    if (!enabled_.exchange(false)) {
        return;  // Already disabled
    }
 
    stop_requested_.store(true);
 
    // Wake up the worker thread
    cv_.notify_all();
 
    if (wait_for_completion && worker_thread_.joinable()) {
        worker_thread_.join();
    } else if (worker_thread_.joinable()) {
        worker_thread_.detach();
    }
 
    integration::logger_adapter::info("Auto prefetch service stopped");
}

References cv_, enabled_, kcenon::pacs::integration::logger_adapter::info(), stop_requested_, and worker_thread_.

Referenced by disable(), and ~auto_prefetch_service().

Here is the call graph for this function:

Here is the caller graph for this function:

study_exists_locally()

auto kcenon::pacs::workflow::auto_prefetch_service::study_exists_locally ( const std::string & study_uid ) -> bool

nodiscardprivate

Check if study already exists locally.

Parameters

study_uid

Study Instance UID

Returns

true if study is already in local storage

Definition at line 694 of file auto_prefetch_service.cpp.

            {
    // Query local database to check if study exists
    auto study_record = database_.find_study(study_uid);
    return study_record.has_value();
}

time_until_next_cycle()

auto kcenon::pacs::workflow::auto_prefetch_service::time_until_next_cycle ( ) const -> std::optional<std::chrono::seconds>

nodiscard

Get the time until the next scheduled prefetch cycle.

Returns

Duration until next cycle, or nullopt if not scheduled

Definition at line 212 of file auto_prefetch_service.cpp.

                                     {
    if (!enabled_.load()) {
        return std::nullopt;
    }
 
    std::lock_guard<std::mutex> lock(mutex_);
    auto now = std::chrono::steady_clock::now();
    if (next_cycle_time_ <= now) {
        return std::chrono::seconds{0};
    }
 
    return std::chrono::duration_cast<std::chrono::seconds>(
        next_cycle_time_ - now);
}

References enabled_, mutex_, and next_cycle_time_.

trigger_cycle()

void kcenon::pacs::workflow::auto_prefetch_service::trigger_cycle

(

)

Trigger next cycle immediately.

Wakes up the background thread to run a prefetch cycle immediately, without waiting for the scheduled interval.

Definition at line 156 of file auto_prefetch_service.cpp.

                                          {
    std::lock_guard<std::mutex> lock(mutex_);
    next_cycle_time_ = std::chrono::steady_clock::now();
    cv_.notify_one();
}

References cv_, mutex_, and next_cycle_time_.

trigger_for_worklist()

void kcenon::pacs::workflow::auto_prefetch_service::trigger_for_worklist

(

const std::vector< storage::worklist_item > &

worklist_items

)

Trigger prefetch for worklist items.

Queues prefetch requests for all patients in the provided worklist items. Useful for batch prefetching.

Parameters

worklist_items

Vector of worklist items to prefetch for

Definition at line 151 of file auto_prefetch_service.cpp.

                                                           {
    on_worklist_query(worklist_items);
}

References on_worklist_query().

Here is the call graph for this function:

update_stats()

void kcenon::pacs::workflow::auto_prefetch_service::update_stats ( const prefetch_result & result )

private

Update cumulative statistics.

Parameters

result

Result from latest operation

Definition at line 791 of file auto_prefetch_service.cpp.

                                                                      {
    cumulative_stats_ += result;
}

References cumulative_stats_.

Referenced by run_loop().

Here is the caller graph for this function:

Member Data Documentation

config_

prefetch_service_config kcenon::pacs::workflow::auto_prefetch_service::config_

private

Service configuration.

Definition at line 573 of file auto_prefetch_service.h.

Referenced by auto_prefetch_service(), auto_prefetch_service(), auto_prefetch_service(), get_prefetch_criteria(), get_prefetch_interval(), run_loop(), set_cycle_complete_callback(), set_error_callback(), set_prefetch_criteria(), set_prefetch_interval(), and start().

cumulative_stats_

prefetch_result kcenon::pacs::workflow::auto_prefetch_service::cumulative_stats_

private

Cumulative statistics.

Definition at line 606 of file auto_prefetch_service.h.

Referenced by get_cumulative_stats(), and update_stats().

cv_

std::condition_variable kcenon::pacs::workflow::auto_prefetch_service::cv_

private

Condition variable for sleep/wake.

Definition at line 582 of file auto_prefetch_service.h.

Referenced by on_worklist_query(), run_loop(), stop(), and trigger_cycle().

cycle_in_progress_

std::atomic<bool> kcenon::pacs::workflow::auto_prefetch_service::cycle_in_progress_ {false}

private

Flag indicating a cycle is in progress.

Definition at line 591 of file auto_prefetch_service.h.

{false};

Referenced by run_loop().

cycles_count_

std::atomic<std::size_t> kcenon::pacs::workflow::auto_prefetch_service::cycles_count_ {0}

private

Number of completed cycles.

Definition at line 609 of file auto_prefetch_service.h.

{0};

Referenced by cycles_completed(), and run_loop().

database_

storage::index_database& kcenon::pacs::workflow::auto_prefetch_service::database_

private

Reference to PACS index database.

Definition at line 564 of file auto_prefetch_service.h.

enabled_

std::atomic<bool> kcenon::pacs::workflow::auto_prefetch_service::enabled_ {false}

private

Flag indicating service is enabled.

Definition at line 588 of file auto_prefetch_service.h.

{false};

Referenced by is_enabled(), start(), stop(), and time_until_next_cycle().

executor_

std::shared_ptr<kcenon::common::interfaces::IExecutor> kcenon::pacs::workflow::auto_prefetch_service::executor_

private

IExecutor for task execution (recommended, Issue #487)

Definition at line 570 of file auto_prefetch_service.h.

last_result_

std::optional<prefetch_result> kcenon::pacs::workflow::auto_prefetch_service::last_result_

private

Last prefetch result.

Definition at line 603 of file auto_prefetch_service.h.

Referenced by get_last_result(), and run_loop().

mutex_

std::mutex kcenon::pacs::workflow::auto_prefetch_service::mutex_

mutableprivate

Mutex for thread synchronization.

Definition at line 579 of file auto_prefetch_service.h.

Referenced by get_cumulative_stats(), get_last_result(), get_prefetch_criteria(), get_prefetch_interval(), run_loop(), set_cycle_complete_callback(), set_error_callback(), set_prefetch_criteria(), set_prefetch_interval(), time_until_next_cycle(), and trigger_cycle().

next_cycle_time_

std::chrono::steady_clock::time_point kcenon::pacs::workflow::auto_prefetch_service::next_cycle_time_

private

Time of next scheduled cycle.

Definition at line 612 of file auto_prefetch_service.h.

Referenced by run_loop(), start(), time_until_next_cycle(), and trigger_cycle().

queue_mutex_

std::mutex kcenon::pacs::workflow::auto_prefetch_service::queue_mutex_

mutableprivate

Mutex for request queue.

Definition at line 600 of file auto_prefetch_service.h.

Referenced by pending_requests(), and queue_request().

queued_patients_

std::set<std::string> kcenon::pacs::workflow::auto_prefetch_service::queued_patients_

private

Set of patient IDs currently in queue (for deduplication)

Definition at line 597 of file auto_prefetch_service.h.

Referenced by queue_request().

request_queue_

std::queue<prefetch_request> kcenon::pacs::workflow::auto_prefetch_service::request_queue_

private

Queue of pending prefetch requests.

Definition at line 594 of file auto_prefetch_service.h.

Referenced by pending_requests(), queue_request(), and run_loop().

stop_requested_

std::atomic<bool> kcenon::pacs::workflow::auto_prefetch_service::stop_requested_ {false}

private

Flag to signal shutdown.

Definition at line 585 of file auto_prefetch_service.h.

{false};

Referenced by run_loop(), start(), and stop().

thread_pool_

std::shared_ptr<kcenon::thread::thread_pool> kcenon::pacs::workflow::auto_prefetch_service::thread_pool_

private

Thread pool for parallel prefetches (optional, legacy)

Definition at line 567 of file auto_prefetch_service.h.

worker_thread_

std::thread kcenon::pacs::workflow::auto_prefetch_service::worker_thread_

private

Background worker thread.

Definition at line 576 of file auto_prefetch_service.h.

Referenced by start(), and stop().

---

The documentation for this class was generated from the following files:

• include/kcenon/pacs/workflow/auto_prefetch_service.h
• src/workflow/auto_prefetch_service.cpp