Add NSet service command

Author: steventux

src/services/dicom/__init__.py

@@ -6,5 +6,6 @@
 PENDING = 0xFF00
 INVALID_ATTRIBUTE = 0x0106
 DUPLICATE_SOP_INSTANCE = 0x0111
+UNKNOWN_SOP_INSTANCE = 0x0112
 MISSING_ATTRIBUTE = 0x0120
 PROCESSING_FAILURE = 0x0110

src/services/mwl/n_set.py

@@ -0,0 +1,52 @@
+import logging
+
+from pydicom import Dataset
+from pynetdicom.events import Event
+from pynetdicom.sop_class import ModalityPerformedProcedureStep  # pyright: ignore[reportAttributeAccessIssue]
+
+from services.dicom import PROCESSING_FAILURE, SUCCESS, UNKNOWN_SOP_INSTANCE
+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(f"MPPS N-SET: Received request for SOP Instance UID: {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 PROCESSING_FAILURE, None
+
+            worklist_item = self.storage.get_worklist_item_by_mpps_instance_uid(requested_sop_instance_uid)
+            if not worklist_item:
+                logger.warning(f"MPPS N-SET: No worklist item found for SOP Instance UID: {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(f"Database updated: {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(f"Error in handle_set: {str(e)}", exc_info=True)
+            return PROCESSING_FAILURE, None

src/services/storage.py

@@ -521,3 +521,18 @@ def mpps_instance_exists(self, mpps_instance_uid: str) -> bool:
         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) -> Optional[WorklistItem]:
+        """Get a worklist item by its associated MPPS instance UID."""
+        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/services/mwl/test_n_set.py

@@ -0,0 +1,95 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+from pydicom.dataset import Dataset
+from pydicom.uid import generate_uid
+from pynetdicom.sop_class import ModalityPerformedProcedureStep  # pyright: ignore[reportAttributeAccessIssue]
+
+from services.dicom import (
+    PROCESSING_FAILURE,
+    SUCCESS,
+    UNKNOWN_SOP_INSTANCE,
+)
+from services.mwl.n_set import NSet
+
+
+@patch(f"{NSet.__module__}.MWLStorage")
+class TestNSet:
+    @pytest.fixture
+    def event(self):
+        event = MagicMock()
+        event.request = MagicMock()
+        event.attribute_list = Dataset()
+        return event
+
+    @pytest.fixture
+    def requested_sop_instance_uid(self):
+        return generate_uid()
+
+    def test_missing_status_returns_processing_failure(self, mock_storage, event):
+        event.request.RequestedSOPInstanceUID = generate_uid()
+
+        # No PerformedProcedureStepStatus set
+        status, ds = NSet(mock_storage).call(event)
+
+        assert status == PROCESSING_FAILURE
+        assert ds is None
+
+    def test_unknown_sop_instance_returns_unknown(self, mock_storage, event, requested_sop_instance_uid):
+        event.request.RequestedSOPInstanceUID = requested_sop_instance_uid
+        event.attribute_list.PerformedProcedureStepStatus = "IN PROGRESS"
+
+        mock_storage.get_worklist_item_by_mpps_instance_uid.return_value = None
+
+        status, ds = NSet(mock_storage).call(event)
+
+        assert status == UNKNOWN_SOP_INSTANCE
+        assert ds is None
+        mock_storage.get_worklist_item_by_mpps_instance_uid.assert_called_once_with(requested_sop_instance_uid)
+
+    def test_database_update_failure_returns_processing_failure(self, mock_storage, event, requested_sop_instance_uid):
+        event.request.RequestedSOPInstanceUID = requested_sop_instance_uid
+        event.attribute_list.PerformedProcedureStepStatus = "COMPLETED"
+
+        worklist_item = MagicMock()
+        worklist_item.accession_number = "ACC123"
+        mock_storage.get_worklist_item_by_mpps_instance_uid.return_value = worklist_item
+
+        mock_storage.update_status.return_value = None
+
+        status, ds = NSet(mock_storage).call(event)
+
+        assert status == PROCESSING_FAILURE
+        assert ds is None
+        mock_storage.update_status.assert_called_once_with("ACC123", "COMPLETED")
+
+    def test_successful_nset_returns_success_and_dataset(self, mock_storage, event, requested_sop_instance_uid):
+        event.request.RequestedSOPInstanceUID = requested_sop_instance_uid
+        event.attribute_list.PerformedProcedureStepStatus = "COMPLETED"
+
+        worklist_item = MagicMock()
+        worklist_item.accession_number = "ACC123"
+        mock_storage.get_worklist_item_by_mpps_instance_uid.return_value = worklist_item
+
+        mock_storage.update_status.return_value = 1001  # mock message id
+
+        status, ds = NSet(mock_storage).call(event)
+
+        assert status == SUCCESS
+        assert isinstance(ds, Dataset)
+        assert ds.SOPClassUID == ModalityPerformedProcedureStep
+        assert ds.SOPInstanceUID == requested_sop_instance_uid
+        assert ds.PerformedProcedureStepStatus == "COMPLETED"
+
+        mock_storage.update_status.assert_called_once_with("ACC123", "COMPLETED")
+
+    def test_exception_returns_processing_failure(self, mock_storage, event):
+        event.request.RequestedSOPInstanceUID = generate_uid()
+        event.attribute_list.PerformedProcedureStepStatus = "COMPLETED"
+
+        mock_storage.get_worklist_item_by_mpps_instance_uid.side_effect = Exception("DB error")
+
+        status, ds = NSet(mock_storage).call(event)
+
+        assert status == PROCESSING_FAILURE
+        assert ds is None

tests/services/test_storage.py

@@ -494,3 +494,39 @@ def test_mpps_instance_not_exists(self, mock_db, tmp_dir):
         mock_connection.reset_mock()
 
         assert subject.mpps_instance_exists(generate_uid()) is False
+
+    def test_get_worklist_item_by_mpps_instance_uid(self, mock_db, tmp_dir, result):
+        mpps_instance_uid = "some-mpps-uid"
+        mock_cursor = MagicMock()
+        mock_cursor.fetchone.return_value = result
+        mock_connection = MagicMock()
+        mock_connection.execute.return_value = mock_cursor
+        mock_db.connect.return_value = mock_connection
+
+        subject = MWLStorage(tmp_dir)
+        mock_connection.reset_mock()
+
+        worklist_item = subject.get_worklist_item_by_mpps_instance_uid(mpps_instance_uid)
+
+        mock_connection.execute.assert_called_once_with(
+            (
+                "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,),
+        )
+        assert worklist_item == WorklistItem(**result)
+
+    def test_get_worklist_item_by_mpps_instance_uid_returns_none(self, mock_db, tmp_dir):
+        mpps_instance_uid = "some-mpps-uid"
+        mock_cursor = MagicMock()
+        mock_cursor.fetchone.return_value = None
+        mock_connection = MagicMock()
+        mock_connection.execute.return_value = mock_cursor
+        mock_db.connect.return_value = mock_connection
+
+        subject = MWLStorage(tmp_dir)
+
+        assert subject.get_worklist_item_by_mpps_instance_uid(mpps_instance_uid) is None