Advanced Preprocessing

This example shows how to build custom preprocessing pipelines and fine-tune preprocessing parameters for different imaging modalities and analysis tasks.

Custom Preprocessing Pipelines

Using the Preprocessor Class Directly

Instead of relying on FeatureExtractor defaults, create custom pipelines:

from medtda import Preprocessor
from medtda.loaders import load_image, load_mask

# Load data
image, img_meta = load_image('ct_scan.nii.gz')
mask, mask_meta = load_mask('roi_mask.nii.gz')

# Create custom preprocessor
preprocessor = Preprocessor(
    normalize=True,
    normalize_method='robust',
    spacing=(1.0, 1.0, 1.0),
    crop_to_roi=True,
    roi_padding=10
)

# Apply preprocessing
processed_image = preprocessor.preprocess(image, mask)

print(f"Original shape: {image.shape}")
print(f"Processed shape: {processed_image.shape}")
print(f"Value range: [{processed_image.min():.3f}, {processed_image.max():.3f}]")

Output:

Original shape: (512, 512, 200)
Processed shape: (127, 154, 89)
Value range: [-1.243, 3.567]

Chaining Multiple Preprocessors

from medtda import Preprocessor
import numpy as np

# Step 1: Coarse preprocessing
preprocessor1 = Preprocessor(
    window=(40, 400),      # CT soft tissue window
    spacing=(2.0, 2.0, 2.0),  # Coarse resampling
    crop_to_roi=True
)

# Step 2: Fine preprocessing
preprocessor2 = Preprocessor(
    normalize=True,
    normalize_method='zscore'
)

# Apply in sequence
image1 = preprocessor1.preprocess(image, mask)
image2 = preprocessor2.preprocess(image1)

print("Two-stage preprocessing complete")

Modality-Specific Preprocessing

CT Scans

Soft Tissue Analysis:

from medtda import Preprocessor

# Soft tissue window: [-60, 340] HU
preprocessor = Preprocessor(
    window=(-60, 340),
    normalize=True,
    normalize_method='robust',
    spacing=(1.0, 1.0, 1.0),
    crop_to_roi=True
)

processed = preprocessor.preprocess(ct_image, organ_mask)

Lung Analysis:

# Lung window: [-1000, 400] HU
preprocessor = Preprocessor(
    window=(-1000, 400),
    normalize=True,
    normalize_method='minmax',
    spacing=(1,0, 1.0, 1.0)
)

processed = preprocessor.preprocess(ct_image, lung_mask)

Bone Analysis:

# Bone window: [300, 1800] HU
preprocessor = Preprocessor(
    window=(300, 1800),
    normalize=True,
    normalize_method='robust',
    spacing=(1.0, 1.0, 1.0)
)

processed = preprocessor.preprocess(ct_image, bone_mask)

MRI Scans

T1-weighted:

from medtda import Preprocessor

# T1: zscore normalization recommended
preprocessor = Preprocessor(
    normalize=True,
    normalize_method='zscore',
    spacing=(1.0, 1.0, 1.0),
    crop_to_roi=True,
    roi_padding=5
)

processed = preprocessor.preprocess(t1_image, brain_mask)

T2-weighted:

# T2: robust normalization for outliers
preprocessor = Preprocessor(
    normalize=True,
    normalize_method='robust',
    spacing=(1.0, 1.0, 1.0),
    crop_to_roi=True
)

processed = preprocessor.preprocess(t2_image, roi_mask)

FLAIR:

# FLAIR: may need intensity clipping
preprocessor = Preprocessor(
    clip_percentiles=(1, 99),  # Remove extreme outliers
    normalize=True,
    normalize_method='robust',
    spacing=(1.0, 1.0, 1.0)
)

processed = preprocessor.preprocess(flair_image, lesion_mask)

PET Scans

# PET: typically log transform helpful
from medtda import Preprocessor
import numpy as np

preprocessor = Preprocessor(
    normalize=False,  # Do custom normalization
    spacing=(2.0, 2.0, 2.0),
    crop_to_roi=True
)

# Preprocess
processed = preprocessor.preprocess(pet_image, mask)

# Custom log transformation
processed = np.log1p(processed)  # log(1 + x)

# Then normalize
processed = (processed - processed.mean()) / processed.std()

Microscopy Images

Histopathology (H&E):

from medtda import Preprocessor
from medtda.loaders import load_image

# Load RGB image
image, metadata = load_image('he_slide.tiff')

# Convert to grayscale intensity
if image.ndim == 3 and image.shape[-1] == 3:
    # Simple RGB to grayscale
    image = 0.299 * image[..., 0] + 0.587 * image[..., 1] + 0.114 * image[..., 2]

