[ENH] Cleaning of Iotools (aramis-lab#1243)

* delete data_handling.py which was replaced with a module

* rework api a bit

* add unit tests

clinica/iotools/converter_utils.py

@@ -1,7 +1,12 @@
 from os import PathLike
 from typing import List, Optional
 
-__all__ = ["replace_sequence_chars"]
+__all__ = [
+    "MissingModsTracker",
+    "replace_sequence_chars",
+    "write_longitudinal_analysis",
+    "write_statistics",
+]
 
 
 def replace_sequence_chars(sequence_name: str) -> str:

clinica/iotools/utils/__init__.py

@@ -0,0 +1,7 @@
+from .center_nifti import center_nifti
+from .merge_tsv import merge_tsv
+
+__all__ = [
+    "center_nifti",
+    "merge_tsv",
+]

clinica/iotools/utils/center_nifti.py

@@ -0,0 +1,92 @@
+from os import PathLike
+from typing import Iterable, Optional, Union
+
+__all__ = ["center_nifti"]
+
+
+def center_nifti(
+    bids_directory: Union[str, PathLike],
+    output_bids_directory: Union[str, PathLike],
+    modalities: Optional[Iterable[str]] = None,
+    center_all_files: bool = False,
+    overwrite_existing_files: bool = False,
+):
+    """Center NIfTI files in a BIDS dataset.
+
+    Parameters
+    ----------
+    bids_directory : PathLike
+        The path to the BIDS directory.
+
+    output_bids_directory : PathLike
+        The path to
+
+    modalities : Iterable of str, optional
+        The modalities
+
+    center_all_files : bool, optional
+        Whether to center all file or not.
+        Default=False.
+
+    overwrite_existing_files : bool, optional
+        If True and if the output BIDS directory already contain files,
+        they might be overwritten. If False, the output BIDS has to be empty
+        or non-existing otherwise a ClinicaExistingDatasetError will be raised.
+
+    Notes
+    -----
+    This tool is mainly useful as a preprocessing step of SPM. In some cases, SPM is not able to segment T1 volumes
+    because their respective center is not aligned with the origin of the world coordinate system. By default, only
+    images detected as problematic are converted from INPUT_BIDS_DIRECTORY to OUTPUT_BIDS_DIRECTORY, whilst the others
+    are copied verbatim.
+    """
+    import time
+    from pathlib import Path
+
+    from clinica.iotools.utils.data_handling import (
+        center_all_nifti,
+        write_list_of_files,
+    )
+    from clinica.utils.exceptions import ClinicaExistingDatasetError
+    from clinica.utils.stream import cprint, log_and_raise
+
+    bids_directory = Path(bids_directory)
+    output_bids_directory = Path(output_bids_directory)
+    if output_bids_directory.exists():
+        files = [
+            file.name
+            for file in output_bids_directory.iterdir()
+            if not file.name.startswith(".")
+        ]
+        if files and not overwrite_existing_files:
+            raise ClinicaExistingDatasetError(output_bids_directory)
+    cprint("Clinica is now centering the requested images.", lvl="info")
+
+    centered_files = center_all_nifti(
+        bids_directory,
+        output_bids_directory,
+        modalities,
+        center_all_files,
+    )
+    # Write list of created files
+    timestamp = time.strftime("%Y%m%d-%H%M%S", time.localtime(time.time()))
+    log_file = output_bids_directory / f"centered_nifti_list_{timestamp}.txt"
+    if not write_list_of_files(centered_files, log_file):
+        log_and_raise(f"Could not create log file {log_file}.", IOError)
+
+    cprint(
+        f"{len(centered_files)} NIfTI files/images of BIDS folder:\n"
+        f"\t{bids_directory}\n"
+        f"for the modalities {modalities} have been centered in output folder:\n"
+        f"\t{output_bids_directory}",
+        lvl="info",
+    )
+    if log_file.is_file():
+        cprint(
+            f"The list of centered NIfTI files is available here: {log_file}.",
+            lvl="info",
+        )
+    cprint(
+        "Please note that the rest of the input BIDS folder has also been copied to the output folder.",
+        lvl="info",
+    )

clinica/iotools/utils/cli.py

@@ -35,67 +35,36 @@ def center_nifti(
     modalities: Optional[List[str]] = None,
     center_all_files: bool = False,
 ) -> None:
-    """Center NIfTI files in a BIDS dataset.
-
-    This tool is mainly useful as a preprocessing step of SPM. In some cases, SPM is not able to segment T1 volumes
-    because their respective center is not aligned with the origin of the world coordinate system. By default, only
-    images detected as problematic are converted from INPUT_BIDS_DIRECTORY to OUTPUT_BIDS_DIRECTORY, whilst the others
-    are copied verbatim.
-    """
+    """Center NIfTI files in a BIDS dataset."""
     import sys
-    import time
-    from os import listdir, makedirs
-    from os.path import isfile, join
 
-    from clinica.iotools.utils.data_handling import (
-        center_all_nifti,
-        write_list_of_files,
-    )
-    from clinica.utils.stream import cprint
+    from clinica.utils.exceptions import ClinicaExistingDatasetError
+
+    from .center_nifti import center_nifti as center_nifti_
 
-    # check that output_folder does not exist, or is an empty folder
     try:
-        makedirs(output_bids_directory)
-    except FileExistsError:
-        file_list = [
-            file for file in listdir(output_bids_directory) if not file.startswith(".")
-        ]
-        if file_list:
-            click.echo(
-                "Target BIDS directory is not empty. Existing files may be overwritten."
+        center_nifti_(
+            bids_directory,
+            output_bids_directory,
+            modalities=modalities,
+            center_all_files=center_all_files,
+            overwrite_existing_files=False,
+        )
+    except ClinicaExistingDatasetError:
+        click.echo(
+            "Target BIDS directory is not empty. Existing files may be overwritten."
+        )
+        if click.confirm("Do you wish to continue?"):
+            center_nifti_(
+                bids_directory,
+                output_bids_directory,
+                modalities=modalities,
+                center_all_files=center_all_files,
+                overwrite_existing_files=True,
             )
-            if not click.confirm("Do you wish to continue?"):
-                click.echo("Clinica will now exit...")
-                sys.exit(0)
-
-    cprint("Clinica is now centering the requested images.")
-
-    centered_files = center_all_nifti(
-        bids_directory,
-        output_bids_directory,
-        modalities,
-        center_all_files,
-    )
-
-    # Write list of created files
-    timestamp = time.strftime("%Y%m%d-%H%M%S", time.localtime(time.time()))
-    log_file = join(output_bids_directory, "centered_nifti_list_" + timestamp + ".txt")
-    # If an error happen while creating the file, the function returns Nan
-    if not write_list_of_files(centered_files, log_file):
-        cprint("Could not create log file.")
-
-    # Final message
-    cprint(
-        f"{str(len(centered_files))} NIfTI files/images of BIDS folder:\n"
-        f"\t{bids_directory}\n"
-        f"for the modalities {modalities} have been centered in output folder:\n"
-        f"\t{output_bids_directory}"
-    )
-    if isfile(log_file):
-        cprint(f"The list of centered NIfTI files is available here: {log_file}.")
-    cprint(
-        "Please note that the rest of the input BIDS folder has also been copied to the output folder."
-    )
+        else:
+            click.echo("Clinica will now exit...")
+            sys.exit(0)
 
 
 @cli.command()
@@ -248,32 +217,19 @@ def merge_tsv(
     ignore_session_scan_files: bool = False,
 ) -> None:
     """Merge clinical data into a single TSV file."""
+    from .merge_tsv import merge_tsv as merge_tsv_
 
-    from clinica.iotools.utils.data_handling import create_merge_file
-    from clinica.utils.inputs import check_bids_folder
-
-    check_bids_folder(bids_directory)
-
-    atlas_selection = []
-    if volume_atlas_selection is not None:
-        atlas_selection += volume_atlas_selection
-    if freesurfer_atlas_selection is not None:
-        atlas_selection += freesurfer_atlas_selection
-    if group_selection == ():
-        group_selection = None
-    if pet_tracers_selection == ():
-        pet_tracers_selection = None
-
-    create_merge_file(
+    merge_tsv_(
         bids_directory,
         output_tsv,
-        caps_dir=caps_directory,
+        caps_directory=caps_directory,
         pipelines=pipelines,
-        ignore_scan_files=ignore_scan_files,
-        ignore_sessions_files=ignore_session_scan_files,
-        atlas_selection=atlas_selection,
+        volume_atlas_selection=volume_atlas_selection,
+        freesurfer_atlas_selection=freesurfer_atlas_selection,
         pvc_restriction=pvc_restriction,
-        tsv_file=subjects_sessions_tsv,
+        pet_tracers_selection=pet_tracers_selection,
         group_selection=group_selection,
-        tracers_selection=pet_tracers_selection,
+        subjects_sessions_tsv=subjects_sessions_tsv,
+        ignore_scan_files=ignore_scan_files,
+        ignore_session_scan_files=ignore_session_scan_files,
     )