enh: implement an interface to compare costs function from registration with original and flipped functional image to detect left-right flip

enh: propagate the flip_info through the workflow so it can be integrated into the visual report

Author: celprov

fmriprep/interfaces/reports.py

@@ -218,6 +218,12 @@ class FunctionalSummaryInputSpec(TraitedSpec):
         desc='Whether to initialize registration with the "header"'
         ' or by centering the volumes ("t1w" or "t2w")',
     )
+    flip_info = traits.Dict(
+        traits.Enum('lr_flip_warning', 'cost_original', 'cost_flipped'),
+        traits.Either(traits.Bool(), traits.Float()),
+        desc='Left-right flip check warning and registration costs',
+        mandatory=True
+    )
     tr = traits.Float(desc='Repetition time', mandatory=True)
     dummy_scans = traits.Either(traits.Int(), None, desc='number of dummy scans specified by user')
     algo_dummy_scans = traits.Int(desc='number of dummy scans determined by algorithm')
@@ -369,3 +375,40 @@ def get_world_pedir(ornt, pe_direction):
         f'Orientation: {ornt}; PE dir: {pe_direction}'
     )
     return 'Could not be determined - assuming Anterior-Posterior'
+
+
+class _CheckFlipInputSpec(BaseInterfaceInputSpec):
+    cost_original = File(
+        exists=True,
+        mandatory=True,
+        desc='cost associated with registration of BOLD to original T1w images',
+    )
+    cost_flipped = File(
+        exists=True,
+        mandatory=True,
+        desc='cost associated with registration of BOLD to the flipped T1w images',
+    )
+
+
+class _CheckFlipOutputSpec(TraitedSpec):
+    flip_info = traits.Dict(
+        traits.Enum('warning', 'cost_original', 'cost_flipped'),
+        traits.Either(traits.Bool(), traits.Float()),
+        desc='Left-right flip check warning and registration costs',
+        mandatory=True
+    )
+
+
+class CheckFlip(SimpleInterface):
+    """Check for a LR flip by comparing registration cost functions."""
+
+    input_spec = _CheckFlipInputSpec
+    output_spec = _CheckFlipOutputSpec
+
+    def _run_interface(self, runtime):
+        self._results['flip_info'] = {
+            'warning': self.inputs.cost_flipped < self.inputs.cost_original,
+            'cost_original': self.inputs.cost_original,
+            'cost_flipped': self.inputs.cost_flipped,
+        }
+        return runtime

fmriprep/workflows/bold/registration.py

@@ -127,6 +127,8 @@ def init_bold_reg_wf(
         Affine transform from T1 space to BOLD space (ITK format)
     fallback
         Boolean indicating whether BBR was rejected (mri_coreg registration returned)
+    flip_info
+        Information regarding whether a left-right flip was detected
 
     See Also
     --------
@@ -153,7 +155,9 @@ def init_bold_reg_wf(
     )
 
     outputnode = pe.Node(
-        niu.IdentityInterface(fields=['itk_bold_to_t1', 'itk_t1_to_bold', 'fallback']),
+        niu.IdentityInterface(
+            fields=['itk_bold_to_t1', 'itk_t1_to_bold', 'fallback', 'flip_info']
+        ),
         name='outputnode',
     )
 
@@ -187,6 +191,7 @@ def init_bold_reg_wf(
             ('outputnode.itk_bold_to_t1', 'itk_bold_to_t1'),
             ('outputnode.itk_t1_to_bold', 'itk_t1_to_bold'),
             ('outputnode.fallback', 'fallback'),
+            ('outputnode.flip_info', 'flip_info'),
         ]),
     ])  # fmt:skip
 
@@ -267,6 +272,8 @@ def init_bbreg_wf(
         Affine transform from T1 space to BOLD space (ITK format)
     fallback
         Boolean indicating whether BBR was rejected (mri_coreg registration returned)
+    flip_info
+        Information regarding whether a left-right flip was detected
 
     """
     from nipype.interfaces.freesurfer import BBRegister
@@ -275,6 +282,7 @@ def init_bbreg_wf(
     from niworkflows.interfaces.nitransforms import ConcatenateXFMs
 
     from fmriprep.interfaces.patches import FreeSurferSource, MRICoreg
+    from fmriprep.interfaces.reports import CheckFlip
 
     workflow = Workflow(name=name)
     workflow.__desc__ = """\
@@ -309,7 +317,7 @@ def init_bbreg_wf(
         name='inputnode',
     )
     outputnode = pe.Node(
-        niu.IdentityInterface(['itk_bold_to_t1', 'itk_t1_to_bold', 'fallback']),
+        niu.IdentityInterface(['itk_bold_to_t1', 'itk_t1_to_bold', 'fallback', 'flip_info']),
         name='outputnode',
     )
 
@@ -358,6 +366,9 @@ def init_bbreg_wf(
         name='bbregister',
         mem_gb=12,
     )
+
+    check_flip = pe.Node(CheckFlip(), name='check_flip')
+
     transforms = pe.Node(niu.Merge(2), run_without_submitting=True, name='transforms')
     # In cases where Merge(2) only has `in1` or `in2` defined
     # output list will just contain a single element
@@ -411,7 +422,10 @@ def init_bbreg_wf(
                                          ('subject_id', 'subject_id')]),
         (inputnode, lr_flip), [('in_file', 'in_file')],
         (lr_flip, bbregister_flipped), [('out_file', 'source_file')],
+        (bbregister, check_flip), [('min_cost_file', 'cost_original')],
+        (bbregister_flipped, check_flip), [('min_cost_file', 'cost_flipped')],
         (bbregister, transforms, [('out_lta_file', 'in1')]),
+        (check_flip, outputnode, [('flip_info', 'flip_info')]),
     ])  # fmt:skip
 
     # Short-circuit workflow building, use boundary-based registration