Use argument-list subprocess calls for NiftyReg commands

brainreg/core/backend/niftyreg/registration.py

@@ -38,20 +38,25 @@ def __init__(self, paths, registration_params, n_processes=None):
         self.hemispheres_img_path = paths.hemispheres
 
     def _prepare_openmp_thread_flag(self):
-        self.openmp_flag = "-omp {}".format(self.n_processes)
+        self.openmp_flag = ["-omp", str(self.n_processes)]
 
     def _prepare_affine_reg_cmd(self):
-        cmd = "{} {} -flo {} -ref {} -aff {} -res {}".format(
+        cmd = [
             self.reg_params.affine_reg_program_path,
-            self.reg_params.format_affine_params().strip(),
-            '"' + self.brain_of_atlas_img_path + '"',
-            '"' + self.dataset_img_path + '"',
-            '"' + self.paths.affine_matrix_path + '"',
-            '"' + self.paths.affine_registered_atlas_brain_path + '"',
-        )
+            *self.reg_params.format_affine_params().split(),
+            "-flo",
+            self.brain_of_atlas_img_path,
+            "-ref",
+            self.dataset_img_path,
+            "-aff",
+            self.paths.affine_matrix_path,
+            "-res",
+            self.paths.affine_registered_atlas_brain_path,
+        ]
 
         if self.n_processes is not None:
-            cmd += " " + self.openmp_flag
+            cmd.extend(self.openmp_flag)
+
         return cmd
 
     def register_affine(self):
@@ -75,18 +80,24 @@ def register_affine(self):
             )
 
     def _prepare_freeform_reg_cmd(self):
-        cmd = "{} {} -aff {} -flo {} -ref {} -cpp {} -res {}".format(
+        cmd = [
             self.reg_params.freeform_reg_program_path,
-            self.reg_params.format_freeform_params().strip(),
-            '"' + self.paths.affine_matrix_path + '"',
-            '"' + self.brain_of_atlas_img_path + '"',
-            '"' + self.dataset_img_path + '"',
-            '"' + self.paths.control_point_file_path + '"',
-            '"' + self.paths.freeform_registered_atlas_brain_path + '"',
-        )
+            *self.reg_params.format_freeform_params().split(),
+            "-aff",
+            self.paths.affine_matrix_path,
+            "-flo",
+            self.brain_of_atlas_img_path,
+            "-ref",
+            self.dataset_img_path,
+            "-cpp",
+            self.paths.control_point_file_path,
+            "-res",
+            self.paths.freeform_registered_atlas_brain_path,
+        ]
 
         if self.n_processes is not None:
-            cmd += " " + self.openmp_flag
+            cmd.extend(self.openmp_flag)
+
         return cmd
 
     def register_freeform(self):
@@ -114,12 +125,12 @@ def generate_inverse_transforms(self):
         self.register_inverse_freeform()
 
     def _prepare_invert_affine_cmd(self):
