Overwrite transform

CPAC/error_handler/exceptions.py

@@ -0,0 +1,14 @@
+class SchemaError(Exception):
+    """Exception raised for errors in the schema."""
+
+    def __init__(self, message):
+        self.message = message
+        super().__init__(self.message)
+
+
+class NodeBlockError(Exception):
+    """Exception raised for errors in the node block."""
+
+    def __init__(self, message):
+        self.message = message
+        super().__init__(self.message)

CPAC/pipeline/schema.py

@@ -1388,6 +1388,22 @@ def schema(config_dict):
                 " Try turning one option off.\n "
             )
             raise ExclusiveInvalid(msg)
+
+        overwrite = partially_validated["registration_workflows"][
+            "anatomical_registration"
+        ]["overwrite_transform"]
+
+        if (
+            overwrite["run"]
+            and overwrite["using"]
+            in partially_validated["registration_workflows"]["anatomical_registration"][
+                "registration"
+            ]["using"]
+        ):
+            raise ExclusiveInvalid(
+                "[!] Overwrite transform is found same as the anatomical registration method! "
+                "No need to overwrite transform with the same registration method."
+            )
     except KeyError:
         pass
     try:

CPAC/pipeline/test/test_schema_validation.py

@@ -113,3 +113,34 @@ def test_pipeline_name():
     """Test that pipeline_name sucessfully sanitizes."""
     c = Configuration({"pipeline_setup": {"pipeline_name": ":va:lid    name"}})
     assert c["pipeline_setup", "pipeline_name"] == "valid_name"
+
+
+@pytest.mark.parametrize(
+    "registration_using",
+    [
+        list(combo)
+        for _ in [
+            list(combinations(["ANTS", "FSL", "FSL-linear"], i)) for i in range(1, 4)
+        ]
+        for combo in _
+    ],
+)
+def test_overwrite_transform(registration_using):
+    """Test that if overwrite transform method is already a registration method."""
+    # pylint: disable=invalid-name
+    d = {
+        "registration_workflows": {
+            "anatomical_registration": {
+                "registration": {"using": registration_using},
+                "overwrite_transform": {"run": "On", "using": "FSL"},
+            }
+        }
+    }
+    if "FSL" not in registration_using:
+        Configuration(d)  # validates without exception
+    else:
+        with pytest.raises(ExclusiveInvalid) as e:
+            Configuration(d)
+        assert "Overwrite transform is found same as the registration method" in str(
+            e.value
+        )

CPAC/registration/registration.py

@@ -24,6 +24,7 @@
 from nipype.interfaces.afni import utils as afni_utils
 
 from CPAC.anat_preproc.lesion_preproc import create_lesion_preproc
+from CPAC.error_handler.exceptions import NodeBlockError
 from CPAC.func_preproc.utils import chunk_ts, split_ts_chunks
 from CPAC.pipeline import nipype_pipeline_engine as pe
 from CPAC.pipeline.nodeblock import nodeblock
@@ -3079,6 +3080,12 @@ def overwrite_transform_anat_to_template(wf, cfg, strat_pool, pipe_num, opt=None
             "from-template_to-T1w_mode-image_xfm": (merge_inv_xfms, "merged_file"),
         }
 
+    else:
+        outputs = {}
+        raise NodeBlockError(
+            "Invalid registration tool or option provided. Please make sure the registration tool is ANTs and the option is FSL."
+        )
+
     return (wf, outputs)