[DTOSS 12685] Emulate modality workflow

Dockerfile

@@ -24,6 +24,7 @@ RUN uv sync --frozen --no-dev
 # Copy source code
 COPY src/ ./src/
 COPY scripts/ ./scripts/
+COPY sample_images/ ./sample_images/
 
 # Set environment variables with defaults
 ENV PACS_AET=SCREENING_PACS \

compose.yml

@@ -92,6 +92,29 @@ services:
       pacs:
         condition: service_healthy
     restart: unless-stopped
+    network_mode: host
+
+  modality_emulator:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    container_name: modality-emulator
+    command: ["uv", "run", "python", "-m", "src.modality_emulator"]
+    volumes:
+      - pacs-storage:/var/lib/pacs/storage
+      - pacs-db:/var/lib/pacs
+    environment:
+      - PACS_DB_PATH=/var/lib/pacs/pacs.db
+      - PACS_STORAGE_PATH=/var/lib/pacs/storage
+      - MWL_DB_PATH=/var/lib/pacs/worklist.db
+      - LOG_LEVEL=INFO
+    depends_on:
+      mwl:
+        condition: service_healthy
+      pacs:
+        condition: service_healthy
+    restart: unless-stopped
+    network_mode: host
 
 volumes:
   pacs-storage:

docs/adr/ADR-006_Modality_emulator.md

@@ -0,0 +1,36 @@
+# ADR-006: Use a modality emulator for non-production environments
+
+Date: 2026-06-04
+
+Status: Accepted
+
+## Context
+
+The Manage Breast Screening service sends worklist items corresponding to screening appointments as the appointment starts. These are received by the gateway via Azure Relay and processed to trigger DICOM API uploads of the screening study data. The worklist items are queried via DICOM C-FIND requests to the gateway MWL server and stored on the gateway PACS server using the DICOM C-STORE protocol.
+
+In production this query is performed by the modality, which is a physical machine in the hospital network that runs software to query the MWL and display results to radiographers. The modality then sends the resulting study DICOM files to the Gateway PACS server using C-STORE. The PACS server then compresses the image data in the DICOM files and uploads it to the DICOM API hosted on Manage.
+
+In non-production environments, there is no modality available to perform this query. This means that testing the full flow of worklist items through the gateway and into the DICOM API is difficult without deploying to production or using complex stubs/mocks.
+
+## Decision
+
+To facilitate end-to-end testing of the worklist flow in non-production environments, we will implement a simple modality emulator. This will be a lightweight service that simulates the behavior of a real modality by performing queries against the MWL server and storing the emulated study data on the Gateway PACS server, mimicking the behavior of a real modality.
+
+The emulator runs on the same host as the MWL and PACS servers in non-production environments. It runs as a Python process that makes a C-FIND request at a configurable polling interval, checking for new worklist items. When it finds a new item, it creates a DICOM dataset from the worklist details, includes some sample image data and sends them to the PACS server via C-STORE.
+
+The sample image data corresponds to the 4 main laterality and view position combinations (LCC, LMLO, RCC, RMLO) to allow testing of the laterality and view position logic in the gateway images screens as the DICOM files are uploaded to Manage.
+
+## Consequences
+
+We no longer have to use a third party emulator or do any UI work to test the happy path of worklist items flowing through the gateway and into the DICOM API in non-production environments. This also allows us to make more comprehensive end-to-end tests that cover the full flow, including the C-FIND and C-STORE interactions with the MWL and PACS servers.
+
+### Positive Consequences
+
+- **End-to-end testing:** Enables testing of the full worklist flow in non-production environments without needing a physical modality or complex stubs/mocks.
+- **Simplified test setup:** Developers can run the emulator locally or in test environments to simulate modality behavior without additional infrastructure.
+- **Improved test coverage:** Allows for more comprehensive tests that cover the interactions with the MWL and PACS servers, as well as the gateway's processing logic for worklist items.
+- **Configurable behavior:** The emulator can be configured to simulate different scenarios, such as varying polling intervals or different worklist item details, to test edge cases and error handling in the gateway.
+
+### Negative Consequences
+
+- The emulator and sample images are packaged with every release. This increases the size of the release artifact and may have implications for storage and distribution.

scripts/bash/deploy_arc_ring.sh

@@ -82,7 +82,8 @@ PACS_AET=SCREENING_PACS
 PACS_PORT=4244
 PACS_STORAGE_PATH=${BASE_PATH}/data/storage
 PACS_DB_PATH=${BASE_PATH}/data/pacs.db
