diff --git a/src/pacs_main.py b/src/pacs_main.py index b0df9320..5dfc7640 100644 --- a/src/pacs_main.py +++ b/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() diff --git a/src/server.py b/src/server.py index ece83807..1f327a60 100644 --- a/src/server.py +++ b/src/server.py @@ -33,6 +33,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 +43,12 @@ 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.ae = None self.block = block @@ -56,7 +59,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).call), + ] logger.info(f"PACS server listening on 0.0.0.0:{self.port}") logger.info(f"Storage: {self.storage.storage_root}") diff --git a/src/services/dicom/c_store.py b/src/services/dicom/c_store.py index 899d102d..acfe45ad 100644 --- a/src/services/dicom/c_store.py +++ b/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 or ValidationFailureNotifier() 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) diff --git a/src/services/dicom/validation_failure_notifier.py b/src/services/dicom/validation_failure_notifier.py new file mode 100644 index 00000000..b8bc2822 --- /dev/null +++ b/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.patch( + 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 diff --git a/tests/integration/test_end_to_end_relay_to_upload.py b/tests/integration/test_end_to_end_relay_to_upload.py index 2828ea2c..a699041c 100644 --- a/tests/integration/test_end_to_end_relay_to_upload.py +++ b/tests/integration/test_end_to_end_relay_to_upload.py @@ -85,12 +85,13 @@ def mwl_server(self, mwl_storage): return server @pytest.fixture - def pacs_server(self, pacs_storage): + def pacs_server(self, pacs_storage, mwl_storage): """PACS server using the shared storage.""" server = PACSServer.__new__(PACSServer) server.ae_title = "SCREENING_PACS" server.port = 4244 server.storage = pacs_storage + server.mwl_storage = mwl_storage server.ae = None server.block = False return server diff --git a/tests/integration/test_send_c_store_to_gateway.py b/tests/integration/test_send_c_store_to_gateway.py index 707e0e5f..8e7c69ac 100644 --- a/tests/integration/test_send_c_store_to_gateway.py +++ b/tests/integration/test_send_c_store_to_gateway.py @@ -12,7 +12,9 @@ class TestSendCStoreToGateway: @pytest.fixture(autouse=True) def with_pacs_server(self, tmp_dir): - server = PACSServer("SCREENING_PACS", 4244, tmp_dir, f"{tmp_dir}/test.db", block=False) + server = PACSServer( + "SCREENING_PACS", 4244, tmp_dir, f"{tmp_dir}/test.db", block=False, mwl_db_path=f"{tmp_dir}/worklist.db" + ) server.start() yield diff --git a/tests/services/dicom/test_c_store.py b/tests/services/dicom/test_c_store.py index 679cd0fd..11e0df5d 100644 --- a/tests/services/dicom/test_c_store.py +++ b/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() diff --git a/tests/test_server.py b/tests/test_server.py index 173a492a..23a76197 100644 --- a/tests/test_server.py +++ b/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()