Merge pull request #19 from NHSDigital/feat/implement-n-create

Implement N-CREATE and N-SET

Author: steventux

compose.yml

@@ -79,7 +79,8 @@ services:
       - pacs-storage:/var/lib/pacs/storage
       - pacs-db:/var/lib/pacs
     environment:
-      - CLOUD_API_ENDPOINT=${CLOUD_API_ENDPOINT:-http://host.docker.internal:8000/api/dicom/upload/}
+      - CLOUD_API_ENDPOINT=${CLOUD_API_ENDPOINT:-http://host.docker.internal:8000/api/v1/dicom}
+      - CLOUD_API_TOKEN=${CLOUD_API_TOKEN:-testtoken}
       - PACS_DB_PATH=/var/lib/pacs/pacs.db
       - PACS_STORAGE_PATH=/var/lib/pacs/storage
       - MWL_DB_PATH=/var/lib/pacs/worklist.db

src/server.py

@@ -8,11 +8,16 @@
 import logging
 
 from pynetdicom import AE, StoragePresentationContexts, evt
-from pynetdicom.sop_class import ModalityWorklistInformationFind  # type: ignore[attr-defined]
+from pynetdicom.sop_class import (
+    ModalityPerformedProcedureStep,  # type: ignore[attr-defined]
+    ModalityWorklistInformationFind,  # type: ignore[attr-defined]
+)
 
 from services.dicom.c_echo import CEcho
 from services.dicom.c_store import CStore
 from services.mwl.c_find import CFindHandler
+from services.mwl.n_create import NCreate
+from services.mwl.n_set import NSet
 from services.storage import MWLStorage, PACSStorage
 
 logger = logging.getLogger(__name__)
@@ -98,8 +103,13 @@ def start(self):
 
         self.ae = AE(ae_title=self.ae_title)
         self.ae.add_supported_context(ModalityWorklistInformationFind)
+        self.ae.add_supported_context(ModalityPerformedProcedureStep)
 
-        handlers = [(evt.EVT_C_FIND, CFindHandler(self.storage).call)]
+        handlers = [
+            (evt.EVT_C_FIND, CFindHandler(self.storage).call),
+            (evt.EVT_N_CREATE, NCreate(self.storage).call),
+            (evt.EVT_N_SET, NSet(self.storage).call),
+        ]
 
         logger.info(f"MWL server listening on 0.0.0.0:{self.port}")
         logger.info(f"Database: {self.storage.db_path}")

src/services/dicom/__init__.py

@@ -4,3 +4,8 @@
 SUCCESS = 0x0000
 FAILURE = 0xC000
 PENDING = 0xFF00
+INVALID_ATTRIBUTE = 0x0106
+DUPLICATE_SOP_INSTANCE = 0x0111
+UNKNOWN_SOP_INSTANCE = 0x0112
+MISSING_ATTRIBUTE = 0x0120
+PROCESSING_FAILURE = 0x0110

src/services/init_worklist_db.sql

@@ -26,7 +26,7 @@ CREATE TABLE IF NOT EXISTS worklist_items (
     procedure_code TEXT,
 
     -- Status tracking
-    status TEXT DEFAULT 'SCHEDULED' CHECK(status IN ('SCHEDULED', 'IN_PROGRESS', 'COMPLETED', 'DISCONTINUED')),
+    status TEXT DEFAULT 'SCHEDULED' CHECK(status IN ('SCHEDULED', 'IN PROGRESS', 'COMPLETED', 'DISCONTINUED')),
 
     -- DICOM identifiers
     study_instance_uid TEXT,

src/services/mwl/__init__.py

@@ -1,5 +1,10 @@
 """Modality Worklist (MWL) services for DICOM worklist management."""
 
-from services.mwl.c_find import CFindHandler
+from enum import Enum
 
-__all__ = ["CFindHandler"]
+
+class MWLStatus(Enum):
+    SCHEDULED = "SCHEDULED"
+    IN_PROGRESS = "IN PROGRESS"
+    COMPLETED = "COMPLETED"
+    DISCONTINUED = "DISCONTINUED"

src/services/mwl/n_create.py

@@ -0,0 +1,75 @@
+import logging
+
+from pydicom.dataset import Dataset
+from pynetdicom.events import Event
+from pynetdicom.sop_class import ModalityPerformedProcedureStep  # pyright: ignore[reportAttributeAccessIssue]
+
+from services.dicom import DUPLICATE_SOP_INSTANCE, INVALID_ATTRIBUTE, MISSING_ATTRIBUTE, PROCESSING_FAILURE, SUCCESS
+from services.mwl import MWLStatus
+from services.storage import MWLStorage
+
+logger = logging.getLogger(__name__)
+
+
+class NCreate:
+    def __init__(self, storage: MWLStorage):
+        self.storage = storage
+
+    def call(self, event: Event) -> tuple[int, Dataset | None]:
+        """Handle N-CREATE request for MPPS (start of procedure)."""
+        ds = Dataset()
+        try:
+            req = event.request
+
+            affected_sop_instance_uid = getattr(req, "AffectedSOPInstanceUID", None)
+
+            if affected_sop_instance_uid is None:
+                return INVALID_ATTRIBUTE, None
+
+            if self.storage.mpps_instance_exists(affected_sop_instance_uid):
+                return DUPLICATE_SOP_INSTANCE, None
+
+            attr_list = event.attribute_list
+            status = getattr(attr_list, "PerformedProcedureStepStatus", None)
+
+            if not status:
+                logger.warning("MPPS N-CREATE: Missing PerformedProcedureStepStatus in request")
+                return MISSING_ATTRIBUTE, None
+
+            if status.upper() != MWLStatus.IN_PROGRESS.value:
+                logger.warning("MPPS N-CREATE: Invalid PerformedProcedureStepStatus value: %s", status)
+                return INVALID_ATTRIBUTE, None
+
+            ds.SOPClassUID = ModalityPerformedProcedureStep
+            ds.SOPInstanceUID = affected_sop_instance_uid
+            ds.update(attr_list)
+
+            scheduled_step_sequence = getattr(attr_list, "ScheduledStepAttributesSequence", [])
+            if len(scheduled_step_sequence) == 0:
+                logger.warning("MPPS N-CREATE: Missing ScheduledStepAttributesSequence in request")
+                return MISSING_ATTRIBUTE, None
+
+            sps = attr_list.ScheduledStepAttributesSequence[0]
+            accession_number = sps.get("AccessionNumber")
+
+            logger.info("MPPS N-CREATE: Started procedure for Accession Number: %s", accession_number)
+
+            if not accession_number:
+                logger.warning("MPPS N-CREATE: Missing Accession Number in ScheduledStepAttributesSequence")
+                return MISSING_ATTRIBUTE, None
+
+            source_message_id = self.storage.update_status(
+                accession_number, MWLStatus.IN_PROGRESS.value, ds.SOPInstanceUID
+            )
+            if source_message_id:
+                logger.info("Worklist item updated: %s -> %s", accession_number, MWLStatus.IN_PROGRESS.value)
+            else:
+                logger.warning("Could not find accession %s in database", accession_number)
+
+        except Exception as e:
+            logger.error("Error in handle_create: %s", str(e), exc_info=True)
+            return PROCESSING_FAILURE, None
+
+        # Success - return the created dataset
+        logger.info("MPPS N-CREATE successful, returning dataset")
+        return SUCCESS, ds

src/services/mwl/n_set.py

@@ -0,0 +1,59 @@
+import logging
+
+from pydicom import Dataset
+from pynetdicom.events import Event
+from pynetdicom.sop_class import ModalityPerformedProcedureStep  # pyright: ignore[reportAttributeAccessIssue]
+
+from services.dicom import INVALID_ATTRIBUTE, MISSING_ATTRIBUTE, PROCESSING_FAILURE, SUCCESS, UNKNOWN_SOP_INSTANCE
+from services.mwl import MWLStatus
+from services.storage import MWLStorage
+
+logger = logging.getLogger(__name__)
+
+
+class NSet:
+    def __init__(self, storage: MWLStorage):
+        self.storage = storage
+
+    def call(self, event: Event) -> tuple[int, Dataset | None]:
+        try:
+            req = event.request
+            requested_sop_instance_uid = getattr(req, "RequestedSOPInstanceUID", None)
+            logger.info("MPPS N-SET: Received request for SOP Instance UID: %s", requested_sop_instance_uid)
+
+            mod_list = event.attribute_list
+            status = mod_list.get("PerformedProcedureStepStatus")
+            if not status:
+                logger.warning("MPPS N-SET: Missing PerformedProcedureStepStatus in request")
+                return MISSING_ATTRIBUTE, None
+
+            if status not in [MWLStatus.COMPLETED.value, MWLStatus.DISCONTINUED.value]:
+                logger.warning("MPPS N-SET: Invalid PerformedProcedureStepStatus: %s", status)
+                return INVALID_ATTRIBUTE, None
+
+            worklist_item = self.storage.get_worklist_item_by_mpps_instance_uid(requested_sop_instance_uid)
+            if not worklist_item:
+                logger.warning(
+                    "MPPS N-SET: No worklist item found for SOP Instance UID: %s", requested_sop_instance_uid
+                )
+                return UNKNOWN_SOP_INSTANCE, None
+
+            accession_number = worklist_item.accession_number
+
+            source_message_id = self.storage.update_status(accession_number, status)
+            if source_message_id:
+                logger.info("Database updated: %s -> %s", accession_number, status)
+
+                ds = Dataset()
+                ds.SOPClassUID = ModalityPerformedProcedureStep
+                ds.SOPInstanceUID = requested_sop_instance_uid
+                ds.update(mod_list)
+
+                logger.info("MPPS N-SET successful")
+                return SUCCESS, ds
+            else:
+                logger.warning("MPPS N-SET: Failed to update database with new status")
+                return PROCESSING_FAILURE, None
+        except Exception as e:
+            logger.error("Error in handle_set: %s", str(e), exc_info=True)
+            return PROCESSING_FAILURE, None

src/services/storage.py

@@ -7,6 +7,8 @@
 from pathlib import Path
 from typing import Dict, List, Optional
 
+from services.mwl import MWLStatus
+
 logger = logging.getLogger(__name__)
 
 
@@ -288,7 +290,7 @@ class WorklistItem:
     patient_name: str = field(doc="Name of the patient. Lastname^Firstname.")
     scheduled_date: str = field(doc="Date the procedure is scheduled for.")
     scheduled_time: str = field(doc="Time the procedure is scheduled for.")
-    status: str = field(doc="Status of the worklist item", default="SCHEDULED")
+    status: str = field(doc="Status of the worklist item", default=MWLStatus.SCHEDULED.value)
     source_message_id: Optional[str] = field(
         default=None, doc="Message ID from system which created this worklist item", hash=True
     )
@@ -429,7 +431,7 @@ def update_status(
 
         Args:
             accession_number: The accession number to update
-            status: New status (SCHEDULED, IN_PROGRESS, COMPLETED, DISCONTINUED)
+            status: New status (SCHEDULED, IN PROGRESS, COMPLETED, DISCONTINUED)
             mpps_instance_uid: Optional MPPS instance UID
 
         Returns:
@@ -515,3 +517,27 @@ def get_source_message_id(self, accession_number: str) -> Optional[str]:
             )
             row = cursor.fetchone()
             return row["source_message_id"] if row and row["source_message_id"] else None
+
+    def mpps_instance_exists(self, mpps_instance_uid: str) -> bool:
+        """Check if an MPPS instance UID already exists in any worklist item."""
+        with self._get_connection() as conn:
+            cursor = conn.execute("SELECT 1 FROM worklist_items WHERE mpps_instance_uid = ?", (mpps_instance_uid,))
+            return cursor.fetchone() is not None
+
+    def get_worklist_item_by_mpps_instance_uid(self, mpps_instance_uid: str | None) -> Optional[WorklistItem]:
+        """Get a worklist item by its associated MPPS instance UID."""
+        if mpps_instance_uid is None:
+            return None
+
+        with self._get_connection() as conn:
+            cursor = conn.execute(
+                (
+                    "SELECT accession_number, modality, patient_birth_date, patient_id, "
+                    "patient_name, patient_sex, procedure_code, scheduled_date, scheduled_time, "
+                    "source_message_id, study_description, study_instance_uid, status, mpps_instance_uid "
+                    "FROM worklist_items WHERE mpps_instance_uid = ?"
+                ),
+                (mpps_instance_uid,),
+            )
+            row = cursor.fetchone()
+            return WorklistItem(**row) if row else None

tests/integration/test_n_create_updates_worklist_status.py

@@ -0,0 +1,76 @@
+import datetime
+
+import pytest
+from pydicom.dataset import Dataset
+from pydicom.uid import generate_uid
+from pynetdicom import AE
+from pynetdicom.sop_class import ModalityPerformedProcedureStep  # pyright: ignore[reportAttributeAccessIssue]
+
+from server import MWLServer
+from services.dicom import SUCCESS
+from services.storage import MWLStorage, WorklistItem
+
+
+class TestNCreateUpdatesWorklistStatus:
+    @pytest.fixture(autouse=True)
+    def with_mwl_server(self, tmp_dir):
+        server = MWLServer("SCREENING_MWL", 4243, f"{tmp_dir}/test.db", block=False)
+        server.start()
+
+        yield
+
+        server.stop()
+
+    @pytest.fixture
+    def worklist_item(self):
+        return WorklistItem(
+            accession_number="ACC123",
+            patient_id="999123456",
+            patient_name="SMITH^JANE",
+            patient_birth_date="19800101",
+            patient_sex="F",
+            scheduled_date="20240101",
+            scheduled_time="090000",
+            modality="MG",
+            procedure_code="12345-6",
+            study_description="MAMMOGRAPHY SCREENING",
+            study_instance_uid=generate_uid(),
+            source_message_id="MSGID123456",
+        )
+
+    def test_n_create_updates_worklist_status(self, tmp_dir, worklist_item):
+        storage = MWLStorage(f"{tmp_dir}/test.db")
+        study_instance_uid = generate_uid()
+        accession_number = storage.store_worklist_item(worklist_item)
+
+        ae = AE(ae_title="MODALITY_SCU")
+        ae.add_requested_context(ModalityPerformedProcedureStep)
+
+        assoc = ae.associate("localhost", 4243, ae_title="SCREENING_MWL")
+
+        mpps_ds = Dataset()
+        mpps_instance_uid = generate_uid()
+        mpps_ds.SOPClassUID = ModalityPerformedProcedureStep
+        mpps_ds.SOPInstanceUID = mpps_instance_uid
+        mpps_ds.PerformedProcedureStepStatus = "IN PROGRESS"
+        now = datetime.datetime.now()
+        mpps_ds.PerformedProcedureStepStartDate = now.strftime("%Y%m%d")
+        mpps_ds.PerformedProcedureStepStartTime = now.strftime("%H%M%S")
+        mpps_ds.Modality = "MG"
+
+        scheduled_step_seq = Dataset()
+        scheduled_step_seq.StudyInstanceUID = study_instance_uid
+        scheduled_step_seq.AccessionNumber = accession_number
+
+        mpps_ds.ScheduledStepAttributesSequence = [scheduled_step_seq]
+        mpps_ds.PerformedSeriesSequence = []
+
+        response = assoc.send_n_create(mpps_ds, ModalityPerformedProcedureStep, mpps_instance_uid)
+
+        assert response[0].Status == SUCCESS
+
+        updated_item = storage.get_worklist_item(accession_number)
+
+        assert updated_item is not None
+        assert updated_item.mpps_instance_uid == mpps_instance_uid
+        assert updated_item.status == "IN PROGRESS"