Allow MWL C-FIND filter by patient name

Including support for wildcard matching

Author: carlosmartinez

src/services/mwl/c_find.py

@@ -40,20 +40,26 @@ def call(self, event: evt.Event) -> Iterator[Tuple[int, Dataset | None]]:
 
         query_patient_id = identifier.get("PatientID")
         anonymised_patient_id = f"*******{query_patient_id[7:]}" if query_patient_id else "None"
+        query_patient_name = identifier.get("PatientName")
 
         procedure_sequence = identifier.get("ScheduledProcedureStepSequence", [{}])
         query_modality = procedure_sequence[0].get("Modality")
         query_date = procedure_sequence[0].get("ScheduledProcedureStepStartDate")
 
         logger.debug(
-            "Query parameters: modality=%s, date=%s, patient_id=%s", query_modality, query_date, anonymised_patient_id
+            "Query parameters: modality=%s, date=%s, patient_id=%s, patient_name=%s",
+            query_modality,
+            query_date,
+            anonymised_patient_id,
+            query_patient_name,
         )
 
         try:
             items = self.storage.find_worklist_items(
                 modality=query_modality if query_modality else None,
                 scheduled_date=query_date if query_date else None,
                 patient_id=query_patient_id if query_patient_id else None,
+                patient_name=str(query_patient_name) if query_patient_name else None,
             )
 
             logger.info("Found %s matching worklist items", len(items))

src/services/storage.py

