C-STORE validation failure notifies Manage API

Via ValidationFailureNotifier#notify

Author: carlosmartinez

src/pacs_main.py

@@ -29,8 +29,9 @@ def main():
     pacs_port = int(os.getenv("PACS_PORT", "4244"))
     pacs_storage_path = os.getenv("PACS_STORAGE_PATH", "/var/lib/pacs/storage")
     pacs_db_path = os.getenv("PACS_DB_PATH", "/var/lib/pacs/pacs.db")
+    mwl_db_path = os.getenv("MWL_DB_PATH", "/var/lib/pacs/worklist.db")
 
-    pacs_server = PACSServer(pacs_aet, pacs_port, pacs_storage_path, pacs_db_path, block=True)
+    pacs_server = PACSServer(pacs_aet, pacs_port, pacs_storage_path, pacs_db_path, block=True, mwl_db_path=mwl_db_path)
 
     try:
         pacs_server.start()

src/server.py

@@ -15,6 +15,7 @@
 
 from services.dicom.c_echo import CEcho
 from services.dicom.c_store import CStore
+from services.dicom.validation_failure_notifier import ValidationFailureNotifier
 from services.mwl.c_find import CFind
 from services.mwl.n_create import NCreate
 from services.mwl.n_set import NSet
@@ -33,6 +34,7 @@ def __init__(
         storage_path: str = "/var/lib/pacs/storage",
         db_path: str = "/var/lib/pacs/pacs.db",
         block: bool = True,
+        mwl_db_path: str = "/var/lib/pacs/worklist.db",
     ):
         """
         Initialize PACS server.
@@ -42,10 +44,13 @@ def __init__(
             port: Port to listen on
             storage_path: Directory for DICOM file storage
             db_path: Path to SQLite database
+            mwl_db_path: Path to the MWL SQLite database (for failure notification lookups)
         """
         self.ae_title = ae_title
         self.port = port
         self.storage = PACSStorage(db_path, storage_path)
+        self.mwl_storage = MWLStorage(mwl_db_path)
+        self.notifier = ValidationFailureNotifier()
         self.ae = None
         self.block = block
 
@@ -56,7 +61,10 @@ def start(self):
         self.ae = AE(ae_title=self.ae_title)
         self.ae.supported_contexts = StoragePresentationContexts
 
-        handlers = [(evt.EVT_C_ECHO, CEcho().call), (evt.EVT_C_STORE, CStore(self.storage).call)]
+        handlers = [
+            (evt.EVT_C_ECHO, CEcho().call),
+            (evt.EVT_C_STORE, CStore(self.storage, mwl_storage=self.mwl_storage, notifier=self.notifier).call),
+        ]
 
         logger.info(f"PACS server listening on 0.0.0.0:{self.port}")
         logger.info(f"Storage: {self.storage.storage_root}")

src/services/dicom/c_store.py

@@ -10,8 +10,9 @@
 
 from services.dicom import FAILURE, SUCCESS
 from services.dicom.image_compressor import ImageCompressor
+from services.dicom.validation_failure_notifier import ValidationFailureNotifier
 from services.dicom.validator import DicomValidationError, DicomValidator
-from services.storage import InstanceExistsError, PACSStorage
+from services.storage import InstanceExistsError, MWLStorage, PACSStorage
 
 logger = logging.getLogger(__name__)
 
@@ -27,10 +28,14 @@ def __init__(
         storage: PACSStorage,
         compressor: ImageCompressor | None = None,
         validator: DicomValidator | None = None,
+        mwl_storage: MWLStorage | None = None,
+        notifier: ValidationFailureNotifier | None = None,
     ):
         self.storage = storage
         self.compressor = compressor or ImageCompressor()
         self.validator = validator or DicomValidator()
+        self.mwl_storage = mwl_storage
+        self.notifier = notifier
 
     def call(self, event: Event) -> int:
         try:
@@ -42,24 +47,27 @@ def call(self, event: Event) -> int:
                 return FAILURE
 
             sop_instance_uid = ds.get("SOPInstanceUID", "")
+            accession_number = ds.get("AccessionNumber", "")
+            patient_id = ds.get("PatientID")
+            patient_name = str(ds.get("PatientName", ""))
+
             if not sop_instance_uid:
                 logger.error("Missing SOPInstanceUID")
