Merge pull request #59 from jjjermiah/47-feature-refactor-getpatients…

…-and-add-getpatient

47 feature refactor getpatients and add getpatient

CHANGELOG.md

@@ -2,6 +2,29 @@
 
 
 
+## v0.10.9 (2024-01-28)
+
+### Build
+
+* build(GHA): separate Codecov ([`9368df9`](https://github.com/jjjermiah/nbia-toolkit/commit/9368df94ae88f005023357153d4c25cb29e33cee))
+
+### Chore
+
+* chore: remove excess comments ([`2e52ef4`](https://github.com/jjjermiah/nbia-toolkit/commit/2e52ef49b3c3558e725e9cde2cc1c3ab220e4dc4))
+
+### Fix
+
+* fix(gha): invalid workflow error ([`1ba3d1e`](https://github.com/jjjermiah/nbia-toolkit/commit/1ba3d1e8804dc61a00c1ebca9867183e129a8efb))
+
+### Unknown
+
+* Merge pull request #56 from jjjermiah/development
+
+Development ([`fb09228`](https://github.com/jjjermiah/nbia-toolkit/commit/fb09228e9096a5987627bd23d6cb0768ff451e1d))
+
+* Merge branch 'main' of https://github.com/jjjermiah/NBIA-toolkit into development ([`a7c7518`](https://github.com/jjjermiah/nbia-toolkit/commit/a7c751887f6e9d91e5c8e6d41ef639c6a8e1890c))
+
+
 ## v0.10.8 (2024-01-28)
 
 ### Build

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nbiatoolkit"
-version = "0.10.8"
+version = "0.10.9"
 description = "A python package to query the National Biomedical Imaging Archive (NBIA) database."
 authors = ["Jermiah Joseph"]
 license = "MIT"

setup.py

@@ -1,6 +1,6 @@
 from setuptools import setup
 
-__version__ = "0.10.8"
+__version__ = "0.10.9"
 
 setup(
     name="nbiatoolkit",

src/nbiatoolkit/__init__.py

@@ -7,7 +7,7 @@
 # this file is used to define the __all__ variable
 
 # set __version__ variable
-__version__ = "0.10.8"
+__version__ = "0.10.9"
 
 # import the modules
 from .nbia import NBIAClient

src/nbiatoolkit/auth.py

@@ -93,7 +93,7 @@ def __init__(
         self.refresh_expiry = None
         self.scope = None
 
-    def getToken(self) -> Union[dict, int]:
+    def getToken(self) -> Union[dict, None]:
         """
         Retrieves the access token from the API.
 
@@ -112,7 +112,7 @@ def getToken(self) -> Union[dict, int]:
         """
         # Check if the access token is valid and not expired
         if self.access_token is not None:
-            return 401 if self.access_token == -1 else self.access_token
+            return None if self.access_token == None else self.access_token
 
         # Prepare the request data
         data = {
@@ -129,7 +129,7 @@ def getToken(self) -> Union[dict, int]:
             response = requests.post(token_url, data=data)
             response.raise_for_status()  # Raise an HTTPError for bad responses
         except requests.exceptions.RequestException as e:
-            self.access_token = -1
+            self.access_token = None
             raise requests.exceptions.RequestException(
                 f"Failed to get access token. Status code:\
                     {response.status_code}"

src/nbiatoolkit/dicomsort/dicomsort.py

@@ -1,17 +1,19 @@
 import re, os, sys, shutil
 import pydicom
 
-from pydicom.filereader import InvalidDicomError
+from pydicom.errors import InvalidDicomError
 
 from .helper_functions import parseDICOMKeysFromFormat, sanitizeFileName, truncateUID
 
+from typing import Optional
+
 
 class DICOMSorter:
     def __init__(
         self,
         sourceDir: str,
         destinationDir: str,
-        targetPattern: str = None,
+        targetPattern: str = "%PatientName/%SeriesNumber-%SeriesInstanceUID/%InstanceNumber.dcm",
         truncateUID: bool = True,
         sanitizeFilename: bool = True,
     ):
@@ -24,9 +26,7 @@ def __init__(
         self.truncateUID = truncateUID
         self.sanitizeFilename = sanitizeFilename
 
-    def generateFilePathFromDICOMAttributes(
-        self, dataset: pydicom.dataset.FileDataset
-    ) -> str:
+    def generateFilePathFromDICOMAttributes(self, dataset: pydicom.FileDataset) -> str:
         """
         Generate a file path for the DICOM file by formatting DICOM attributes.
         """
@@ -57,7 +57,9 @@ def sortSingleDICOMFile(
         assert option in ["copy", "move"], "Invalid option: symlink not implemented yet"
 
         try:
-            dataset = pydicom.dcmread(filePath, stop_before_pixels=True)
+            dataset: pydicom.FileDataset = pydicom.dcmread(
+                filePath, stop_before_pixels=True
+            )
         except InvalidDicomError as e:
             print(f"Error reading file {filePath}: {e}")
             return False

src/nbiatoolkit/logger/logger.py

@@ -1,14 +1,15 @@
 import logging
 from logging.handlers import TimedRotatingFileHandler
 import os
+from typing import Optional
 
 
 def setup_logger(
     name: str,
     log_level: str = "INFO",
     console_logging: bool = False,
-    log_file: str = None,
-    log_dir: str = None,
+    log_file: Optional[str] = None,
+    log_dir: Optional[str] = None,
     log_format: str = "%(asctime)s | %(name)s | %(levelname)s | %(message)s",
     datefmt: str = "%y-%m-%d %H:%M",
 ) -> logging.Logger:

src/nbiatoolkit/nbia.py

@@ -35,88 +35,186 @@ def __init__(
         self.log.debug("Setting up OAuth2 client... with username %s", username)
 
         self._oauth2_client = OAuth2(username=username, password=password)
-        self._api_headers = self._oauth2_client.getToken()
+
+        try:
+            self._api_headers = self._oauth2_client.getToken()
+        except Exception as e:
+            self.log.error("Error retrieving access token: %s", e)
+            self._api_headers = None
+            raise e
 
     @property
     def headers(self):
         return self._api_headers
 
-    def query_api(self, endpoint: NBIA_ENDPOINTS, params: dict = {}) -> dict:
+    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
         try:
             response = requests.get(url=query_url, headers=self.headers, params=params)
+            response.raise_for_status()  # Raise an HTTPError for bad responses
+        except requests.exceptions.RequestException as e:
+            self.log.error("Error querying API: %s", e)
+            raise e
+
+        if response.status_code != 200:
+            self.log.error(
+                "Error querying API: %s %s", response.status_code, response.reason
+            )
+            raise requests.exceptions.RequestException(
+                f"Failed to get access token. Status code:\
+                    {response.status_code}"
+            )
+
+        try:
             if response.headers.get("Content-Type") == "application/json":
-                response_data = response.json()
+                response_json: dict | list = response.json()
+                return response_json
             else:
                 # If response is binary data, return raw response
-                response_data = response.content
+                response_data: bytes = response.content
+                return response_data
         except JSONDecodeError as j:
+            self.log.debug("Response: %s", response.text)
             if response.text == "":
                 self.log.error("Response text is empty.")
-                return response
-            self.log.error("Error parsing response as JSON: %s", j)
-            self.log.debug("Response: %s", response.text)
+            else:
+                self.log.error("Error parsing response as JSON: %s", j)
+            raise j
         except Exception as e:
             self.log.error("Error querying API: %s", e)
             raise e
 
-        return response_data
-
-    def getCollections(self, prefix: str = "") -> list[dict[str]]:
+    def getCollections(self, prefix: str = "") -> Union[list[str], None]:
         response = self.query_api(NBIA_ENDPOINTS.GET_COLLECTIONS)
+
+        if not isinstance(response, list):
+            self.log.error("Expected list, but received: %s", type(response))
+            return None
+
         collections = []
         for collection in response:
             name = collection["Collection"]
             if name.lower().startswith(prefix.lower()):
                 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 = "") -> list[dict[str, int]]:
+    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):
+            self.log.error("Expected list, but received: %s", type(response))
+            return None
+
         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 = "") -> list:
+    def getBodyPartCounts(
+        self, Collection: str = "", Modality: str = ""
+    ) -> Union[list[dict[str, int]], None]:
         PARAMS = self.parsePARAMS(locals())
 
         response = self.query_api(
             endpoint=NBIA_ENDPOINTS.GET_BODY_PART_PATIENT_COUNT, params=PARAMS
         )
 
+        if not isinstance(response, list):
+            self.log.error("Expected list, but received: %s", type(response))
+            return None
+
         bodyparts = []
         for bodypart in response:
             bodyparts.append(
-                {"BodyPartExamined": bodypart["criteria"], "Count": bodypart["count"]}
+                {
+                    "BodyPartExamined": bodypart["criteria"],
+                    "Count": int(bodypart["count"]),
+                }
             )
         return bodyparts
 
-    def getPatients(self, Collection: str, Modality: str) -> list:
-        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,
-        )
-
-        patientList = [_["PatientId"] for _ in response]
-        return patientList
-
     def getSeries(
         self,
         Collection: str = "",
@@ -127,15 +225,15 @@ def getSeries(
         BodyPartExamined: str = "",
         ManufacturerModelName: str = "",
         Manufacturer: str = "",
-    ) -> list:
-        PARAMS = dict()
-
-        for key, value in locals().items():
-            if (value != "") and (key != "self"):
-                PARAMS[key] = value
+    ) -> Union[list[dict[str, str]], None]:
+        PARAMS = self.parsePARAMS(locals())
 
         response = self.query_api(endpoint=NBIA_ENDPOINTS.GET_SERIES, params=PARAMS)
 
+        if not isinstance(response, list):
+            self.log.error("Expected list, but received: %s", type(response))
+            return None
+
         return response
 
     def downloadSeries(
@@ -258,4 +356,4 @@ def parsePARAMS(self, params: dict) -> dict:
     pprint(all)
 
     sub = client.getCollections(prefix="aCrin")
-    pprint(sub)
+    pprint(sub)