Add modality emulator

This standalone python script polls the worklist database every 5 seconds for new worklist items scheduled for today.
It creates a DICOM dataset with the standard 4 images and the worklist patient info and then performs a C-STORE.
The C-STORE will then be picked up by the dicom uploader and the sample images sent back to Manage.

Author: steventux

sample_images/LCC.jpg

@@ -0,0 +1,187 @@
+import datetime
+import logging
+import os
+import time
+from functools import cached_property
+
+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, ModalityWorklistInformationFind
+
+from environment import Environment
+from services.dicom import PENDING, SUCCESS
+from services.mwl import MWLStatus
+from services.storage import MWLStorage, PACSStorage
+
+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__)
+
+
+MWL_DB_PATH = os.getenv("MWL_DB_PATH", "/var/lib/pacs/worklist.db")
+PACS_DB_PATH = os.getenv("PACS_DB_PATH", "/var/lib/pacs/pacs.db")
+PACS_STORAGE_PATH = os.getenv("PACS_STORAGE_PATH", "/var/lib/pacs/storage")
+MWL_AET = os.getenv("MWL_AET", "SCREENING_MWL")
+MWL_ADDRESS = os.getenv("MWL_ADDRESS", "localhost")
+MWL_PORT = int(os.getenv("MWL_PORT", "4243"))
+PACS_AET = os.getenv("PACS_AET", "SCREENING_PACS")
+PACS_ADDRESS = os.getenv("PACS_ADDRESS", "localhost")
+PACS_PORT = int(os.getenv("PACS_PORT", "4244"))
+DICOM_LATERALITIES = ["L", "R"]
+DICOM_VIEWS = ["CC", "MLO"]
+SAMPLE_IMAGES_PATH = os.getenv("SAMPLE_IMAGES_PATH", "/app/sample_images")
+EMULATED_PROCEDURE_DURATION_SECONDS = int(os.getenv("EMULATED_PROCEDURE_DURATION_SECONDS", "5"))
+
+
+class DicomExample:
+    def __init__(self, dataset: Dataset, 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:
+        img_path = f"{SAMPLE_IMAGES_PATH}/{self.laterality}{self.view}.jpg"
+        img = Image.open(img_path).convert("L")
+        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 = Dataset()
+        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 = "MG"
+        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, pacs_storage: PACSStorage):
+        self.mwl_storage = mwl_storage
+        self.pacs_storage = pacs_storage
+
+    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_ADDRESS, MWL_PORT, ae_title=MWL_AET)
+        pacs_assoc = ae.associate(PACS_ADDRESS, PACS_PORT, ae_title=PACS_AET)
+
+        if mwl_assoc.is_established and pacs_assoc.is_established:
+            logger.info(f"Connected to MWL server {MWL_ADDRESS}:{MWL_PORT} ({MWL_AET})")
+            logger.info(f"Connected to PACS server {PACS_ADDRESS}:{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:
+                if status == PENDING:
+                    accession_number = ds.AccessionNumber
+                    study_instance_uid = generate_uid()
+                    logger.info(f"Found worklist item: {accession_number} for patient {ds.PatientID}")
+                    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)
+                            pacs_assoc.send_c_store(dicom_example.data)
+                            logger.info(
+                                f"Sent DICOM for {laterality}{view} of worklist item {accession_number}. Series# {series_number}"
+                            )
+                            series_number += 1
+
+                    self.mwl_storage.update_status(accession_number, MWLStatus.COMPLETED.value)
+                    logger.info(f"Completed processing for worklist item {accession_number}")
+                elif status == SUCCESS:
+                    logger.info("C-FIND query completed successfully")
+                else:
+                    logger.error(f"C-FIND query failed with status: {status}")
+
+        else:
+            logger.error("Failed to make MWL and PACS associations")
+
+        mwl_assoc.release()
+        pacs_assoc.release()
+
+    @cached_property
+    def c_find_dataset(self) -> Dataset:
+        date_today = datetime.date.today()
+        one_hour_ago = datetime.datetime.now() - datetime.timedelta(hours=1)
+        ds = Dataset()
+        sps_dataset = Dataset()
+        sps_dataset.Modality = "MG"
+        sps_dataset.ScheduledProcedureStepStartDate = date_today.strftime("%Y%m%d")
+        sps_dataset.ScheduledProcedureStepStartTime = one_hour_ago.strftime("%H%M%S-")
+        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)
+    pacs_storage = PACSStorage(db_path=PACS_DB_PATH)
+    emulator = ModalityEmulator(mwl_storage, pacs_storage)
+    ae = AE(ae_title="ModalityEmulator")
+    ae.add_requested_context(DigitalMammographyXRayImageStorageForPresentation)
+    ae.add_requested_context(ModalityWorklistInformationFind)
+
+    while True:
+        try:
+            emulator.process_worklist_items(ae)
+            time.sleep(5)
+        except KeyboardInterrupt:
+            logger.warning("\n Modality Emulator shutting down...")
+            break
+
+
+if __name__ == "__main__":
+    main()

