Package common

config.py

mercure’s configuration management, used by various mercure modules

common.config.check_folders() → bool

Checks if all required folders for handling the DICOM files exist.

common.config.mercure: Config

common.config.read_config() → Config

Reads the configuration settings (rules, targets, general settings) from the configuration file. The configuration will only be updated if the file has changed compared the the last function call. If the configuration file is locked by another process, an exception will be raised.

common.config.read_tagslist() → None

Reads the list of supported DICOM tags with example values, displayed the UI.

common.config.save_config() → None

Saves the current configuration in a file on the disk. Raises an exception if the file has been locked by another process.

common.config.write_configfile(json_content) → None

Rewrites the config file using the JSON data passed as argument. Used by the config editor of the webgui.

constants.py

mercure-wide constants, used for standardizing key names and extensions.

class common.constants.mercure_actions

Bases: object

BOTH = 'both'

DISCARD = 'discard'

NOTIFICATION = 'notification'

PROCESS = 'process'

ROUTE = 'route'

class common.constants.mercure_config

Bases: object

MODULES = 'modules'

RULES = 'rules'

TARGETS = 'targets'

class common.constants.mercure_defs

Bases: object

SEPARATOR = '#'

VERSION = '0.4.0-beta.4'

class common.constants.mercure_events(value)

Bases: IntEnum

An enumeration.

COMPLETED = 1

ERROR = 2

RECEIVED = 0

class common.constants.mercure_folders

Bases: object

DISCARD = 'discard_folder'

ERROR = 'error_folder'

INCOMING = 'incoming_folder'

OUTGOING = 'outgoing_folder'

PROCESSING = 'processing_folder'

STUDIES = 'studies_folder'

SUCCESS = 'success_folder'

class common.constants.mercure_info

Bases: object

ACC = 'acc'

ACTION = 'action'

MERCURE_APPLIANCE = 'mercure_appliance'

MERCURE_SERVER = 'mercure_server'

MERCURE_VERSION = 'mercure_version'

MRN = 'mrn'

TRIGGERED_RULES = 'triggered_rules'

UID = 'uid'

UID_TYPE = 'uid_type'

class common.constants.mercure_names

Bases: object

DCM = '.dcm'

DCMFILTER = '*.dcm'

ERROR = '.error'

FORCE_COMPLETE = '.force-complete'

HALT = 'HALT'

LOCK = '.lock'

PROCESSING = '.processing'

RUNNING = '.running'

SENDLOG = 'sent.txt'

TAGS = '.tags'

TASKFILE = 'task.json'

class common.constants.mercure_options

Bases: object

FALSE = 'False'

INVALID = '#@INVALID@#'

MISSING = 'MISSING'

NORMAL = 'normal'

OFFPEAK = 'offpeak'

SERIES = 'series'

STUDY = 'study'

TRUE = 'True'

URGENT = 'urgent'

class common.constants.mercure_rule

Bases: object

ACTION = 'action'

ACTION_TRIGGER = 'action_trigger'

DISABLED = 'disabled'

FALLBACK = 'fallback'

NOTIFICATION_PAYLOAD = 'notification_payload'

NOTIFICATION_TRIGGER_COMPLETION = 'notification_trigger_completion'

NOTIFICATION_TRIGGER_ERROR = 'notification_trigger_error'

NOTIFICATION_TRIGGER_RECEPTION = 'notification_trigger_reception'

NOTIFICATION_WEBHOOK = 'notification_webhook'

PRIORITY = 'priority'

PROCESSING_MODULE = 'processing_module'

RULE = 'rule'

STUDY_FORCE_COMPLETION_ACTION = 'study_force_completion_action'

STUDY_TRIGGER = 'study_trigger'

STUDY_TRIGGER_CONDITION = 'study_trigger_condition'

STUDY_TRIGGER_CONDITION_RECEIVED_SERIES = 'received_series'

STUDY_TRIGGER_CONDITION_TIMEOUT = 'timeout'

TARGET = 'target'

class common.constants.mercure_sections

Bases: object

DISPATCH = 'dispatch'

FILES = 'files'

INFO = 'info'

JOURNAL = 'journal'

NOTIFICATION = 'notification'

PROCESS = 'process'

STUDY = 'study'

class common.constants.mercure_study

Bases: object

COMPLETE_FORCE = 'complete_force'

