Service webgui

webgui.py

The web-based graphical user interface of mercure.

class webgui.CSPMiddleware(app: Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]], dispatch: Optional[Callable[[Request, Callable[[Request], Awaitable[Response]]], Awaitable[Response]]] = None)

Bases: BaseHTTPMiddleware

async dispatch(request, call_next)

webgui.DEBUG_MODE: bool

class webgui.ExtendedUser(username: str, is_admin: bool = False)

Bases: SimpleUser

property is_admin: bool

webgui.SECRET_KEY: Secret

class webgui.SessionAuthBackend

Bases: AuthenticationBackend

async authenticate(request)

webgui.WEBGUI_HOST: str

webgui.WEBGUI_PORT: int

async webgui.configuration(request) → Response

Shows the current configuration of the mercure appliance.

async webgui.configuration_edit(request) → Response

Shows a configuration editor

async webgui.configuration_edit_post(request) → Response

Updates the configuration after post from editor

async webgui.control_services(request) → Response

webgui.create_app() → Starlette

async webgui.delete_old_tests() → Response

async webgui.emergency_response(request) → Response

Shows emergency message about invalid configuration.

async webgui.error(request)

An example error. Switch the debug setting to see either tracebacks or 500 pages.

webgui.get_nomad_logs(service, log_size: int) → bytes

Reads the service log when running a nomad-type installation.

async webgui.get_service_status(runtime) → List[Dict[str, Any]]

async webgui.homepage(request) → Response

Renders the index page that shows information about the system status.

webgui.launch_emergency_app() → None

Launches a minimal application to inform the user about the incorrect configuration

webgui.lifespan(app)

async webgui.login(request) → Response

Shows the login page.

async webgui.login_post(request) → Response

Evaluate the submitted login information. Redirects to index page if login information valid, otherwise back to login. On the first login, the user will be directed to the settings page and asked to change the password.

async webgui.logout(request)

Logouts the users by clearing the session cookie.

webgui.main(args=['-b', 'html', '-d', '_build/doctrees', '.', '_build/html']) → None

async webgui.not_found(request, exc) → Response

Return an HTTP 404 page.

webgui.read_webgui_config() → Config

async webgui.self_test(request) → Response

generate a test rule

async webgui.self_test_cleanup(test_id: str, delay: int = 60) → None

Delete the rules and targets for this test after a delay

async webgui.self_test_notification(request) → Response

async webgui.server_error(request, exc) → Response

Return an HTTP 500 page.

async webgui.settings_edit(request) → Response

Shows the settings for the current user. Renders the same template as the normal user edit, but with parameter own_settings=True.

async webgui.show_first_log(request) → Response

Get the first service entry and forward to corresponding log entry point.

async webgui.show_log(request) → Response

Render the log for the given service. The time range can be specified via URL parameters.

async webgui.shutdown() → None

webgui.startup(app: Starlette)

api.py

API backend functions for AJAX querying from the web frontend.

async webinterface.api.find_tasks(request)

async webinterface.api.get_series(request)

async webinterface.api.get_series_events(request)

async webinterface.api.get_task_info(request)

async webinterface.api.get_tasks(request)

async webinterface.api.get_tests(request)

async webinterface.api.task_process_logs(request)

async webinterface.api.task_process_results(request)

async webinterface.api.test(request)

common.py

Helper functions for the graphical user interface of mercure.

async webinterface.common.async_run(cmd, **params) → Tuple[Optional[int], bytes, bytes]

Executes the given command in a way compatible with ayncio.

async webinterface.common.async_run_exec(*args, **params) → Tuple[Optional[int], bytes, bytes]

Executes the given command in a way compatible with ayncio.

webinterface.common.get_csp_nonce(request=None) → dict

Returns the CSP nonce for the current request.

webinterface.common.get_mercure_version(request) → dict

webinterface.common.get_user_information(request) → dict

Returns dictionary of values that should always be passed to the templates when the user is logged in.

webinterface.common.strip_untrusted(input: Union[str, list]) → Any

exception webinterface.dicom_client.DicomClientBadStatus

Bases: Exception

exception webinterface.dicom_client.DicomClientCouldNotAssociate

Bases: Exception

exception webinterface.dicom_client.DicomClientCouldNotFind

Bases: Exception

class webinterface.dicom_client.SimpleDicomClient(host, port, called_aet, calling_aet, out_dir)