-        cmd = "{} -invAff {} {}".format(
+        return [
             self.reg_params.transform_program_path,
-            '"' + self.paths.affine_matrix_path + '"',
-            '"' + self.paths.invert_affine_matrix_path + '"',
-        )
-        return cmd
+            "-invAff",
+            self.paths.affine_matrix_path,
+            self.paths.invert_affine_matrix_path,
+        ]
 
     def generate_inverse_affine(self):
         """
@@ -144,20 +155,24 @@ def generate_inverse_affine(self):
             )
 
     def _prepare_inverse_freeform_reg_cmd(self):
-        cmd = "{} {} -aff {} -flo {} -ref {} -cpp {} -res {}".format(
+        cmd = [
             self.reg_params.freeform_reg_program_path,
-            self.reg_params.format_freeform_params().strip(),
-            '"' + self.paths.invert_affine_matrix_path + '"',
-            '"' + self.dataset_img_path + '"',
-            '"' + self.brain_of_atlas_img_path + '"',
-            '"' + self.paths.inverse_control_point_file_path + '"',
-            '"'
-            + self.paths.inverse_freeform_registered_atlas_brain_path
-            + '"',
-        )
+            *self.reg_params.format_freeform_params().split(),
+            "-aff",
+            self.paths.invert_affine_matrix_path,
+            "-flo",
+            self.dataset_img_path,
+            "-ref",
+            self.brain_of_atlas_img_path,
+            "-cpp",
+            self.paths.inverse_control_point_file_path,
+            "-res",
+            self.paths.inverse_freeform_registered_atlas_brain_path,
+        ]
 
         if self.n_processes is not None:
-            cmd += " " + self.openmp_flag
+            cmd.extend(self.openmp_flag)
+
         return cmd
 
     def register_inverse_freeform(self):
@@ -183,37 +198,44 @@ def register_inverse_freeform(self):
             )
 
     def _prepare_segmentation_cmd(self, floating_image_path, dest_img_path):
-        cmd = "{} {} -cpp {} -flo {} -ref {} -res {}".format(
+        return [
             self.reg_params.segmentation_program_path,
-            self.reg_params.format_segmentation_params().strip(),
-            '"' + self.paths.control_point_file_path + '"',
-            '"' + floating_image_path + '"',
-            '"' + self.dataset_img_path + '"',
-            '"' + dest_img_path + '"',
-        )
-        return cmd
+            *self.reg_params.format_segmentation_params().split(),
+            "-cpp",
+            self.paths.control_point_file_path,
+            "-flo",
+            floating_image_path,
+            "-ref",
+            self.dataset_img_path,
+            "-res",
+            dest_img_path,
+        ]
 
     def _prepare_inverse_registration_cmd(
         self, floating_image_path, dest_img_path
     ):
-        cmd = "{} {} -cpp {} -flo {} -ref {} -res {}".format(
+        return [
             self.reg_params.segmentation_program_path,
-            self.reg_params.format_segmentation_params().strip(),
-            '"' + self.paths.inverse_control_point_file_path + '"',
-            '"' + floating_image_path + '"',
-            '"' + self.brain_of_atlas_img_path + '"',
-            '"' + dest_img_path + '"',
-        )
-        return cmd
+            *self.reg_params.format_segmentation_params().split(),
+            "-cpp",
+            self.paths.inverse_control_point_file_path,
+            "-flo",
+            floating_image_path,
+            "-ref",
+            self.brain_of_atlas_img_path,
+            "-res",
+            dest_img_path,
+        ]
 
     def _prepare_deformation_field_cmd(self, deformation_field_path):
-        cmd = "{} -def {} {} -ref {}".format(
+        return [
             self.reg_params.transform_program_path,
-            '"' + self.paths.control_point_file_path + '"',
-            '"' + deformation_field_path + '"',
-            '"' + self.paths.downsampled_filtered + '"',
-        )
-        return cmd
+            "-def",
+            self.paths.control_point_file_path,
+            deformation_field_path,
+            "-ref",
+            self.paths.downsampled_filtered,
+        ]
 
     def segment(self):
         """
@@ -234,7 +256,7 @@ def segment(self):
                 self.paths.segmentation_error_file,
             )
         except SafeExecuteCommandError as err:
-            SegmentationError("Segmentation failed; {}".format(err))
+            raise SegmentationError("Segmentation failed; {}".format(err))
 
     def register_hemispheres(self):
         """
@@ -256,7 +278,7 @@ def register_hemispheres(self):
                 self.paths.segmentation_error_file,
             )
         except SafeExecuteCommandError as err:
-            SegmentationError("Segmentation failed; {}".format(err))
+            raise SegmentationError("Segmentation failed; {}".format(err))
 
     def transform_to_standard_space(self, image_path, destination_path):
         """

pyproject.toml

