Package routing

routing.generate_taskfile

generate_taskfile.py

Helper functions for generating task files in json format, which describe the job to be done and maintain a journal of the executed actions.

routing.generate_taskfile.add_dispatching(task_id: str, uid: str, applied_rule: str, tags_list: Dict[str, str], target: str) → Optional[TaskDispatch]

Adds information about the desired dispatching step into the task file, which is evaluated by the dispatcher. For series-level dispatching, the target information is provided in string “target”, as dispatch operations from multiple rules to the same target are combined (to avoid double sending). In all other cases, the applied_rule is provided and the target information is taken from the rule definition.

routing.generate_taskfile.add_info(uid: str, uid_type: typing_extensions.Literal[series, study], triggered_rules: Dict[str, typing_extensions.Literal[True]], applied_rule: str, tags_list: Dict[str, str]) → TaskInfo

Adds general information into the task file

routing.generate_taskfile.add_processing(uid: str, applied_rule: str, tags_list: Dict[str, str]) → Optional[Union[TaskProcessing, List[TaskProcessing]]]

Adds information about the desired processing step into the task file, which is evaluated by the processing module

routing.generate_taskfile.add_study(uid: str, uid_type: typing_extensions.Literal[series, study], applied_rule: str, tags_list: Dict[str, str]) → Optional[TaskStudy]

Adds study information into the task file. Returns nothing if the task is a series-level task

routing.generate_taskfile.compose_task(task_id: str, uid: str, uid_type: typing_extensions.Literal[series, study], triggered_rules: Dict[str, typing_extensions.Literal[True]], applied_rule: str, tags_list: Dict[str, str], target: str) → Task

Composes the JSON content that is written into a task file when submitting a job (for processing, dispatching, or both)

routing.generate_taskfile.create_series_task(task_id: str, folder_name: str, triggered_rules: Dict[str, typing_extensions.Literal[True]], applied_rule: str, series_UID: str, tags_list: Dict[str, str], target: str) → bool

Writes a task file for the received series, containing all information needed by the processor and dispatcher. Additional information is written into the file as well

routing.generate_taskfile.create_study_task(task_id: str, folder_name: str, triggered_rules: Dict[str, typing_extensions.Literal[True]], applied_rule: str, study_UID: str, tags_list: Dict[str, str]) → bool

Generate task file with information on the study

routing.generate_taskfile.update_study_task(task_id: str, folder_name: str, triggered_rules: Dict[str, typing_extensions.Literal[True]], applied_rule: str, study_UID: str, tags_list: Dict[str, str]) → Tuple[bool, str]

Update the study task file with information from the latest received series

routing.route_series

route_series.py

Provides functions for routing/processing of series. For study-level processing, series will be pushed into study folders.

routing.route_series.get_triggered_rules(task_id: str, tagList: Dict[str, str]) → Literal[]]]

Evaluates the routing rules and returns a list with triggered rules.

routing.route_series.push_files(task_id: str, file_list: List[str], target_path: str, copy_files: bool) → bool

Copies or moves the given files to the target path. If copy_files is True, files are copied, otherwise moved. Note that this function does not create a lock file (this needs to be done by the calling function).

routing.route_series.push_series_complete(task_id: str, file_list: List[str], series_UID: str, destination: str, discard_rule: str, copy_files: bool, *, tagsList_encoding_error=False) → None

Moves all files of the series into either the “discard” or “success” folders, which both are periodically cleared.

routing.route_series.push_series_serieslevel(task_id: str, triggered_rules: Dict[str, typing_extensions.Literal[True]], file_list: List[str], series_UID: str, tags_list: Dict[str, str]) → None

Prepeares all series-level routings for the current series.

routing.route_series.push_series_studylevel(task_id: str, triggered_rules: Dict[str, typing_extensions.Literal[True]], file_list: List[str], series_UID: str, tags_list: Dict[str, str]) → None

Prepeares study-level routing for the current series.

routing.route_series.push_serieslevel_notification(task_id: str, triggered_rules: Dict[str, typing_extensions.Literal[True]], file_list: List[str], series_UID: str, tags_list: Dict[str, str]) → bool

routing.route_series.push_serieslevel_outgoing(task_id: str, triggered_rules: Dict[str, typing_extensions.Literal[True]], file_list: List[str], series_UID: str, tags_list: Dict[str, str], selected_targets: Dict[str, List[str]]) → None

Move the DICOM files of the series to a separate subfolder for each target in the outgoing folder.

routing.route_series.push_serieslevel_processing(task_id: str, triggered_rules: Dict[str, typing_extensions.Literal[True]], file_list: List[str], series_UID: str, tags_list: Dict[str, str]) → bool

routing.route_series.push_serieslevel_routing(task_id: str, triggered_rules: Dict[str, typing_extensions.Literal[True]], file_list: List[str], series_UID: str, tags_list: Dict[str, str]) → None

routing.route_series.remove_series(task_id: str, file_list: List[str]) → bool

Deletes the given files from the incoming folder.

routing.route_series.route_error_files() → None

Looks for error files, moves these files and the corresponding DICOM files to the error folder, and sends an alert to the bookkeeper instance.

routing.route_series.trigger_serieslevel_notification(current_rule: str, tags_list: Dict[str, str], event: mercure_events, task_id: str) → None

routing.route_studies

route_studies.py

Provides functions for routing and processing of studies (consisting of multiple series).

routing.route_studies.check_study_series(task: TaskHasStudy, required_series: str) → bool

Checks if all series required for study completion have been received

routing.route_studies.check_study_timeout(task: TaskHasStudy, pending_series: Dict[str, float]) → bool

Checks if the duration since the last series of the study was received exceeds the study completion timeout

routing.route_studies.is_study_complete(folder: str, pending_series: Dict[str, float]) → bool

Returns true if the study in the given folder is ready for processing, i.e. if the completeness criteria of the triggered rule has been met

routing.route_studies.is_study_locked(folder: str) → bool

Returns true if the given folder is locked, i.e. if another process is already working on the study

routing.route_studies.move_study_folder(task_id: Optional[str], study: str, destination: str) → bool

Moves the study subfolder to the specified destination with proper locking of the folders

routing.route_studies.push_studylevel_dispatch(study: str, task: Task) → bool

Pushes the study folder to the dispatchter, including the generated task file containing the destination information

routing.route_studies.push_studylevel_error(study: str) → None

Pushes the study folder to the error folder after unsuccessful processing

routing.route_studies.push_studylevel_notification(study: str, task: Task) → bool

Executes the study-level reception notification

routing.route_studies.push_studylevel_processing(study: str, task: Task) → bool

Pushes the study folder to the processor, including the generated task file containing the processing instructions

routing.route_studies.remove_study_folder(task_id: Optional[str], study: str, lock: FileLock) → bool

Removes a study folder containing nothing but the lock file (called during cleanup after all files have been moved somewhere else already)

routing.route_studies.route_studies(pending_series: Dict[str, float]) → None

Searches for completed studies and initiates the routing of the completed studies

routing.route_studies.trigger_studylevel_notification(study: str, task: Task, event: mercure_events) → bool