COMPLETE_FORCE_ACTION = 'complete_force_action'

COMPLETE_REQUIRED_SERIES = 'complete_required_series'

COMPLETE_TRIGGER = 'complete_trigger'

CREATION_TIME = 'creation_time'

LAST_RECEIVE_TIME = 'last_receive_time'

RECEIVED_SERIES = 'received_series'

STUDY_UID = 'study_uid'

class common.event_types.FailStage(value)

Bases: StringEnum

Enum for the stages a task can fail at.

DISPATCHING = 'DISPATCHING'

PROCESSING = 'PROCESSING'

ROUTING = 'ROUTING'

class common.event_types.StringEnum(value)

Bases: Enum

An enum class that can be converted to a string based on the name, so str(enum.FOO) == “FOO”

class common.event_types.m_events(value)

Bases: StringEnum

Event types for general mercure monitoring.

BOOT = 'BOOT'

CONFIG_UPDATE = 'CONFIG_UPDATE'

PROCESSING = 'PROCESSING'

SHUTDOWN = 'SHUTDOWN'

SHUTDOWN_REQUEST = 'SHUTDOWN_REQUEST'

UNKNOWN = 'UNKNOWN'

class common.event_types.severity(value)

Bases: Enum

Severity level associated to the mercure events.

CRITICAL = 3

ERROR = 2

INFO = 0

WARNING = 1

class common.event_types.task_event(value)

Bases: StringEnum

Event types for monitoring everything related to one specific series.

ASSIGN = 'ASSIGN'

CLEAN = 'CLEAN'

COMPLETE = 'COMPLETE'

COPY = 'COPY'

DELEGATE = 'DELEGATE'

DISCARD = 'DISCARD'

DISPATCH_BEGIN = 'DISPATCH_BEGIN'

DISPATCH_COMPLETE = 'DISPATCH_COMPLETE'

ERROR = 'ERROR'

MOVE = 'MOVE'

NOTIFICATION = 'NOTIFICATION'

PROCESS_BEGIN = 'PROCESS_BEGIN'

PROCESS_COMPLETE = 'PROCESS_COMPLETE'

PROCESS_MODULE_BEGIN = 'PROCESS_MODULE_BEGIN'

PROCESS_MODULE_COMPLETE = 'PROCESS_MODULE_COMPLETE'

PROCESS_RESTART = 'PROCESS_RESTART'

REGISTER = 'REGISTER'

REMOVE = 'REMOVE'

SUSPEND = 'SUSPEND'

UNKNOWN = 'UNKNOWN'

class common.event_types.w_events(value)

Bases: StringEnum

Event types for monitoring the webgui activity.

CONFIG_EDIT = 'CONFIG_EDIT'

LOGIN = 'LOGIN'

LOGIN_FAIL = 'LOGIN_FAIL'

LOGOUT = 'LOGOUT'

RULE_CREATE = 'RULE_CREATE'

RULE_DELETE = 'RULE_DELETE'

RULE_EDIT = 'RULE_EDIT'

SERVICE_CONTROL = 'SERVICE_CONTROL'

TARGET_CREATE = 'TARGET_CREATE'

TARGET_DELETE = 'TARGET_DELETE'

TARGET_EDIT = 'TARGET_EDIT'

UNKNOWN = 'UNKNOWN'

USER_CREATE = 'USER_CREATE'

USER_DELETE = 'USER_DELETE'

USER_EDIT = 'USER_EDIT'

common.generate_test_series.generate_file(study: str, series: str, slice_number: int, acc: str, study_uid: str, desc: str, image: Any, orientation: List[List[float]] = [[1, 0, 0], [0, 1, 0]], patient_name: Optional[str] = 'Julia^Set', patient_id: Optional[str] = 'JULIATEST', series_number: int = 1) → Dataset

common.generate_test_series.generate_series(k: Union[str, Path], n: int, orientation: List[List[float]] = [[1, 0, 0], [0, 1, 0]], series_description: Optional[str] = 'Julia set', series_number=1) → List[Path]

common.generate_test_series.generate_several_protocols(base_path: Union[str, Path], protocols=['PROT1', 'PROT2', 'PROT1_DL', 'PROT2_DL']) → List[Path]