-LOG_LEVEL=INFO"
+LOG_LEVEL=INFO
+SAMPLE_IMAGES_PATH=${BASE_PATH}/current/sample_images"
 
   # Cross-platform base64 encoding (works on macOS and Linux)
   ENV_CONTENT_B64=$(printf '%s' "$ENV_CONTENT" | base64 | tr -d '\n')
@@ -105,6 +106,7 @@ LOG_LEVEL=INFO"
     --arg pyver  "$PYTHON_VERSION" \
     --arg envb64 "$ENV_CONTENT_B64" \
     --arg token  "$GITHUB_TOKEN" \
+    --arg env    "$ENVIRONMENT" \
     '{
       location: $loc,
       properties: {
@@ -113,7 +115,8 @@ LOG_LEVEL=INFO"
           { name: "ReleaseTag",    value: $tag   },
           { name: "PythonVersion", value: $pyver },
           { name: "EnvContentB64", value: $envb64 },
-          { name: "GitHubToken",   value: $token }
+          { name: "GitHubToken",   value: $token },
+          { name: "Environment",   value: $env }
         ],
         runAsSystem:      true,
         timeoutInSeconds: 1800

scripts/bash/package_release.sh

@@ -61,7 +61,7 @@ echo ""
 
 # ── Validate required files ───────────────────────────────────────────────────
 
-REQUIRED_FILES=("src/" "scripts/python/" "pyproject.toml" "uv.lock" "README.md")
+REQUIRED_FILES=("src/" "sample_images/" "scripts/python/" "pyproject.toml" "uv.lock" "README.md" "LICENCE.md")
 
 for item in "${REQUIRED_FILES[@]}"; do
     if [[ ! -e "${REPO_ROOT}/${item}" ]]; then

scripts/powershell/deploy.ps1

@@ -43,7 +43,10 @@ param(
     [string]$GitHubToken,
 
     [Parameter()]
-    [string]$EnvContentB64 = ""
+    [string]$EnvContentB64 = "",
+
+    [Parameter()]
+    [string]$Environment = "prod"
 )
 
 Set-StrictMode -Version Latest
@@ -337,6 +340,9 @@ $services = @(
     @{ Name = "Gateway-MWL"; Script = "mwl_main.py" },
     @{ Name = "Gateway-Upload"; Script = "upload_main.py" }
 )
+if (($Environment -ne "prod") -and ($Environment -ne "preprod")) {
+    $services += @{ Name = "Gateway-Emulator"; Script = "modality_emulator.py" }
+}
 
 # -- Extraction ---------------------------------------------------------------
 

scripts/powershell/rollback.ps1

@@ -61,7 +61,8 @@ $services = @(
     @{ Name = "Gateway-Relay"; Script = "relay_listener.py" },
     @{ Name = "Gateway-PACS"; Script = "pacs_main.py" },
     @{ Name = "Gateway-MWL"; Script = "mwl_main.py" },
-    @{ Name = "Gateway-Upload"; Script = "upload_main.py" }
+    @{ Name = "Gateway-Upload"; Script = "upload_main.py" },
+    @{ Name = "Gateway-Emulator"; Script = "modality_emulator.py" }
 )
 
 # -- Resolve current version --------------------------------------------------

src/modality_emulator.py