sample_images/LMLO.jpg

@@ -0,0 +1,195 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+from pydicom.dataset import Dataset
+from pydicom.uid import DigitalMammographyXRayImageStorageForPresentation, ExplicitVRLittleEndian, generate_uid
+
+from modality_emulator import (
+    DICOM_LATERALITIES,
+    DICOM_VIEWS,
+    DicomExample,
+    ModalityEmulator,
+    main,
+)
+
+
+class TestDicomExample:
+    @patch("modality_emulator.Image.open")
+    @patch("modality_emulator.generate_uid")
+    def test_generate_dicom_creates_valid_dataset(
+        self,
+        mock_generate_uid,
+        mock_image_open,
+    ):
+        study_instance_uid = generate_uid()
+        sop_instance_uid = generate_uid()
+        implementation_class_uid = generate_uid()
+        series_instance_uid = generate_uid()
+
+        mock_generate_uid.side_effect = [
+            sop_instance_uid,
+            implementation_class_uid,
+            series_instance_uid,
+        ]
+
+        mock_image = MagicMock()
+        mock_image.convert.return_value = mock_image
+        mock_image.size = (100, 200)
+
+        mock_image_open.return_value = mock_image
+
+        ds = Dataset()
+        ds.AccessionNumber = "ACC123"
+        ds.PatientID = "PAT001"
+        ds.PatientName = "Jane^Doe"
+        ds.PatientBirthDate = "19800101"
+        ds.PatientSex = "F"
+        sps = Dataset()
+        sps.ScheduledProcedureStepStartDate = "20260514"
+        sps.ScheduledProcedureStepStartTime = "090000"
+        ds.ScheduledProcedureStepSequence = [sps]
+
+        with patch("modality_emulator.np.array") as mock_np_array:
+            mock_pixel_array = MagicMock()
+            mock_pixel_array.tobytes.return_value = b"\x01\x02"
+            mock_np_array.return_value = mock_pixel_array
+
+            dicom = DicomExample(
+                dataset=ds,
+                laterality="L",
+                view="CC",
+                study_instance_uid=study_instance_uid,
+                series_number=1,
+            )
+
+        assert isinstance(dicom.data, Dataset)
+        assert dicom.data.PatientID == "PAT001"
+        assert dicom.data.PatientName == "Jane^Doe"
+        assert dicom.data.ImageLaterality == "L"
+        assert dicom.data.ViewPosition == "CC"
+        assert dicom.data.Rows == 200
+        assert dicom.data.Columns == 100
+        assert dicom.data.StudyInstanceUID == study_instance_uid
+        assert dicom.data.SOPInstanceUID == sop_instance_uid
+        assert dicom.data.SeriesInstanceUID == series_instance_uid
+        assert dicom.data.file_meta.TransferSyntaxUID == ExplicitVRLittleEndian
+        assert dicom.data.file_meta.ImplementationClassUID == implementation_class_uid
+        assert dicom.data.file_meta.MediaStorageSOPClassUID == DigitalMammographyXRayImageStorageForPresentation
+        assert dicom.data.file_meta.MediaStorageSOPInstanceUID == sop_instance_uid
+        assert dicom.data.StudyDate == "20260514"
+        assert dicom.data.StudyTime == "090000"
+
+
+class TestModalityEmulator:
+    @patch("modality_emulator.time.sleep")
+    @patch("modality_emulator.generate_uid")
+    @patch("modality_emulator.DicomExample")
+    def test_process_worklist_items_sends_all_dicoms(
+        self,
+        mock_dicom_example,
+        mock_generate_uid,
+        _,
+    ):
+        mock_generate_uid.return_value = "1.2.3.study"  # gitleaks: ignore
+
+        mwl_storage = MagicMock()
+        pacs_storage = MagicMock()
+
+        emulator = ModalityEmulator(mwl_storage, pacs_storage)
+
+        ds = Dataset()
+        ds.AccessionNumber = "ACC123"
+        ds.PatientID = "PAT001"
+        ds.PatientName = "Jane^Doe"
+        ds.PatientSex = "F"
+        sps = Dataset()
+        sps.ScheduledProcedureStepStartDate = "20260514"
+        sps.ScheduledProcedureStepStartTime = "090000"
+        ds.scheduled_procedure_step_sequence = [sps]
+
+        mock_dicom_example.return_value.data = ds
+
+        mwl_assoc = MagicMock()
+        mwl_assoc.is_established = True
+        mwl_assoc.send_c_find.return_value = [(0xFF00, ds), (0x0000, None)]
+
+        pacs_assoc = MagicMock()
+        pacs_assoc.is_established = True
+
+        ae = MagicMock()
+        ae.associate.side_effect = [mwl_assoc, pacs_assoc]
+
+        emulator.process_worklist_items(ae)
+
+        expected_send_count = len(DICOM_LATERALITIES) * len(DICOM_VIEWS)
+
+        assert mwl_assoc.send_c_find.call_count == 1
+        assert pacs_assoc.send_c_store.call_count == expected_send_count
+
+        mwl_storage.update_status.assert_called_once_with(
+            "ACC123",
+            "COMPLETED",
+        )
+
+        mwl_assoc.release.assert_called_once()
+        pacs_assoc.release.assert_called_once()
+
+    @patch("modality_emulator.time.sleep")
+    def test_process_worklist_items_returns_when_no_items(
+        self,
+        mock_sleep,
+    ):
+        mwl_storage = MagicMock()
+        pacs_storage = MagicMock()
+        emulator = ModalityEmulator(mwl_storage, pacs_storage)
+        ae = MagicMock()
+
+        mwl_assoc = MagicMock()
+        mwl_assoc.is_established = True
+        mwl_assoc.send_c_find.return_value = [(0x0000, None)]
+
+        pacs_assoc = MagicMock()
+        pacs_assoc.is_established = True
+
+        ae = MagicMock()
+        ae.associate.side_effect = [mwl_assoc, pacs_assoc]
+
+        emulator.process_worklist_items(ae)
+
+        mock_sleep.assert_not_called()
+        pacs_assoc.send_c_store.assert_not_called()
+        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,
+        _,
+    ):
+        mwl_storage = MagicMock()
+        pacs_storage = MagicMock()
+
+        emulator = ModalityEmulator(mwl_storage, pacs_storage)
+
+        mwl_assoc = MagicMock()
+        mwl_assoc.is_established = True
+
+        pacs_assoc = MagicMock()
+        pacs_assoc.is_established = False
+
+        ae = MagicMock()
+        ae.associate.side_effect = [mwl_assoc, pacs_assoc]
+
+        emulator.process_worklist_items(ae)
+
+        mwl_assoc.send_c_find.assert_not_called()
+        pacs_assoc.send_c_store.assert_not_called()
+        mwl_storage.update_status.assert_not_called()
+        mwl_assoc.release.assert_called_once()
+        pacs_assoc.release.assert_called_once()
+
+    @patch("modality_emulator.Environment", production=True)
+    def test_main_raises_in_production(self, _):
+        with pytest.raises(RuntimeError, match="Modality Emulator should not be run in production environment"):
+            main()