wip

clinica/pipelines/anatomical/freesurfer/t1/pipeline.py

@@ -4,6 +4,7 @@
 from nipype import config
 
 from clinica.pipelines.engine import Pipeline
+from clinica.utils.input_files import QueryPattern
 
 cfg = dict(execution={"parameterize_dirs": False})
 config.update_config(cfg)
@@ -21,13 +22,20 @@ def get_processed_images(
         caps_directory: Path, subjects: List[str], sessions: List[str]
     ) -> List[str]:
         from clinica.utils.filemanip import extract_image_ids
-        from clinica.utils.input_files import T1_FS_DESTRIEUX
+        from clinica.utils.input_files import (
+            Parcellation,
+            QueryPatternName,
+            query_pattern_factory,
+        )
         from clinica.utils.inputs import clinica_file_reader
 
         image_ids: List[str] = []
         if caps_directory.is_dir():
+            pattern = query_pattern_factory(
+                QueryPatternName.T1_FREESURFER_SEGMENTATION
+            )(Parcellation.DESTRIEUX)
             t1_freesurfer_files, _ = clinica_file_reader(
-                subjects, sessions, caps_directory, T1_FS_DESTRIEUX
+                subjects, sessions, caps_directory, pattern
             )
             image_ids = extract_image_ids(t1_freesurfer_files)
         return image_ids
@@ -96,7 +104,7 @@ def _build_input_node(self):
             extract_subjects_sessions_from_filename,
             save_participants_sessions,
         )
-        from clinica.utils.input_files import T1W_NII
+        from clinica.utils.input_files import QueryPatternName, query_pattern_factory
         from clinica.utils.inputs import clinica_file_filter
         from clinica.utils.stream import cprint
         from clinica.utils.ux import print_images_to_process
@@ -130,8 +138,9 @@ def _build_input_node(self):
                     to_process_ids
                 )
 
+        pattern = query_pattern_factory(QueryPatternName.T1W)()
         t1w_files, self.subjects, self.sessions = clinica_file_filter(
-            self.subjects, self.sessions, self.bids_directory, T1W_NII
+            self.subjects, self.sessions, self.bids_directory, pattern
         )
 
         if not t1w_files:

clinica/pipelines/dwi/preprocessing/t1/pipeline.py

@@ -4,6 +4,12 @@
 from nipype import config
 
 from clinica.pipelines.dwi.preprocessing.engine import DWIPreprocessingPipeline
+from clinica.utils.input_files import (
+    DWIFileType,
+    QueryPatternName,
+    query_pattern_factory,
+)
+from clinica.utils.inputs import clinica_file_reader
 
 cfg = dict(execution={"parameterize_dirs": False})
 config.update_config(cfg)
@@ -34,13 +40,14 @@ def get_processed_images(
         caps_directory: Path, subjects: List[str], sessions: List[str]
     ) -> List[str]:
         from clinica.utils.filemanip import extract_image_ids
-        from clinica.utils.input_files import DWI_PREPROC_NII
-        from clinica.utils.inputs import clinica_file_reader
 
         image_ids: List[str] = []
         if caps_directory.is_dir():
+            pattern = query_pattern_factory(QueryPatternName.DWI_PREPROC)(
+                DWIFileType.NII
+            )
             preproc_files, _ = clinica_file_reader(
-                subjects, sessions, caps_directory, DWI_PREPROC_NII
+                subjects, sessions, caps_directory, pattern
             )
             image_ids = extract_image_ids(preproc_files)
         return image_ids
@@ -90,21 +97,20 @@ def filter_qc(self) -> tuple[list[str], list[str]]:
         from clinica.pipelines.dwi.preprocessing.utils import check_dwi_volume
         from clinica.pipelines.dwi.utils import DWIDataset
         from clinica.utils.bids import BIDSFileName
-        from clinica.utils.input_files import (
-            DWI_BVAL,
-            DWI_BVEC,
-            DWI_NII,
-        )
         from clinica.utils.inputs import clinica_list_of_files_reader
         from clinica.utils.stream import cprint
 
         subjects = []
         sessions = []