@@ -0,0 +1,215 @@
+import datetime
+import logging
+import os
+import time
+
+import numpy as np
+from PIL import Image
+from pydicom.dataset import Dataset, FileMetaDataset
+from pydicom.uid import ExplicitVRLittleEndian, generate_uid
+from pynetdicom import AE
+from pynetdicom.sop_class import (
+    DigitalMammographyXRayImageStorageForPresentation,  # type: ignore
+    ModalityWorklistInformationFind,  # type: ignore
+)
+
+from environment import Environment
+from services.dicom import PENDING, PENDING_WARNING, SUCCESS
+from services.mwl import MWLStatus
+from services.storage import MWLStorage
+
+logging.basicConfig(
+    level=os.getenv("LOG_LEVEL", "INFO").upper(),
+    format=os.getenv("LOG_FORMAT", "%(asctime)s - %(name)s - %(levelname)s - %(message)s"),
+)
+
+logger = logging.getLogger(__name__)
+
+
+DICOM_LATERALITIES = ["L", "R"]
+DICOM_VIEWS = ["CC", "MLO"]
+EMULATED_PROCEDURE_DURATION_SECONDS = int(os.getenv("EMULATED_PROCEDURE_DURATION_SECONDS", "5"))
+MODALITY = "MG"
+MWL_AET = os.getenv("MWL_AET", "SCREENING_MWL")
+MWL_DB_PATH = os.getenv("MWL_DB_PATH", "/var/lib/pacs/worklist.db")
+MWL_HOST = os.getenv("MWL_HOST", "localhost")
+MWL_PORT = int(os.getenv("MWL_PORT", "4243"))
+PACS_AET = os.getenv("PACS_AET", "SCREENING_PACS")
+PACS_DB_PATH = os.getenv("PACS_DB_PATH", "/var/lib/pacs/pacs.db")
+PACS_HOST = os.getenv("PACS_HOST", "localhost")
+PACS_PORT = int(os.getenv("PACS_PORT", "4244"))
+SAMPLE_IMAGES_PATH = os.getenv("SAMPLE_IMAGES_PATH", "sample_images")
+
+
+class DicomExample:
+    def __init__(
+        self, dataset: Dataset | None, laterality: str, view: str, study_instance_uid: str, series_number: int
+    ):
+        self.dataset = dataset
+        self.laterality = laterality
+        self.view = view
+        self.study_instance_uid = study_instance_uid
+        self.series_number = series_number
+        self.data = self.generate_dicom()
+
+    def generate_dicom(self) -> Dataset:
+        ds = Dataset()
+        if not self.dataset:
+            logger.error("No dataset provided for DICOM generation")
+            return ds
+
+        img_path = f"{SAMPLE_IMAGES_PATH}/L{self.view}.jpg"
+        img = Image.open(img_path).convert("L")
+        if self.laterality == "R":
+            img = img.transpose(Image.Transpose.FLIP_LEFT_RIGHT)
+        columns, rows = img.size
+        pixel_array = np.array(img, dtype=np.uint8)
+        pixel_bytes = pixel_array.tobytes()
+        if len(pixel_bytes) % 2 != 0:
+            pixel_bytes += b"\x00"
+
+        file_meta = FileMetaDataset()
+        file_meta.MediaStorageSOPClassUID = DigitalMammographyXRayImageStorageForPresentation
+        file_meta.MediaStorageSOPInstanceUID = generate_uid()
+        file_meta.ImplementationClassUID = generate_uid()
+        file_meta.TransferSyntaxUID = ExplicitVRLittleEndian
+
+        ds.SOPClassUID = file_meta.MediaStorageSOPClassUID
+        ds.SOPInstanceUID = file_meta.MediaStorageSOPInstanceUID
+        ds.SamplesPerPixel = 1
+        ds.PhotometricInterpretation = "MONOCHROME2"
+        ds.Rows = rows
+        ds.Columns = columns
+        ds.BitsAllocated = 8
+        ds.BitsStored = 8
+        ds.HighBit = 7
+        ds.PixelRepresentation = 0
+        ds.PixelData = pixel_bytes
+        ds.ImageLaterality = self.laterality
+        ds.ViewPosition = self.view
+
+        ds.AccessionNumber = self.dataset.AccessionNumber
+        ds.PatientID = self.dataset.PatientID
+        ds.PatientName = self.dataset.PatientName
+        ds.PatientBirthDate = self.dataset.PatientBirthDate
+        ds.PatientSex = self.dataset.PatientSex
+        scheduled_step = self.dataset.ScheduledProcedureStepSequence[0]
+        ds.StudyDate = scheduled_step.ScheduledProcedureStepStartDate
+        ds.StudyTime = scheduled_step.ScheduledProcedureStepStartTime
+        ds.StudyInstanceUID = self.study_instance_uid
+        ds.StudyID = f"STUDY{self.study_instance_uid[-8:]}"
+        ds.SeriesInstanceUID = generate_uid()
+        ds.SeriesNumber = self.series_number
+        ds.Modality = MODALITY
+        ds.InstanceNumber = 1
+        ds.file_meta = file_meta
+
+        logger.debug(f"Generated DICOM for worklist item {self.dataset.AccessionNumber} - {self.laterality}{self.view}")
+        logger.debug(f"{ds}")
+
+        return ds
+
+
+class ModalityEmulator:
+    def __init__(self, mwl_storage: MWLStorage):
+        self.mwl_storage = mwl_storage
+        self.processed_items = set()
+
+    def process_worklist_items(self, ae: AE):
+        """
+        Queries the MWL for items scheduled for today and sends generated DICOM files to the PACS server for each item.
+        """
+        mwl_assoc = ae.associate(MWL_HOST, MWL_PORT, ae_title=MWL_AET)
+        pacs_assoc = ae.associate(PACS_HOST, PACS_PORT, ae_title=PACS_AET)
+
+        if mwl_assoc.is_established and pacs_assoc.is_established:
+            logger.info(f"Connected to MWL server {MWL_HOST}:{MWL_PORT} ({MWL_AET})")
+            logger.info(f"Connected to PACS server {PACS_HOST}:{PACS_PORT} ({PACS_AET})")
+
+            logger.info("Querying MWL for scheduled items...")
+            responses = mwl_assoc.send_c_find(self.c_find_dataset, query_model=ModalityWorklistInformationFind)
+            for status, ds in responses:
+                status_code = getattr(status, "Status", SUCCESS)
+
+                if status_code in (PENDING, PENDING_WARNING):
+                    accession_number = getattr(ds, "AccessionNumber", "UNKNOWN")
+
+                    if accession_number in self.processed_items:
+                        logger.info(f"Skipping already processed worklist item {accession_number}")
+                        continue
+
+                    self.processed_items.add(accession_number)
+                    study_instance_uid = generate_uid()
+                    series_number = 1
+                    for laterality in DICOM_LATERALITIES:
+                        for view in DICOM_VIEWS:
+                            logger.info(
+                                f"Processing worklist item {accession_number} - generating DICOM for {laterality}{view}"
+                            )
+                            dicom_example = DicomExample(ds, laterality, view, study_instance_uid, series_number)
+                            dataset = dicom_example.data
+
+                            if getattr(dataset, "SOPInstanceUID", None) is None:
+                                logger.error(
+                                    f"Skipping DICOM generation for {laterality}{view} of worklist item {accession_number}"
+                                )
+                                continue
+
+                            pacs_assoc.send_c_store(dataset)
+                            logger.info(
+                                f"Sent DICOM for {laterality}{view} of worklist item {accession_number}. Series# {series_number}"
+                            )
+                            series_number += 1
+
+                    time.sleep(1)  # Allow C-STORE operations to complete before updating status
+
+                    self.mwl_storage.update_status(accession_number, MWLStatus.COMPLETED.value)
+                    logger.info(f"Completed processing for worklist item {accession_number}")
+                elif status_code == SUCCESS:
+                    logger.info("C-FIND query completed successfully")
+                else:
+                    logger.error(f"C-FIND query failed with status: 0x{status_code:04X}")
+
+        else:
+            logger.error("Failed to make MWL and PACS associations")
+
+        mwl_assoc.release()
+        pacs_assoc.release()
+
+    @property
+    def c_find_dataset(self) -> Dataset:
+        date_today = datetime.date.today()
+        ds = Dataset()
+        sps_dataset = Dataset()
+        sps_dataset.Modality = MODALITY
+        sps_dataset.ScheduledProcedureStepStartDate = date_today.strftime("%Y%m%d")
+        sps_dataset.ScheduledProcedureStepStartTime = "000000-"
+        ds.ScheduledProcedureStepSequence = [sps_dataset]
+        return ds
+
+
+def main():
+    if Environment().production:
+        raise RuntimeError("Modality Emulator should not be run in production environment")
+
+    logger.info("Modality Emulator Starting...")
+    mwl_storage = MWLStorage(db_path=MWL_DB_PATH)
+    emulator = ModalityEmulator(mwl_storage)
+    ae = AE(ae_title="ModalityEmulator")
+    ae.add_requested_context(DigitalMammographyXRayImageStorageForPresentation)
+    ae.add_requested_context(ModalityWorklistInformationFind)
+
+    while True:
+        try:
+            time.sleep(EMULATED_PROCEDURE_DURATION_SECONDS)
+            emulator.process_worklist_items(ae)
+        except KeyboardInterrupt:
+            logger.warning("\n Modality Emulator shutting down...")
+            break
+        except Exception:
+            logger.exception("Error in Modality Emulator")
+            time.sleep(5)  # Wait before retrying to avoid tight error loop
+
+
+if __name__ == "__main__":
+    main()

src/services/dicom/__init__.py

@@ -7,6 +7,7 @@
 SUCCESS = 0x0000
 FAILURE = 0xC000
 PENDING = 0xFF00
+PENDING_WARNING = 0xFF01
 INVALID_ATTRIBUTE = 0x0106
 DUPLICATE_SOP_INSTANCE = 0x0111
 UNKNOWN_SOP_INSTANCE = 0x0112

src/services/storage.py

@@ -367,7 +367,7 @@ def find_worklist_items(
             "source_message_id, study_description, study_instance_uid, status, mpps_instance_uid "
             "FROM worklist_items"
         )
-        where_clauses = []
+        where_clauses = ["status NOT IN ('COMPLETED', 'DISCONTINUED')"]
         params = []
 
         if accession_number: