[ENH] Add the possibility to use ANTsPy instead of ANTs for T1Linear and FlairLinear

clinica/pipelines/engine.py

@@ -381,6 +381,7 @@ def __init__(
         base_dir: Optional[str] = None,
         parameters: Optional[dict] = None,
         name: Optional[str] = None,
+        ignore_dependencies: Optional[List[str]] = None,
     ):
         """Init a Pipeline object.
 
@@ -407,6 +408,10 @@ def __init__(
         name : str, optional
             Pipeline name. Defaults to None.
 
+        ignore_dependencies : List of str
+            List of names of dependencies whose installation checking procedure should be ignored.
+            Defaults to None (i.e. all dependencies will be checked).
+
         Raises
         ------
         RuntimeError: [description]
@@ -446,6 +451,7 @@ def __init__(
 
         self._name = name or self.__class__.__name__
         self._parameters = parameters or {}
+        self._ignore_dependencies = ignore_dependencies or []
 
         if not self._bids_directory:
             if not self._caps_directory:
@@ -768,18 +774,19 @@ def _check_dependencies(self):
         if not self.info:
             self._load_info()
         for d in self.info["dependencies"]:
-            if d["type"] == "software":
-                check_software(d["name"])
-            elif d["type"] == "binary":
-                check_binary(d["name"])
-            elif d["type"] == "toolbox":
-                pass
-            elif d["type"] == "pipeline":
-                pass
-            else:
-                raise Exception(
-                    f"Pipeline.check_dependencies() Unknown dependency type: '{d['type']}'."
-                )
+            if d["name"] not in self._ignore_dependencies:
+                if d["type"] == "software":
+                    check_software(d["name"])
+                elif d["type"] == "binary":
+                    check_binary(d["name"])
+                elif d["type"] == "toolbox":
+                    pass
+                elif d["type"] == "pipeline":
+                    pass
+                else:
+                    raise Exception(
+                        f"Pipeline.check_dependencies() Unknown dependency type: '{d['type']}'."
+                    )
         self._check_custom_dependencies()
 
         return self

clinica/pipelines/t1_linear/anat_linear_pipeline.py

@@ -1,7 +1,7 @@
 # Use hash instead of parameters for iterables folder names
 # Otherwise path will be too long and generate OSError
 from pathlib import Path
-from typing import List
+from typing import List, Optional
 
 from nipype import config
 
@@ -24,6 +24,44 @@ class AnatLinear(Pipeline):
         A clinica pipeline object containing the  AnatLinear pipeline.
     """
 
+    def __init__(
+        self,
+        bids_directory: Optional[str] = None,
+        caps_directory: Optional[str] = None,
+        tsv_file: Optional[str] = None,
+        overwrite_caps: Optional[bool] = False,
+        base_dir: Optional[str] = None,
+        parameters: Optional[dict] = None,
+        name: Optional[str] = None,
+        ignore_dependencies: Optional[List[str]] = None,
+        use_antspy: bool = False,
+    ):
+        from clinica.utils.stream import cprint
+
+        super().__init__(
+            bids_directory=bids_directory,
+            caps_directory=caps_directory,
+            tsv_file=tsv_file,
+            overwrite_caps=overwrite_caps,
+            base_dir=base_dir,
+            parameters=parameters,
+            ignore_dependencies=ignore_dependencies,
+            name=name,
+        )
+        self.use_antspy = use_antspy
+        if self.use_antspy:
+            self._ignore_dependencies.append("ants")
+            cprint(
+                (
+                    "The AnatLinear pipeline has been configured to use ANTsPy instead of ANTs.\n"
+                    "This means that no installation of ANTs is required, but the antspyx Python "
+                    "package must be installed in your environment.\nThis functionality has been "
+                    "introduced in Clinica 0.9.0 and is considered experimental.\n"
+                    "Please report any issue or unexpected results to the Clinica developer team."
+                ),
+                lvl="warning",
+            )
+
     @staticmethod
     def get_processed_images(
         caps_directory: Path, subjects: List[str], sessions: List[str]
@@ -215,6 +253,10 @@ def _build_core_nodes(self):
         import nipype.pipeline.engine as npe
         from nipype.interfaces import ants
 
+        from clinica.pipelines.t1_linear.tasks import (
+            run_ants_registration_task,
+            run_n4biasfieldcorrection_task,
+        )
         from clinica.pipelines.tasks import crop_nifti_task, get_filename_no_ext_task
 
         from .anat_linear_utils import print_end_pipeline
@@ -228,30 +270,61 @@ def _build_core_nodes(self):
             name="ImageID",
         )
 
-        # The core (processing) nodes
-        # =====================================
-
         # 1. N4biascorrection by ANTS. It uses nipype interface.
         n4biascorrection = npe.Node(
             name="n4biascorrection",
-            interface=ants.N4BiasFieldCorrection(dimension=3, save_bias=True),
+            interface=(
+                nutil.Function(
+                    function=run_n4biasfieldcorrection_task,
+                    input_names=[
+                        "input_image",
+                        "bspline_fitting_distance",
+                        "output_prefix",
+                        "output_dir",
+                        "save_bias",
+                        "verbose",
+                    ],
+                    output_names=["output_image"],
+                )
+                if self.use_antspy
+                else ants.N4BiasFieldCorrection(dimension=3)
+            ),
         )
-
+        n4biascorrection.inputs.save_bias = True
+        if self.use_antspy:
+            n4biascorrection.inputs.output_dir = str(self.base_dir)
+            n4biascorrection.inputs.verbose = True
         if self.name == "t1-linear":
             n4biascorrection.inputs.bspline_fitting_distance = 600
         else:
             n4biascorrection.inputs.bspline_fitting_distance = 100
 
         # 2. `RegistrationSynQuick` by *ANTS*. It uses nipype interface.
         ants_registration_node = npe.Node(
-            name="antsRegistrationSynQuick", interface=ants.RegistrationSynQuick()
+            name="antsRegistrationSynQuick",
+            interface=(
+                nutil.Function(
+                    function=run_ants_registration_task,
+                    input_names=[
+                        "fixed_image",
+                        "moving_image",
+                        "random_seed",
+                        "output_prefix",
+                        "output_dir",
+                    ],
+                    output_names=["warped_image", "out_matrix"],
+                )
+                if self.use_antspy
+                else ants.RegistrationSynQuick()
+            ),
         )
         ants_registration_node.inputs.fixed_image = self.ref_template
-        ants_registration_node.inputs.transform_type = "a"
-        ants_registration_node.inputs.dimension = 3
+        if not self.use_antspy:
+            ants_registration_node.inputs.transform_type = "a"
+            ants_registration_node.inputs.dimension = 3
 
-        if random_seed := self.parameters.get("random_seed", None):
-            ants_registration_node.inputs.random_seed = random_seed
+        random_seed = self.parameters.get("random_seed", None)
+        ants_registration_node.inputs.random_seed = random_seed or 0
 
         # 3. Crop image (using nifti). It uses custom interface, from utils file
 
@@ -301,6 +374,16 @@ def _build_core_nodes(self):
                 (self.input_node, print_end_message, [("anat", "anat")]),
             ]
         )
+        if self.use_antspy:
+            self.connect(
+                [
+                    (
+                        image_id_node,
+                        n4biascorrection,
+                        [("image_id", "output_prefix")],
+                    ),
+                ]
+            )
         if not (self.parameters.get("uncropped_image")):
             self.connect(
                 [