@@ -21,7 +21,7 @@ requires-python = ">=3.11"
 dependencies = [
     "brainglobe-atlasapi>=2.0.1",
     "brainglobe-space>=1.0.0",
-    "brainglobe-utils>=0.5.0",
+    "brainglobe-utils>=0.10.0",
     "fancylog>=0.6.0",
     "numpy",
     "scikit-image>=0.24.0",

tests/tests/test_backend/test_registration.py

@@ -0,0 +1,65 @@
+from unittest.mock import Mock, patch
+
+import pytest
+from brainglobe_utils.general.system import SafeExecuteCommandError
+
+from brainreg.core.backend.niftyreg.registration import (
+    BrainRegistration,
+    SegmentationError,
+)
+
+
+def _make_registration():
+    """
+    Create a minimal BrainRegistration instance with mocked paths
+    and registration parameters.
+    """
+    paths = Mock()
+    paths.segmentation_log_file = "seg.log"
+    paths.segmentation_error_file = "seg.err"
+    paths.registered_atlas_img_path = "registered_atlas.nii"
+    paths.control_point_file_path = "cpp.nii"
+    paths.downsampled_filtered = "ref.nii"
+    paths.annotations = "atlas.nii"
+
+    reg_params = Mock()
+    reg_params.segmentation_program_path = "reg_resample"
+    reg_params.format_segmentation_params.return_value = ""
+
+    return BrainRegistration(
+        paths=paths,
+        registration_params=reg_params,
+        n_processes=None,
+    )
+
+
+def test_segment_raises_segmentation_error():
+    """
+    Ensure SegmentationError is raised when the command fails.
+    """
+    reg = _make_registration()
+
+    with patch(
+        "brainreg.core.backend.niftyreg.registration.safe_execute_command",
+        side_effect=SafeExecuteCommandError("command failed"),
+    ):
+        with pytest.raises(SegmentationError):
+            reg.segment()
+
+
+def test_register_hemispheres_raises_segmentation_error():
+    """
+    Ensure SegmentationError is raised when hemisphere registration fails.
+    """
+    reg = _make_registration()
+
+    # extra path used by register_hemispheres
+    reg.paths.hemispheres = "hemi.nii"
+    reg.paths.registered_hemispheres_img_path = "registered_hemi.nii"
+
+    with patch(
+        "brainreg.core.backend.niftyreg.registration.safe_execute_command",
+        side_effect=SafeExecuteCommandError("command failed"),
+    ):
+        with pytest.raises(SegmentationError):
+            reg.register_hemispheres()

tests/tests/test_integration/test_registration.py

@@ -84,6 +84,34 @@ def hemisphere_output_path(tmp_path_factory):
     return test_output_dir
 
 
+@pytest.fixture(scope="session")
+def whole_brain_output_path_with_spaces(tmp_path_factory):
+    test_output_dir = tmp_path_factory.mktemp("output dir with spaces")
+
+    brainreg_args = [
+        "brainreg",
+        str(whole_brain_data_dir),
+        str(test_output_dir),
+        "-v",
+        whole_brain_voxel_sizes[0],
+        whole_brain_voxel_sizes[1],
+        whole_brain_voxel_sizes[2],
+        "--orientation",
+        "psl",
+        "--n-free-cpus",
+        "0",
+        "--atlas",
+        "allen_mouse_100um",
+        "-a",
+        str(whole_brain_data_dir),
+    ]
+
+    sys.argv = brainreg_args
+    brainreg_run()
+
+    return test_output_dir
+
+
 @pytest.mark.parametrize(
     "image",
     [
@@ -126,3 +154,21 @@ def test_hemisphere(hemisphere_output_path, image):
         image, hemisphere_output_path, hemisphere_expected_output_dir
     )
     check_volumes_equal(hemisphere_output_path, hemisphere_expected_output_dir)
+
+
+def test_whole_brain_with_space_in_output_path(
+    whole_brain_output_path_with_spaces,
+):
+    """
+    Ensure brainreg runs successfully when output path contains spaces.
+    """
+    expected_files = [
+        "registered_atlas.tiff",
+        "registered_hemispheres.tiff",
+        "downsampled_standard.tiff",
+    ]
+
+    for fname in expected_files:
+        assert (
+            whole_brain_output_path_with_spaces / fname
+        ).exists(), f"{fname} was not created"