Merge pull request #73 from mathdugre/deprecate/ppmi-nifti

[DEPRECATE] PPMI download of NIfTI files.

livingpark_utils/dataset/convert.py

@@ -0,0 +1,54 @@
+"""Utility functions to convert datasets between different file formats."""
+from pathlib import Path
+from typing import Any
+
+from nipype import config
+from nipype import logging
+from nipype.interfaces.dcm2nii import Dcm2niix
+
+log_dir = Path("logs").absolute()
+log_dir.mkdir(parents=True, exist_ok=True, mode=0o755)
+config.update_config({"logging": {"log_directory": log_dir, "log_to_file": True}})
+config.enable_debug_mode()
+logging.update_logging(config)
+
+
+class fileConversionError(Exception):
+    """Exception to report a failed file conversion."""
+
+    pass
+
+
+def dcm2niix(filenames: list[Path] | Path, output_dir: Path) -> Any:
+    """Convert DICOM files to NIfTI using dcm2niix.
+
+    Parameters
+    ----------
+    filenames : list[Path] | Path
+        File to convert. Can either be a list of files or a directory.
+    output_dir : Path, optional
+        Path to the output folder
+
+    Returns
+    -------
+    Any
+        If conversion is successful, returns the converter. Otherwise, None.
+
+    Raises
+    ------
+    fileConversionError
+        When a file conversion fails.
+    """
+    converter = Dcm2niix()
+    if isinstance(filenames, Path) and filenames.is_dir():
+        converter.inputs.source_dir = filenames
+    else:
+        converter.inputs.source_names = filenames
+
+    output_dir.mkdir(parents=True, exist_ok=True, mode=0o755)
+    converter.inputs.output_dir = output_dir
+
+    try:
+        return converter.run()
+    except Exception as e:
+        raise fileConversionError(e)

livingpark_utils/dataset/ppmi.py

@@ -1,9 +1,9 @@
 """Provide utilies to work with the PPMI dataset."""
 import datetime
-import glob
 import logging
 import os.path
 import re
+from pathlib import Path
 from pprint import pprint
 
 import numpy as np
@@ -121,66 +121,65 @@ def clean_protocol_description(desc: str) -> str:
     str
         Protocol description. Example: "MPRAGE GRAPPA"
     """
-    return re.sub(r"_+", "_", re.sub(r"[\s()/-]", "_", desc)).strip("_")
+    return re.sub(r"[\s()/]", "_", desc)
 
 
 def find_nifti_file_in_cache(
     subject_id: str,
     event_id: str,
-    protocol_description: str,
+    description: str,
     *,
     cache_dir: str = ".cache",
     base_dir: str = "inputs",
+    debug: bool = True,
 ) -> str:
     """Return cached nifti files, if any.
 
     Search for nifti file matching `subject_id`, `event_id` and
-    `protocol_description` in the cache directory.
-    If not found, search for nifti file matching `subject_id` and `event_id` only,
-    and return it if a single file is found.
+    `description` in the cache directory.
 
     Parameters
     ----------
     subject_id: str
         Subject ID
     event_id: str
         Event ID. Example: BL
-    protocol_description: str
+    description: str
         Protocol description. Example: "MPRAGE GRAPPA"
 
     Returns
     -------
     str:
         File name matching the `subject_id`, `event_id`, and if possible
-        `protocol_description`. Empty string if no matching file is found.
+        `description`. Empty string if no matching file is found.
     """
-    expression = os.path.join(
-        cache_dir,
-        base_dir,
-        f"sub-{subject_id}",
-        f"ses-{event_id}",
-        "anat",
-        f"PPMI_*{clean_protocol_description(protocol_description)}_br_raw_*.nii",
+    expression = f"PPMI_{subject_id}_{clean_protocol_description(description)}.nii.gz"
+    filenames = list(
+        Path(cache_dir, base_dir, f"sub-{subject_id}", f"ses-{event_id}", "anat").rglob(
+            expression
+        )
     )
-    files = glob.glob(expression)
-    if len(files) > 1:
-        logging.warning(
-            f"""More than 1 Nifti file matched by {expression}
+
+    match len(filenames):
+        case 0:
+            if debug:
+                logging.warning(
+                    f"""No Nifti file matched by {expression}
 {subject_id=}
 {event_id=}
-protocol_description={clean_protocol_description(protocol_description)}
+description={clean_protocol_description(description)}
 """
-        )
-        return ""
-    elif len(files) == 1:
-        return files[0]
-
-    else:
-        logging.warning(
-            f"""No Nifti file matched by {expression}
+                )
+        case 1:
+            return filenames[0].as_posix()
+        case _:
+            if debug:
+                logging.warning(
+                    f"""More than 1 Nifti file matched by {expression}
 {subject_id=}
 {event_id=}
-protocol_description={clean_protocol_description(protocol_description)}
+description={clean_protocol_description(description)}
 """
-        )
-        return ""
+                )
+
+    return ""

livingpark_utils/download/DownloaderABC.py

@@ -48,7 +48,7 @@ def get_study_files(
         Returns
         -------
         tuple[list[str], list[str]]
-            Tuple with the successful and missing study files, respectlively.
+            Tuple with the successful and missing study files, respectively.
 
         Raises
         ------
@@ -100,7 +100,7 @@ def get_T1_nifti_files(
         -------
         tuple[pd.DataFrame, pd.DataFrame]
             Tuple with the successful and missing T1 NIfTI file identifiers,
-            respectlively.
+            respectively.
 
         Raises
         ------