+        patterns = [
+            query_pattern_factory(QueryPatternName.DWI)(file_type)
+            for file_type in (DWIFileType.NII, DWIFileType.BVEC, DWIFileType.BVAL)
+        ]
         list_bids_files = clinica_list_of_files_reader(
             self.subjects,
             self.sessions,
             self.bids_directory,
-            [DWI_NII, DWI_BVEC, DWI_BVAL],
+            patterns,
             raise_exception=True,
         )
         for dwi_image_file, b_vectors_file, b_values_file in zip(*list_bids_files):
@@ -134,22 +140,25 @@ def _build_input_node(self):
         import nipype.pipeline.engine as npe
 
         from clinica.utils.filemanip import save_participants_sessions
-        from clinica.utils.input_files import (
-            DWI_BVAL,
-            DWI_BVEC,
-            DWI_JSON,
-            DWI_NII,
-            T1W_NII,
-        )
         from clinica.utils.inputs import clinica_list_of_files_reader
         from clinica.utils.stream import cprint
         from clinica.utils.ux import print_images_to_process
 
+        patterns = [
+            query_pattern_factory(QueryPatternName.DWI)(file_type)
+            for file_type in (
+                DWIFileType.NII,
+                DWIFileType.JSON,
+                DWIFileType.BVEC,
+                DWIFileType.BVAL,
+            )
+        ]
+        patterns.insert(0, query_pattern_factory(QueryPatternName.T1W)())
         list_bids_files = clinica_list_of_files_reader(
             self.subjects,
             self.sessions,
             self.bids_directory,
-            [T1W_NII, DWI_NII, DWI_JSON, DWI_BVEC, DWI_BVAL],
+            patterns,
             raise_exception=True,
         )
 

clinica/pipelines/pet/engine.py

@@ -1,4 +1,5 @@
 from clinica.pipelines.engine import Pipeline
+from clinica.utils.input_files import QueryPattern
 from clinica.utils.pet import ReconstructionMethod, Tracer
 from clinica.utils.stream import log_and_raise
 
@@ -22,7 +23,7 @@ def _check_pipeline_parameters(self) -> None:
         else:
             self.parameters["reconstruction_method"] = None
 
-    def _get_pet_scans_query(self) -> dict:
+    def _get_pet_scans_query(self) -> QueryPattern:
         """Return the query to retrieve PET scans."""
         from clinica.utils.input_files import bids_pet_nii
 

clinica/pipelines/pet/linear/pipeline.py

@@ -60,13 +60,9 @@ def _build_input_node(self):
         import nipype.pipeline.engine as npe
 
         from clinica.pipelines.pet.utils import get_suvr_mask
-        from clinica.utils.exceptions import (
-            ClinicaBIDSError,
-            ClinicaCAPSError,
-            ClinicaException,
-        )
+        from clinica.utils.exceptions import ClinicaBIDSError, ClinicaCAPSError
         from clinica.utils.image import get_mni_template
-        from clinica.utils.input_files import T1W_NII, T1W_TO_MNI_TRANSFORM
+        from clinica.utils.input_files import QueryPatternName, query_pattern_factory
         from clinica.utils.inputs import (
             clinica_file_reader,
             format_clinica_file_reader_errors,
@@ -91,25 +87,24 @@ def _build_input_node(self):
                 )
             )
 
-        # T1w file:
+        pattern = query_pattern_factory(QueryPatternName.T1W)()
         t1w_files, t1w_errors = clinica_file_reader(
-            self.subjects, self.sessions, self.bids_directory, T1W_NII
+            self.subjects, self.sessions, self.bids_directory, pattern
         )
         if t1w_errors:
             raise ClinicaBIDSError(
-                format_clinica_file_reader_errors(t1w_errors, T1W_NII)
+                format_clinica_file_reader_errors(t1w_errors, pattern)
             )
 
         # Inputs from t1-linear pipeline
         # Transformation files from T1w files to MNI:
