Add clinical DICOM preprocessing utilities for CT/MRI with unit tests

Author: Hitendrasinhdata7

monai/tests/test_clinical_preprocessing.py

@@ -1,46 +1,129 @@
+"""Unit tests for clinical DICOM preprocessing utilities."""
+
 import numpy as np
+from unittest.mock import patch, MagicMock
 import pytest
 
 from monai.transforms import ScaleIntensityRange, NormalizeIntensity
+from monai.transforms.clinical_preprocessing import (
+    get_ct_preprocessing_pipeline,
+    get_mri_preprocessing_pipeline,
+    preprocess_dicom_series,
+)
 
 
-def test_ct_windowing_range_and_shape():
+def test_ct_windowing_range_and_shape_direct():
+    """Test ScaleIntensityRange transform on sample CT data."""
     rng = np.random.default_rng(0)
-
-    sample_ct = rng.integers(
-        -1024, 2048, size=(64, 64, 64), dtype=np.int16
-    )
-
-    transform = ScaleIntensityRange(
-        a_min=-1000,
-        a_max=400,
-        b_min=0.0,
-        b_max=1.0,
-        clip=True,
-    )
-
-    output = transform(sample_ct)
-    output = np.asarray(output)
+    sample_ct = rng.integers(-1024, 2048, size=(64, 64, 64), dtype=np.int16)
+    transform = ScaleIntensityRange(a_min=-1000, a_max=400, b_min=0.0, b_max=1.0, clip=True)
+    output = np.asarray(transform(sample_ct))
 
     assert output.shape == sample_ct.shape
     assert np.isfinite(output).all()
     assert output.min() >= -1e-6
     assert output.max() <= 1.0 + 1e-6
 
 
-def test_mri_normalization_mean_std():
+def test_mri_normalization_mean_std_direct():
+    """Test NormalizeIntensity transform on sample MRI data."""
     rng = np.random.default_rng(0)
-
     sample_mri = rng.random((64, 64, 64), dtype=np.float32)
-
     transform = NormalizeIntensity(nonzero=True)
+    output = np.asarray(transform(sample_mri))
 
-    output = transform(sample_mri)
-    output = np.asarray(output)
+    assert output.shape == sample_mri.shape
+    assert np.isclose(float(output.mean()), 0.0, atol=0.1)
+    assert np.isclose(float(output.std()), 1.0, atol=0.1)
 
-    mean_val = float(output.mean())
-    std_val = float(output.std())
 
-    assert output.shape == sample_mri.shape
-    assert np.isclose(mean_val, 0.0, atol=0.1)
-    assert np.isclose(std_val, 1.0, atol=0.1)
+@patch("monai.transforms.clinical_preprocessing.LoadImage")
+def test_ct_pipeline(mock_loadimage):
+    """Test get_ct_preprocessing_pipeline returns correct transform sequence."""
+    pipeline = get_ct_preprocessing_pipeline()
+    assert len(pipeline.transforms) == 3
+    assert pipeline.transforms[0].__class__.__name__ == "LoadImage"
+    assert pipeline.transforms[1].__class__.__name__ == "EnsureChannelFirst"
+    assert pipeline.transforms[2].__class__.__name__ == "ScaleIntensityRange"
+
+
+@patch("monai.transforms.clinical_preprocessing.LoadImage")
+def test_mri_pipeline(mock_loadimage):
+    """Test get_mri_preprocessing_pipeline returns correct transform sequence."""
+    pipeline = get_mri_preprocessing_pipeline()
+    assert len(pipeline.transforms) == 3
+    assert pipeline.transforms[0].__class__.__name__ == "LoadImage"
+    assert pipeline.transforms[1].__class__.__name__ == "EnsureChannelFirst"
+    assert pipeline.transforms[2].__class__.__name__ == "NormalizeIntensity"
+
+
+@patch("monai.transforms.clinical_preprocessing.get_ct_preprocessing_pipeline")
+def test_preprocess_dicom_series_ct(mock_pipeline):
+    """Test preprocess_dicom_series with CT modality."""
+    mock_transform = MagicMock()
+    mock_pipeline.return_value = mock_transform
+    preprocess_dicom_series("dummy_path.dcm", "CT")
+    mock_pipeline.assert_called_once()
+    mock_transform.assert_called_once_with("dummy_path.dcm")
+
+
+@patch("monai.transforms.clinical_preprocessing.get_ct_preprocessing_pipeline")
+def test_preprocess_dicom_series_ct_lowercase(mock_pipeline):
+    """Test preprocess_dicom_series with lowercase CT modality."""
+    mock_transform = MagicMock()
+    mock_pipeline.return_value = mock_transform
+    preprocess_dicom_series("dummy_path.dcm", "ct")
+    mock_pipeline.assert_called_once()
+    mock_transform.assert_called_once_with("dummy_path.dcm")
+
+
+@patch("monai.transforms.clinical_preprocessing.get_mri_preprocessing_pipeline")
+def test_preprocess_dicom_series_mri(mock_pipeline):
+    """Test preprocess_dicom_series with MRI modality."""
+    mock_transform = MagicMock()
+    mock_pipeline.return_value = mock_transform
+    preprocess_dicom_series("dummy_path.dcm", "MRI")
+    mock_pipeline.assert_called_once()
+    mock_transform.assert_called_once_with("dummy_path.dcm")
+
+
+@patch("monai.transforms.clinical_preprocessing.get_mri_preprocessing_pipeline")
+def test_preprocess_dicom_series_mr(mock_pipeline):
+    """Test preprocess_dicom_series with MR modality."""
+    mock_transform = MagicMock()
+    mock_pipeline.return_value = mock_transform
+    preprocess_dicom_series("dummy_path.dcm", "MR")
+    mock_pipeline.assert_called_once()
+    mock_transform.assert_called_once_with("dummy_path.dcm")
+
+
+def test_preprocess_dicom_series_invalid_modality():
+    """Test preprocess_dicom_series raises ValueError for unsupported modality."""
+    with pytest.raises(ValueError) as exc:
+        preprocess_dicom_series("dummy_path.dcm", "PET")
+    assert "Unsupported modality" in str(exc.value)
+    assert "PET" in str(exc.value)
+
+
+def test_preprocess_dicom_series_invalid_type():
+    """Test preprocess_dicom_series raises TypeError for non-string modality."""
+    with pytest.raises(TypeError) as exc:
+        preprocess_dicom_series("dummy_path.dcm", 123)
+    assert "modality must be a string" in str(exc.value)
+
+
+def test_preprocess_dicom_series_none_modality():
+    """Test preprocess_dicom_series raises TypeError for None modality."""
+    with pytest.raises(TypeError) as exc:
+        preprocess_dicom_series("dummy_path.dcm", None)
+    assert "modality must be a string" in str(exc.value)
+
+
+@patch("monai.transforms.clinical_preprocessing.get_ct_preprocessing_pipeline")
+def test_preprocess_dicom_series_whitespace(mock_pipeline):
+    """Test preprocess_dicom_series handles whitespace in modality."""
+    mock_transform = MagicMock()
+    mock_pipeline.return_value = mock_transform
+    preprocess_dicom_series("dummy_path.dcm", "  CT  ")
+    mock_pipeline.assert_called_once()
+    mock_transform.assert_called_once_with("dummy_path.dcm")

