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,163 @@
+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,
+)
+
+from environment import Environment
+from models import WorklistItem
+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")
+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, worklist_item: WorklistItem, laterality: str, view: str, study_instance_uid: str):
+        self.worklist_item = worklist_item
+        self.laterality = laterality
+        self.view = view
+        self.data = self.generate_dicom(study_instance_uid)
+
+    def generate_dicom(self, study_instance_uid: str) -> 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.worklist_item.accession_number
+        ds.PatientID = self.worklist_item.patient_id
+        ds.PatientName = self.worklist_item.patient_name
+        ds.PatientBirthDate = self.worklist_item.patient_birth_date
+        ds.PatientSex = self.worklist_item.patient_sex
+        ds.StudyDate = self.worklist_item.scheduled_date
+        ds.StudyTime = self.worklist_item.scheduled_time
+        ds.StudyInstanceUID = study_instance_uid
+        ds.StudyID = self.worklist_item.accession_number
+        ds.SeriesInstanceUID = generate_uid()
+        ds.SeriesNumber = 1
+        ds.Modality = "MG"
+        ds.InstanceNumber = 1
+        ds.file_meta = file_meta
+
+        logger.debug(
+            f"Generated DICOM for worklist item {self.worklist_item.accession_number} - {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.
+        """
+        today_formatted = datetime.date.today().strftime("%Y%m%d")
+        worklist_items = self.mwl_storage.find_worklist_items(
+            scheduled_date=today_formatted,
+        )
+
+        if not worklist_items:
+            return
+
+        time.sleep(EMULATED_PROCEDURE_DURATION_SECONDS)
+
+        assoc = ae.associate(PACS_ADDRESS, PACS_PORT, ae_title=PACS_AET)
+
+        if assoc.is_established:
+            logger.info(f"Connected to server {PACS_ADDRESS}:{PACS_PORT} ({PACS_AET})")
+
+            for item in worklist_items:
+                study_instance_uid = generate_uid()
+                for laterality in DICOM_LATERALITIES:
+                    for view in DICOM_VIEWS:
+                        logger.info(
+                            f"Processing worklist item {item.accession_number} - generating DICOM for {laterality}{view}"
+                        )
+                        dicom_example = DicomExample(item, laterality, view, study_instance_uid)
+                        assoc.send_c_store(dicom_example.data)
+                        logger.info(f"Sent DICOM for {laterality}{view} of worklist item {item.accession_number}")
+
+                self.mwl_storage.update_status(item.accession_number, "COMPLETED")
+                logger.info(f"Completed processing for worklist item {item.accession_number}")
+
+            assoc.release()
+        else:
+            logger.error(f"Failed to connect to server {PACS_ADDRESS}:{PACS_PORT} ({PACS_AET})")
+
+
+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)
+
+    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,167 @@
+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
+
+        worklist_item = MagicMock()
+        worklist_item.accession_number = "ACC123"
+        worklist_item.patient_id = "PAT001"
+        worklist_item.patient_name = "Jane^Doe"
+        worklist_item.patient_birth_date = "19800101"
+        worklist_item.patient_sex = "F"
+        worklist_item.scheduled_date = "20260514"
+        worklist_item.scheduled_time = "090000"
+
+        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(
+                worklist_item=worklist_item,
+                laterality="L",
+                view="CC",
+                study_instance_uid=study_instance_uid,
+            )
+
+        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
+
+
+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()
+
+        worklist_item = MagicMock()
+        worklist_item.accession_number = "ACC123"
+
+        mwl_storage.find_worklist_items.return_value = [worklist_item]
+
+        emulator = ModalityEmulator(mwl_storage, pacs_storage)
+
+        assoc = MagicMock()
+        assoc.is_established = True
+
+        ae = MagicMock()
+        ae.associate.return_value = assoc
+
+        mock_dataset = MagicMock()
+        mock_dicom_example.return_value.data = mock_dataset
+
+        emulator.process_worklist_items(ae)
+
+        expected_send_count = len(DICOM_LATERALITIES) * len(DICOM_VIEWS)
+
+        assert assoc.send_c_store.call_count == expected_send_count
+
+        mwl_storage.update_status.assert_called_once_with(
+            "ACC123",
+            "COMPLETED",
+        )
+
+        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()
+        mwl_storage.find_worklist_items.return_value = []
+        emulator = ModalityEmulator(mwl_storage, pacs_storage)
+        ae = MagicMock()
+
+        emulator.process_worklist_items(ae)
+
+        ae.associate.assert_not_called()
+        mock_sleep.assert_not_called()
+
+    @patch("modality_emulator.time.sleep")
+    def test_process_worklist_items_handles_failed_association(
+        self,
+        _,
+    ):
+        mwl_storage = MagicMock()
+        pacs_storage = MagicMock()
+        worklist_item = MagicMock()
+        worklist_item.accession_number = "ACC123"
+
+        mwl_storage.find_worklist_items.return_value = [worklist_item]
+
+        emulator = ModalityEmulator(mwl_storage, pacs_storage)
+
+        assoc = MagicMock()
+        assoc.is_established = False
+
+        ae = MagicMock()
+        ae.associate.return_value = assoc
+
+        emulator.process_worklist_items(ae)
+
+        assoc.send_c_store.assert_not_called()
+        mwl_storage.update_status.assert_not_called()
+        assoc.release.assert_not_called()
+
+    @patch("modality_emulator.Environment", production=True)
+    def test_main_raises_in_production(self, mock_environment):
+        with pytest.raises(RuntimeError, match="Modality Emulator should not be run in production environment"):
+            main()