rebase from main and format for func files

Author: jpillai00

src/rbc/core/functional/__init__.py

@@ -7,7 +7,7 @@
 must be corrected before analysis.
 """
 
-from .initialization import truncate_trs, scale
+from .initialization import scale, truncate_trs
 from .motion import generate_motion_reference, motion_correction
 
-__all__ = ["truncate_trs", "scale", "generate_motion_reference", "motion_correction"]
+__all__ = ["truncate_trs", "scale", "generate_motion_reference", "motion_correction"]

src/rbc/core/functional/initialization.py

@@ -4,8 +4,9 @@
 
 from niwrap import afni
 
+
 def truncate_trs(
-    in_file: Path, output_fname: str, start_tr: int 
+    in_file: Path, output_fname: str, start_tr: int
 ) -> afni.V3dcalcOutputs:
     """Remove first N TRs from BOLD timeseries using AFNI 3dcalc.
 
@@ -19,17 +20,14 @@ def truncate_trs(
     """
     return afni.v_3dcalc(
         dataset_a=afni.v_3dcalc_dataset_a_file(
-            file=in_file,
-            selectors_=f"[{start_tr}..$]"
+            file=in_file, selectors_=f"[{start_tr}..$]"
         ),
         expression="a",
         prefix=output_fname,
     )
 
 
-def scale(
-    in_file: Path, scale_factor: float = 0.1
-) -> afni.V3drefitOutputs:
+def scale(in_file: Path, scale_factor: float = 0.1) -> afni.V3drefitOutputs:
     """Scale BOLD voxel dimensions using AFNI 3drefit.
 
     Args:
@@ -42,4 +40,4 @@ def scale(
     return afni.v_3drefit(
         in_file=in_file,
         xyzscale=scale_factor,
-    )
+    )

src/rbc/core/functional/motion.py

@@ -6,9 +6,7 @@
 from niwrap import afni, fsl
 
 
-def generate_motion_reference(
-    in_file: Path, output_fname: str
-) -> afni.V3dcalcOutputs:
+def generate_motion_reference(in_file: Path, output_fname: str) -> afni.V3dcalcOutputs:
     """Creates reference volume for motion correction by extracting middle volume.
 
     Args:
@@ -18,23 +16,17 @@ def generate_motion_reference(
     Returns:
         AFNI 3dcalc output object.
     """
+    total_vols = afni.v_3dinfo(dataset=[in_file], nv=True)
 
-    total_vols = afni.v_3dinfo(
-        dataset=[in_file], 
-        nv=True
-    )
-    
     mid_vol = (int(total_vols.info[0])) // 2
 
     return afni.v_3dcalc(
-        dataset_a=afni.v_3dcalc_dataset_a_file(
-            file=in_file,
-            selectors_=f"[{mid_vol}]"
-        ),
+        dataset_a=afni.v_3dcalc_dataset_a_file(file=in_file, selectors_=f"[{mid_vol}]"),
         expression="a",
         prefix=output_fname,
     )
 
+
 def motion_correction(
     in_file: Path, ref_file: Path, output_prefix: str
 ) -> SimpleNamespace:
@@ -46,9 +38,8 @@ def motion_correction(
         output_prefix: Prefix for output files.
 
     Returns:
-        Namespace with paths to motion corrected BOLD, motion parameters, displacement files, and directory to transformation matrices.
+        Namespace with paths to BOLD motion correction outputs and transformation matrices.
     """
-
     mc_result = fsl.mcflirt(
         in_file=in_file,
         ref_file=ref_file,
@@ -59,12 +50,14 @@ def motion_correction(
         out_file=output_prefix,
     )
 
-    motion_mat_dir = [d for d in Path(mc_result.root).iterdir() if d.is_dir() and d.suffix == ".mat"]
+    motion_mat_dir = [
+        d for d in Path(mc_result.root).iterdir() if d.is_dir() and d.suffix == ".mat"
+    ]
 
     return SimpleNamespace(
-        bold = Path(mc_result.out_file),
-        par = Path(mc_result.par_file),
-        rms_rel = Path(mc_result.rmsrel_files),
-        rms_abs = Path(mc_result.rmsabs_files),
-        mat_dir=motion_mat_dir[0]
-    )
+        bold=Path(mc_result.out_file),
+        par=Path(mc_result.par_file),
+        rms_rel=Path(mc_result.rmsrel_files),
+        rms_abs=Path(mc_result.rmsabs_files),
+        mat_dir=motion_mat_dir[0],
+    )

src/rbc/core/utils.py

@@ -39,14 +39,17 @@ def get_base_entities(in_file: Path) -> dict[str, str]:
         A string-mapping of BIDS entities to values.
     """
     file_entities = parse_bids_entities(in_file)
-    return {k: v for k, v in file_entities.items() if k in ["sub", "ses", "task", "run"]}
+    return {
+        k: v for k, v in file_entities.items() if k in ["sub", "ses", "task", "run"]
+    }
 
 
 def rename(in_file: str | Path, new_name: str | Path) -> Path:
     """Rename a file, keeping it in the same directory."""
     in_file = Path(in_file)
     return in_file.rename(in_file.with_name(Path(new_name).name))
 
+
 def save_directory(in_dir: str | Path, out_dir: str | Path, name: str) -> Path:
     """Save a directory by copying it to a new location.
 
@@ -62,4 +65,4 @@ def save_directory(in_dir: str | Path, out_dir: str | Path, name: str) -> Path:
         raise ValueError(f"Input path {in_dir} is not a directory.")
 
     shutil.copytree(in_dir, target, dirs_exist_ok=True)
-    return target
+    return target

src/rbc/workflows/functional.py

@@ -6,43 +6,45 @@
 import niwrap_helper
 
 from rbc.core.common import reorient
-from rbc.core.functional import truncate_trs, generate_motion_reference, motion_correction
+from rbc.core.functional import (
+    generate_motion_reference,
+    motion_correction,
+    truncate_trs,
+)
 from rbc.core.utils import get_base_entities, rename, save_directory
 
 
-
 def single_session(in_bold: Path, output_dir: Path, start_tr: int = 2) -> None:
     """Workflow for preprocessing functional data.
-    
+
     Args:
         in_bold: Input BOLD timeseries to process.
         output_dir: Parent output directory to save data to.
         start_tr: Number of initial TRs to remove (default: 2).
     """
-
     bids_entities = get_base_entities(in_bold)
     bids = partial(niwrap_helper.bids_path, **bids_entities)
-    
+
     reoriented_bold = reorient(
         in_file=in_bold,
-        output_fname=str(bids(desc="reoriented", suffix="bold", ext=".nii.gz")) 
+        output_fname=str(bids(desc="reoriented", suffix="bold", ext=".nii.gz")),
     )
 
     truncated_bold = truncate_trs(
         in_file=reoriented_bold.out_file,
         output_fname=str(bids(suffix="bold", ext=".nii.gz")),
-        start_tr=start_tr
-    )  
-    
-    motion_reference = generate_motion_reference( 
+        start_tr=start_tr,
+    )
+
+    motion_reference = generate_motion_reference(
         in_file=truncated_bold.output_file,
-        output_fname=str(bids(suffix="sbref", ext=".nii.gz"))
+        output_fname=str(bids(suffix="sbref", ext=".nii.gz")),
     )
 
     motion_corrected = motion_correction(
         in_file=truncated_bold.output_file,
         ref_file=motion_reference.output_file,
-        output_prefix=str(bids())
+        output_prefix=str(bids()),
     )
 
     # Prep files to save
@@ -54,20 +56,22 @@ def single_session(in_bold: Path, output_dir: Path, start_tr: int = 2) -> None:
         (motion_corrected.par, "movementParameters", "motion", ".1D"),
         (motion_corrected.rms_rel, "relsDisplacement", "motion", ".rms"),
         (motion_corrected.rms_abs, "absDisplacement", "motion", ".rms"),
-    ]   
+    ]
 
     func_out_dir = output_dir / bids(datatype="func", directory=True)
 
     save_directory(
-        motion_corrected.mat_dir, 
-        func_out_dir, 
-        bids(desc="motion", suffix="mat")
+        motion_corrected.mat_dir, func_out_dir, bids(desc="motion", suffix="mat")
     )
 
     renamed_files = [
-        rename(out_file, bids(desc=desc, suffix=suffix, ext=ext) if desc else bids(suffix=suffix, ext=ext))
+        rename(
+            out_file,
+            bids(desc=desc, suffix=suffix, ext=ext)
+            if desc
+            else bids(suffix=suffix, ext=ext),
+        )
         for out_file, desc, suffix, ext in outputs
     ]
 
     niwrap_helper.save(renamed_files, out_dir=func_out_dir)
-