common.generate_test_series.generate_test_series(pt: complex = - 0.3 + 0j, n: int = 10, orientation: List[List[float]] = [[1, 0, 0], [0, 1, 0]], accession: Optional[str] = None, study_id: Optional[str] = None, patient_name: Optional[str] = None, patient_id: Optional[str] = None, series_description: Optional[str] = None, series_number: int = 1) → List[Dataset]

common.generate_test_series.nums(n: int, source: Optional[str] = None) → str

helper.py

Various internal helper functions for mercure.

class common.helper.AsyncTimer(interval: int, func)

Bases: object

run_until_complete(loop=None) → None

start() → None

stop() → None

Signal to stop after the current

class common.helper.FileLock(path_for_lockfile: Path)

Bases: object

Helper class that implements a file lock. The lock file will be removed also from the destructor so that no spurious lock files remain if exceptions are raised.

free() → None

lockCreated = False

class common.helper.RepeatedTimer(interval: float, function: Callable, exit_function: Callable, *args, **kwargs)

Bases: object

Helper class for running a continuous timer that is suspended while the worker function is running

start() → None

Starts the timer for triggering the calllback after the defined interval.

stop() → None

Stops the timer and executes the defined exit callback function.

common.helper.g_log(*args, **kwargs) → None

common.helper.get_now_str() → str

Returns the current time as string with mercure-wide formatting

common.helper.get_runner() → str

Returns the name of the mechanism that is used for running mercure in the current installation (systemd, docker, nomad).

common.helper.is_terminated() → bool

Checks if the process will terminate after the current task.

common.helper.localize_log_timestamps(logstring: str, config) → str

common.helper.send_to_graphite(*args, **kwargs) → None

Wrapper for asynchronous graphite call to avoid wait time of main loop.

common.helper.send_to_influxdb(*args, **kwargs) → None

Wrapper for asynchronous influxdb call to avoid wait time of main loop.

common.helper.trigger_terminate() → None

Trigger that the processing loop should terminate after finishing the currently active task.

common.helper.validate_folders(config) → Tuple[bool, str]

Send data to InfluxDB metrics server.

class common.influxdb.Sender(host: str, token: str, org: str, bucket: str, prefix: str, log_sends: bool = False, raise_send_errors: bool = False)

Bases: object

build_message(metric: str, value: Union[int, float], timestamp: Optional[float]) → Point

Build an InfluxDB message to send and return it.

send(metric: str, value: Union[int, float], timestamp: Optional[float] = None)

Send given metric and (int or float) value to InfluxDB host.

send_message(message: Point)

send_socket(message: Point)

common.influxdb.init(*args, **kwargs) → None

Initialize default Sender instance with given args.

common.influxdb.send(*args, **kwargs)

Send message using default Sender instance.

class common.log_helpers.BookkeeperHandler(level=30)

Bases: Handler

emit(record: LogRecord) → None

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.

class common.log_helpers.ExceptionsKeywordArgumentAdapter(logger: Logger, extra: dict)

Bases: KeywordArgumentAdapter

clearTask() → None

process(msg, kwargs) → Tuple[str, MutableMapping[str, Any]]

Process the logging message and keyword arguments passed in to a logging call to insert contextual information. You can either manipulate the message itself, the keyword args or both. Return the message and kwargs modified (or not) to suit your needs.

Normally, you’ll only need to override this one method in a LoggerAdapter subclass for your specific needs.

setTask(task_id: str) → None

common.log_helpers.clear_task_decorator(func)

common.log_helpers.clear_task_decorator_async(func)

common.log_helpers.get_logformat() → str

Returns the format that should be used for log messages. Includes the time for docker and nomad, but not for systemd as journalctl already outputs the time of the log events.

common.log_helpers.get_logger() → ExceptionsKeywordArgumentAdapter

common.log_helpers.get_loglevel() → int

Returns the logging level that should be used for printing messages.

monitor.py

Helper functions and definitions for monitoring mercure’s operations via the bookkeeper module.

exception common.monitor.MonitorHTTPError(status_code: int, message: str)

Bases: Exception

Exception raised when a HTTP error occurs.

async common.monitor.async_post(endpoint: str, **kwargs)

async common.monitor.async_send_task_event(event: task_event, task_id: str, file_count: int, target: str, info: str)

common.monitor.configure(module, instance, address) → None

