tst: keep expanding coverage, add tiny doctests

sdcflows/utils/misc.py

@@ -0,0 +1,37 @@
+"""Basic miscelaneous utilities."""
+
+
+def front(inlist):
+    """
+    Pop from a list or tuple, otherwise return untouched.
+
+    Examples
+    --------
+    >>> front([1, 0])
+    1
+
+    >>> front("/path/somewhere")
+    '/path/somewhere'
+
+    """
+    if isinstance(inlist, (list, tuple)):
+        return inlist[0]
+    return inlist
+
+
+def last(inlist):
+    """
+    Return the last element from a list or tuple, otherwise return untouched.
+
+    Examples
+    --------
+    >>> last([1, 0])
+    0
+
+    >>> last("/path/somewhere")
+    '/path/somewhere'
+
+    """
+    if isinstance(inlist, (list, tuple)):
+        return inlist[-1]
+    return inlist

sdcflows/workflows/apply/registration.py

@@ -65,6 +65,7 @@ def init_coeff2epi_wf(
     from packaging.version import parse as parseversion, Version
     from niworkflows.interfaces.fixes import FixHeaderRegistration as Registration
     from ...interfaces.bspline import TransformCoefficients
+    from ...utils.misc import front as _pop
 
     workflow = Workflow(name=name)
     workflow.__desc__ = """\
@@ -120,15 +121,9 @@ def init_coeff2epi_wf(
     workflow.connect([
         (inputnode, map_coeff, [("fmap_coeff", "in_coeff"),
                                 ("fmap_ref", "fmap_ref")]),
-        (coregister, map_coeff, [(("forward_transforms", _flatten), "transform")]),
+        (coregister, map_coeff, [(("forward_transforms", _pop), "transform")]),
         (map_coeff, outputnode, [("out_coeff", "fmap_coeff")]),
     ])
     # fmt: on
 
     return workflow
-
-
-def _flatten(inlist):
-    if isinstance(inlist, str):
-        return inlist
-    return inlist[0]

sdcflows/workflows/fit/pepolar.py

@@ -75,14 +75,7 @@ def init_topup_wf(omp_nthreads=1, debug=False, name="pepolar_estimate_wf"):
     )
     outputnode = pe.Node(
         niu.IdentityInterface(
-            fields=[
-                "fmap",
-                "fmap_ref",
-                "fmap_coeff",
-                "jacobians",
-                "xfms",
-                "out_warps",
-            ]
+            fields=["fmap", "fmap_ref", "fmap_coeff", "jacobians", "xfms", "out_warps"]
         ),
         name="outputnode",
     )
@@ -166,40 +159,50 @@ def init_3dQwarp_wf(omp_nthreads=1, name="pepolar_estimate_wf"):
     )
     from niworkflows.interfaces.freesurfer import StructuralReference
     from niworkflows.func.util import init_enhance_and_skullstrip_bold_wf
+    from ...utils.misc import front as _front, last as _last
     from ...interfaces.utils import Flatten
 
     workflow = Workflow(name=name)
     workflow.__desc__ = f"""{_PEPOLAR_DESC} \
 with `3dQwarp` @afni (AFNI {''.join(['%02d' % v for v in afni.Info().version() or []])}).
 """
 
-    inputnode = pe.Node(niu.IdentityInterface(fields=["in_data", "metadata"]),
-                        name="inputnode")
+    inputnode = pe.Node(
+        niu.IdentityInterface(fields=["in_data", "metadata"]), name="inputnode"
+    )
 
     outputnode = pe.Node(
         niu.IdentityInterface(fields=["fmap", "fmap_ref"]), name="outputnode"
     )
 
     flatten = pe.Node(Flatten(), name="flatten")
-    sort_pe = pe.Node(niu.Function(
-        function=_sorted_pe, output_names=["sorted", "qwarp_args"]),
-        name="sort_pe", run_without_submitting=True)
-
-    merge_pes = pe.MapNode(StructuralReference(
-        auto_detect_sensitivity=True,
-        initial_timepoint=1,
-        fixed_timepoint=True,  # Align to first image
-        intensity_scaling=True,
-        # 7-DOF (rigid + intensity)
-        no_iteration=True,
-        subsample_threshold=200,
-        out_file='template.nii.gz'),
-        name='merge_pes',
+    sort_pe = pe.Node(
+        niu.Function(function=_sorted_pe, output_names=["sorted", "qwarp_args"]),
+        name="sort_pe",
+        run_without_submitting=True,
+    )
+
+    merge_pes = pe.MapNode(
+        StructuralReference(
+            auto_detect_sensitivity=True,
+            initial_timepoint=1,
+            fixed_timepoint=True,  # Align to first image
+            intensity_scaling=True,
+            # 7-DOF (rigid + intensity)
+            no_iteration=True,
+            subsample_threshold=200,
+            out_file="template.nii.gz",
+        ),
+        name="merge_pes",
         iterfield=["in_files"],
     )
 
-    pe0_wf = init_enhance_and_skullstrip_bold_wf(omp_nthreads=omp_nthreads, name="pe0_wf")
-    pe1_wf = init_enhance_and_skullstrip_bold_wf(omp_nthreads=omp_nthreads, name="pe1_wf")
+    pe0_wf = init_enhance_and_skullstrip_bold_wf(
+        omp_nthreads=omp_nthreads, name="pe0_wf"
+    )
+    pe1_wf = init_enhance_and_skullstrip_bold_wf(
+        omp_nthreads=omp_nthreads, name="pe1_wf"
+    )
 
     align_pes = pe.Node(
         Registration(
@@ -228,11 +231,7 @@ def init_3dQwarp_wf(omp_nthreads=1, name="pepolar_estimate_wf"):
     cphdr_warp = pe.Node(CopyHeader(), name="cphdr_warp", mem_gb=0.01)
 
     unwarp_reference = pe.Node(
-        ApplyTransforms(
-            dimension=3,
-            float=True,
-            interpolation="LanczosWindowedSinc",
-        ),
+        ApplyTransforms(dimension=3, float=True, interpolation="LanczosWindowedSinc",),
         name="unwarp_reference",
     )
 
@@ -278,7 +277,15 @@ def _fix_hdr(in_file, newpath=None):
 
 
 def _pe2fsl(metadata):
-    """Convert ijk notation to xyz."""
+    """
+    Convert ijk notation to xyz.
+
+    Example
+    -------
+    >>> _pe2fsl([{"PhaseEncodingDirection": "j-"}, {"PhaseEncodingDirection": "i"}])
+    ['y-', 'x']
+
+    """
     return [
         m["PhaseEncodingDirection"]
         .replace("i", "x")
@@ -289,6 +296,29 @@ def _pe2fsl(metadata):
 
 
 def _sorted_pe(inlist):
+    """
+    Generate suitable inputs to ``3dQwarp``.
+
+    Example
+    -------
+    >>> paths, args = _sorted_pe([
+    ...     ("dir-AP_epi.nii.gz", {"PhaseEncodingDirection": "j-"}),
+    ...     ("dir-AP_bold.nii.gz", {"PhaseEncodingDirection": "j-"}),
+    ...     ("dir-PA_epi.nii.gz", {"PhaseEncodingDirection": "j"}),
+    ...     ("dir-PA_bold.nii.gz", {"PhaseEncodingDirection": "j"}),
+    ...     ("dir-AP_sbref.nii.gz", {"PhaseEncodingDirection": "j-"}),
+    ...     ("dir-PA_sbref.nii.gz", {"PhaseEncodingDirection": "j"}),
+    ... ])
+    >>> paths[0]
+    ['dir-AP_epi.nii.gz', 'dir-AP_bold.nii.gz', 'dir-AP_sbref.nii.gz']
+
+    >>> paths[1]
+    ['dir-PA_epi.nii.gz', 'dir-PA_bold.nii.gz', 'dir-PA_sbref.nii.gz']
+
+    >>> args
+    '-noXdis -noZdis'
+
+    """
     out_ref = [inlist[0][0]]
     out_opp = []
 
@@ -302,20 +332,9 @@ def _sorted_pe(inlist):
         else:
             raise ValueError("Cannot handle orthogonal PE encodings.")
 
-    return [out_ref, out_opp], {
-        "i": "-noYdis -noZdis",
-        "j": "-noXdis -noZdis",
-        "k": "-noXdis -noYdis",
-    }[ref_pe[0]]
-
-
-def _front(inlist):
-    if isinstance(inlist, (list, tuple)):
-        return inlist[0]
-    return inlist
-
-
-def _last(inlist):
-    if isinstance(inlist, (list, tuple)):
-        return inlist[-1]
-    return inlist
+    return (
+        [out_ref, out_opp],
+        {"i": "-noYdis -noZdis", "j": "-noXdis -noZdis", "k": "-noXdis -noYdis"}[
+            ref_pe[0]
+        ],
+    )

sdcflows/workflows/fit/syn.py

@@ -120,6 +120,7 @@ def init_syn_sdc_wf(
         FixHeaderRegistration as Registration,
     )
     from niworkflows.interfaces.nibabel import Binarize
+    from ...utils.misc import front as _pop
 
     workflow = Workflow(name=name)
     workflow.__desc__ = f"""\
@@ -230,7 +231,3 @@ def _fixed_masks_arg(mask):
 
     """
     return ["NULL", mask]
-
-
-def _pop(inlist):
-    return inlist[0]