Shim Notify Gear

The shim_notify.py script runs as the main entrypoint of the Flywheel gear. It performs Z-shim quality checks on Siemens MRI data and notifies relevant users via email.

Each email recipient entry should include:

[[recipients]]
from = "scanner-alerts@example.com"
to = "tech@example.com"
host = "smtp.example.com"

Workflow summary

  1. Extract the first DICOM file from the ZIP archive.

  2. Parse Siemens CSA headers to extract the Z-shim offset.

  3. Determine scanner type from StationName.

  4. Compare Z value to scanner-specific threshold.

  5. Notify recipients with pass/fail result.

  6. Optionally update Flywheel session metadata (if run within a gear context).

Source Code Reference

Shim QC Script

This script checks the Z-shim offset from a Siemens CSA DICOM header, maps the scanner serial number to a PRISMA label, and sends a notification email indicating whether the shim value meets the threshold.

It loads station map and threshold values from a TOML file.

class shim_notify.Curator(**kwargs)[source]

Bases: FileCurator

curate_file(file_: Dict[str, Any])[source]

Curates a file.

Parameters:

file (Dict or flywheel.FileEntry) -- The file entry object to curate

shim_notify.first_dicom_from_zip(zfname: str) → Dataset[source]
shim_notify.load_toml_config(path: str) → dict[source]

Load station map and z thresholds from a TOML file.

Parameters:

path (str) -- Path to the TOML config file.

Returns:

Dictionary with 'station_map' and 'z_thresholds'.

Return type:

dict

shim_notify.main(zip_path: str, email_configs: list[dict], config_path: str)[source]

Run shim QC check and send email notifications.

Parameters:

  • zip_path (str) -- Path to ZIP file containing DICOMs.

  • email_configs (list of dict) -- Email configuration dictionary list.

  • config_path (str) -- Path to shim_config.toml with station map and thresholds.

shim_notify.notify_message(scanner: str, z: float, z_thresholds: dict) → str[source]
shim_notify.read_emails(toml_path: str) → list[dict][source]
shim_notify.read_z(dcm: Dataset) → float[source]
shim_notify.send_email(subject: str, body: str, sender: str, recipient: str, host: str = 'localhost')[source]
shim_notify.station_to_name(station_id: str, station_map: dict) → str[source]
shim_notify.update_db(fw, dest_id, z: float)[source]