monai/transforms/clinical_preprocessing.py

@@ -1,4 +1,7 @@
+"""Clinical DICOM preprocessing utilities for CT and MRI modalities."""
+
 from typing import Union
+from os import PathLike
 
 from monai.transforms import (
     Compose,
@@ -8,10 +11,18 @@
     NormalizeIntensity,
 )
 
+SUPPORTED_MODALITIES = "CT, MR, MRI"
+
 
-def get_ct_preprocessing_pipeline():
+def get_ct_preprocessing_pipeline() -> Compose:
     """
-    CT preprocessing pipeline using standard HU windowing.
+    Build a CT preprocessing pipeline using standard HU windowing.
+
+    The pipeline applies LoadImage, EnsureChannelFirst, and ScaleIntensityRange
+    with HU window [-1000, 400] normalized to [0.0, 1.0].
+
+    Returns:
+        Compose: A composed transform pipeline for CT preprocessing.
     """
     return Compose(
         [
@@ -28,9 +39,15 @@ def get_ct_preprocessing_pipeline():
     )
 
 
-def get_mri_preprocessing_pipeline():
+def get_mri_preprocessing_pipeline() -> Compose:
     """
-    MRI preprocessing pipeline using intensity normalization.
+    Build an MRI preprocessing pipeline using intensity normalization.
+
+    The pipeline applies LoadImage, EnsureChannelFirst, and NormalizeIntensity
+    with nonzero=True to normalize only non-zero voxels.
+
+    Returns:
+        Compose: A composed transform pipeline for MRI preprocessing.
     """
     return Compose(
         [
@@ -42,21 +59,25 @@ def get_mri_preprocessing_pipeline():
 
 
 def preprocess_dicom_series(
-    dicom_path: Union[str, bytes],
+    dicom_path: Union[str, bytes, PathLike],
     modality: str,
-):
+) -> "MetaTensor":
     """
     Preprocess a DICOM series based on modality.
 
     Args:
         dicom_path: Path to DICOM file or directory.
-        modality: CT, MR, or MRI.
+        modality: Imaging modality. Supported values: "CT", "MR", "MRI" (case-insensitive).
 
     Returns:
-        Preprocessed image.
+        MetaTensor: Preprocessed image with intensity values normalized based on modality.
+
+    Raises:
+        TypeError: If modality is not a string.
+        ValueError: If modality is not one of the supported values.
     """
     if not isinstance(modality, str):
-        raise TypeError("modality must be a string")
+        raise TypeError(f"modality must be a string, got {type(modality).__name__}")
 
     modality = modality.strip().upper()
 
@@ -65,6 +86,6 @@ def preprocess_dicom_series(
     elif modality in ("MR", "MRI"):
         transform = get_mri_preprocessing_pipeline()
     else:
-        raise ValueError("Unsupported modality")
+        raise ValueError(f"Unsupported modality: {modality}. Supported values: {SUPPORTED_MODALITIES}")
 
     return transform(dicom_path)