Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return super(ANTSRegistrationRPT, self)._post_run_hook(runtime)
class _ANTSApplyTransformsInputSpecRPT(
nrc._SVGReportCapableInputSpec, ants.resampling.ApplyTransformsInputSpec
):
pass
class _ANTSApplyTransformsOutputSpecRPT(
reporting.ReportCapableOutputSpec, ants.resampling.ApplyTransformsOutputSpec
):
pass
class ANTSApplyTransformsRPT(nrc.RegistrationRC, ApplyTransforms):
input_spec = _ANTSApplyTransformsInputSpecRPT
output_spec = _ANTSApplyTransformsOutputSpecRPT
def _post_run_hook(self, runtime):
self._fixed_image = self.inputs.reference_image
self._moving_image = self.aggregate_outputs(runtime=runtime).output_image
NIWORKFLOWS_LOG.info(
"Report - setting fixed (%s) and moving (%s) images",
self._fixed_image,
self._moving_image,
)
return super(ANTSApplyTransformsRPT, self)._post_run_hook(runtime)
class _ApplyTOPUPInputSpecRPT(
return super(RobustMNINormalizationRPT, self)._post_run_hook(runtime)
class _ANTSRegistrationInputSpecRPT(
nrc._SVGReportCapableInputSpec, ants.registration.RegistrationInputSpec
):
pass
class _ANTSRegistrationOutputSpecRPT(
reporting.ReportCapableOutputSpec, ants.registration.RegistrationOutputSpec
):
pass
class ANTSRegistrationRPT(nrc.RegistrationRC, Registration):
input_spec = _ANTSRegistrationInputSpecRPT
output_spec = _ANTSRegistrationOutputSpecRPT
def _post_run_hook(self, runtime):
self._fixed_image = self.inputs.fixed_image[0]
self._moving_image = self.aggregate_outputs(runtime=runtime).warped_image
NIWORKFLOWS_LOG.info(
"Report - setting fixed (%s) and moving (%s) images",
self._fixed_image,
self._moving_image,
)
return super(ANTSRegistrationRPT, self)._post_run_hook(runtime)
class _ANTSApplyTransformsInputSpecRPT(
self._fixed_image,
self._moving_image,
)
return super(MRICoregRPT, self)._post_run_hook(runtime)
class _SimpleBeforeAfterInputSpecRPT(nrc._SVGReportCapableInputSpec):
before = File(exists=True, mandatory=True, desc="file before")
after = File(exists=True, mandatory=True, desc="file after")
wm_seg = File(desc="reference white matter segmentation mask")
before_label = traits.Str("before", usedefault=True)
after_label = traits.Str("after", usedefault=True)
class SimpleBeforeAfterRPT(nrc.RegistrationRC, nrc.ReportingInterface):
input_spec = _SimpleBeforeAfterInputSpecRPT
def _post_run_hook(self, runtime):
""" there is not inner interface to run """
self._fixed_image_label = self.inputs.after_label
self._moving_image_label = self.inputs.before_label
self._fixed_image = self.inputs.after
self._moving_image = self.inputs.before
self._contour = self.inputs.wm_seg if isdefined(self.inputs.wm_seg) else None
NIWORKFLOWS_LOG.info(
"Report - setting before (%s) and after (%s) images",
self._fixed_image,
self._moving_image,
)
return super(SimpleBeforeAfterRPT, self)._post_run_hook(runtime)
)
return super(ApplyTOPUPRPT, self)._post_run_hook(runtime)
class _FUGUEInputSpecRPT(nrc._SVGReportCapableInputSpec, fsl.preprocess.FUGUEInputSpec):
wm_seg = File(argstr="-wmseg %s", desc="reference white matter segmentation mask")
class _FUGUEOutputSpecRPT(
reporting.ReportCapableOutputSpec, fsl.preprocess.FUGUEOutputSpec
):
pass
class FUGUERPT(nrc.RegistrationRC, fsl.FUGUE):
input_spec = _FUGUEInputSpecRPT
output_spec = _FUGUEOutputSpecRPT
def _post_run_hook(self, runtime):
self._fixed_image_label = "after"
self._moving_image_label = "before"
self._fixed_image = self.aggregate_outputs(runtime=runtime).unwarped_file
self._moving_image = self.inputs.in_file
self._contour = self.inputs.wm_seg if isdefined(self.inputs.wm_seg) else None
NIWORKFLOWS_LOG.info(
"Report - setting corrected (%s) and warped (%s) images",
self._fixed_image,
self._moving_image,
)
return super(FUGUERPT, self)._post_run_hook(runtime)
return super(ANTSApplyTransformsRPT, self)._post_run_hook(runtime)
class _ApplyTOPUPInputSpecRPT(
nrc._SVGReportCapableInputSpec, fsl.epi.ApplyTOPUPInputSpec
):
wm_seg = File(argstr="-wmseg %s", desc="reference white matter segmentation mask")
class _ApplyTOPUPOutputSpecRPT(
reporting.ReportCapableOutputSpec, fsl.epi.ApplyTOPUPOutputSpec
):
pass
class ApplyTOPUPRPT(nrc.RegistrationRC, fsl.ApplyTOPUP):
input_spec = _ApplyTOPUPInputSpecRPT
output_spec = _ApplyTOPUPOutputSpecRPT
def _post_run_hook(self, runtime):
self._fixed_image_label = "after"
self._moving_image_label = "before"
self._fixed_image = index_img(
self.aggregate_outputs(runtime=runtime).out_corrected, 0
)
self._moving_image = index_img(self.inputs.in_files[0], 0)
self._contour = self.inputs.wm_seg if isdefined(self.inputs.wm_seg) else None
NIWORKFLOWS_LOG.info(
"Report - setting corrected (%s) and warped (%s) images",
self._fixed_image,
self._moving_image,
)
File,
default=True,
usedefault=True,
argstr="--lta %s",
min_ver="5.2.0",
desc="write the transformation matrix in LTA format",
)
class _BBRegisterOutputSpecRPT(
reporting.ReportCapableOutputSpec, fs.preprocess.BBRegisterOutputSpec
):
pass
class BBRegisterRPT(nrc.RegistrationRC, fs.BBRegister):
input_spec = _BBRegisterInputSpecRPT
output_spec = _BBRegisterOutputSpecRPT
def _post_run_hook(self, runtime):
outputs = self.aggregate_outputs(runtime=runtime)
mri_dir = os.path.join(self.inputs.subjects_dir, self.inputs.subject_id, "mri")
target_file = os.path.join(mri_dir, "brainmask.mgz")
# Apply transform for simplicity
mri_vol2vol = fs.ApplyVolTransform(
source_file=self.inputs.source_file,
target_file=target_file,
lta_file=outputs.out_lta_file,
interp="nearest",
)
res = mri_vol2vol.run()
return super(BBRegisterRPT, self)._post_run_hook(runtime)
class _MRICoregInputSpecRPT(
nrc._SVGReportCapableInputSpec, fs.registration.MRICoregInputSpec
):
pass
class _MRICoregOutputSpecRPT(
reporting.ReportCapableOutputSpec, fs.registration.MRICoregOutputSpec
):
pass
class MRICoregRPT(nrc.RegistrationRC, fs.MRICoreg):
input_spec = _MRICoregInputSpecRPT
output_spec = _MRICoregOutputSpecRPT
def _post_run_hook(self, runtime):
outputs = self.aggregate_outputs(runtime=runtime)
mri_dir = None
if isdefined(self.inputs.subject_id):
mri_dir = os.path.join(
self.inputs.subjects_dir, self.inputs.subject_id, "mri"
)
if isdefined(self.inputs.reference_file):
target_file = self.inputs.reference_file
else:
target_file = os.path.join(mri_dir, "brainmask.mgz")
)
class _RobustMNINormalizationInputSpecRPT(
nrc._SVGReportCapableInputSpec, _RobustMNINormalizationInputSpec
):
pass
class _RobustMNINormalizationOutputSpecRPT(
reporting.ReportCapableOutputSpec, _RobustMNINormalizationOutputSpec
):
pass
class RobustMNINormalizationRPT(nrc.RegistrationRC, RobustMNINormalization):
input_spec = _RobustMNINormalizationInputSpecRPT
output_spec = _RobustMNINormalizationOutputSpecRPT
def _post_run_hook(self, runtime):
# We need to dig into the internal ants.Registration interface
self._fixed_image = self._get_ants_args()["fixed_image"]
if isinstance(self._fixed_image, (list, tuple)):
self._fixed_image = self._fixed_image[0] # get first item if list
if self._get_ants_args().get("fixed_image_mask") is not None:
self._fixed_image_mask = self._get_ants_args().get("fixed_image_mask")
self._moving_image = self.aggregate_outputs(runtime=runtime).warped_image
NIWORKFLOWS_LOG.info(
"Report - setting fixed (%s) and moving (%s) images",
self._fixed_image,
self._moving_image,
)
return super(FUGUERPT, self)._post_run_hook(runtime)
class _FLIRTInputSpecRPT(nrc._SVGReportCapableInputSpec, fsl.preprocess.FLIRTInputSpec):
pass
class _FLIRTOutputSpecRPT(
reporting.ReportCapableOutputSpec, fsl.preprocess.FLIRTOutputSpec
):
pass
class FLIRTRPT(nrc.RegistrationRC, fsl.FLIRT):
input_spec = _FLIRTInputSpecRPT
output_spec = _FLIRTOutputSpecRPT
def _post_run_hook(self, runtime):
self._fixed_image = self.inputs.reference
self._moving_image = self.aggregate_outputs(runtime=runtime).out_file
self._contour = self.inputs.wm_seg if isdefined(self.inputs.wm_seg) else None
NIWORKFLOWS_LOG.info(
"Report - setting fixed (%s) and moving (%s) images",
self._fixed_image,
self._moving_image,
)
return super(FLIRTRPT, self)._post_run_hook(runtime)