Configures the connection to the bookkeeper module. If not called, events will not be transmitted to the bookkeeper.

async common.monitor.do_post(endpoint, kwargs, catch_errors=False) → None

async common.monitor.find_tasks(request) → Any

async common.monitor.get(endpoint: str, payload: Any = {}) → Any

async common.monitor.get_series(series_uid='') → Any

async common.monitor.get_task_events(task_id='') → Any

async common.monitor.get_task_info(task_id='') → Any

async common.monitor.get_tasks() → Any

async common.monitor.get_tests() → Any

common.monitor.post(endpoint: str, **kwargs) → None

common.monitor.send_event(event: m_events, severity: severity = severity.INFO, description: str = '') → None

Sends information about general mercure events to the bookkeeper (e.g., during module start).

common.monitor.send_process_logs(task_id: str, module_name: str, logs: str) → None

common.monitor.send_processor_output(task: Task, task_processing: TaskProcessing, index: int, output: dict) → None

common.monitor.send_register_series(tags: Dict[str, str]) → None

Registers a received series on the bookkeeper. This should be called when a series has been fully received and the DICOM tags have been parsed.

common.monitor.send_register_task(task_id: str, series_uid: str, parent_id: Optional[str] = None) → None

Registers a new task on the bookkeeper. This should be called whenever a new task has been created.

common.monitor.send_task_event(event: task_event, task_id: str, file_count: int, target: str, info: str) → None

Send an event related to a specific series to the bookkeeper.

common.monitor.send_update_task(task: Task) → None

Registers a new task on the bookkeeper. This should be called whenever a new task has been created.

common.monitor.send_update_task_tags(id, info_dict) → None

common.monitor.send_webgui_event(event: w_events, user: str, description='') → None

Sends information about an event on the webgui to the bookkeeper.

common.monitor.set_api_key() → None

common.monitor.task_event_payload(event: task_event, task_id: str, file_count: int, target, info)

async common.monitor.task_process_logs(task_id='') → Any

async common.monitor.task_process_results(task_id='') → Any

notification.py

Helper functions for triggering webhook calls.

common.notification.get_task_custom_notification(task: Task) → Optional[str]

common.notification.get_task_requested_notification(task: Task) → bool

common.notification.parse_payload(payload: str, event: mercure_events, rule_name: str, task_id: str, details: str = '', context: dict = {}, *, task: Optional[Task] = None, tags_list: Optional[Dict[str, str]] = None) → str

common.notification.post(url: str, payload: Any) → None

common.notification.send_email(address: str, payload: str, event: mercure_events, rule_name: str, rule_type: str) → None

common.notification.send_email_helper(to: str, subject: str, content: str, rule_type='plain') → None

common.notification.send_webhook(url: str, payload: str) → None

common.notification.setup() → bool

Load the SSL certificate if it is configured (after the configuration has been read).

common.notification.trigger_notification_for_rule(rule_name: str, task_id: str, event: mercure_events, *, task: Task, details: Optional[str] = '', send_always: bool = False)

common.notification.trigger_notification_for_rule(rule_name: str, task_id: str, event: mercure_events, *, tags_list: Dict[str, str], details: Optional[str] = '', send_always: bool = False)

rule_evaluation.py

Helper functions for evaluating routing rules and study-completion conditions.

common.rule_evaluation.eval_rule(rule: str, tags_dict: Dict[str, str]) → Any

Parses the given rule, replaces all tag variables with values from the given tags dictionary, and evaluates the rule. If the rule is invalid, an exception will be raised.

common.rule_evaluation.parse_completion_series(task_id: str, completion_str: str, received_series: list) → bool

Evaluates the configuration string defining which series are required using the list of received series as input. Returns true if all required series have arrived, otherwise false is returned.

common.rule_evaluation.parse_rule(rule: str, tags: Dict[str, str]) → Tuple[bool, Optional[str], Optional[str]]

common.rule_evaluation.replace_tags(rule: str, tags: Dict[str, str]) → Any

Replaces all tags with format @tagname@ in the given rule string with the corresponding values from the currently processed series (stored in the second argument).

common.rule_evaluation.test_completion_series(value: str) → str

Tests if the given string with the list of series required for study completion has valid format. If so, True is returned as string, otherwise the error description is returned.

exception common.tags_rule_interface.TagNotFoundException

Bases: Exception