# Preprocess
preprocessor = Preprocessor(
    normalize=True,
    normalize_method='minmax',
    spacing=None  # Already at right resolution
)

processed = preprocessor.preprocess(image)

Immunofluorescence:

# Single channel fluorescence
preprocessor = Preprocessor(
    clip_percentiles=(1, 99.5),  # Remove hot pixels
    normalize=True,
    normalize_method='robust',
    spacing=None
)

processed = preprocessor.preprocess(fluorescence_image)

Normalization Methods Comparison

MinMax Normalization

from medtda import Preprocessor
import numpy as np

preprocessor = Preprocessor(
    normalize=True,
    normalize_method='minmax'
)

processed = preprocessor.preprocess(image)

print(f"Range: [{processed.min()}, {processed.max()}]")
# Output: Range: [0.0, 1.0]

When to use:

• Images with known, fixed value ranges

• When you want features in[0, 1]

• Visualization purposes

Z-score Normalization

preprocessor = Preprocessor(
    normalize=True,
    normalize_method='zscore'
)

processed = preprocessor.preprocess(image)

print(f"Mean: {processed.mean():.3f}, Std: {processed.std():.3f}")
# Output: Mean: 0.000, Std: 1.000

When to use:

• MRI scans (Gaussian-ish distributions)

• When maintaining relative intensities matters

• ML models expecting zero-mean data

Robust Normalization

preprocessor = Preprocessor(
    normalize=True,
    normalize_method='robust'
)

processed = preprocessor.preprocess(image)

# Uses median and IQR instead of mean and std
print(f"Median: {np.median(processed):.3f}")
# Output: Median: 0.000

When to use:

• CT scans with outliers

• Images with artifacts

• When robustness to outliers is important

Custom Normalization

from medtda import Preprocessor
import numpy as np

# Use Preprocessor without normalization
preprocessor = Preprocessor(
    normalize=False,
    spacing=(1.0, 1.0, 1.0)
)

processed = preprocessor.preprocess(image, mask)

# Apply custom normalization
# Example: Percentile-based
p1, p99 = np.percentile(processed, [1, 99])
processed = np.clip(processed, p1, p99)
processed = (processed - p1) / (p99 - p1)

print("Custom percentile normalization applied")

Resampling Strategies

Isotropic Resampling

from medtda import Preprocessor

# Resample to 1mm isotropic
preprocessor = Preprocessor(
    spacing=(1.0, 1.0, 1.0),
    normalize=True
)

processed = preprocessor.preprocess(image)

print(f"Resampled to isotropic 1mm³ voxels")

Anisotropic Resampling

# Different spacing per axis
# Example: maintain high in-plane resolution, downsample slice direction
preprocessor = Preprocessor(
    spacing=(0.5, 0.5, 2.0),  # xy high-res, z low-res
    normalize=True
)

processed = preprocessor.preprocess(image)

Preserve Original Spacing

# Don't resample
preprocessor = Preprocessor(
    spacing=None,  # Keep original
    normalize=True
)

processed = preprocessor.preprocess(image)

Adaptive Spacing

from medtda import Preprocessor
from medtda.loaders import load_image

image, metadata = load_image('scan.nii.gz')
original_spacing = metadata.get('spacing', (1.0, 1.0, 1.0))

# Decide spacing based on image properties
if original_spacing[0] < 0.5:
    # High resolution, downsample
    new_spacing = (1.0, 1.0, 1.0)
else:
    # Low resolution, keep or slightly improve
    new_spacing = original_spacing

preprocessor = Preprocessor(spacing=new_spacing)
processed = preprocessor.preprocess(image)

ROI Extraction and Cropping

Basic ROI Cropping

from medtda import Preprocessor

preprocessor = Preprocessor(
    crop_to_roi=True,
    roi_padding=0  # Tight crop
)

processed = preprocessor.preprocess(image, mask)
print(f"Cropped to ROI: {processed.shape}")

ROI with Padding

# Add context around ROI
preprocessor = Preprocessor(
    crop_to_roi=True,
    roi_padding=10  # 10 voxels padding
)

processed = preprocessor.preprocess(image, mask)
print(f"Cropped with 10-voxel padding: {processed.shape}")

Multiple ROIs

from medtda import Preprocessor
import numpy as np

# Multi-label mask: 0=background, 1=organ1, 2=organ2, 3=tumor
mask = load_mask('multilabel_mask.nii.gz')[0]

# Extract tumor (label 3)
preprocessor = Preprocessor(
    crop_to_roi=True,
    roi_padding=5,
    label=3  # Extract tumor only
)

processed = preprocessor.preprocess(image, mask)
print("Tumor ROI extracted")

Process Each ROI Separately

from medtda import Preprocessor
import numpy as np