Bases: object

called_aet: str

findscu(accession_number, search_filters={}) → List[Dataset]

getscu(accession_number: str, search_filters: Dict[str, List[str]]) → Iterator[Dataset]

handle_store(event)

host: str

output_dir: str

port: int

modules.py

Modules page for the graphical user interface of mercure.

class webinterface.modules.BadRequestResponse(*args)

Bases: PlainTextResponse

class webinterface.modules.ServerErrorResponse(*args)

Bases: PlainTextResponse

async webinterface.modules.add_module(request)

Creates a new module and forwards the user to the module edit page.

async webinterface.modules.delete_module(request)

Deletes the module with the given module name.

async webinterface.modules.edit_module(request)

Show the module edit page for the given module name.

async webinterface.modules.edit_module_POST(request)

Save the settings for the given module name.

async webinterface.modules.save_module(form, name) → None

Save the settings for the module with the given name.

async webinterface.modules.show_modules(request)

Shows all installed modules

queue.py

Queue page for the graphical user interface of mercure.

class webinterface.queue.RestartTaskErrors(value)

Bases: str, Enum

An enumeration.

CURRENTLY_PROCESSING = 'currently_processing'

FAILED_TO_ADD_PROCESSING = 'failed_to_add_processing'

NO_AS_RECEIVED = 'no_as_received'

NO_DISPATCH_STATUS = 'no_dispatch_status'

NO_RULE_APPLIED = 'no_rule_applied'

NO_TASK_FILE = 'no_task_file'

TASK_NOT_READY = 'not_ready'

WRONG_JOB_TYPE = 'wrong_type'

async webinterface.queue.force_study_complete(request)

webinterface.queue.get_fail_stage(taskfile_folder: Path) → str

async webinterface.queue.get_jobinfo(request)

webinterface.queue.is_dispatch_failure(taskfile_folder: Path) → bool

Determines if a task in the error folder is a dispatch failure.

webinterface.queue.restart_dispatch(taskfile_folder: Path, outgoing_folder: Path) → dict

async webinterface.queue.restart_job(request)

Restarts a failed job. This endpoint handles both dispatch and processing failures.

webinterface.queue.restart_processing_task(task_id: str, source_folder: Path, is_error: bool = False) → Dict

Restarts a processing task by moving it from the source folder (error or success) to the processing folder.

Args:

task_id: The ID of the task to restart source_folder: Path to the task folder in the source directory (error or success) is_error: Whether the source folder is the error folder (True) or success folder (False)

Returns:

Dict with success or error information

async webinterface.queue.set_queues_status(request)

async webinterface.queue.show_jobs_fail(request)

async webinterface.queue.show_jobs_processing(request)

async webinterface.queue.show_jobs_routing(request)

async webinterface.queue.show_jobs_studies(request)

async webinterface.queue.show_queues(request)

Shows all installed modules

async webinterface.queue.show_queues_status(request)

rules.py

Rules page for the graphical user interface of mercure.

async webinterface.rules.add_rule(request) → Response

Creates a new routing rule and forwards the user to the rule edit page.

async webinterface.rules.duplicate_rule(request) → Response

Duplicates an existing routing rule.

async webinterface.rules.rules(request) → Response

Show all defined routing rules. Can be executed by all logged-in users.

async webinterface.rules.rules_delete_post(request) → Response

Deletes the given routing rule

async webinterface.rules.rules_edit(request) → Response

Shows the edit page for the given routing rule.

async webinterface.rules.rules_edit_post(request) → Response

Updates the settings for the given routing rule.

async webinterface.rules.rules_test(request) → Response

Evalutes if a given routing rule is valid. The rule and testing dictionary have to be passed as form parameters.

async webinterface.rules.rules_test_completionseries(request) → Response

Evalutes if a given value for the series list for study completion is valid.

services.py

Helper functions for controlling the services from the graphical user interface of mercure.

webinterface.services.read_services() → None

Reads the list of configured services from the configuration file. This list normally does not have to be changed, but can be modified if multiple instances of individual services should be used to increase performance.

targets.py

Targets page for the graphical user interface of mercure.

async webinterface.targets.add_target(request) → Response

Creates a new target.

async webinterface.targets.show_targets(request) → Response

Shows all configured targets.