class common.tags_rule_interface.Tags(input_dict: Dict[str, str])

Bases: object

tags_accessed() → Set[str]

tagslist.py

Helper functions for displaying a list of DICOM tags available for routing in the graphical user interface of mercure.

types.py

Definitions for using TypedDicts throughout mercure.

class common.types.Compat

Bases: object

get(item, els=None) → Any

class common.types.Config(*, appliance_name: str, appliance_color: str = '#FFF', port: int, accept_compressed_images: bool, incoming_folder: str, studies_folder: str, outgoing_folder: str, success_folder: str, error_folder: str, discard_folder: str, processing_folder: str, jobs_folder: str, router_scan_interval: int, dispatcher_scan_interval: int, cleaner_scan_interval: int, retention: int, emergency_clean_percentage: int, retry_delay: int, retry_max: int, series_complete_trigger: int, study_complete_trigger: int, study_forcecomplete_trigger: int, dicom_receiver: DicomReceiverConfig = DicomReceiverConfig(additional_tags={}), graphite_ip: str, graphite_port: int, influxdb_host: str, influxdb_token: str, influxdb_org: str, influxdb_bucket: str, bookkeeper: str, offpeak_start: str, offpeak_end: str, targets: Dict[str, Target], rules: Dict[str, Rule], modules: Dict[str, Module], process_runner: Literal['docker', 'nomad', ''] = '', processing_runtime: Optional[str] = None, bookkeeper_api_key: Optional[str] = None, features: Dict[str, bool], processing_logs: ProcessingLogsConfig = ProcessingLogsConfig(discard_logs=False, logs_file_store=None), email_notification_from: str = 'mercure@mercure.mercure', support_root_modules: Optional[bool] = False, webhook_certificate_location: Optional[str] = None, phi_notifications: Optional[bool] = False, server_time: str = 'UTC', local_time: str = 'UTC', dicom_retrieve: DicomRetrieveConfig = DicomRetrieveConfig(dicom_nodes=[], destination_folders=[]), store_sample_dicom_tags: bool = False)

Bases: BaseModel, Compat

accept_compressed_images: bool

appliance_color: str

appliance_name: str

bookkeeper: str

bookkeeper_api_key: Optional[str]

cleaner_scan_interval: int

dicom_receiver: DicomReceiverConfig

dicom_retrieve: DicomRetrieveConfig

discard_folder: str

dispatcher_scan_interval: int

email_notification_from: str

emergency_clean_percentage: int

error_folder: str

features: Dict[str, bool]

graphite_ip: str

graphite_port: int

incoming_folder: str

influxdb_bucket: str

influxdb_host: str

influxdb_org: str

influxdb_token: str

jobs_folder: str

local_time: str

modules: Dict[str, Module]

offpeak_end: str

offpeak_start: str

outgoing_folder: str

phi_notifications: Optional[bool]

port: int

process_runner: Literal['docker', 'nomad', '']

processing_folder: str

processing_logs: ProcessingLogsConfig

processing_runtime: Optional[str]

retention: int

retry_delay: int

retry_max: int

router_scan_interval: int

rules: Dict[str, Rule]

series_complete_trigger: int

server_time: str

store_sample_dicom_tags: bool

studies_folder: str

study_complete_trigger: int

study_forcecomplete_trigger: int

success_folder: str

support_root_modules: Optional[bool]

targets: Dict[str, Target]

webhook_certificate_location: Optional[str]

class common.types.DicomDestination(*, name: str, path: str)

Bases: BaseModel

name: str

path: str

class common.types.DicomNodeBase(*, name: str)

Bases: BaseModel

classmethod get_name() → str

name: str

classmethod validate(v)

Parse the target as any of the known target types.

class common.types.DicomReceiverConfig(*, additional_tags: Dict[str, str] = {})

Bases: BaseModel

additional_tags: Dict[str, str]

class common.types.DicomRetrieveConfig(*, dicom_nodes: List[DicomNodeBase] = [], destination_folders: List[DicomDestination] = [])

Bases: BaseModel

destination_folders: List[DicomDestination]

dicom_nodes: List[DicomNodeBase]

