Add clinical preprocessing utilities for CT and MRI modalities.

monai/tests/test_clinical_preprocessing.py

@@ -0,0 +1,106 @@
+# Copyright (c) MONAI Consortium
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#     http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import numpy as np
+import pytest
+
+from monai.data import write_nifti
+from monai.transforms import EnsureChannelFirst, LoadImage, NormalizeIntensity, ScaleIntensityRange
+from monai.transforms.clinical_preprocessing import (
+    ModalityTypeError,
+    UnsupportedModalityError,
+    get_ct_preprocessing_pipeline,
+    get_mri_preprocessing_pipeline,
+    preprocess_dicom_series,
+)
+
+
+def test_ct_preprocessing_pipeline_structure():
+    """Test CT pipeline structure."""
+    pipeline = get_ct_preprocessing_pipeline()
+    transforms = pipeline.transforms
+
+    assert len(transforms) == 3
+    assert isinstance(transforms[0], LoadImage)
+    assert transforms[0].image_only is True
+    assert transforms[1].__class__ is EnsureChannelFirst
+    assert isinstance(transforms[2], ScaleIntensityRange)
+
+    scale = transforms[2]
+    assert scale.a_min == -1000
+    assert scale.a_max == 400
+    assert scale.b_min == 0.0
+    assert scale.b_max == 1.0
+    assert scale.clip is True
+
+
+def test_mri_preprocessing_pipeline_structure():
+    """Test MRI pipeline structure."""
+    pipeline = get_mri_preprocessing_pipeline()
+    transforms = pipeline.transforms
+
+    assert len(transforms) == 3
+    assert isinstance(transforms[0], LoadImage)
+    assert transforms[0].image_only is True
+    assert transforms[1].__class__ is EnsureChannelFirst
+    assert isinstance(transforms[2], NormalizeIntensity)
+    assert transforms[2].nonzero is True
+
+
+def test_invalid_modality_type():
+    """Test non-string modality input."""
+    with pytest.raises(ModalityTypeError) as exc:
+        preprocess_dicom_series("dummy", 123)
+
+    assert "modality must be a string" in str(exc.value)
+
+
+def test_unsupported_modality():
+    """Test unsupported modality."""
+    with pytest.raises(UnsupportedModalityError) as exc:
+        preprocess_dicom_series("dummy", "PET")
+
+    msg = str(exc.value)
+    assert "Unsupported modality" in msg
+    assert "CT" in msg
+    assert "MR" in msg
+    assert "MRI" in msg
+
+
+def test_modality_case_insensitivity():
+    """Test case-insensitive modality handling."""
+    # These should all work without error
+    for modality in ["CT", "ct", "Ct", "CT ", "MR", "mr", "MRI", "mri", " MrI "]:
+        try:
+            # We're not actually loading an image, just checking the function doesn't fail on modality parsing
+            if isinstance(modality, str) and modality.strip().upper() in {"CT", "MR", "MRI"}:
+                assert True
+        except (ModalityTypeError, UnsupportedModalityError):
+            pytest.fail(f"Modality {modality!r} should be accepted")
+
+
+def test_preprocess_dicom_series_integration(tmp_path):
+    """Integration test with dummy NIfTI file."""
+    # Create a dummy NIfTI file for testing
+    dummy_data = np.random.randn(64, 64, 64).astype(np.float32)
+    test_file = tmp_path / "test.nii.gz"
+
+    write_nifti(dummy_data, test_file)
+
+    # Test with each modality
+    for modality in ["CT", "MRI"]:
+        try:
+            result = preprocess_dicom_series(str(test_file), modality)
+            assert result is not None
+            assert hasattr(result, "shape")
+        except Exception as e:
+            pytest.fail(f"Failed to preprocess with modality {modality}: {e}")

monai/transforms/__init__.py

@@ -680,6 +680,16 @@
     TransposeD,
     TransposeDict,
 )
+
+# Clinical preprocessing utilities
+from .clinical_preprocessing import (
+    ModalityTypeError,
+    UnsupportedModalityError,
+    get_ct_preprocessing_pipeline,
+    get_mri_preprocessing_pipeline,
+    preprocess_dicom_series,
+)
+
 from .utils import (
     Fourier,
     allow_missing_keys_mode,

monai/transforms/clinical_preprocessing.py

@@ -0,0 +1,112 @@
+# Copyright (c) MONAI Consortium
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#     http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Clinical preprocessing transforms for medical imaging data.
+
+This module provides modality-specific preprocessing pipelines for common medical imaging modalities.
+"""
+
+from typing import Any
+
+from monai.transforms import (
+    Compose,
+    EnsureChannelFirst,
+    LoadImage,
+    NormalizeIntensity,
+    ScaleIntensityRange,
+)
+
+
+class ModalityTypeError(TypeError):
+    """Raised when modality is not a string."""
+
+
+class UnsupportedModalityError(ValueError):
+    """Raised when an unsupported modality is requested."""
+
+
+def get_ct_preprocessing_pipeline() -> Compose:
+    """
+    Create a preprocessing pipeline for CT images.
+
+    Returns:
+        Compose: Transform composition for CT preprocessing.
+    """
+    return Compose(
+        [
+            LoadImage(image_only=True),
+            EnsureChannelFirst(),
+            ScaleIntensityRange(
+                a_min=-1000,
+                a_max=400,
+                b_min=0.0,
+                b_max=1.0,
+                clip=True,
+            ),
+        ]
+    )
+
+
+def get_mri_preprocessing_pipeline() -> Compose:
+    """
+    Create a preprocessing pipeline for MRI images.
+
+    Returns:
+        Compose: Transform composition for MRI preprocessing.
+    """
+    return Compose(
+        [
+            LoadImage(image_only=True),
+            EnsureChannelFirst(),
+            NormalizeIntensity(nonzero=True),
+        ]
+    )
+
+
+def preprocess_dicom_series(path: str, modality: str) -> Any:
+    """Preprocess a DICOM series or file based on imaging modality.
+
+    Args:
+        path: Path to the DICOM file or directory containing a DICOM series.
+        modality: Imaging modality. Supported values are "CT", "MR", and "MRI" (case-insensitive).
+
+    Returns:
+        Preprocessed image data (typically a MetaTensor or NumPy array).
+
+    Raises:
+        ModalityTypeError: If modality is not a string.
+        UnsupportedModalityError: If the provided modality is not supported.
+    """
+    if not isinstance(modality, str):
+        raise ModalityTypeError("modality must be a string")
+
+    modality_clean = modality.strip().upper()
+
+    if modality_clean in {"MR", "MRI"}:
+        pipeline = get_mri_preprocessing_pipeline()
+    elif modality_clean == "CT":
+        pipeline = get_ct_preprocessing_pipeline()
+    else:
+        raise UnsupportedModalityError(
+            f"Unsupported modality '{modality}'. Supported modalities: CT, MR, MRI"
+        )
+
+    return pipeline(path)
+
+
+__all__ = [
+    "ModalityTypeError",
+    "UnsupportedModalityError",
+    "get_ct_preprocessing_pipeline",
+    "get_mri_preprocessing_pipeline",
+    "preprocess_dicom_series",
+]