tessl/pypi-torchio

Tools for medical image processing with PyTorch

Overview

Eval results

Files

Transform Composition

Name: tessl/pypi-torchio
Author: tessl

Tools for combining and organizing transforms into pipelines, including sequential composition, random selection from transform groups, and custom lambda transforms. These utilities enable flexible and powerful data processing pipelines.

Capabilities

Sequential Composition

Sequential application of multiple transforms in a specified order, the most common way to build preprocessing and augmentation pipelines.

class Compose(Transform):
    """
    Sequential composition of multiple transforms.
    
    Applies transforms in the specified order, passing the output of each
    transform as input to the next. Essential for building preprocessing
    and augmentation pipelines.
    
    Parameters:
    - transforms: Sequence of Transform instances to apply
    """
    def __init__(self, transforms: Sequence[Transform]): ...
    
    def __len__(self) -> int:
        """Return number of transforms in composition"""
    
    def __getitem__(self, index: int) -> Transform:
        """Get transform at specified index"""
    
    def __iter__(self):
        """Iterate through transforms"""

Usage example:

import torchio as tio

# Create preprocessing pipeline
preprocessing = tio.Compose([
    tio.ToCanonical(),                      # 1. Standardize orientation
    tio.Resample(1),                        # 2. Resample to 1mm isotropic
    tio.CropOrPad((128, 128, 64)),         # 3. Standardize shape
    tio.ZNormalization(),                   # 4. Normalize intensities
])

# Create augmentation pipeline
augmentation = tio.Compose([
    tio.RandomFlip(axes=(0,)),             # 1. Random horizontal flip
    tio.RandomAffine(                       # 2. Random affine transform
        scales=(0.9, 1.1),
        degrees=(-5, 5)
    ),
    tio.RandomNoise(std=(0, 0.1)),         # 3. Add random noise
])

# Combine preprocessing and augmentation
full_pipeline = tio.Compose([
    preprocessing,
    augmentation
])

# Apply to subject
subject = tio.Subject(
    t1=tio.ScalarImage('t1.nii.gz'),
    seg=tio.LabelMap('segmentation.nii.gz')
)

transformed = full_pipeline(subject)

Random Transform Selection

Randomly selects and applies one transform from a group of transforms, useful for introducing controlled randomness in augmentation pipelines.

class OneOf(Transform):
    """
    Randomly selects one transform from a group to apply.
    
    Enables probabilistic application of different augmentation strategies,
    allowing for varied augmentation while maintaining control over frequency.
    
    Parameters:
    - transforms: Dictionary mapping Transform instances to their probabilities,
                 or sequence of transforms (equal probability)
    """
    def __init__(
        self,
        transforms: Union[dict[Transform, float], Sequence[Transform]]
    ): ...

Usage example:

# Random selection between different augmentation strategies
intensity_augmentation = tio.OneOf({
    tio.RandomNoise(std=(0, 0.1)): 0.3,           # 30% chance
    tio.RandomBlur(std=(0, 1)): 0.3,              # 30% chance
    tio.RandomGamma(log_gamma=(-0.3, 0.3)): 0.4,  # 40% chance
})

# Random medical imaging artifacts
artifact_simulation = tio.OneOf([
    tio.RandomMotion(degrees=2),                   # Equal probability
    tio.RandomGhosting(intensity=(0.5, 1)),        # Equal probability
    tio.RandomSpike(num_spikes=(1, 3)),           # Equal probability
])

# Combine in pipeline
pipeline = tio.Compose([
    tio.ToCanonical(),
    tio.RandomFlip(),
    intensity_augmentation,                        # One of noise/blur/gamma
    artifact_simulation,                           # One of motion/ghost/spike
])

subject = tio.Subject(t1=tio.ScalarImage('t1.nii.gz'))
augmented = pipeline(subject)

Custom Lambda Transforms

Apply custom functions as transforms, enabling integration of user-defined processing operations into TorchIO pipelines.

class Lambda(Transform):
    """
    Apply a user-defined function as a transform.
    
    Enables integration of custom operations into TorchIO pipelines
    while maintaining transform history and compatibility.
    
    Parameters:
    - function: Callable that transforms tensors or subjects
    - types_to_apply: Types to apply transform to (None for all)
    """
    def __init__(
        self,
        function: Callable,
        types_to_apply: tuple[type, ...] = None
    ): ...

Usage example:

import torch
import torchio as tio

# Custom function for tensor processing
def custom_intensity_scaling(tensor):
    """Custom intensity scaling function"""
    return tensor * 1.5 + 0.1

# Custom function for subject processing
def add_computed_field(subject):
    """Add computed field to subject"""
    if 't1' in subject and 't2' in subject:
        # Compute T1/T2 ratio
        t1_data = subject['t1'].data
        t2_data = subject['t2'].data
        ratio = t1_data / (t2_data + 1e-6)  # Avoid division by zero
        
        # Create new image with ratio
        ratio_image = tio.ScalarImage(tensor=ratio, affine=subject['t1'].affine)
        subject['t1_t2_ratio'] = ratio_image
    
    return subject