+                self._notify_failure(accession_number, "Missing SOPInstanceUID")
                 return FAILURE
 
-            patient_id = ds.get("PatientID")
             if not patient_id:
                 logger.error("Missing PatientID")
+                self._notify_failure(accession_number, "Missing PatientID")
                 return FAILURE
 
-            accession_number = ds.get("AccessionNumber", "")
-            patient_name = str(ds.get("PatientName", ""))
-
             # Validate dataset before compression
             try:
                 self.validator.validate_dataset(ds)
                 self.validator.validate_pixel_data(ds)
             except DicomValidationError as e:
                 logger.error(f"DICOM validation failed: {e}")
+                self._notify_failure(accession_number, f"DICOM validation failed: {e}")
                 return FAILURE
 
             # Compress dataset before storing
@@ -71,6 +79,7 @@ def call(self, event: Event) -> int:
                 self.validator.validate_bytes(dicom_bytes)
             except DicomValidationError as e:
                 logger.error(f"Serialized DICOM invalid: {e}")
+                self._notify_failure(accession_number, f"Serialized DICOM invalid: {e}")
                 return FAILURE
 
             self.storage.store_instance(
@@ -100,3 +109,16 @@ def dataset_to_bytes(self, ds: Dataset) -> bytes:
             dcmwrite(buffer, ds, enforce_file_format=True)
             buffer.seek(0)
             return buffer.read()
+
+    def _notify_failure(self, accession_number: str, error: str) -> None:
+        if not self.mwl_storage or not self.notifier:
+            return
+
+        source_message_id = self.mwl_storage.get_source_message_id(accession_number)
+        if not source_message_id:
+            logger.warning(
+                f"Cannot report validation failure: no worklist item found for accession {accession_number!r}"
+            )
+            return
+
+        self.notifier.notify(source_message_id, error)

src/services/dicom/validation_failure_notifier.py

@@ -0,0 +1,52 @@
+"""Notifier for DICOM C-STORE validation failures.
+
+Reports validation failures to the Manage Breast Screening HTTP API.
+"""
+
+import logging
+import os
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+
+class ValidationFailureNotifier:
+    def __init__(self, api_endpoint: str | None = None, timeout: int = 30, verify_ssl: bool = True):
+        self.api_endpoint = api_endpoint or os.getenv("CLOUD_API_ENDPOINT", "http://localhost:8000/api/v1/dicom")
+        self.timeout = timeout
+        self.verify_ssl = verify_ssl
+
+    def headers(self) -> dict:
+        return {
+            "Authorization": f"Bearer {os.getenv('CLOUD_API_TOKEN', '')}",
+        }
+
+    def notify(self, source_message_id: str, error: str) -> bool:
+        try:
+            logger.info(f"Reporting validation failure for action {source_message_id}")
+
+            response = requests.post(
+                f"{self.api_endpoint}/{source_message_id}/failure",
+                json={"error": error},
+                timeout=self.timeout,
+                verify=self.verify_ssl,
+                headers=self.headers(),
+            )
+
+            if response.status_code == 200:
+                logger.info(f"Validation failure reported for action {source_message_id}")
+                return True
+            else:
+                logger.error(
+                    f"Failed to report validation failure for {source_message_id}: "
+                    f"status {response.status_code}, body: {response.text}"
+                )
+                return False
+
+        except requests.exceptions.Timeout:
+            logger.error(f"Timeout reporting validation failure for {source_message_id} after {self.timeout}s")
+            return False
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Error reporting validation failure for {source_message_id}: {e}", exc_info=True)
+            return False

tests/services/dicom/test_c_store.py

@@ -8,6 +8,9 @@
 from services.dicom import FAILURE, SUCCESS
 from services.dicom.c_store import CStore
 from services.dicom.image_compressor import ImageCompressor
+from services.dicom.validation_failure_notifier import ValidationFailureNotifier
+from services.dicom.validator import DicomValidationError, DicomValidator
+from services.storage import MWLStorage
 
 
 class TestCStore:
@@ -104,3 +107,34 @@ def test_compression_applied_on_storage(self, mock_storage, mock_event):
         stored_bytes = mock_storage.store_instance.call_args[0][1]
         stored_ds = pydicom.dcmread(BytesIO(stored_bytes), force=True)
         assert stored_ds.file_meta.TransferSyntaxUID == JPEG2000
+
+    def test_validation_failure_notifies_manage(self, mock_storage, mock_event):
+        """When validation fails and accession is in MWL, notify manage."""
+        mock_validator = Mock(spec=DicomValidator)
+        mock_validator.validate_dataset.side_effect = DicomValidationError("Missing required tag")
+
+        mock_mwl = Mock(spec=MWLStorage)
+        mock_mwl.get_source_message_id.return_value = "action-uuid-123"
+
+        mock_notifier = Mock(spec=ValidationFailureNotifier)
+
+        subject = CStore(mock_storage, validator=mock_validator, mwl_storage=mock_mwl, notifier=mock_notifier)
+        assert subject.call(mock_event) == FAILURE
+
+        mock_notifier.notify.assert_called_once_with("action-uuid-123", "DICOM validation failed: Missing required tag")
+        mock_mwl.get_source_message_id.assert_called_once_with("ABC123")
+
+    def test_validation_failure_accession_not_in_mwl(self, mock_storage, mock_event):
+        """When accession is not in MWL, validation failure returns FAILURE without calling notify."""
+        mock_validator = Mock(spec=DicomValidator)
+        mock_validator.validate_dataset.side_effect = DicomValidationError("Missing required tag")
+
+        mock_mwl = Mock(spec=MWLStorage)
+        mock_mwl.get_source_message_id.return_value = None
+
+        mock_notifier = Mock(spec=ValidationFailureNotifier)
+
+        subject = CStore(mock_storage, validator=mock_validator, mwl_storage=mock_mwl, notifier=mock_notifier)
+        assert subject.call(mock_event) == FAILURE
+
+        mock_notifier.notify.assert_not_called()

tests/test_server.py

@@ -10,34 +10,41 @@
 from server import MWLServer, PACSServer
 
 
+@patch(f"{PACSServer.__module__}.MWLStorage")
 @patch(f"{PACSServer.__module__}.PACSStorage")
 class TestPACSServer:
-    def test_init(self, mock_storage, tmp_dir):
-        subject = PACSServer("Custom AE Title", 2222, tmp_dir, f"{tmp_dir}/test.db", False)
+    def test_init(self, mock_pacs_storage, mock_mwl_storage, tmp_dir):
+        subject = PACSServer(
+            "Custom AE Title", 2222, tmp_dir, f"{tmp_dir}/test.db", False, mwl_db_path=f"{tmp_dir}/worklist.db"
+        )
 
         assert subject.ae_title == "Custom AE Title"
         assert subject.port == 2222
-        assert subject.storage == mock_storage.return_value
+        assert subject.storage == mock_pacs_storage.return_value
+        assert subject.mwl_storage == mock_mwl_storage.return_value
         assert subject.ae is None
         assert subject.block is False
 
-        mock_storage.assert_called_once_with(f"{tmp_dir}/test.db", tmp_dir)
+        mock_pacs_storage.assert_called_once_with(f"{tmp_dir}/test.db", tmp_dir)
+        mock_mwl_storage.assert_called_once_with(f"{tmp_dir}/worklist.db")
 
-    def test_init_defaults(self, mock_storage):
+    def test_init_defaults(self, mock_pacs_storage, mock_mwl_storage):
         subject = PACSServer()
 
         assert subject.ae_title == "SCREENING_PACS"
         assert subject.port == 4244
-        assert subject.storage == mock_storage.return_value
+        assert subject.storage == mock_pacs_storage.return_value
+        assert subject.mwl_storage == mock_mwl_storage.return_value
         assert subject.ae is None
         assert subject.block is True
 
-        mock_storage.assert_called_once_with("/var/lib/pacs/pacs.db", "/var/lib/pacs/storage")
+        mock_pacs_storage.assert_called_once_with("/var/lib/pacs/pacs.db", "/var/lib/pacs/storage")
+        mock_mwl_storage.assert_called_once_with("/var/lib/pacs/worklist.db")
 
     @patch(f"{PACSServer.__module__}.AE")
     @patch(f"{PACSServer.__module__}.CEcho")
     @patch(f"{PACSServer.__module__}.CStore")
-    def test_start(self, mock_c_store, mock_c_echo, mock_ae, _):
+    def test_start(self, mock_c_store, mock_c_echo, mock_ae, _mock_pacs_storage, _mock_mwl_storage):
         subject = PACSServer()
         subject.start()