♻️ SSOT register_{symmetric_?}FSL_anat_to_template

Author: shnizzedy

CPAC/registration/registration.py

@@ -2256,52 +2256,23 @@ def bold_to_T1template_xfm_connector(
     return (wf, outputs)
 
 
-@nodeblock(
-    name="register_FSL_anat_to_template",
-    config=["registration_workflows", "anatomical_registration"],
-    switch=["run"],
-    option_key=["registration", "using"],
-    option_val=["FSL", "FSL-linear"],
-    inputs=[
-        (
-            [
-                "desc-preproc_T1w",
-                "longitudinal-template_space-longitudinal_desc-reorient_T1w",
-                "longitudinal-template_space-longitudinal_desc-head_T1w",
-            ],
-            [
-                "desc-brain_T1w",
-                "longitudinal-template_space-longitudinal_desc-brain_T1w",
-            ],
-        ),
-        "T1w-template",
-        "T1w-brain-template",
-        "FNIRT-T1w-template",
-        "FNIRT-T1w-brain-template",
-        "template-ref-mask",
-    ],
-    outputs={
-        "space-template_desc-preproc_T1w": {"Template": "T1w-brain-template"},
-        "space-template_desc-head_T1w": {"Template": "T1w-template"},
-        "space-template_desc-T1w_mask": {"Template": "T1w-template"},
-        "space-template_desc-T1wT2w_biasfield": {"Template": "T1w-template"},
-        "from-T1w_to-template_mode-image_desc-linear_xfm": {"Template": "T1w-template"},
-        "from-template_to-T1w_mode-image_desc-linear_xfm": {"Template": "T1w-template"},
-        "from-T1w_to-template_mode-image_xfm": {"Template": "T1w-template"},
-        "from-T1w_to-template_mode-image_warp": {"Template": "T1w-template"},
-        "from-longitudinal_to-template_mode-image_desc-linear_xfm": {
-            "Template": "T1w-template"
-        },
-        "from-template_to-longitudinal_mode-image_desc-linear_xfm": {
-            "Template": "T1w-template"
-        },
-        "from-longitudinal_to-template_mode-image_xfm": {"Template": "T1w-template"},
-    },
-)
-def register_FSL_anat_to_template(wf, cfg, strat_pool, pipe_num, opt=None):
-    """Register T1w to template with FSL."""
+def _register_FSL_anat_to_template(
+    wf: pe.Workflow,
+    cfg: Configuration,
+    strat_pool: ResourcePool,
+    pipe_num: int,
+    opt: str,
+    symmetric: bool,
+) -> NODEBLOCK_RETURN:
+    """Register T1w to symmetric template with FSL."""
+    symmetric_str = "_symmetric_" if symmetric else ""
+    template_str = "template-symmetric" if symmetric else "template"
     fsl, outputs = FSL_registration_connector(
-        f"register_{opt}_anat_to_template_{pipe_num}", cfg, orig="T1w", opt=opt
+        f"register_{opt}_anat_to_template_{symmetric_str}_{pipe_num}",
+        cfg,
+        orig="T1w",
+        opt=opt,
+        symmetric=symmetric,
     )
 
     fsl.inputs.inputspec.interpolation = cfg.registration_workflows[
@@ -2325,16 +2296,16 @@ def register_FSL_anat_to_template(wf, cfg, strat_pool, pipe_num, opt=None):
         ]["ref_resolution"]
         == cfg.registration_workflows["anatomical_registration"]["resolution_for_anat"]
     ):
-        node, out = strat_pool.get_data("T1w-brain-template")
+        node, out = strat_pool.get_data(f"T1w-brain-{template_str}")
         wf.connect(node, out, fsl, "inputspec.reference_brain")
 
-        node, out = strat_pool.get_data("T1w-template")
+        node, out = strat_pool.get_data(f"T1w-{template_str}")
         wf.connect(node, out, fsl, "inputspec.reference_head")
     else:
-        node, out = strat_pool.get_data("FNIRT-T1w-brain-template")
+        node, out = strat_pool.get_data(f"FNIRT-T1w-brain-{template_str}")
         wf.connect(node, out, fsl, "inputspec.reference_brain")
 
-        node, out = strat_pool.get_data("FNIRT-T1w-template")
+        node, out = strat_pool.get_data(f"FNIRT-T1w-{template_str}")
         wf.connect(node, out, fsl, "inputspec.reference_head")
 
     node, out = strat_pool.get_data(
@@ -2346,23 +2317,80 @@ def register_FSL_anat_to_template(wf, cfg, strat_pool, pipe_num, opt=None):
     )
     wf.connect(node, out, fsl, "inputspec.input_head")
 
-    node, out = strat_pool.get_data("template-ref-mask")
+    node, out = strat_pool.get_data(
+        "dilated-symmetric-brain-mask" if symmetric else "template-ref-mask"
+    )
     wf.connect(node, out, fsl, "inputspec.reference_mask")
 
     if "space-longitudinal" in brain:
         for key in list(outputs.keys()):
             if "from-T1w" in key:
                 new_key = key.replace("from-T1w", "from-longitudinal")
                 outputs[new_key] = outputs[key]
-                # del outputs[key]
+                del outputs[key]
             if "to-T1w" in key:
                 new_key = key.replace("to-T1w", "to-longitudinal")
                 outputs[new_key] = outputs[key]
-                # del outputs[key]
+                del outputs[key]
 
     return wf, outputs
 
 