# Multi-label mask
mask, _ = load_mask('organs_mask.nii.gz')
unique_labels = np.unique(mask)
unique_labels = unique_labels[unique_labels > 0]  # Skip background

results = {}

for label_id in unique_labels:
    preprocessor = Preprocessor(
        crop_to_roi=True,
        label=label_id,
        normalize=True
    )

    processed = preprocessor.preprocess(image, mask)
    results[label_id] = processed

    print(f"Label {label_id}: shape {processed.shape}")

Intensity Windowing

Fixed Window

from medtda import Preprocessor

# CT soft tissue window
preprocessor = Preprocessor(
    window=(-60, 340),  # HU units
    normalize=True
)

processed = preprocessor.preprocess(ct_image)

Multiple Windows

# Process same image with different windows
windows = {
    'soft_tissue': (-60, 340),
    'lung': (-1000, 400),
    'bone': (300, 1800)
}

results = {}

for name, (w_min, w_max) in windows.items():
    preprocessor = Preprocessor(
        window=(w_min, w_max),
        normalize=True
    )
    results[name] = preprocessor.preprocess(ct_image)
    print(f"Processed {name} window")

Adaptive Windowing

import numpy as np
from medtda import Preprocessor

# Compute window from image statistics
mean = np.mean(image[mask > 0])  # Within ROI
std = np.std(image[mask > 0])

# Define window as mean ± 3*std
window = (mean - 3*std, mean + 3*std)

preprocessor = Preprocessor(
    window=window,
    normalize=True
)

processed = preprocessor.preprocess(image, mask)
print(f"Adaptive window: [{window[0]:.1f}, {window[1]:.1f}]")

Clip Percentiles

from medtda import Preprocessor

# Clip extreme values
preprocessor = Preprocessor(
    clip_percentiles=(1, 99),  # Remove bottom 1% and top 1%
    normalize=True
)

processed = preprocessor.preprocess(image)

Handling Missing or Invalid Data

NaN/Inf Handling

from medtda import Preprocessor
import numpy as np

# Image has NaN or Inf values
print(f"NaN count: {np.isnan(image).sum()}")
print(f"Inf count: {np.isinf(image).sum()}")

# Replace NaN/Inf
image_clean = np.nan_to_num(image, nan=0.0, posinf=0.0, neginf=0.0)

# Then preprocess
preprocessor = Preprocessor(normalize=True)
processed = preprocessor.preprocess(image_clean)

Mask Validation

from medtda import Preprocessor
from medtda.loaders import validate_compatibility
import numpy as np

# Check mask validity
if mask is not None:
    # Check shapes match
    validate_compatibility(image, mask)

    # Check mask is binary or labeled
    unique_values = np.unique(mask)
    print(f"Mask labels: {unique_values}")

    # Check mask is not empty
    if np.sum(mask > 0) == 0:
        raise ValueError("Mask is empty!")

    # Proceed with preprocessing
    preprocessor = Preprocessor(crop_to_roi=True)
    processed = preprocessor.preprocess(image, mask)

Empty ROI Handling

import numpy as np
from medtda import Preprocessor

def safe_preprocess(image, mask=None, min_voxels=100):
    """Preprocess with safety checks."""

    if mask is not None:
        roi_voxels = np.sum(mask > 0)

        if roi_voxels < min_voxels:
            print(f"Warning: ROI too small ({roi_voxels} voxels)")
            return None

    preprocessor = Preprocessor(
        crop_to_roi=(mask is not None),
        normalize=True
    )

    return preprocessor.preprocess(image, mask)

# Use it
processed = safe_preprocess(image, mask)

if processed is not None:
    print("Preprocessing successful")
else:
    print("Preprocessing skipped (ROI too small)")

Performance Optimization

Memory-Efficient Processing

from medtda import Preprocessor
import numpy as np

# For large images, downsample aggressively
preprocessor = Preprocessor(
    spacing=(2.0, 2.0, 2.0),  # Reduce resolution
    crop_to_roi=True,          # Reduce FOV
    roi_padding=5,
    normalize=True
)

processed = preprocessor.preprocess(large_image, mask)

print(f"Original size: {large_image.nbytes / 1e6:.1f} MB")
print(f"Processed size: {processed.nbytes / 1e6:.1f} MB")

Batch Preprocessing

from medtda import Preprocessor
from pathlib import Path
import numpy as np

# Create one preprocessor for all images
preprocessor = Preprocessor(
    normalize=True,
    spacing=(1.0, 1.0, 1.0),
    crop_to_roi=True
)

# Process multiple images
image_dir = Path('data/images')
mask_dir = Path('data/masks')

results = []

