|
1 | | -"""Clinical DICOM preprocessing utilities for CT and MRI modalities.""" |
2 | | - |
3 | | -from typing import Union |
4 | | -from os import PathLike |
5 | | - |
6 | | -from monai.data import MetaTensor |
7 | | -from monai.transforms import ( |
8 | | - Compose, |
9 | | - LoadImage, |
10 | | - EnsureChannelFirst, |
11 | | - ScaleIntensityRange, |
12 | | - NormalizeIntensity, |
| 1 | +import pytest |
| 2 | +from monai.transforms import LoadImage, EnsureChannelFirst, ScaleIntensityRange, NormalizeIntensity |
| 3 | +from monai.transforms.clinical_preprocessing import ( |
| 4 | + get_ct_preprocessing_pipeline, |
| 5 | + get_mri_preprocessing_pipeline, |
| 6 | + preprocess_dicom_series, |
| 7 | + UnsupportedModalityError, |
| 8 | + ModalityTypeError, |
13 | 9 | ) |
14 | 10 |
|
15 | | -SUPPORTED_MODALITIES = ("CT", "MR", "MRI") |
16 | | - |
17 | | - |
18 | | -class UnsupportedModalityError(ValueError): |
19 | | - """Raised when an unsupported modality is provided.""" |
20 | | - pass |
21 | | - |
22 | | - |
23 | | -class ModalityTypeError(TypeError): |
24 | | - """Raised when modality is not a string.""" |
25 | | - pass |
26 | | - |
27 | | - |
28 | | -def get_ct_preprocessing_pipeline() -> Compose: |
29 | | - """Return a CT preprocessing pipeline.""" |
30 | | - return Compose([ |
31 | | - LoadImage(image_only=True), |
32 | | - EnsureChannelFirst(), |
33 | | - ScaleIntensityRange(a_min=-1000, a_max=400, b_min=0.0, b_max=1.0, clip=True), |
34 | | - ]) |
35 | | - |
36 | | - |
37 | | -def get_mri_preprocessing_pipeline() -> Compose: |
38 | | - """Return an MRI preprocessing pipeline.""" |
39 | | - return Compose([ |
40 | | - LoadImage(image_only=True), |
41 | | - EnsureChannelFirst(), |
42 | | - NormalizeIntensity(nonzero=True), |
43 | | - ]) |
44 | | - |
45 | | - |
46 | | -def preprocess_dicom_series( |
47 | | - dicom_path: Union[str, bytes, PathLike], |
48 | | - modality: str, |
49 | | -) -> MetaTensor: |
50 | | - """Preprocess a DICOM series according to modality (CT or MRI). |
51 | | -
|
52 | | - Args: |
53 | | - dicom_path (Union[str, bytes, PathLike]): Path to DICOM series. |
54 | | - modality (str): Modality type, must be one of 'CT', 'MR', 'MRI'. |
55 | | -
|
56 | | - Returns: |
57 | | - MetaTensor: Preprocessed image tensor. |
58 | | -
|
59 | | - Raises: |
60 | | - ModalityTypeError: If modality is not a string. |
61 | | - UnsupportedModalityError: If modality is not supported. |
62 | | - """ |
63 | | - if not isinstance(modality, str): |
64 | | - raise ModalityTypeError(f"modality must be a string, got {type(modality).__name__}") |
65 | | - |
66 | | - modality = modality.strip().upper() |
67 | | - |
68 | | - if modality == "CT": |
69 | | - transform = get_ct_preprocessing_pipeline() |
70 | | - elif modality in ("MR", "MRI"): |
71 | | - transform = get_mri_preprocessing_pipeline() |
72 | | - else: |
73 | | - raise UnsupportedModalityError( |
74 | | - f"Unsupported modality: {modality}. Supported values: {', '.join(SUPPORTED_MODALITIES)}" |
75 | | - ) |
76 | 11 |
|
77 | | - return transform(dicom_path) |
| 12 | +def test_ct_preprocessing_pipeline(): |
| 13 | + """Test CT preprocessing pipeline returns expected transform composition and parameters.""" |
| 14 | + pipeline = get_ct_preprocessing_pipeline() |
| 15 | + assert hasattr(pipeline, 'transforms') |
| 16 | + assert len(pipeline.transforms) == 3 |
| 17 | + assert isinstance(pipeline.transforms[0], LoadImage) |
| 18 | + assert isinstance(pipeline.transforms[1], EnsureChannelFirst) |
| 19 | + assert isinstance(pipeline.transforms[2], ScaleIntensityRange) |
| 20 | + |
| 21 | + # Verify CT-specific HU window parameters |
| 22 | + scale_transform = pipeline.transforms[2] |
| 23 | + assert scale_transform.a_min == -1000 |
| 24 | + assert scale_transform.a_max == 400 |
| 25 | + assert scale_transform.b_min == 0.0 |
| 26 | + assert scale_transform.b_max == 1.0 |
| 27 | + assert scale_transform.clip is True |
| 28 | + |
| 29 | + |
| 30 | +def test_mri_preprocessing_pipeline(): |
| 31 | + """Test MRI preprocessing pipeline returns expected transform composition and parameters.""" |
| 32 | + pipeline = get_mri_preprocessing_pipeline() |
| 33 | + assert hasattr(pipeline, 'transforms') |
| 34 | + assert len(pipeline.transforms) == 3 |
| 35 | + assert isinstance(pipeline.transforms[0], LoadImage) |
| 36 | + assert isinstance(pipeline.transforms[1], EnsureChannelFirst) |
| 37 | + assert isinstance(pipeline.transforms[2], NormalizeIntensity) |
| 38 | + |
| 39 | + # Verify MRI-specific normalization parameter |
| 40 | + normalize_transform = pipeline.transforms[2] |
| 41 | + assert normalize_transform.nonzero is True |
| 42 | + |
| 43 | + |
| 44 | +def test_preprocess_dicom_series_invalid_modality(): |
| 45 | + """Test preprocess_dicom_series raises UnsupportedModalityError for unsupported modality.""" |
| 46 | + with pytest.raises(UnsupportedModalityError, match=r"Unsupported modality.*PET.*CT, MR, MRI"): |
| 47 | + preprocess_dicom_series("dummy_path.dcm", "PET") |
| 48 | + |
| 49 | + |
| 50 | +def test_preprocess_dicom_series_invalid_type(): |
| 51 | + """Test preprocess_dicom_series raises ModalityTypeError for non-string modality.""" |
| 52 | + with pytest.raises(ModalityTypeError, match=r"modality must be a string, got int"): |
| 53 | + preprocess_dicom_series("dummy_path.dcm", 123) |
0 commit comments