Run ruff format.

Author: tsalo

src/fmripost_template/cli/parser.py

@@ -185,11 +185,11 @@ def _bids_filter(value, parser):
         type=BIDSFilter,
         metavar='FILE',
         help=(
-            "A JSON file describing custom BIDS input filters using PyBIDS. "
-            "For further details, please check out "
-            "https://fmriprep.readthedocs.io/en/"
-            f"{currentv.base_version if is_release else 'latest'}/faq.html#"
-            "how-do-I-select-only-certain-files-to-be-input-to-fMRIPrep"
+            'A JSON file describing custom BIDS input filters using PyBIDS. '
+            'For further details, please check out '
+            'https://fmriprep.readthedocs.io/en/'
+            f'{currentv.base_version if is_release else "latest"}/faq.html#'
+            'how-do-I-select-only-certain-files-to-be-input-to-fMRIPrep'
         ),
     )
     g_bids.add_argument(
@@ -546,8 +546,7 @@ def parse_args(args=None, namespace=None):
         from fmripost_template.utils.bids import validate_input_dir
 
         build_log.info(
-            'Making sure the input data is BIDS compliant '
-            '(warnings can be ignored in most cases).'
+            'Making sure the input data is BIDS compliant (warnings can be ignored in most cases).'
         )
         validate_input_dir(config.environment.exec_env, opts.bids_dir, opts.participant_label)
 
@@ -567,8 +566,8 @@ def parse_args(args=None, namespace=None):
     missing_subjects = participant_label - set(all_subjects)
     if missing_subjects:
         parser.error(
-            "One or more participant labels were not found in the BIDS directory: "
-            f"{', '.join(missing_subjects)}."
+            'One or more participant labels were not found in the BIDS directory: '
+            f'{", ".join(missing_subjects)}.'
         )
 
     config.execution.participant_label = sorted(participant_label)

src/fmripost_template/cli/workflow.py

@@ -58,11 +58,11 @@ def build_workflow(config_file, retval):
     notice_path = Path(pkgrf('fmripost_template', 'data/NOTICE'))
     if notice_path.exists():
         banner[0] += '\n'
-        banner += [f"License NOTICE {'#' * 50}"]
+        banner += [f'License NOTICE {"#" * 50}']
         banner += [f'fMRIPost-template {version}']
         banner += notice_path.read_text().splitlines(keepends=False)[1:]
         banner += ['#' * len(banner[1])]
-    build_log.log(25, f"\n{' ' * 9}".join(banner))
+    build_log.log(25, f'\n{" " * 9}'.join(banner))
 
     # warn if older results exist: check for dataset_description.json in output folder
     msg = check_pipeline_version(
@@ -110,7 +110,7 @@ def build_workflow(config_file, retval):
     if config.execution.derivatives:
         init_msg += [f'Searching for derivatives: {config.execution.derivatives}.']
 
-    build_log.log(25, f"\n{' ' * 11}* ".join(init_msg))
+    build_log.log(25, f'\n{" " * 11}* '.join(init_msg))
 
     retval['workflow'] = init_fmripost_template_wf()
 

src/fmripost_template/config.py

@@ -430,7 +430,7 @@ class execution(_Config):
     the command line) as spatial references for outputs."""
     reports_only = False
     """Only build the reports, based on the reportlets found in a cached working directory."""
-    run_uuid = f"{strftime('%Y%m%d-%H%M%S')}_{uuid4()}"
+    run_uuid = f'{strftime("%Y%m%d-%H%M%S")}_{uuid4()}'
     """Unique identifier of this particular run."""
     participant_label = None
     """List of participant identifiers that are to be preprocessed."""

src/fmripost_template/tests/run_local_tests.py

@@ -83,7 +83,7 @@ def run_command(command, env=None):
 
     if process.returncode != 0:
         raise RuntimeError(
-            f'Non zero return code: {process.returncode}\n' f'{command}\n\n{process.stdout.read()}'
+            f'Non zero return code: {process.returncode}\n{command}\n\n{process.stdout.read()}'
         )
 
 

src/fmripost_template/tests/utils.py

@@ -131,7 +131,7 @@ def run_command(command, env=None):
 
     if process.returncode != 0:
         raise Exception(
-            f'Non zero return code: {process.returncode}\n' f'{command}\n\n{process.stdout.read()}'
+            f'Non zero return code: {process.returncode}\n{command}\n\n{process.stdout.read()}'
         )
 
 

src/fmripost_template/utils/utils.py

@@ -43,7 +43,7 @@ def _get_wf_name(bold_fname, prefix):
 
     fname = split_filename(bold_fname)[1]
     fname_nosub = '_'.join(fname.split('_')[1:-1])
-    return f"{prefix}_{fname_nosub.replace('-', '_')}_wf"
+    return f'{prefix}_{fname_nosub.replace("-", "_")}_wf'
 
 
 def update_dict(orig_dict, new_dict):
@@ -70,3 +70,99 @@ def update_dict(orig_dict, new_dict):
             updated_dict[key] = value
 
     return updated_dict
+
+
+def find_shortest_path(space_pairs, start, end):
+    """Find the shortest path between two spaces in a list of space pairs.
+
+    Parameters
+    ----------
+    space_pairs : list of tuples
+        List of tuples where each tuple contains two spaces of the form (from, to).
+    start : str
+        The starting space.
+    end : str
+        The ending space.
+
+    Returns
+    -------
+    list
+        List of indices that represent the shortest path between the two spaces.
+
+    Raises
+    ------
+    ValueError
+        If no path exists between the two spaces.
+
+    Examples
+    --------
+    >>> space_pairs = [("a", "b"), ("a", "c"), ("b", "d"), ("c", "d")]
+    >>> find_shortest_path(space_pairs, "a", "d")
+    [0, 2]
+    """
+    from collections import deque
+
+    # Create a graph from the space pairs and keep track of indices
+    graph = {}
+    index_map = {}
+    for index, (src, dst) in enumerate(space_pairs):
+        if src not in graph:
+            graph[src] = []
+        graph[src].append(dst)
+        if src not in index_map:
+            index_map[src] = []
+        index_map[src].append(index)
+
+    # Perform BFS to find the shortest path
+    queue = deque([(start, [start], [])])
+    visited = set()
+
+    while queue:
+        current, path, indices = queue.popleft()
+        if current == end:
+            return indices
+        if current not in visited:
+            visited.add(current)
+            for neighbor, idx in zip(
+                graph.get(current, []), index_map.get(current, []), strict=False
+            ):
+                queue.append((neighbor, path + [neighbor], indices + [idx]))
+
+    raise ValueError(f'No path exists between {start} and {end}')
+
+
+def get_transforms(source, target, local_transforms=None):
+    """Get the transforms required to go from source to target space."""
+    import templateflow.api as tflow
+    from bids.layout import Entity, parse_file_entities
+
+    query = [
+        Entity('template', 'tpl-([a-zA-Z0-9]+)'),
+        Entity('from', 'from-([a-zA-Z0-9]+)'),
+    ]
+
+    all_transforms = local_transforms or []
+
+    templates = tflow.get_templates()
+    for template in templates:
+        template_transforms = tflow.get(template, suffix='xfm', extension='h5')
+        if not isinstance(template_transforms, list):
+            template_transforms = [template_transforms]
+        all_transforms += template_transforms
+
+    links = []
+    for transform in all_transforms:
+        entities = parse_file_entities(transform, entities=query)
+        link = (entities['from'], entities['template'])
+        links.append(link)
+
+    path = None
+    try:
+        path = find_shortest_path(links, source, target)
+        print('Shortest path:', path)
+    except ValueError as e:
+        print(e)
+
+    selected_transforms = [all_transforms[i] for i in path]
+
+    return selected_transforms

src/fmripost_template/workflows/base.py

@@ -182,10 +182,10 @@ def init_single_subject_wf(subject_id: str):
     if not subject_data['bold']:
         task_id = config.execution.task_id
         raise RuntimeError(
-            f"No BOLD images found for participant {subject_id} and "
-            f"task {task_id if task_id else '<all>'}. "
-            "All workflows require BOLD images. "
-            f"Please check your BIDS filters: {config.execution.bids_filters}."
+            f'No BOLD images found for participant {subject_id} and '
+            f'task {task_id if task_id else "<all>"}. '
+            'All workflows require BOLD images. '
+            f'Please check your BIDS filters: {config.execution.bids_filters}.'
         )
 
     config.loggers.workflow.info(