for img_path in image_dir.glob('*.nii.gz'):
    mask_path = mask_dir / img_path.name

    image, _ = load_image(img_path)
    mask, _ = load_mask(mask_path)

    processed = preprocessor.preprocess(image, mask)
    results.append(processed)

    print(f"Processed {img_path.name}")

print(f"Total images processed: {len(results)}")

Caching Preprocessed Data

from medtda import Preprocessor
from medtda.loaders import load_image, load_mask
import numpy as np
from pathlib import Path

def preprocess_and_cache(image_path, mask_path, cache_dir='cache'):
    """Preprocess and cache result."""

    # Check cache
    cache_path = Path(cache_dir) / f"{Path(image_path).stem}_processed.npy"

    if cache_path.exists():
        print(f"Loading from cache: {cache_path}")
        return np.load(cache_path)

    # Preprocess
    image, _ = load_image(image_path)
    mask, _ = load_mask(mask_path)

    preprocessor = Preprocessor(
        normalize=True,
        spacing=(1.0, 1.0, 1.0),
        crop_to_roi=True
    )

    processed = preprocessor.preprocess(image, mask)

    # Cache
    cache_path.parent.mkdir(exist_ok=True)
    np.save(cache_path, processed)
    print(f"Cached to: {cache_path}")

    return processed

# Use it
processed = preprocess_and_cache('image.nii.gz', 'mask.nii.gz')

Complete Preprocessing Workflows

Workflow 1: Clinical CT Analysis

from medtda import Preprocessor, FeatureExtractor
from medtda.loaders import load_image, load_mask
import numpy as np

def clinical_ct_pipeline(image_path, mask_path):
    """Complete CT preprocessing for clinical analysis."""

    # Load
    image, img_meta = load_image(image_path)
    mask, mask_meta = load_mask(mask_path)

    print("Original image:")
    print(f"  Shape: {image.shape}")
    print(f"  Spacing: {img_meta.get('spacing')}")
    print(f"  Value range: [{image.min()}, {image.max()}]")

    # Stage 1: Windowing
    preprocessor1 = Preprocessor(
        window=(-60, 340),  # Soft tissue
        normalize=False
    )
    image = preprocessor1.preprocess(image)
    print("Stage 1: Windowing complete")

    # Stage 2: Resampling
    preprocessor2 = Preprocessor(
        spacing=(1.0, 1.0, 1.0),
        normalize=False
    )
    image = preprocessor2.preprocess(image)
    print("Stage 2: Resampling complete")

    # Stage 3: ROI extraction
    preprocessor3 = Preprocessor(
        crop_to_roi=True,
        roi_padding=5,
        normalize=False
    )
    image = preprocessor3.preprocess(image, mask)
    print(f"Stage 3: ROI extracted (shape: {image.shape})")

    # Stage 4: Normalization
    preprocessor4 = Preprocessor(
        normalize=True,
        normalize_method='robust'
    )
    image = preprocessor4.preprocess(image)
    print("Stage 4: Normalization complete")

    print(f"\nFinal image:")
    print(f"  Shape: {image.shape}")
    print(f"  Value range: [{image.min():.3f}, {image.max():.3f}]")
    print(f"  Mean: {image.mean():.3f}, Std: {image.std():.3f}")

    return image

# Use it
processed = clinical_ct_pipeline('ct_scan.nii.gz', 'lesion_mask.nii.gz')

Workflow 2: Research MRI Pipeline

from medtda import Preprocessor
import numpy as np

def research_mri_pipeline(image_path, mask_path):
    """MRI preprocessing for research."""

    image, _ = load_image(image_path)
    mask, _ = load_mask(mask_path)

    # All-in-one preprocessing
    preprocessor = Preprocessor(
        clip_percentiles=(1, 99),      # Remove outliers
        normalize=True,
        normalize_method='zscore',
        spacing=(1.0, 1.0, 1.0),
        crop_to_roi=True,
        roi_padding=3
    )

    processed = preprocessor.preprocess(image, mask)

    # Quality check
    print("Preprocessing complete:")
    print(f"  Shape: {processed.shape}")
    print(f"  Mean: {processed.mean():.6f} (expect ~0)")
    print(f"  Std: {processed.std():.6f} (expect ~1)")
    print(f"  Range: [{processed.min():.3f}, {processed.max():.3f}]")

    # Check for issues
    if np.isnan(processed).any():
        print("WARNING: NaN values detected!")
    if np.isinf(processed).any():
        print("WARNING: Inf values detected!")

    return processed

Next Steps

• Basic Usage - Basic feature extraction

• Batch Workflow - Processing multiple images

• Preprocessing - Complete preprocessing guide

See Also

• Preprocessor - Preprocessor API reference

• User Guide - User guide

• Quick Start - Quick start