class common.types.DicomTLSTarget(*, target_type: Literal['dicomtls'] = 'dicomtls', contact: Optional[str] = '', comment: str = '', direction: Optional[Literal['pull', 'push', 'both']] = 'push', ip: str, port: str, aet_target: str, aet_source: Optional[str] = '', pass_sender_aet: Optional[bool] = False, pass_receiver_aet: Optional[bool] = False, tls_key: str, tls_cert: str, ca_cert: str)

Bases: Target

aet_source: Optional[str]

aet_target: str

ca_cert: str

ip: str

pass_receiver_aet: Optional[bool]

pass_sender_aet: Optional[bool]

port: str

property short_description: str

target_type: Literal['dicomtls']

tls_cert: str

tls_key: str

class common.types.DicomTarget(*, target_type: Literal['dicom'] = 'dicom', contact: Optional[str] = '', comment: str = '', direction: Optional[Literal['pull', 'push', 'both']] = 'push', ip: str, port: str, aet_target: str, aet_source: Optional[str] = '', pass_sender_aet: Optional[bool] = False, pass_receiver_aet: Optional[bool] = False)

Bases: Target

aet_source: Optional[str]

aet_target: str

ip: str

pass_receiver_aet: Optional[bool]

pass_sender_aet: Optional[bool]

port: str

property short_description: str

target_type: Literal['dicom']

class common.types.DicomWebTarget(*, target_type: Literal['dicomweb'] = 'dicomweb', contact: Optional[str] = '', comment: str = '', direction: Optional[Literal['pull', 'push', 'both']] = 'push', url: str, qido_url_prefix: Optional[str] = None, wado_url_prefix: Optional[str] = None, stow_url_prefix: Optional[str] = None, access_token: Optional[str] = None, http_user: Optional[str] = None, http_password: Optional[str] = None)

Bases: Target

access_token: Optional[str]

http_password: Optional[str]

http_user: Optional[str]

qido_url_prefix: Optional[str]

property short_description: str

stow_url_prefix: Optional[str]

target_type: Literal['dicomweb']

url: str

wado_url_prefix: Optional[str]

class common.types.DummyTarget(*, target_type: Literal['dummy'] = 'dummy', contact: Optional[str] = '', comment: str = '', direction: Optional[Literal['pull', 'push', 'both']] = 'push')

Bases: Target

target_type: Literal['dummy']

class common.types.EmptyDict

Bases: TypedDict

class common.types.FolderTarget(*, target_type: Literal['folder'] = 'folder', contact: Optional[str] = '', comment: str = '', direction: Optional[Literal['pull', 'push', 'both']] = 'push', folder: str, file_filter: Optional[str] = None)

Bases: Target

file_filter: Optional[str]

folder: str

property short_description: str

target_type: Literal['folder']

class common.types.Module(*, docker_tag: Optional[str] = '', additional_volumes: Optional[str] = '', environment: Optional[str] = '', docker_arguments: Optional[str] = '', settings: Dict[str, Any] = {}, contact: Optional[str] = '', comment: Optional[str] = '', constraints: Optional[str] = '', resources: Optional[str] = '', requires_root: Optional[bool] = False)

Bases: BaseModel, Compat

additional_volumes: Optional[str]

comment: Optional[str]

constraints: Optional[str]

contact: Optional[str]

docker_arguments: Optional[str]

docker_tag: Optional[str]

environment: Optional[str]

requires_root: Optional[bool]

resources: Optional[str]

settings: Dict[str, Any]

class common.types.ProcessingLogsConfig(*, discard_logs: bool = False, logs_file_store: Optional[str] = None)

Bases: BaseModel

discard_logs: bool

logs_file_store: Optional[str]

class common.types.RsyncTarget(*, target_type: Literal['rsync'] = 'rsync', contact: Optional[str] = '', comment: str = '', direction: Optional[Literal['pull', 'push', 'both']] = 'push', folder: str, user: str, host: str, password: Optional[str] = None, run_on_complete: bool = False)

Bases: Target

folder: str

host: str

password: Optional[str]

run_on_complete: bool

property short_description: str

target_type: Literal['rsync']

user: str