+        pattern = query_pattern_factory(QueryPatternName.T1W_TO_MNI_TRANSFORM)()
         t1w_to_mni_transformation_files, t1w_to_mni_errors = clinica_file_reader(
-            self.subjects, self.sessions, self.caps_directory, T1W_TO_MNI_TRANSFORM
+            self.subjects, self.sessions, self.caps_directory, pattern
         )
         if t1w_to_mni_errors:
             raise ClinicaCAPSError(
-                format_clinica_file_reader_errors(
-                    t1w_to_mni_errors, T1W_TO_MNI_TRANSFORM
-                )
+                format_clinica_file_reader_errors(t1w_to_mni_errors, pattern)
             )
 
         if len(self.subjects):

clinica/pipelines/pet/volume/pipeline.py

@@ -3,6 +3,7 @@
 from nipype import config
 
 from clinica.pipelines.pet.engine import PETPipeline
+from clinica.utils.input_files import QueryPatternName, query_pattern_factory
 
 # Use hash instead of parameters for iterables folder names
 # Otherwise path will be too long and generate OSError
@@ -90,13 +91,6 @@ def _build_input_node(self):
         from clinica.pipelines.pet.utils import get_suvr_mask, read_psf_information
         from clinica.utils.exceptions import ClinicaException
         from clinica.utils.filemanip import save_participants_sessions
-        from clinica.utils.input_files import (
-            T1W_NII,
-            t1_volume_deformation_to_template,
-            t1_volume_final_group_template,
-            t1_volume_native_tpm,
-            t1_volume_native_tpm_in_mni,
-        )
         from clinica.utils.inputs import (
             clinica_file_reader,
             clinica_group_reader,
@@ -130,15 +124,14 @@ def _build_input_node(self):
 
         # PET from BIDS directory
         # Native T1w-MRI
-
         try:
             pet_bids, t1w_bids = clinica_list_of_files_reader(
                 self.subjects,
                 self.sessions,
                 self.bids_directory,
                 [
                     self._get_pet_scans_query(),
-                    T1W_NII,
+                    query_pattern_factory(QueryPatternName.T1W)(),
                 ],
             )
         except ClinicaException as e:
@@ -151,7 +144,9 @@ def _build_input_node(self):
                 self.sessions,
                 self.caps_directory,
                 [
-                    t1_volume_native_tpm_in_mni(tissue_number, False)
+                    query_pattern_factory(QueryPatternName.T1_VOLUME_NATIVE_TPM)(
+                        tissue_number, modulation=False, mni_space=True
+                    )
                     for tissue_number in self.parameters["mask_tissues"]
                 ],
             )
@@ -166,26 +161,26 @@ def _build_input_node(self):
             all_errors += e
 
         # Flowfields
+        pattern = query_pattern_factory(
+            QueryPatternName.T1_VOLUME_DEFORMATION_TO_TEMPLATE
+        )(self.parameters["group_label"])
         flowfields_caps, flowfields_errors = clinica_file_reader(
             self.subjects,
             self.sessions,
             self.caps_directory,
-            t1_volume_deformation_to_template(self.parameters["group_label"]),
+            pattern,
         )
         if flowfields_errors:
             all_errors.append(
-                format_clinica_file_reader_errors(
-                    flowfields_errors,
-                    t1_volume_deformation_to_template(self.parameters["group_label"]),
-                )
+                format_clinica_file_reader_errors(flowfields_errors, pattern)
             )
 
         # Dartel Template
         try:
