remove factory and adapt to GroupID new type

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

@@ -51,7 +51,12 @@ def get_to_process_with_atlases(
             get_processed_images,
         )
         from clinica.utils.filemanip import extract_image_ids
-        from clinica.utils.input_files import T1_FS_DESTRIEUX
+        from clinica.utils.image import HemiSphere
+        from clinica.utils.input_files import (
+            Parcellation,
+            get_t1_freesurfer_segmentation,
+            get_t1_freesurfer_statistics,
+        )
         from clinica.utils.inputs import clinica_file_reader
 
         part_ids, sess_ids, list_long_id = grab_image_ids_from_caps_directory(
@@ -66,13 +71,6 @@ def get_to_process_with_atlases(
 
         if caps_directory.is_dir():
             for atlas in atlas_list:
-                atlas_info = dict(
-                    {
-                        "pattern": f"t1/freesurfer_cross_sectional/sub-*_ses-*/stats/rh.{atlas}.stats",
-                        "description": f"{atlas}-based segmentation",
-                        "needed_pipeline": "t1-freesurfer",
-                    }
-                )
                 t1_freesurfer_longitudinal_output = get_processed_images(
                     caps_directory, part_ids, sess_ids, list_long_id
                 )
@@ -87,13 +85,13 @@ def get_to_process_with_atlases(
                     subjects,
                     sessions,
                     caps_directory,
-                    T1_FS_DESTRIEUX,
+                    get_t1_freesurfer_segmentation(Parcellation.DESTRIEUX),
                 )
                 t1_freesurfer_files, _ = clinica_file_reader(
                     subjects,
                     sessions,
                     caps_directory,
-                    atlas_info,
+                    get_t1_freesurfer_statistics(atlas, HemiSphere.RIGHT),
                 )
                 image_ids = extract_image_ids(t1_freesurfer_files)
                 image_ids_2 = extract_image_ids(t1_freesurfer_output)

clinica/pipelines/anatomical/freesurfer/longitudinal/correction/pipeline.py

@@ -61,8 +61,8 @@ def _build_input_node(self):
         from clinica.utils.exceptions import ClinicaException
         from clinica.utils.input_files import (
             Parcellation,
-            QueryPatternName,
-            query_pattern_factory,
+            get_t1_freesurfer_segmentation,
+            get_t1_freesurfer_template,
         )
         from clinica.utils.inputs import (
             clinica_file_reader,
@@ -123,15 +123,11 @@ def _build_input_node(self):
                 ) = extract_subject_session_longitudinal_ids_from_filename(
                     to_process_ids
                 )
-        pattern_segmentation = query_pattern_factory(
-            QueryPatternName.T1_FREESURFER_SEGMENTATION
-        )(Parcellation.DESTRIEUX)
+        pattern_segmentation = get_t1_freesurfer_segmentation(Parcellation.DESTRIEUX)
         _, errors_destrieux = clinica_file_reader(
             self.subjects, self.sessions, self.caps_directory, pattern_segmentation
         )
-        pattern_template = query_pattern_factory(
-            QueryPatternName.T1_FREESURFER_TEMPLATE
-        )(Parcellation.DESTRIEUX)
+        pattern_template = get_t1_freesurfer_template(Parcellation.DESTRIEUX)
         _, errors_t_destrieux = clinica_file_reader(
             self.subjects, list_long_id, self.caps_directory, pattern_template
         )

clinica/pipelines/anatomical/freesurfer/longitudinal/template/pipeline.py

@@ -4,8 +4,8 @@
 from clinica.pipelines.engine import Pipeline
 from clinica.utils.input_files import (
     Parcellation,
-    QueryPatternName,
-    query_pattern_factory,
+    get_t1_freesurfer_segmentation,
+    get_t1_freesurfer_template,
 )
 
 
@@ -34,11 +34,11 @@ def get_processed_images(
         ]
         image_ids: List[str] = []
         if caps_directory.is_dir():
-            pattern = query_pattern_factory(QueryPatternName.T1_FREESURFER_TEMPLATE)(
-                Parcellation.DESTRIEUX
-            )
             t1_freesurfer_files, _ = clinica_file_reader(
-                list_participant_id, list_long_id, caps_directory, pattern
+                list_participant_id,
+                list_long_id,
+                caps_directory,
+                get_t1_freesurfer_template(Parcellation.DESTRIEUX),
             )
             image_ids = [
                 re.search(r"(sub-[a-zA-Z0-9]+)_(long-[a-zA-Z0-9]+)", file).group()
@@ -153,11 +153,11 @@ def _build_input_node(self):
                 self.subjects, self.sessions = extract_subjects_sessions_from_filename(
                     to_process_ids
                 )
-        pattern = query_pattern_factory(QueryPatternName.T1_FREESURFER_SEGMENTATION)(
-            Parcellation.DESTRIEUX
-        )
         _, self.subjects, self.sessions = clinica_file_filter(
-            self.subjects, self.sessions, self.caps_directory, pattern
+            self.subjects,
+            self.sessions,
+            self.caps_directory,
+            get_t1_freesurfer_segmentation(Parcellation.DESTRIEUX),
         )
         long_ids = get_participants_long_id(self.subjects, self.sessions)
         save_part_sess_long_ids_to_tsv(

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

@@ -4,7 +4,6 @@
 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)
@@ -24,18 +23,17 @@ def get_processed_images(
         from clinica.utils.filemanip import extract_image_ids
         from clinica.utils.input_files import (
             Parcellation,
-            QueryPatternName,
-            query_pattern_factory,
+            get_t1_freesurfer_segmentation,
         )
         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, pattern
+                subjects,
+                sessions,
+                caps_directory,
+                get_t1_freesurfer_segmentation(Parcellation.DESTRIEUX),
             )
             image_ids = extract_image_ids(t1_freesurfer_files)
         return image_ids
@@ -104,7 +102,7 @@ def _build_input_node(self):
             extract_subjects_sessions_from_filename,
             save_participants_sessions,
         )
-        from clinica.utils.input_files import QueryPatternName, query_pattern_factory
+        from clinica.utils.input_files import get_t1w_mri
         from clinica.utils.inputs import clinica_file_filter
         from clinica.utils.stream import cprint
         from clinica.utils.ux import print_images_to_process
@@ -138,11 +136,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, pattern
+            self.subjects, self.sessions, self.bids_directory, get_t1w_mri()
         )
-
         if not t1w_files:
             raise ClinicaException("Empty dataset or already processed")
 

clinica/pipelines/dwi/connectome/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)
@@ -54,50 +55,54 @@ def get_output_fields(self) -> List[str]:
         """
         return ["response", "fod", "tracts", "nodes", "connectomes"]
 
-    def _build_input_node(self):
-        """Build and connect an input node to the pipeline."""
-        import re
-
-        import nipype.interfaces.utility as nutil
-        import nipype.pipeline.engine as npe
-
-        from clinica.utils.exceptions import ClinicaCAPSError
-        from clinica.utils.filemanip import save_participants_sessions
+    @staticmethod
+    def _get_input_patterns() -> list[QueryPattern]:
         from clinica.utils.input_files import (
             DWIFileType,
             Parcellation,
-            QueryPatternName,
-            query_pattern_factory,
+            get_dwi_preprocessed_brainmask,
+            get_dwi_preprocessed_file,
+            get_t1_freesurfer_extracted_brain,
+            get_t1_freesurfer_segmentation,
+            get_t1_freesurfer_segmentation_white_matter,
         )
-        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
 
-        # Read CAPS files
-        patterns = [
-            query_pattern_factory(QueryPatternName.T1_FREESURFER_WHITE_MATTER)()
-        ]
+        patterns = [get_t1_freesurfer_segmentation_white_matter()]
         patterns.extend(
             [
-                query_pattern_factory(QueryPatternName.T1_FREESURFER_SEGMENTATION)(p)
+                get_t1_freesurfer_segmentation(p)
                 for p in (Parcellation.DESIKAN, Parcellation.DESTRIEUX)
             ]
         )
-        patterns.append(query_pattern_factory(QueryPatternName.T1_FREESURFER_BRAIN)())
+        patterns.append(get_t1_freesurfer_extracted_brain())
         patterns.extend(
             [
-                query_pattern_factory(QueryPatternName.DWI_PREPROC)(file_type)
+                get_dwi_preprocessed_file(file_type)
                 for file_type in (DWIFileType.NII, DWIFileType.BVEC, DWIFileType.BVAL)
             ]
         )
-        patterns.append(query_pattern_factory(QueryPatternName.DWI_PREPROC_BRAINMASK)())
+        patterns.append(get_dwi_preprocessed_brainmask())
+        return patterns
+
+    def _build_input_node(self):
+        """Build and connect an input node to the pipeline."""
+        import re
+
+        import nipype.interfaces.utility as nutil
+        import nipype.pipeline.engine as npe
+
+        from clinica.utils.exceptions import ClinicaCAPSError
+        from clinica.utils.filemanip import save_participants_sessions
+        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
+
         list_caps_files = clinica_list_of_files_reader(
             self.subjects,
             self.sessions,
             self.caps_directory,
-            patterns,
+            self._get_input_patterns(),
         )
-
         # Check space of DWI dataset
         dwi_file_spaces = [
             re.search(

clinica/pipelines/dwi/dti/pipeline.py

@@ -70,18 +70,18 @@ def _build_input_node(self):
         from clinica.utils.filemanip import save_participants_sessions
         from clinica.utils.input_files import (
             DWIFileType,
-            QueryPatternName,
-            query_pattern_factory,
+            get_dwi_preprocessed_brainmask,
+            get_dwi_preprocessed_file,
         )
         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_PREPROC)(file_type)
+            get_dwi_preprocessed_file(file_type)
             for file_type in (DWIFileType.NII, DWIFileType.BVEC, DWIFileType.BVAL)
         ]
-        patterns.append(query_pattern_factory(QueryPatternName.DWI_PREPROC_BRAINMASK)())
+        patterns.append(get_dwi_preprocessed_brainmask())
         list_caps_files = clinica_list_of_files_reader(
             self.subjects,
             self.sessions,

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

@@ -6,8 +6,10 @@
 from clinica.pipelines.dwi.preprocessing.engine import DWIPreprocessingPipeline
 from clinica.utils.input_files import (
     DWIFileType,
-    QueryPatternName,
-    query_pattern_factory,
+    get_dwi_file,
+    get_dwi_fmap_magnitude1_file,
+    get_dwi_fmap_phasediff_file,
+    get_dwi_preprocessed_file,
 )
 
 # Use hash instead of parameters for iterables folder names
@@ -39,11 +41,11 @@ def get_processed_images(
 
         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, pattern
+                subjects,
+                sessions,
+                caps_directory,
+                get_dwi_preprocessed_file(DWIFileType.NII),
             )
             image_ids = extract_image_ids(preproc_files)
         return image_ids
@@ -109,20 +111,18 @@ def _build_input_node(self):
         from clinica.utils.ux import print_images_to_process
 
         patterns = [
-            query_pattern_factory(QueryPatternName.DWI)(file_type)
+            get_dwi_file(file_type)
             for file_type in (
                 DWIFileType.NII,
                 DWIFileType.BVEC,
                 DWIFileType.BVAL,
                 DWIFileType.JSON,
             )
         ]
-        patterns.append(
-            query_pattern_factory(QueryPatternName.DWI_FMAP_MAGNITUDE1)(DWIFileType.NII)
-        )
+        patterns.append(get_dwi_fmap_magnitude1_file(DWIFileType.NII))
         patterns.extend(
             [
-                query_pattern_factory(QueryPatternName.DWI_FMAP_PHASEDIFF)(file_type)
+                get_dwi_fmap_phasediff_file(file_type)
                 for file_type in (DWIFileType.NII, DWIFileType.JSON)
             ]
         )

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

@@ -6,8 +6,9 @@
 from clinica.pipelines.dwi.preprocessing.engine import DWIPreprocessingPipeline
 from clinica.utils.input_files import (
     DWIFileType,
-    QueryPatternName,
-    query_pattern_factory,
+    get_dwi_file,
+    get_dwi_preprocessed_file,
+    get_t1w_mri,
 )
 from clinica.utils.inputs import clinica_file_reader
 
@@ -43,11 +44,11 @@ def get_processed_images(
 
         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, pattern
+                subjects,
+                sessions,
+                caps_directory,
+                get_dwi_preprocessed_file(DWIFileType.NII),
             )
             image_ids = extract_image_ids(preproc_files)
         return image_ids
@@ -103,7 +104,7 @@ def filter_qc(self) -> tuple[list[str], list[str]]:
         subjects = []
         sessions = []
         patterns = [
-            query_pattern_factory(QueryPatternName.DWI)(file_type)
+            get_dwi_file(file_type)
             for file_type in (DWIFileType.NII, DWIFileType.BVEC, DWIFileType.BVAL)
         ]
         list_bids_files = clinica_list_of_files_reader(
@@ -145,15 +146,15 @@ def _build_input_node(self):
         from clinica.utils.ux import print_images_to_process
 
         patterns = [
-            query_pattern_factory(QueryPatternName.DWI)(file_type)
+            get_dwi_file(file_type)
             for file_type in (
                 DWIFileType.NII,
                 DWIFileType.JSON,
                 DWIFileType.BVEC,
                 DWIFileType.BVAL,
             )
         ]
-        patterns.insert(0, query_pattern_factory(QueryPatternName.T1W)())
+        patterns.insert(0, get_t1w_mri())
         list_bids_files = clinica_list_of_files_reader(
             self.subjects,
             self.sessions,