class common.types.Rule(*, rule: str = 'False', target: Union[str, List[str]] = '', disabled: bool = False, fallback: bool = False, contact: str = '', comment: str = '', tags: str = '', action: Literal['route', 'both', 'process', 'discard', 'notification'] = 'route', action_trigger: Literal['series', 'study'] = 'series', study_trigger_condition: Literal['timeout', 'received_series'] = 'timeout', study_force_completion_action: Literal['discard', 'proceed', 'ignore'] = 'discard', study_trigger_series: str = '', priority: Literal['normal', 'urgent', 'offpeak'] = 'normal', processing_module: Union[str, List[str]] = '', processing_settings: Union[List[Dict[str, Any]], Dict[str, Any]] = {}, processing_retain_images: bool = False, notification_email: str = '', notification_webhook: str = '', notification_payload: str = '', notification_payload_body: str = '', notification_email_body: str = '', notification_email_type: str = 'plain', notification_trigger_reception: bool = True, notification_trigger_completion: bool = True, notification_trigger_completion_on_request: bool = False, notification_trigger_error: bool = True)

Bases: BaseModel, Compat

action: Literal['route', 'both', 'process', 'discard', 'notification']

action_trigger: Literal['series', 'study']

comment: str

contact: str

disabled: bool

fallback: bool

notification_email: str

notification_email_body: str

notification_email_type: str

notification_payload: str

notification_payload_body: str

notification_trigger_completion: bool

notification_trigger_completion_on_request: bool

notification_trigger_error: bool

notification_trigger_reception: bool

notification_webhook: str

priority: Literal['normal', 'urgent', 'offpeak']

processing_module: Union[str, List[str]]

processing_retain_images: bool

processing_settings: Union[List[Dict[str, Any]], Dict[str, Any]]

rule: str

study_force_completion_action: Literal['discard', 'proceed', 'ignore']

study_trigger_condition: Literal['timeout', 'received_series']

study_trigger_series: str

tags: str

target: Union[str, List[str]]

class common.types.S3Target(*, target_type: Literal['s3'] = 's3', contact: Optional[str] = '', comment: str = '', direction: Optional[Literal['pull', 'push', 'both']] = 'push', region: str, bucket: str, prefix: str, access_key_id: str, secret_access_key: str)

Bases: Target

access_key_id: str

bucket: str

prefix: str

region: str

secret_access_key: str

property short_description: str

target_type: Literal['s3']

class common.types.SftpTarget(*, target_type: Literal['sftp'] = 'sftp', contact: Optional[str] = '', comment: str = '', direction: Optional[Literal['pull', 'push', 'both']] = 'push', folder: str, user: str, host: str, password: Optional[str] = None)

Bases: Target

folder: str

host: str

password: Optional[str]

property short_description: str

target_type: Literal['sftp']

user: str

class common.types.Target(*, target_type: Any = None, contact: Optional[str] = '', comment: str = '', direction: Optional[Literal['pull', 'push', 'both']] = 'push')

Bases: BaseModel, Compat

comment: str

contact: Optional[str]

direction: Optional[Literal['pull', 'push', 'both']]

classmethod get_name() → str

property short_description: str

target_type: Any

classmethod validate(v)

Parse the target as any of the known target types.

class common.types.Task(*, info: TaskInfo, id: str, dispatch: Union[TaskDispatch, EmptyDict] = {}, process: Union[TaskProcessing, EmptyDict, List[TaskProcessing]] = {}, study: Union[TaskStudy, EmptyDict] = {}, nomad_info: Optional[Any] = None)

Bases: BaseModel, Compat

class Config

Bases: object

extra = 'forbid'

dispatch: Union[TaskDispatch, EmptyDict]

classmethod from_file(file_or_path: Union[PathLike, TextIOWrapper]) → Task

id: str

info: TaskInfo

nomad_info: Optional[Any]

process: Union[TaskProcessing, EmptyDict, List[TaskProcessing]]

study: Union[TaskStudy, EmptyDict]

to_file(file_or_path: Union[PathLike, TextIOWrapper]) → None

class common.types.TaskDispatch(*, target_name: Union[str, List[str]], status: Union[Dict[str, TaskDispatchStatus], EmptyDict] = {}, retries: Optional[int] = 0, next_retry_at: Optional[float] = 0, series_uid: Optional[str] = None)

Bases: BaseModel, Compat

next_retry_at: Optional[float]

retries: Optional[int]

series_uid: Optional[str]

status: Union[Dict[str, TaskDispatchStatus], EmptyDict]

target_name: Union[str, List[str]]

class common.types.TaskDispatchStatus(*, state: Literal['waiting', 'complete', 'error'], time: str)