-            final_template = clinica_group_reader(
-                self.caps_directory,
-                t1_volume_final_group_template(self.parameters["group_label"]),
+            pattern = query_pattern_factory(QueryPatternName.T1_VOLUME_GROUP_TEMPLATE)(
+                self.parameters["group_label"]
             )
+            final_template = clinica_group_reader(self.caps_directory, pattern)
         except ClinicaException as e:
             all_errors.append(e)
 
@@ -204,14 +199,17 @@ def _build_input_node(self):
         if self.parameters["apply_pvc"]:
             # pvc tissues input
             try:
+                patterns = [
+                    query_pattern_factory(QueryPatternName.T1_VOLUME_NATIVE_TPM)(
+                        tissue_number, modulation=False, mni_space=False
+                    )
+                    for tissue_number in self.parameters["pvc_mask_tissues"]
+                ]
                 pvc_tissues_input = clinica_list_of_files_reader(
                     self.subjects,
                     self.sessions,
                     self.caps_directory,
-                    [
-                        t1_volume_native_tpm(tissue_number)
-                        for tissue_number in self.parameters["pvc_mask_tissues"]
-                    ],
+                    patterns,
                 )
                 pvc_tissues_input_final = []
                 for subject_tissue_list in zip(*pvc_tissues_input):

clinica/pipelines/t1_linear/anat_linear_pipeline.py

@@ -72,16 +72,17 @@ def get_processed_images(
         caps_directory: Path, subjects: List[str], sessions: List[str]
     ) -> List[str]:
         from clinica.utils.filemanip import extract_image_ids
-        from clinica.utils.input_files import T1W_LINEAR_CROPPED
+        from clinica.utils.input_files import QueryPatternName, query_pattern_factory
         from clinica.utils.inputs import clinica_file_reader
 
         image_ids: List[str] = []
         if caps_directory.is_dir():
+            pattern = query_pattern_factory(QueryPatternName.T1W_LINEAR)(cropped=True)
             cropped_files, _ = clinica_file_reader(
                 subjects,
                 sessions,
                 caps_directory,
-                T1W_LINEAR_CROPPED,
+                pattern,
             )
             image_ids = extract_image_ids(cropped_files)
         return image_ids
@@ -119,10 +120,9 @@ def _build_input_node(self):
         import nipype.interfaces.utility as nutil
         import nipype.pipeline.engine as npe
 
-        from clinica.utils.exceptions import ClinicaBIDSError, ClinicaException
         from clinica.utils.filemanip import extract_subjects_sessions_from_filename
         from clinica.utils.image import get_mni_template
-        from clinica.utils.input_files import T1W_NII, Flair_T2W_NII
+        from clinica.utils.input_files import QueryPatternName, query_pattern_factory
         from clinica.utils.inputs import clinica_file_filter
         from clinica.utils.stream import cprint
         from clinica.utils.ux import print_images_to_process
@@ -155,10 +155,11 @@ def _build_input_node(self):
         # Inputs from anat/ folder
         # ========================
         # anat image file:
-        query = T1W_NII if self.name == "t1-linear" else Flair_T2W_NII
-
+        pattern = query_pattern_factory(
+            QueryPatternName.T1W if self.name == "t1-linear" else QueryPatternName.T2W
+        )()
         anat_files, filtered_subjects, filtered_sessions = clinica_file_filter(
-            self.subjects, self.sessions, self.bids_directory, query
+            self.subjects, self.sessions, self.bids_directory, pattern
         )
         self.subjects = filtered_subjects
         self.sessions = filtered_sessions

clinica/pipelines/t1_volume_tissue_segmentation/t1_volume_tissue_segmentation_pipeline.py

@@ -75,17 +75,17 @@ def _build_input_node(self):
         from clinica.iotools.utils.data_handling import (
             check_volume_location_in_world_coordinate_system,
         )
-        from clinica.utils.exceptions import ClinicaBIDSError, ClinicaException
-        from clinica.utils.input_files import T1W_NII
+        from clinica.utils.input_files import QueryPatternName, query_pattern_factory
         from clinica.utils.inputs import clinica_file_filter
         from clinica.utils.stream import cprint
         from clinica.utils.ux import print_images_to_process
 
         # Inputs from anat/ folder
         # ========================
         # T1w file:
+        pattern = query_pattern_factory(QueryPatternName.T1W)()
         t1w_files, subjects, sessions = clinica_file_filter(
-            self.subjects, self.sessions, self.bids_directory, T1W_NII
+            self.subjects, self.sessions, self.bids_directory, pattern
         )
         self.subjects = subjects
         self.sessions = sessions