# Create lambda transforms
intensity_lambda = tio.Lambda(
    function=custom_intensity_scaling,
    types_to_apply=(tio.ScalarImage,)  # Only apply to scalar images
)

ratio_lambda = tio.Lambda(function=add_computed_field)

# Use in pipeline
pipeline = tio.Compose([
    tio.ZNormalization(),
    intensity_lambda,         # Apply custom scaling
    ratio_lambda,            # Compute T1/T2 ratio
    tio.RandomFlip(),
])

subject = tio.Subject(
    t1=tio.ScalarImage('t1.nii.gz'),
    t2=tio.ScalarImage('t2.nii.gz')
)

processed = pipeline(subject)
# Now subject contains 't1_t2_ratio' image

Advanced Composition Patterns

Examples of advanced composition patterns for complex processing pipelines.

Conditional Augmentation

def conditional_augmentation_pipeline():
    """Create pipeline with conditional augmentation based on image properties"""
    
    def age_based_augmentation(subject):
        """Apply different augmentation based on subject age"""
        age = subject.get('age', 50)  # Default age if not specified
        
        if age < 30:
            # Stronger augmentation for younger subjects
            augment = tio.Compose([
                tio.RandomAffine(degrees=(-10, 10), scales=(0.9, 1.1)),
                tio.RandomElasticDeformation(max_displacement=7.5),
                tio.RandomNoise(std=(0, 0.1)),
            ])
        else:
            # Milder augmentation for older subjects
            augment = tio.Compose([
                tio.RandomAffine(degrees=(-5, 5), scales=(0.95, 1.05)),
                tio.RandomNoise(std=(0, 0.05)),
            ])
        
        return augment(subject)
    
    return tio.Compose([
        tio.ToCanonical(),
        tio.ZNormalization(),
        tio.Lambda(age_based_augmentation),
    ])

Multi-Stage Processing

def multi_stage_pipeline():
    """Create multi-stage processing pipeline"""
    
    # Stage 1: Basic preprocessing
    stage1 = tio.Compose([
        tio.ToCanonical(),
        tio.Resample(1),
        tio.CropOrPad((128, 128, 64)),
    ])
    
    # Stage 2: Intensity normalization
    stage2 = tio.Compose([
        tio.RescaleIntensity(out_min_max=(0, 1)),
        tio.ZNormalization(),
    ])
    
    # Stage 3: Augmentation (applied randomly)
    stage3 = tio.OneOf({
        tio.Compose([  # Spatial augmentation
            tio.RandomFlip(),
            tio.RandomAffine(degrees=(-5, 5))
        ]): 0.5,
        tio.Compose([  # Intensity augmentation
            tio.RandomNoise(std=(0, 0.1)),
            tio.RandomGamma(log_gamma=(-0.3, 0.3))
        ]): 0.5,
    })
    
    return tio.Compose([stage1, stage2, stage3])

Pipeline with Quality Control

def pipeline_with_qc():
    """Pipeline that includes quality control checks"""
    
    def quality_check(subject):
        """Perform quality checks on processed subject"""
        for key, image in subject.get_images(intensity_only=False):
            # Check for extreme values
            data = image.data
            if torch.any(torch.isnan(data)) or torch.any(torch.isinf(data)):
                raise ValueError(f"Invalid values detected in {key}")
            
            # Check shape consistency
            if data.shape[-3:] != (128, 128, 64):
                raise ValueError(f"Unexpected shape in {key}: {data.shape}")
        
        return subject
    
    return tio.Compose([
        tio.ToCanonical(),
        tio.Resample(1),
        tio.CropOrPad((128, 128, 64)),
        tio.ZNormalization(),
        tio.Lambda(quality_check),  # QC after preprocessing
        tio.RandomFlip(),
        tio.RandomNoise(std=(0, 0.05)),
        tio.Lambda(quality_check),  # QC after augmentation
    ])

Usage of advanced patterns:

# Create and use advanced pipelines
conditional_pipeline = conditional_augmentation_pipeline()
multi_stage = multi_stage_pipeline()
qc_pipeline = pipeline_with_qc()

subject = tio.Subject(
    t1=tio.ScalarImage('t1.nii.gz'),
    age=25  # Young subject for conditional augmentation
)

# Apply different processing strategies
processed_conditional = conditional_pipeline(subject)
processed_multi_stage = multi_stage(subject)
processed_with_qc = qc_pipeline(subject)

Transform History and Debugging

All composed transforms maintain history for debugging and reproducibility:

# Apply transform pipeline
subject = tio.Subject(t1=tio.ScalarImage('t1.nii.gz'))
pipeline = tio.Compose([
    tio.ToCanonical(),
    tio.RandomFlip(),
    tio.RandomNoise(std=0.1),
])

transformed = pipeline(subject)

# Access transform history
print("Applied transforms:")
for transform_name, params in transformed.history:
    print(f"  {transform_name}: {params}")

# History enables reproducibility and debugging

Install with Tessl CLI