+@nodeblock(
+    name="register_FSL_anat_to_template",
+    config=["registration_workflows", "anatomical_registration"],
+    switch=["run"],
+    option_key=["registration", "using"],
+    option_val=["FSL", "FSL-linear"],
+    inputs=[
+        (
+            [
+                "desc-preproc_T1w",
+                "longitudinal-template_space-longitudinal_desc-reorient_T1w",
+                "longitudinal-template_space-longitudinal_desc-head_T1w",
+            ],
+            [
+                "desc-brain_T1w",
+                "longitudinal-template_space-longitudinal_desc-brain_T1w",
+            ],
+        ),
+        "T1w-template",
+        "T1w-brain-template",
+        "FNIRT-T1w-template",
+        "FNIRT-T1w-brain-template",
+        "template-ref-mask",
+    ],
+    outputs={
+        "space-template_desc-preproc_T1w": {"Template": "T1w-brain-template"},
+        "space-template_desc-head_T1w": {"Template": "T1w-template"},
+        "space-template_desc-T1w_mask": {"Template": "T1w-template"},
+        "space-template_desc-T1wT2w_biasfield": {"Template": "T1w-template"},
+        "from-T1w_to-template_mode-image_desc-linear_xfm": {"Template": "T1w-template"},
+        "from-template_to-T1w_mode-image_desc-linear_xfm": {"Template": "T1w-template"},
+        "from-T1w_to-template_mode-image_xfm": {"Template": "T1w-template"},
+        "from-T1w_to-template_mode-image_warp": {"Template": "T1w-template"},
+        "from-longitudinal_to-template_mode-image_desc-linear_xfm": {
+            "Template": "T1w-template"
+        },
+        "from-template_to-longitudinal_mode-image_desc-linear_xfm": {
+            "Template": "T1w-template"
+        },
+        "from-longitudinal_to-template_mode-image_xfm": {"Template": "T1w-template"},
+    },
+)
+def register_FSL_anat_to_template(
+    wf: pe.Workflow,
+    cfg: Configuration,
+    strat_pool: ResourcePool,
+    pipe_num: int,
+    opt: str = "",
+) -> NODEBLOCK_RETURN:
+    """Register T1w to template with FSL."""
+    return _register_FSL_anat_to_template(
+        wf, cfg, strat_pool, pipe_num, opt, symmetric=False
+    )
+
+
 @nodeblock(
     name="register_symmetric_FSL_anat_to_template",
     config=["registration_workflows", "anatomical_registration"],
@@ -2409,62 +2437,18 @@ def register_FSL_anat_to_template(wf, cfg, strat_pool, pipe_num, opt=None):
         },
     },
 )
-def register_symmetric_FSL_anat_to_template(wf, cfg, strat_pool, pipe_num, opt=None):
+def register_symmetric_FSL_anat_to_template(
+    wf: pe.Workflow,
+    cfg: Configuration,
+    strat_pool: ResourcePool,
+    pipe_num: int,
+    opt: str = "",
+) -> NODEBLOCK_RETURN:
     """Register T1w to symmetric template with FSL."""
-    fsl, outputs = FSL_registration_connector(
-        f"register_{opt}_anat_to_template_symmetric_{pipe_num}",
-        cfg,
-        orig="T1w",
-        opt=opt,
-        symmetric=True,
+    return _register_FSL_anat_to_template(
+        wf, cfg, strat_pool, pipe_num, opt, symmetric=True
     )
 
-    fsl.inputs.inputspec.interpolation = cfg.registration_workflows[
-        "anatomical_registration"
-    ]["registration"]["FSL-FNIRT"]["interpolation"]
-
-    fsl.inputs.inputspec.fnirt_config = cfg.registration_workflows[
-        "anatomical_registration"
-    ]["registration"]["FSL-FNIRT"]["fnirt_config"]
-
-    connect, brain = strat_pool.get_data(
-        ["desc-brain_T1w", "longitudinal-template_space-longitudinal_desc-brain_T1w"],
-        report_fetched=True,
-    )
-    node, out = connect
-    wf.connect(node, out, fsl, "inputspec.input_brain")
-
-    node, out = strat_pool.get_data("T1w-brain-template-symmetric")
-    wf.connect(node, out, fsl, "inputspec.reference_brain")
-
-    node, out = strat_pool.get_data(
-        [
-            "desc-preproc_T1w",
-            "longitudinal-template_space-longitudinal_desc-reorient_T1w",
-            "longitudinal-template_space-longitudinal_desc-head_T1w",
-        ]
-    )
-    wf.connect(node, out, fsl, "inputspec.input_head")
-
-    node, out = strat_pool.get_data("T1w-template-symmetric")
-    wf.connect(node, out, fsl, "inputspec.reference_head")
-
-    node, out = strat_pool.get_data("dilated-symmetric-brain-mask")
-    wf.connect(node, out, fsl, "inputspec.reference_mask")
-
-    if "space-longitudinal" in brain:
-        for key in list(outputs.keys()):
-            if "from-T1w" in key:
-                new_key = key.replace("from-T1w", "from-longitudinal")
-                outputs[new_key] = outputs[key]
-                del outputs[key]
-            if "to-T1w" in key:
-                new_key = key.replace("to-T1w", "to-longitudinal")
-                outputs[new_key] = outputs[key]
-                del outputs[key]
-
-    return (wf, outputs)
-
 
 @nodeblock(
     name="register_FSL_EPI_to_template",