♻️ DRY up output key collection for group config

Author: shnizzedy

CPAC/pipeline/cpac_group_runner.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2022-2024  C-PAC Developers
+# Copyright (C) 2022-2025  C-PAC Developers
 
 # This file is part of C-PAC.
 
@@ -143,31 +143,12 @@ def gather_nifti_globs(pipeline_output_folder, resource_list, pull_func=False):
     import glob
     import os
 
-    import pandas as pd
-    import pkg_resources as p
+    from CPAC.utils.outputs import group_derivatives
 
     exts = ".nii"
     nifti_globs = []
 
-    keys_tsv = p.resource_filename("CPAC", "resources/cpac_outputs.tsv")
-    try:
-        keys = pd.read_csv(keys_tsv, delimiter="\t")
-    except Exception as e:
-        err = (
-            "\n[!] Could not access or read the cpac_outputs.tsv "
-            f"resource file:\n{keys_tsv}\n\nError details {e}\n"
-        )
-        raise Exception(err)
-
-    derivative_list = list(keys[keys["Sub-Directory"] == "func"]["Resource"])
-    derivative_list = derivative_list + list(
-        keys[keys["Sub-Directory"] == "anat"]["Resource"]
-    )
-
-    if pull_func:
-        derivative_list = derivative_list + list(
-            keys[keys["Space"] == "functional"]["Resource"]
-        )
+    derivative_list = group_derivatives(pull_func)
 
     if len(resource_list) == 0:
         err = "\n\n[!] No derivatives selected!\n\n"
@@ -361,33 +342,14 @@ def create_output_dict_list(
     """Create a dictionary of output filepaths and their associated information."""
     import os
 
-    import pandas as pd
-    import pkg_resources as p
-
     if len(resource_list) == 0:
         err = "\n\n[!] No derivatives selected!\n\n"
         raise Exception(err)
 
     if derivatives is None:
-        keys_tsv = p.resource_filename("CPAC", "resources/cpac_outputs.tsv")
-        try:
-            keys = pd.read_csv(keys_tsv, delimiter="\t")
-        except Exception as e:
-            err = (
-                "\n[!] Could not access or read the cpac_outputs.csv "
-                f"resource file:\n{keys_tsv}\n\nError details {e}\n"
-            )
-            raise Exception(err)
+        from CPAC.utils.outputs import group_derivatives
 
-        derivatives = list(keys[keys["Sub-Directory"] == "func"]["Resource"])
-        derivatives = derivatives + list(
-            keys[keys["Sub-Directory"] == "anat"]["Resource"]
-        )
-
-        if pull_func:
-            derivatives = derivatives + list(
-                keys[keys["Space"] == "functional"]["Resource"]
-            )
+        derivatives = group_derivatives(pull_func)
 
     # remove any extra /'s
     pipeline_output_folder = pipeline_output_folder.rstrip("/")
@@ -752,18 +714,10 @@ def prep_feat_inputs(group_config_file: str) -> dict:
     import os
 
     import pandas as pd
-    import pkg_resources as p
 
-    keys_tsv = p.resource_filename("CPAC", "resources/cpac_outputs.tsv")
-    try:
-        keys = pd.read_csv(keys_tsv, delimiter="\t")
-    except Exception as e:
-        err = (
-            "\n[!] Could not access or read the cpac_outputs.tsv "
-            f"resource file:\n{keys_tsv}\n\nError details {e}\n"
-        )
-        raise Exception(err)
+    from CPAC.utils.outputs import Outputs
 
+    keys = Outputs.reference
     derivatives = list(
         keys[keys["Derivative"] == "yes"][keys["Space"] == "template"][
             keys["Values"] == "z-score"

CPAC/utils/create_fsl_flame_preset.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2018-2024  C-PAC Developers
+# Copyright (C) 2018-2025  C-PAC Developers
 
 # This file is part of C-PAC.
 
@@ -1092,20 +1092,6 @@ def run(
 
     import os
 
-    import pandas as pd
-    import pkg_resources as p
-
-    # make life easy
-    keys_csv = p.resource_filename("CPAC", "resources/cpac_outputs.csv")
-    try:
-        pd.read_csv(keys_csv)
-    except Exception as e:
-        err = (
-            "\n[!] Could not access or read the cpac_outputs.csv "
-            f"resource file:\n{keys_csv}\n\nError details {e}\n"
-        )
-        raise Exception(err)
-
     if derivative_list == "all":
         derivative_list = [
             "alff",

CPAC/utils/outputs.py

@@ -17,6 +17,7 @@
 """Specify the resources that C-PAC writes to the output direcotry."""
 
 from importlib.resources import files
+from typing import ClassVar
 
 import pandas as pd
 
@@ -47,8 +48,12 @@ class Outputs:
         reference[reference["4D Time Series"] == "Yes"]["Resource"]
     )
 
-    anat = list(reference[reference["Sub-Directory"] == "anat"]["Resource"])
-    func = list(reference[reference["Sub-Directory"] == "func"]["Resource"])
+    anat: ClassVar[list[str]] = list(
+        reference[reference["Sub-Directory"] == "anat"]["Resource"]
+    )
+    func: ClassVar[list[str]] = list(
+        reference[reference["Sub-Directory"] == "func"]["Resource"]
+    )
 
     # outputs to send into smoothing, if smoothing is enabled, and
     # outputs to write out if the user selects to write non-smoothed outputs
@@ -65,6 +70,8 @@ class Outputs:
     all_template_filter = _template_filter | _epitemplate_filter | _symtemplate_filter
     all_native_filter = _T1w_native_filter | _bold_native_filter | _long_native_filter
 
+    bold_native: ClassVar[list[str]] = list(reference[_bold_native_filter]["Resource"])
+
     native_nonsmooth = list(
         reference[all_native_filter & _nonsmoothed_filter]["Resource"]
     )
@@ -121,3 +128,11 @@ def _is_gifti(_file_key):
         for gifti in giftis.itertuples()
         if " " in gifti.File
     }
+
+
+def group_derivatives(pull_func: bool = False) -> list[str]:
+    """Gather keys for anatomical and functional derivatives for group analysis."""
+    derivatives: list[str] = Outputs.func + Outputs.anat
+    if pull_func:
+        derivatives = derivatives + Outputs.bold_native
+    return derivatives