Accept worklist.create_test_item actions and process with Modality Emulator

Processes test worklist items immediately sending emulated DICOM files
with sample images via the uploader.

Author: steventux

src/modality_emulator.py

@@ -128,7 +128,7 @@ def __init__(self, mwl_storage: MWLStorage):
         self.mwl_storage = mwl_storage
         self.processed_items = set()
 
-    def process_worklist_items(self, ae: AE):
+    def process_worklist_items(self, ae: AE, patient_name: str | None = None):
         """
         Queries the MWL for items scheduled for today and sends generated DICOM files to the PACS server for each item.
         """
@@ -140,7 +140,8 @@ def process_worklist_items(self, ae: AE):
             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)
+            c_find_dataset = self.c_find_dataset(patient_name=patient_name)
+            responses = mwl_assoc.send_c_find(c_find_dataset, query_model=ModalityWorklistInformationFind)
             for status, ds in responses:
                 status_code = getattr(status, "Status", SUCCESS)
 
@@ -189,15 +190,18 @@ def process_worklist_items(self, ae: AE):
         mwl_assoc.release()
         pacs_assoc.release()
 
-    @property
-    def c_find_dataset(self) -> Dataset:
+    def c_find_dataset(self, patient_name: str | None = None) -> 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]
+
+        if patient_name:
+            ds.PatientName = patient_name
+
         return ds
 
 

src/relay_listener.py

@@ -16,10 +16,16 @@
 
 from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
 from dotenv import load_dotenv
+from pynetdicom import AE
+from pynetdicom.sop_class import (
+    DigitalMammographyXRayImageStorageForPresentation,  # type: ignore
+    ModalityWorklistInformationFind,  # type: ignore
+)
 from websockets.asyncio.client import connect
 from websockets.exceptions import ConnectionClosedError
 
 from environment import Environment
+from modality_emulator import ModalityEmulator
 from services.mwl.create_worklist_item import CreateWorklistItem
 from services.mwl.update_worklist_item_status import UpdateWorklistItemStatus
 from services.storage import MWLStorage
@@ -94,6 +100,10 @@ def process_action(self, payload: dict):
             return {"status": "echo", "payload": payload}
         elif action_name == "worklist.create_item":
             return CreateWorklistItem(self.storage).call(payload)
+        elif action_name == "worklist.create_test_item":
+            result = CreateWorklistItem(self.storage).call(payload)
+            self.process_with_modality_emulator(patient_name=payload.get("patient_name"))
+            return result
         elif action_name == "worklist.update_item_status":
             return UpdateWorklistItemStatus(self.storage).call(payload)
         else:
@@ -106,6 +116,14 @@ def _connect(self):
             compression=None,
         )
 
+    def process_with_modality_emulator(self, patient_name: str | None = None):
+        """Process worklist items with ModalityEmulator."""
+        ae = AE(ae_title="ModalityEmulator")
+        ae.add_requested_context(DigitalMammographyXRayImageStorageForPresentation)
+        ae.add_requested_context(ModalityWorklistInformationFind)
+
+        ModalityEmulator(self.storage).process_worklist_items(ae, patient_name=patient_name)
+
 
 class RelayURI:
     def __init__(self):

tests/test_modality_emulator.py

@@ -187,6 +187,36 @@ def test_process_worklist_items_returns_when_no_items(
         mwl_assoc.release.assert_called_once()
         pacs_assoc.release.assert_called_once()
 
+    @patch.object(ModalityEmulator, "c_find_dataset")
+    def test_process_worklist_items_passes_patient_name_to_c_find_dataset(
+        self,
+        mock_c_find_dataset,
+        success_status,
+    ):
+        """Process worklist items passes patient_name to c_find_dataset."""
+        mwl_storage = MagicMock()
+        emulator = ModalityEmulator(mwl_storage)
+
+        query_ds = Dataset()
+        mock_c_find_dataset.return_value = query_ds
+
+        mwl_assoc = MagicMock()
+        mwl_assoc.is_established = True
+        mwl_assoc.send_c_find.return_value = [(success_status, None)]
+
+        pacs_assoc = MagicMock()
+        pacs_assoc.is_established = True
+
+        ae = MagicMock()
+        ae.associate.side_effect = [mwl_assoc, pacs_assoc]
+
+        emulator.process_worklist_items(ae, patient_name="Jane Doe")
+
+        mock_c_find_dataset.assert_called_once_with(patient_name="Jane Doe")
+        mwl_storage.update_status.assert_not_called()
+        mwl_assoc.release.assert_called_once()
+        pacs_assoc.release.assert_called_once()
+
     @patch("modality_emulator.time.sleep")
     def test_process_worklist_items_handles_failed_association(
         self,

tests/test_relay_listener.py

@@ -114,6 +114,34 @@ def test_process_update_item_status_action(self, storage_instance, listener_payl
 
         storage_instance.update_status.assert_called_once_with("ACC999999", "IN PROGRESS")
 
+    def test_process_create_test_item_action_triggers_modality_emulator(self, storage_instance, listener_payload):
+        """Process create test item action and trigger modality emulator."""
+        subject = RelayListener(storage_instance)
+        payload = dict(listener_payload)
+        payload["action_type"] = "worklist.create_test_item"
+
+        with patch.object(subject, "process_with_modality_emulator") as mock_emulator:
+            response = subject.process_action(payload)
+
+        assert response == {"action_id": "action-12345", "status": "created"}
+        mock_emulator.assert_called_once_with(patient_name=payload.get("patient_name"))
+
+        storage_instance.store_worklist_item.assert_called_once_with(
+            WorklistItem(
+                accession_number="ACC999999",
+                patient_id="999123456",
+                patient_name="SMITH^JANE",
+                patient_birth_date="19900202",
+                patient_sex="F",
+                scheduled_date="20240615",
+                scheduled_time="101500",
+                modality="MG",
+                study_description="MAMMOGRAPHY",
+                source_message_id="action-12345",
+            )
+        )
+
+
     def test_process_action_missing_keys(self, storage_instance, listener_payload):
         """Process action missing keys."""
         subject = RelayListener(storage_instance)