@@ -367,6 +367,7 @@ def find_worklist_items(
         modality: Optional[str] = None,
         scheduled_date: Optional[str] = None,
         patient_id: Optional[str] = None,
+        patient_name: Optional[str] = None,
     ) -> List[WorklistItem]:
         """
         Query worklist items with optional filters.
@@ -375,6 +376,7 @@ def find_worklist_items(
             modality: Filter by modality (e.g., "MG")
             scheduled_date: Filter by scheduled date (YYYYMMDD)
             patient_id: Filter by patient ID
+            patient_name: Filter by patient name
 
         Returns:
             List of WorklistItem instances matching the criteria
@@ -400,6 +402,12 @@ def find_worklist_items(
             where_clauses.append("patient_id = ?")
             params.append(patient_id)
 
+        if patient_name:
+            # Convert DICOM wildcards (* → %, ? → _) to SQL LIKE syntax.
+            sql_pattern = patient_name.replace("*", "%").replace("?", "_")
+            where_clauses.append("UPPER(patient_name) LIKE UPPER(?)")
+            params.append(sql_pattern)
+
         if where_clauses:
             query += " WHERE " + " AND ".join(where_clauses)
 

tests/services/mwl/test_c_find.py

@@ -116,7 +116,9 @@ def test_call_with_modality_filter(self, handler, mock_storage, mock_event):
 
         list(handler.call(mock_event))
 
-        mock_storage.find_worklist_items.assert_called_once_with(modality="MG", scheduled_date=None, patient_id=None)
+        mock_storage.find_worklist_items.assert_called_once_with(
+            modality="MG", scheduled_date=None, patient_id=None, patient_name=None
+        )
 
     def test_call_with_date_filter(self, handler, mock_storage, mock_event):
         sps_item = Dataset()
@@ -127,7 +129,7 @@ def test_call_with_date_filter(self, handler, mock_storage, mock_event):
         list(handler.call(mock_event))
 
         mock_storage.find_worklist_items.assert_called_once_with(
-            modality=None, scheduled_date="20260107", patient_id=None
+            modality=None, scheduled_date="20260107", patient_id=None, patient_name=None
         )
 
     def test_call_with_patient_id_filter(self, handler, mock_storage, mock_event):
@@ -137,7 +139,17 @@ def test_call_with_patient_id_filter(self, handler, mock_storage, mock_event):
         list(handler.call(mock_event))
 
         mock_storage.find_worklist_items.assert_called_once_with(
-            modality=None, scheduled_date=None, patient_id="9876543210"
+            modality=None, scheduled_date=None, patient_id="9876543210", patient_name=None
+        )
+
+    def test_call_with_patient_name_filter(self, handler, mock_storage, mock_event):
+        mock_event.identifier.PatientName = "Smith*"
+        mock_storage.find_worklist_items.return_value = []
+
+        list(handler.call(mock_event))
+
+        mock_storage.find_worklist_items.assert_called_once_with(
+            modality=None, scheduled_date=None, patient_id=None, patient_name="Smith*"
         )
 
     def test_call_handles_storage_exception(self, handler, mock_storage, mock_event):

tests/services/test_mwl_storage_patient_name_search.py

@@ -0,0 +1,71 @@
+"""
+Integration-style tests for MWLStorage.find_worklist_items patient name filtering.
+
+These tests use a real SQLite database to verify that wildcard conversion and case-insensitive
+matching work correctly in practice.
+"""
+
+import pytest
+
+from services.storage import MWLStorage, WorklistItem
+
+
+@pytest.fixture
+def storage(tmp_dir):
+    return MWLStorage(f"{tmp_dir}/worklist.db")
+
+
+@pytest.fixture(autouse=True)
+def items(storage):
+    """Insert a small set of worklist items with varied names."""
+    names = [
+        "SMITH^SARITA",
+        "SMITH^JANE",
+        "JONES^SARITA",
+        "MÜLLER^DILMA",
+    ]
+    for i, name in enumerate(names):
+        storage.store_worklist_item(
+            WorklistItem(
+                accession_number=f"ACC{i:03d}",
+                patient_id=f"999000000{i}",
+                patient_name=name,
+                patient_birth_date="19800101",
+                scheduled_date="20260101",
+                scheduled_time="090000",
+                modality="MG",
+            )
+        )
+    return names
+
+
+class TestPatientNameSearch:
+    def test_trailing_wildcard(self, storage):
+        results = storage.find_worklist_items(patient_name="SMITH*")
+        assert {r.patient_name for r in results} == {"SMITH^SARITA", "SMITH^JANE"}
+
+    def test_wildcard_on_given_name(self, storage):
+        results = storage.find_worklist_items(patient_name="*SARITA")
+        assert {r.patient_name for r in results} == {"SMITH^SARITA", "JONES^SARITA"}
+
+    def test_single_character_wildcard(self, storage):
+        results = storage.find_worklist_items(patient_name="SMITH^J?NE")
+        assert {r.patient_name for r in results} == {"SMITH^JANE"}
+
+    def test_exact_match(self, storage):
+        results = storage.find_worklist_items(patient_name="JONES^SARITA")
+        assert len(results) == 1
+        assert results[0].patient_name == "JONES^SARITA"
+
+    def test_no_match(self, storage):
+        results = storage.find_worklist_items(patient_name="BROWN*")
+        assert results == []
+
+    def test_case_insensitive_match(self, storage):
+        results = storage.find_worklist_items(patient_name="smith*")
+        assert {r.patient_name for r in results} == {"SMITH^SARITA", "SMITH^JANE"}
+
+    @pytest.mark.xfail(reason="SQLite's UPPER() is ASCII-only; non-ASCII case folding requires ICU compilation")
+    def test_case_insensitive_non_ascii(self, storage):
+        results = storage.find_worklist_items(patient_name="müller*")
+        assert {r.patient_name for r in results} == {"MÜLLER^DILMA"}

tests/services/test_storage.py

@@ -283,6 +283,49 @@ def test_find_worklist_items_with_filters(self, mock_db, tmp_dir):
             ["MG", "20240101", "999123456"],
         )
 
+        mock_connection.reset_mock()
+        subject.find_worklist_items(patient_name="Smith*")
+
+        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 UPPER(patient_name) LIKE UPPER(?) ORDER BY scheduled_date, scheduled_time"
+            ),
+            ["Smith%"],
+        )
+
+    @pytest.mark.parametrize(
+        "dicom_pattern, sql_pattern",
+        [
+            ("Smith*", "Smith%"),  # trailing wildcard
+            ("*Smith*", "%Smith%"),  # leading and trailing wildcard
+            ("Sm?th*", "Sm_th%"),  # single-character wildcard combined with trailing
+            ("Smith^Jane", "Smith^Jane"),  # exact name, no wildcards
+        ],
+    )
+    def test_find_worklist_items_patient_name_wildcard_conversion(self, mock_db, tmp_dir, dicom_pattern, sql_pattern):
+        mock_cursor = MagicMock()
+        mock_cursor.fetchall.return_value = []
+        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()
+
+        subject.find_worklist_items(patient_name=dicom_pattern)
+
+        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 UPPER(patient_name) LIKE UPPER(?) ORDER BY scheduled_date, scheduled_time"
+            ),
+            [sql_pattern],
+        )
+
     def test_get_worklist_item(self, mock_db, tmp_dir, result):
         mock_cursor = MagicMock()
         mock_cursor.fetchone.return_value = result