async webinterface.targets.targets_delete_post(request) → Response

Deletes the given target.

async webinterface.targets.targets_edit(request) → Response

Shows the edit page for the given target.

async webinterface.targets.targets_edit_post(request) → Union[RedirectResponse, PlainTextResponse]

Updates the given target using the form values posted with the request.

async webinterface.targets.targets_test_post(request) → Response

Tests the connectivity of the given target by executing ping and c-echo requests.

users.py

Users page and user support functions for the graphical user interface of mercure.

class webinterface.users.User(*args, **kwargs)

Bases: dict

change_password: Literal['True', 'False']

email: str

is_admin: Literal['True', 'False']

password: str

permissions: Any

async webinterface.users.add_new_user(request) → Response

Creates a new user and redirects to the user-edit page.

webinterface.users.create_users() → Dict[str, User]

Create new users file and create seed admin account with name “admin” and password “router”.

webinterface.users.evaluate_password(username, password) → bool

Check if the given password for the given user is correct. Hashed passwords are stored with salt.

webinterface.users.hash_password(password) → str

Hash the password using the passlib library.

webinterface.users.is_admin(username) → bool

Check in the user list if the given user has admin rights.

webinterface.users.needs_change_password(username) → bool

Check if the given user has to change his password after login.

webinterface.users.read_users() → Dict[str, User]

Reads the user list from the configuration file. The file will only be read if it has been updated since the last function call. If the file does not exist, create a new user file.

webinterface.users.save_users() → None

Write the users list into a file on the disk.

async webinterface.users.show_users(request) → Response

Shows all available users.

async webinterface.users.users_delete_post(request) → Response

Deletes the given users.

async webinterface.users.users_edit(request) → Response

Shows the settings for a given user.

async webinterface.users.users_edit_post(request) → Response

Updates the given user with settings passed as form parameters.

Dashboards

common.py

class webinterface.dashboards.common.ClassBasedRQTask(parent: Optional[str] = None, type: str = 'unknown', _job: Optional[rq.job.Job] = None, _queue: str = '')

Bases: object

create_job(connection, rq_options={}, **kwargs) → Job

execute(*args, **kwargs) → Any

static move_to_destination(path: str, destination: Optional[str], job_id: str, node: Union[DicomTarget, DicomWebTarget], force_rule: Optional[str] = None) → None

parent: Optional[str] = None

classmethod queue(connection=None) → Queue

type: str = 'unknown'

class webinterface.dashboards.common.JSONErrorResponse(message: str, status_code: int = 500)

Bases: JSONResponse

async webinterface.dashboards.common.index(request)

webinterface.dashboards.common.invoke_getdcmtags(file: Path, node: Optional[Union[DicomTarget, DicomWebTarget]], force_rule: Optional[str] = None)

dicomweb.py

DICOMweb interface for handling DICOM uploads via STOW-RS protocol.

class webinterface.dashboards.dicomweb.MultipartData(dicoms, zips, form_data)

Bases: object

dicoms: List[bytes]

form_data: List[bytes]

zips: List[bytes]

async webinterface.dashboards.dicomweb.dataset_operation(request: Request)

webinterface.dashboards.dicomweb.extract_zip(zip_file: ZipFile, extract_to: str, force_rule: Optional[str]) → int

async webinterface.dashboards.dicomweb.index(request) → Response

async webinterface.dashboards.dicomweb.parse_multipart_data(request: Request) → MultipartData

async webinterface.dashboards.dicomweb.stow_rs(request: Request) → Response

Handle STOW-RS requests for uploading DICOM instances

async webinterface.dashboards.dicomweb.upload(request)

query_routes.py

async webinterface.dashboards.query_routes.check_accessions(request)

async webinterface.dashboards.query_routes.get_job_info(request)

async webinterface.dashboards.query_routes.post_pause_job(request)

async webinterface.dashboards.query_routes.post_resume_job(request)

async webinterface.dashboards.query_routes.post_retry_job(request)

async webinterface.dashboards.query_routes.query(request)

async webinterface.dashboards.query_routes.query_jobs(request)

Returns a list of all query jobs.

async webinterface.dashboards.query_routes.query_post_batch(request)

Starts a new query job for the given accession number and DICOM node.

simple.py

async webinterface.dashboards.simple.tasks(request)

async webinterface.dashboards.simple.tests(request)