Skip to content
Open
28 changes: 28 additions & 0 deletions brainles_preprocessing/modality.py
Original file line number Diff line number Diff line change
Expand Up @@ -239,6 +239,7 @@ def register(
registration_dir: Union[str, Path],
moving_image_name: str,
step: PreprocessorSteps,
interpolator: Optional[str] = None,
) -> Path:
"""
Register the current modality to a fixed image using the specified registrator.
Expand All @@ -249,9 +250,13 @@ def register(
registration_dir (str or Path): Directory to store registration results.
moving_image_name (str): Name of the moving image.
step (PreprocessorSteps): The current preprocessing step.
interpolator (Optional[str]): Interpolation method used during registration. Only supported by ANTsRegistrator; omit it for other registrators to use the backend default.

Returns:
Path: Path to the registration matrix.

Raises:
ValueError: If an interpolator is passed for a registrator other than ANTsRegistrator.
"""
fixed_image_path = Path(fixed_image_path)
registration_dir = Path(registration_dir)
Expand All @@ -262,12 +267,21 @@ def register(
# Note, add file ending depending on registration backend!
registered_matrix = registration_dir / f"M_{moving_image_name}"

if interpolator is not None and not isinstance(registrator, ANTsRegistrator):
raise ValueError(
f"'interpolator' is only supported by ANTsRegistrator for registration, "
f"got {type(registrator).__name__}. Omit it to use the backend default."
)

interpolator_kwargs = {"interpolator": interpolator} if interpolator else {}

registrator.register(
fixed_image_path=fixed_image_path,
moving_image_path=self.current,
transformed_image_path=registered,
matrix_path=registered_matrix,
log_file_path=str(registered_log),
**interpolator_kwargs,
Comment thread
LMZimmer marked this conversation as resolved.
)
self.current = registered
self.steps[step] = registered
Expand Down Expand Up @@ -364,6 +378,7 @@ def transform(
moving_image_name: str,
transformation_matrix_path: Union[str, Path],
step: PreprocessorSteps,
interpolator: Optional[str] = None,
) -> None:
"""
Transform the current modality using the specified registrator and transformation matrix.
Expand All @@ -375,8 +390,12 @@ def transform(
moving_image_name (str): Name of the moving image.
transformation_matrix_path (str or Path): Path to the transformation matrix.
step (PreprocessorSteps): The current preprocessing step.
interpolator (Optional[str]): Interpolation method used during transformation. "linear" and "nearestNeighbor" are supported by ANTsRegistrator, NiftyRegRegistrator and GreedyRegistrator; further options depend on the chosen registrator. ElastixRegistrator does not support choosing an interpolator.
Returns:
None

Raises:
ValueError: If an interpolator is passed for a registrator that does not support it.
"""
fixed_image_path = Path(fixed_image_path)
registration_dir_path = Path(registration_dir_path)
Expand All @@ -385,6 +404,12 @@ def transform(
transformed = registration_dir_path / f"{moving_image_name}.nii.gz"
transformed_log = registration_dir_path / f"{moving_image_name}.log"

interpolator_kwargs = (
{"interpolator": registrator.normalize_interpolator(interpolator)}
if interpolator
else {}
)

if (
isinstance(registrator, (ANTsRegistrator, NiftyRegRegistrator))
and step == PreprocessorSteps.ATLAS_REGISTERED
Expand All @@ -406,6 +431,7 @@ def transform(
transformation_matrix_path, # atlas registration matrix
],
log_file_path=str(transformed_log),
**interpolator_kwargs,
)
else:
registrator.transform(
Expand All @@ -414,6 +440,7 @@ def transform(
transformed_image_path=str(transformed),
matrix_path=str(transformation_matrix_path),
log_file_path=str(transformed_log),
**interpolator_kwargs,
)

self.current = transformed
Expand Down Expand Up @@ -707,6 +734,7 @@ def deface(
transformed_image_path=mask_path,
matrix_path=atlas_bet_M,
log_file_path=defaced_dir_path / "inverse_transform.log",
interpolator="nearestNeighbor",
)
else:
defacer.deface(
Expand Down
50 changes: 41 additions & 9 deletions brainles_preprocessing/registration/ANTs/ANTs.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@


class ANTsRegistrator(Registrator):
INTERPOLATOR_MAP = {"linear": "linear", "nearestNeighbor": "nearestNeighbor"}

def __init__(
self,
registration_params: Optional[Dict[str, Any]] = None,
Expand Down Expand Up @@ -61,6 +63,7 @@ def register(
transformed_image_path: Union[str, Path],
matrix_path: Union[str, Path],
log_file_path: Union[str, Path],
interpolator: Optional[str] = None,
**kwargs,
) -> None:
"""
Expand All @@ -72,6 +75,7 @@ def register(
transformed_image_path (str or Path): Path to the transformed image (output).
matrix_path (str or Path): Path to the transformation matrix (output).
log_file_path (str or Path): Path to the log file.
interpolator (Optional[str]): Interpolator to use for the transformation. Defaults to the instantiated transformation parameters, or 'linear' if not set.
**kwargs: Additional registration parameters to update the instantiated defaults.

Raises:
Expand All @@ -83,6 +87,10 @@ def register(
# We update the registration parameters with the provided kwargs
registration_kwargs = {**self.registration_params, **kwargs}

interpolator = self._resolve_interpolator(
interpolator, dict(self.transformation_params)
)

# Convert all paths to Path objects
fixed_image_path = Path(fixed_image_path)
moving_image_path = Path(moving_image_path)
Expand Down Expand Up @@ -123,6 +131,7 @@ def register(
transformed_image_path=transformed_image_path,
matrix_path=matrix_path,
log_file_path=log_file_path,
interpolator=interpolator,
)

end_time = datetime.datetime.now()
Expand All @@ -136,6 +145,7 @@ def register(
moving_image_path=moving_image_path,
transformed_image_path=transformed_image_path,
matrix_path=matrix_path,
interpolator=interpolator,
operation_name="registration",
start_time=start_time,
end_time=end_time,
Expand All @@ -148,7 +158,7 @@ def transform(
transformed_image_path: Union[str, Path],
matrix_path: str | Path | List[str | Path],
log_file_path: Union[str, Path],
interpolator: str = "nearestNeighbor",
interpolator: Optional[str] = None,
**kwargs,
) -> None:
"""
Expand All @@ -161,23 +171,20 @@ def transform(
transformed_image_path (str or Path): Path to the transformed image (output).
matrix_path (str or Path or List[str | Path]): Path to the transformation matrix or a list of matrices.
log_file_path (str or Path): Path to the log file.
interpolator (str): Interpolator to use for the transformation. Default is 'nearestNeighbor'.
interpolator (Optional[str]): Interpolator to use for the transformation. Defaults to the instantiated transformation parameters, or 'linear' if not set.
**kwargs: Additional transformation parameters to update the instantiated defaults.
Raises:
AssertionError: If the interpolator is not valid.
FileNotFoundError: If the fixed or moving image paths do not exist.
"""
start_time = datetime.datetime.now()

assert interpolator in VALID_INTERPOLATORS, (
f"Invalid interpolator: {interpolator}. "
f"Valid options are: {', '.join(VALID_INTERPOLATORS)}."
)

# TODO - self.transformation_params
# we update the transformation parameters with the provided kwargs
transform_kwargs = {**self.transformation_params, **kwargs}

interpolator = self._resolve_interpolator(interpolator, transform_kwargs)

# Convert all paths to Path objects
fixed_image_path = Path(fixed_image_path)
moving_image_path = Path(moving_image_path)
Expand Down Expand Up @@ -243,7 +250,7 @@ def inverse_transform(
transformed_image_path: Union[str, Path],
matrix_path: str | Path | List[str | Path],
log_file_path: Union[str, Path],
interpolator: str = "nearestNeighbor",
interpolator: Optional[str] = None,
Comment thread
LMZimmer marked this conversation as resolved.
**kwargs,
) -> None:
"""
Expand All @@ -255,7 +262,7 @@ def inverse_transform(
transformed_image_path (str or Path): Path to the transformed image (output).
matrix_path (str or Path or List[str | Path]): Path to the transformation matrix or a list of matrices.
log_file_path (str or Path): Path to the log file.
interpolator (str): Interpolator to use for the transformation. Default is 'nearestNeighbor'.
interpolator (Optional[str]): Interpolator to use for the transformation. Defaults to the instantiated transformation parameters, or 'linear' if not set.
**kwargs: Additional transformation parameters to update the instantiated defaults.
"""
if not isinstance(matrix_path, list):
Expand All @@ -271,6 +278,31 @@ def inverse_transform(
**kwargs,
)

def _resolve_interpolator(
self, interpolator: Optional[str], transform_kwargs: Dict[str, Any]
) -> str:
"""
Resolve the interpolator to use. The argument takes precedence over the instantiated transformation parameters, which default to 'linear'.

Args:
interpolator (Optional[str]): Interpolator passed to the calling method.
transform_kwargs (Dict[str, Any]): Transformation parameters; the 'interpolator' key is removed.

Returns:
str: The validated interpolator.

Raises:
AssertionError: If the interpolator is not valid.
"""
default_interpolator = transform_kwargs.pop("interpolator", "linear")
interpolator = self.normalize_interpolator(interpolator or default_interpolator)

assert interpolator in VALID_INTERPOLATORS, (
f"Invalid interpolator: {interpolator}. "
f"Valid options are: {', '.join(VALID_INTERPOLATORS)}."
)
return interpolator

@staticmethod
def _log_to_file(
log_file_path: Union[str, Path],
Expand Down
9 changes: 8 additions & 1 deletion brainles_preprocessing/registration/elastix/elastix.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ def transform(
transformed_image_path: str,
matrix_path: str,
log_file_path: Optional[str] = None,
interpolator: str = "linear",
interpolator: Optional[str] = None,
**kwargs
) -> None:
"""
Expand All @@ -92,7 +92,14 @@ def transform(
transformed_image_path (str): Path to the transformed image (output).
matrix_path (str): Path to the transformation matrix (output). This gets overwritten if it already exists.
log_file_path (Optional[str]): Path to the log file.
interpolator (Optional[str]): Not supported by elastix. Must be None.

Raises:
ValueError: If an interpolator is provided.
"""
if interpolator is not None:
self.normalize_interpolator(interpolator)

parameter_object = self.__initialize_parameter_object()

# check if the matrix file exists
Expand Down
13 changes: 11 additions & 2 deletions brainles_preprocessing/registration/greedy/greedy.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,12 @@
from brainles_preprocessing.registration.registrator import Registrator
from brainles_preprocessing.utils import check_and_add_suffix

VALID_INTERPOLATORS = ["NN", "LINEAR", "LABEL"]


class GreedyRegistrator(Registrator):
INTERPOLATOR_MAP = {"linear": "LINEAR", "nearestNeighbor": "NN"}

def __init__(
self,
):
Expand Down Expand Up @@ -72,10 +76,15 @@ def transform(
transformed_image_path (str): Path to the transformed image (output).
matrix_path (str): Path to the transformation matrix (output). This gets overwritten if it already exists.
log_file_path (Optional[str]): Path to the log file, which is not used.
interpolator (Optional[str]): The interpolator to use; one of NN, LINEAR or LABEL. Defaults to LINEAR.
interpolator (Optional[str]): The interpolator to use; one of NN, LINEAR or LABEL, or "linear"/"nearestNeighbor". Defaults to LINEAR.
Raises:
AssertionError: If the interpolator is not valid.
"""
registor = Greedy3D()
interpolator_upper = interpolator.upper()
interpolator_upper = self.normalize_interpolator(interpolator).upper()
assert (
interpolator_upper in VALID_INTERPOLATORS
), f"Invalid interpolator: {interpolator}. Valid options are: {', '.join(VALID_INTERPOLATORS)}"
if "LABEL" in interpolator_upper:
interpolator_upper += " 0.3vox"

Expand Down
9 changes: 6 additions & 3 deletions brainles_preprocessing/registration/niftyreg/niftyreg.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@


class NiftyRegRegistrator(Registrator):
INTERPOLATOR_MAP = {"linear": "1", "nearestNeighbor": "0"}

def __init__(
self,
registration_abspath: str = os.path.dirname(os.path.abspath(__file__)),
Expand Down Expand Up @@ -155,7 +157,7 @@ def transform(
transformed_image_path: str,
matrix_path: str | Path | List[str | Path],
log_file_path: str,
interpolator: str = "0",
interpolator: str = "1",
**kwargs: dict,
) -> None:
"""
Expand All @@ -168,10 +170,11 @@ def transform(
transformed_image_path (str): Path to the transformed image (output).
matrix_path (str | Path | List[str | Path]): Path(s) to the transformation matrix(es).
log_file_path (str): Path to the log file.
interpolator (str): Interpolation order (0, 1, 3, 4) (0=NN, 1=LIN; 3=CUB, 4=SINC). Default is '1' (linear).
interpolator (str): Interpolation order (0, 1, 3, 4) (0=NN, 1=LIN; 3=CUB, 4=SINC) or "linear"/"nearestNeighbor". Default is '1' (linear).
Raises:
AssertionError: If the interpolator is not valid.
"""
interpolator = self.normalize_interpolator(interpolator)
assert (
interpolator in VALID_INTERPOLATORS
), f"Invalid interpolator: {interpolator}. Valid options are: {', '.join([f'{k} ({v})' for k, v in VALID_INTERPOLATORS.items()])}"
Expand Down Expand Up @@ -236,7 +239,7 @@ def inverse_transform(
transformed_image_path: str,
matrix_path: str | Path | List[str | Path],
log_file_path: str,
interpolator: str = "0",
interpolator: str = "1",
) -> None:
"""
Apply inverse transformation using NiftyReg.
Expand Down
28 changes: 26 additions & 2 deletions brainles_preprocessing/registration/registrator.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,36 @@
from abc import ABC, abstractmethod
from typing import Any
from typing import Any, Dict, Optional


class Registrator(ABC):
# TODO probably the init here should be removed?
# def __init__(self, backend):
# self.backend = backend

# Maps the shared interpolator names ("linear", "nearestNeighbor") to the backend-specific value.
# None means the backend does not support choosing an interpolator.
INTERPOLATOR_MAP: Optional[Dict[str, str]] = None

@classmethod
def normalize_interpolator(cls, interpolator: str) -> str:
"""
Map a shared interpolator name to the backend-specific value. Backend-specific values are passed through unchanged.

Args:
interpolator (str): Shared name ("linear", "nearestNeighbor") or backend-specific value.

Returns:
str: The backend-specific interpolator value.

Raises:
ValueError: If the backend does not support choosing an interpolator.
"""
if cls.INTERPOLATOR_MAP is None:
raise ValueError(
f"{cls.__name__} does not support choosing an interpolator."
)
return cls.INTERPOLATOR_MAP.get(interpolator, interpolator)

@abstractmethod
def register(
self,
Expand Down Expand Up @@ -37,7 +61,7 @@ def transform(
matrix_path: Any,
log_file_path: str,
interpolator: str = "default",
**kwargs
**kwargs,
):
"""
Abstract method for transforming images.
Expand Down
Loading
Loading