Bases: BaseModel, Compat

state: Literal['waiting', 'complete', 'error']

time: str

class common.types.TaskHasStudy(*, info: TaskInfo, id: str, dispatch: Union[TaskDispatch, EmptyDict] = {}, process: Union[TaskProcessing, EmptyDict, List[TaskProcessing]] = {}, study: TaskStudy, nomad_info: Optional[Any] = None)

Bases: Task

study: TaskStudy

class common.types.TaskInfo(*, action: Literal['route', 'both', 'process', 'discard', 'notification'], uid: str, uid_type: Literal['series', 'study'], triggered_rules: Union[Dict[str, Literal[True]], str], applied_rule: Optional[str] = None, patient_name: Optional[str] = None, mrn: str, acc: str, sender_address: str = 'MISSING', mercure_version: str, mercure_appliance: str, mercure_server: str, device_serial_number: Optional[str] = None, fail_stage: Optional[FailStage] = None, sender_aet: str = 'MISSING', receiver_aet: str = 'MISSING')

Bases: BaseModel, Compat

acc: str

action: Literal['route', 'both', 'process', 'discard', 'notification']

applied_rule: Optional[str]

device_serial_number: Optional[str]

fail_stage: Optional[FailStage]

mercure_appliance: str

mercure_server: str

mercure_version: str

mrn: str

patient_name: Optional[str]

receiver_aet: str

sender_address: str

sender_aet: str

triggered_rules: Union[Dict[str, Literal[True]], str]

uid: str

uid_type: Literal['series', 'study']

class common.types.TaskProcessing(*, module_name: str, module_config: Optional[Module] = None, settings: Dict[str, Any] = {}, retain_input_images: bool, output: Optional[Dict] = None)

Bases: BaseModel, Compat

module_config: Optional[Module]

module_name: str

output: Optional[Dict]

retain_input_images: bool

settings: Dict[str, Any]

class common.types.TaskStudy(*, study_uid: str, complete_trigger: Optional[Literal['timeout', 'received_series']] = None, complete_required_series: str, creation_time: str, last_receive_time: str, received_series: Optional[List[str]] = None, received_series_uid: Optional[List[str]] = None, complete_force: bool = False, complete_force_action: Optional[Literal['discard', 'proceed', 'ignore']] = 'discard')

Bases: BaseModel, Compat

complete_force: bool

complete_force_action: Optional[Literal['discard', 'proceed', 'ignore']]

complete_required_series: str

complete_trigger: Optional[Literal['timeout', 'received_series']]

creation_time: str

last_receive_time: str

received_series: Optional[List[str]]

received_series_uid: Optional[List[str]]

study_uid: str

class common.types.UnsetRule

Bases: TypedDict

rule: str

class common.types.XnatTarget(*, target_type: Literal['xnat'] = 'xnat', contact: Optional[str] = '', comment: str = '', direction: Optional[Literal['pull', 'push', 'both']] = 'push', project_id: str, host: str, user: str, password: str)

Bases: Target

host: str

password: str

project_id: str

property short_description: str

target_type: Literal['xnat']

user: str

version.py

Semantic version handling for mercure.

class common.version.SemanticVersion

Bases: object

Helper class for handling semantic versioning in mercure.

INVALID = '0.0.0-invalid.0'

STATES = ['invalid', 'dev', 'alpha', 'beta', 'rc', 'stable']

dev = 0

get_image_tag() → str

Returns the image tag that should be used for pulling docker images

get_version_signature() → list

Returns the parsed version number as list of numerical values that can be compared for version consistency check.

get_version_string() → str

Returns the semantic version string. If no valid version number has been found, it will return 0.0.0-invalid.0.

is_dev_version() → bool

Returns True if the current version is a version under development.

is_release() → bool

Returns True if the current version is a release version.

is_valid_version() → bool

Returns True if the version numbering is valid.

major = 0

minor = 0

parse_version_string() → bool

Checks if the version string read from the VERSION file is valid and parses it into numerical values stored in the object. Returns False is the version string is invalid.

patch = 0

read_version_file() → bool

Reads the version string from the file VERSION in mercure’s root folder and parses it. If the file is missing or invalid, the version string is set to ‘0.0.0-invalid.0’ and all numerical version numbers are set to 0.

state = 0

version_string = ''