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][source]

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[source]

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]]][source]

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][source]

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[source]

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[source]

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[source]

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][source]

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[]]][source]

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[source]

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[source]

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[source]

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[source]

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[source]
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[source]

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[source]
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[source]
routing.route_series.remove_series(task_id: str, file_list: List[str]) bool[source]

Deletes the given files from the incoming folder.

routing.route_series.route_error_files() None[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

Pushes the study folder to the error folder after unsuccessful processing

routing.route_studies.push_studylevel_notification(study: str, task: Task) bool[source]

Executes the study-level reception notification

routing.route_studies.push_studylevel_processing(study: str, task: Task) bool[source]

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[source]

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[source]

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[source]