feat: add getModalityValues and refactor getPatients for getPatientsB…

…yCollection

src/nbiatoolkit/logger/logger.py

@@ -3,6 +3,7 @@
 import os
 from typing import Optional
 
+
 def setup_logger(
     name: str,
     log_level: str = "INFO",

src/nbiatoolkit/nbia.py

@@ -47,11 +47,13 @@ def __init__(
     def headers(self):
         return self._api_headers
 
-    def query_api(self, endpoint: NBIA_ENDPOINTS, params: dict = {}) -> Union[list, dict, bytes]:
+    def query_api(
+        self, endpoint: NBIA_ENDPOINTS, params: dict = {}
+    ) -> Union[list, dict, bytes]:
         query_url = NBIA_ENDPOINTS.BASE_URL.value + endpoint.value
 
         self.log.debug("Querying API endpoint: %s", query_url)
-        response : requests.Response
+        response: requests.Response
         try:
             response = requests.get(url=query_url, headers=self.headers, params=params)
             response.raise_for_status()  # Raise an HTTPError for bad responses
@@ -70,11 +72,11 @@ def query_api(self, endpoint: NBIA_ENDPOINTS, params: dict = {}) -> Union[list,
 
         try:
             if response.headers.get("Content-Type") == "application/json":
-                response_json : dict | list = response.json()
+                response_json: dict | list = response.json()
                 return response_json
             else:
                 # If response is binary data, return raw response
-                response_data : bytes = response.content
+                response_data: bytes = response.content
                 return response_data
         except JSONDecodeError as j:
             self.log.debug("Response: %s", response.text)
@@ -87,8 +89,6 @@ def query_api(self, endpoint: NBIA_ENDPOINTS, params: dict = {}) -> Union[list,
             self.log.error("Error querying API: %s", e)
             raise e
 
-
-
     def getCollections(self, prefix: str = "") -> Union[list[str], None]:
         response = self.query_api(NBIA_ENDPOINTS.GET_COLLECTIONS)
 
@@ -103,9 +103,77 @@ def getCollections(self, prefix: str = "") -> Union[list[str], None]:
                 collections.append(name)
         return collections
 
+    def getModalityValues(self, Collection: str = "", BodyPartExamined: str = "") -> Union[list[str], None]:
+        PARAMS = self.parsePARAMS(locals())
+
+        response = self.query_api(
+            endpoint=NBIA_ENDPOINTS.GET_MODALITY_VALUES, params=PARAMS
+        )
+
+        if not isinstance(response, list):
+            self.log.error("Expected list, but received: %s", type(response))
+            return None
+
+        modalities = []
+        for modality in response:
+            modalities.append(modality["Modality"])
+        return modalities
+
+    def getPatients(self, Collection: str = "") -> Union[list[dict[str, str]], None]:
+        assert isinstance(Collection, str), "Collection must be a string"
+
+        PARAMS = self.parsePARAMS(locals())
+
+        response = self.query_api(endpoint=NBIA_ENDPOINTS.GET_PATIENTS, params=PARAMS)
+        if not isinstance(response, list):
+            self.log.error("Expected list, but received: %s", type(response))
+            return None
+
+        patientList = []
+        for patient in response:
+            assert isinstance(patient, dict), "Expected dict, but received: %s" % type(
+                patient
+            )
+            assert "PatientId" in patient, "PatientId not in patient dict"
+            assert isinstance(
+                patient["PatientId"], str
+            ), "PatientId must be a string, but received: %s" % type(
+                patient["PatientId"]
+            )
+            patientList.append(
+                {
+                    "PatientId": patient["PatientId"],
+                    "PatientName": patient["PatientName"],
+                    "PatientSex": patient["PatientSex"],
+                    "Collection": patient["Collection"],
+                }
+            )
+
+        return patientList
+
+    def getPatientsByCollectionAndModality(
+        self, Collection: str, Modality: str
+    ) -> Union[list[str], None]:
+        assert Collection is not None
+        assert Modality is not None
+
+        PARAMS = self.parsePARAMS(locals())
+
+        response = self.query_api(
+            endpoint=NBIA_ENDPOINTS.GET_PATIENT_BY_COLLECTION_AND_MODALITY,
+            params=PARAMS,
+        )
+        if not isinstance(response, list):
+            self.log.error("Expected list, but received: %s", type(response))
+            return None
+
+        patientList = [_["PatientId"] for _ in response]
+        return patientList
 
     # returns a list of dictionaries with the collection name and patient count
-    def getCollectionPatientCount(self, prefix: str = "") -> Union[list[dict[str, int]], None]:
+    def getCollectionPatientCount(
+        self, prefix: str = ""
+    ) -> Union[list[dict[str, int]], None]:
         response = self.query_api(NBIA_ENDPOINTS.GET_COLLECTION_PATIENT_COUNT)
 
         if not isinstance(response, list):
@@ -114,17 +182,19 @@ def getCollectionPatientCount(self, prefix: str = "") -> Union[list[dict[str, in
 
         patientCounts = []
         for collection in response:
-            name = collection["criteria"]
-            if name.lower().startswith(prefix.lower()):
+            CollectionName = collection["criteria"]
+            if CollectionName.lower().startswith(prefix.lower()):
                 patientCounts.append(
                     {
-                        "Collection": collection["criteria"],
+                        "Collection": CollectionName,
                         "PatientCount": int(collection["count"]),
                     }
                 )
         return patientCounts
 
-    def getBodyPartCounts(self, Collection: str = "", Modality: str = "") -> Union[list[dict[str, int]], None]:
+    def getBodyPartCounts(
+        self, Collection: str = "", Modality: str = ""
+    ) -> Union[list[dict[str, int]], None]:
         PARAMS = self.parsePARAMS(locals())
 
         response = self.query_api(
@@ -145,23 +215,6 @@ def getBodyPartCounts(self, Collection: str = "", Modality: str = "") -> Union[l
             )
         return bodyparts
 
-    def getPatients(self, Collection: str, Modality: str) -> Union[list[str], None]:
-        assert Collection is not None
-        assert Modality is not None
-
-        PARAMS = self.parsePARAMS(locals())
-
-        response = self.query_api(
-            endpoint=NBIA_ENDPOINTS.GET_PATIENT_BY_COLLECTION_AND_MODALITY,
-            params=PARAMS,
-        )
-        if not isinstance(response, list):
-            self.log.error("Expected list, but received: %s", type(response))
-            return None
-
-        patientList = [_["PatientId"] for _ in response]
-        return patientList
-
     def getSeries(
         self,
         Collection: str = "",
@@ -173,7 +226,6 @@ def getSeries(
         ManufacturerModelName: str = "",
         Manufacturer: str = "",
     ) -> Union[list[dict[str, str]], None]:
-
         PARAMS = self.parsePARAMS(locals())
 
         response = self.query_api(endpoint=NBIA_ENDPOINTS.GET_SERIES, params=PARAMS)
@@ -304,4 +356,4 @@ def parsePARAMS(self, params: dict) -> dict:
     pprint(all)
 
     sub = client.getCollections(prefix="aCrin")
-    pprint(sub)
+    pprint(sub)

src/nbiatoolkit/utils/nbia_endpoints.py

@@ -12,6 +12,9 @@ class NBIA_ENDPOINTS(Enum):
     GET_COLLECTION_PATIENT_COUNT = "getCollectionValuesAndCounts"
     GET_COLLECTION_DESCRIPTIONS = "getCollectionDescriptions"
 
+    GET_MODALITY_VALUES = "v2/getModalityValues"
+
+    GET_PATIENTS = "v2/getPatient"
     GET_PATIENT_BY_COLLECTION_AND_MODALITY = "v2/getPatientByCollectionAndModality"
     GET_BODY_PART_PATIENT_COUNT = "getBodyPartValuesAndCounts"
     GET_NEW_PATIENTS_IN_COLLECTION = "NewPatientsInCollection"

tests/test_nbia.py

@@ -28,54 +28,26 @@ def nbia_collections(nbia_client):
 @pytest.fixture(scope="session")
 def nbia_patients(nbia_client, nbia_collections):
     patients = nbia_client.getPatients(
-        Collection=nbia_collections[0], Modality = "CT")
+        Collection=nbia_collections[0])
     return patients
 
+@pytest.fixture(scope="session")
+def nbia_ModalityValues(nbia_client, nbia_collections):
+    modalities = nbia_client.getModalityValues(
+        Collection=nbia_collections[0])
+    return modalities
 
-def test_downloadSeries(nbia_client, nbia_collections, nbia_patients):
-    seriesList = nbia_client.getSeries(
-        Collection=nbia_collections[0],
-        PatientID=nbia_patients[0],
-        Modality="CT"
-    )
-    filePattern = '%PatientID/%Modality/%SeriesNumber-%SeriesInstanceUID/%InstanceNumber.dcm'
-    # create a temporary directory
-
-    tempdir_ = TemporaryDirectory()
-    tempdir = tempdir_.name
-
-    nbia_client.downloadSeries(
-        SeriesInstanceUID=seriesList[0]["SeriesInstanceUID"],
-        downloadDir=tempdir,
-        filePattern=filePattern
-    )
-    dir = os.listdir(tempdir)
-
-    assert len(dir) == 1
-    assert dir[0] == nbia_patients[0]
-    assert os.path.isdir(os.path.join(tempdir, dir[0]))
-
-    modality_dir = os.listdir(os.path.join(tempdir, dir[0]))
-    assert len(modality_dir) == 1
-    assert modality_dir[0] == "CT"
-    assert os.path.isdir(os.path.join(tempdir, dir[0], modality_dir[0]))
-
-    series_dir = os.listdir(os.path.join(tempdir, dir[0], modality_dir[0]))
-    assert len(series_dir) == 1
-    # only last 5 digits of SeriesInstanceUID are used
-    assert series_dir[0] == "{}-{}".format(
-        seriesList[0]["SeriesNumber"], seriesList[0]["SeriesInstanceUID"][-5:])
-    assert os.path.isdir(os.path.join(tempdir, dir[0], modality_dir[0], series_dir[0]))
-
-    dicom_dir = os.listdir(os.path.join(
-        tempdir, dir[0], modality_dir[0], series_dir[0]))
-
-    assert len(dicom_dir) == int(seriesList[0]["ImageCount"])
-    for file in dicom_dir:
-        assert file.endswith(".dcm")
-        assert file[:-4].isdigit()
-
+@pytest.fixture(scope="session")
+def nbia_patientsByCollectionAndModality(nbia_client, nbia_collections):
+    patients = nbia_client.getPatientsByCollectionAndModality(
+        Collection=nbia_collections[0], Modality = "CT")
+    return patients
 
+def test_getModalityValues(nbia_ModalityValues):
+    assert isinstance(nbia_ModalityValues, list)
+    assert len(nbia_ModalityValues) > 0
+    assert isinstance(nbia_ModalityValues[0], str)
+    assert len(nbia_ModalityValues[0]) > 0
 
 def test_nbiaclient_access_token(nbia_client):
     assert nbia_client.headers is not None
@@ -114,8 +86,17 @@ def test_getCollectionPatientCount(nbia_client):
 def test_getPatients(nbia_patients):
     assert isinstance(nbia_patients, list)
     assert len(nbia_patients) > 0
-    assert isinstance(nbia_patients[0], str)
-    assert len(nbia_patients[0]) > 0
+    assert isinstance(nbia_patients[0], dict)
+    assert "PatientId" in nbia_patients[0]
+    assert "PatientName" in nbia_patients[0]
+    assert "Collection" in nbia_patients[0]
+    assert "PatientSex" in nbia_patients[0]
+
+def test_getPatientsByCollectionAndModality(nbia_patientsByCollectionAndModality):
+    assert isinstance(nbia_patientsByCollectionAndModality, list)
+    assert len(nbia_patientsByCollectionAndModality) > 0
+    assert isinstance(nbia_patientsByCollectionAndModality[0], str)
+    assert len(nbia_patientsByCollectionAndModality[0]) > 0
 
 def test_getSeries(nbia_client, nbia_collections, nbia_patients):
     seriesList = nbia_client.getSeries(
@@ -144,3 +125,4 @@ def test_fail_getSeries(nbia_client, nbia_collections, nbia_patients):
 
 
     # nbia_client.downloadSeries(
+