diff --git a/src/modality_emulator.py b/src/modality_emulator.py index 398bbd3d..753e40c1 100644 --- a/src/modality_emulator.py +++ b/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,8 +190,7 @@ 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() @@ -198,6 +198,10 @@ def c_find_dataset(self) -> Dataset: 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 diff --git a/src/relay_listener.py b/src/relay_listener.py index d24ba635..5415d8cd 100644 --- a/src/relay_listener.py +++ b/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,10 +100,25 @@ 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) + patient_name = payload.get("parameters", {}).get("worklist_item", {}).get("participant", {}).get("name") + + if not patient_name: + logger.warning("No patient name provided for ModalityEmulator test item processing") + return { + "status": "error", + "message": "No patient name provided for ModalityEmulator test item processing", + } + + self.process_with_modality_emulator(patient_name=patient_name) + + return result elif action_name == "worklist.update_status": return UpdateWorklistItemStatus(self.storage).call(payload) else: - raise ValueError(f"Unsupported action: {action_name}") + logger.error("Unsupported action: %s", action_name) + return {"status": "error", "message": f"Unsupported action: {action_name}"} def _connect(self): """Connect to Azure Relay.""" @@ -106,6 +127,26 @@ 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) + + def _run_emulator(): + try: + ModalityEmulator(self.storage).process_worklist_items(ae, patient_name=patient_name) + except Exception: + logger.exception("Modality emulator processing failed") + + try: + loop = asyncio.get_running_loop() + except RuntimeError: + # Called outside an event loop (e.g. unit tests) + _run_emulator() + else: + loop.create_task(asyncio.to_thread(_run_emulator)) + class RelayURI: def __init__(self): diff --git a/tests/test_modality_emulator.py b/tests/test_modality_emulator.py index 9eebe081..bebfeaff 100644 --- a/tests/test_modality_emulator.py +++ b/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, diff --git a/tests/test_relay_listener.py b/tests/test_relay_listener.py index 3e4de26f..0e4ad64c 100644 --- a/tests/test_relay_listener.py +++ b/tests/test_relay_listener.py @@ -114,6 +114,52 @@ 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["parameters"]["worklist_item"]["participant"]["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_create_test_item_action_without_patient_name_returns_error( + self, storage_instance, listener_payload + ): + """Process create test item action without a patient name returns an error.""" + subject = RelayListener(storage_instance) + payload = dict(listener_payload) + payload["action_type"] = "worklist.create_test_item" + del payload["parameters"]["worklist_item"]["participant"]["name"] + + with patch.object(subject, "process_with_modality_emulator"): + response = subject.process_action(payload) + + assert response == { + "status": "error", + "message": "No patient name provided for ModalityEmulator test item processing", + } + def test_process_action_missing_keys(self, storage_instance, listener_payload): """Process action missing keys.""" subject = RelayListener(storage_instance) @@ -132,15 +178,13 @@ def test_process_action_invalid_type(self, storage_instance, listener_payload): listener_payload["action_type"] = "worklist.unknown_action" - with pytest.raises(ValueError): - response = subject.process_action(listener_payload) - assert response == { - "status": "error", - "action_id": "action-12345", - "error": "Unknown action type: worklist.unknown_action", - } + response = subject.process_action(listener_payload) + assert response == { + "status": "error", + "message": "Unsupported action: worklist.unknown_action", + } - storage_instance.store_worklist_item.assert_not_called() + storage_instance.store_worklist_item.assert_not_called() class TestRelayURIWithDefaultAzureCredential: