Batch Processing

This guide covers processing multiple images efficiently using both Python and the command-line interface.

Overview

Batch processing allows you to:

  • Process dozens to thousands of images automatically

  • Use parallel processing to leverage multiple CPU cores

  • Handle errors gracefully without stopping the entire batch

  • Organize outputs systematically

  • Save configurations for reproducibility

MedTDA provides two approaches:

  1. Python API - Loop over images with FeatureExtractor

  2. Command-Line Interface (CLI) - Powerful batch processing with parallelization

Python Batch Processing

Basic Batch Loop

Process multiple images in a Python loop:

from medtda import FeatureExtractor
import pandas as pd

# Initialize extractor once
extractor = FeatureExtractor(
    normalize=True,
    spacing=(1.0, 1.0, 1.0),
    crop_to_roi=True,
    vectorization_method='PersStats'
)

# List of cases to process
cases = [
    {'id': 'patient001', 'image': 'scan001.nii.gz', 'mask': 'mask001.nii.gz'},
    {'id': 'patient002', 'image': 'scan002.nii.gz', 'mask': 'mask002.nii.gz'},
    {'id': 'patient003', 'image': 'scan003.nii.gz', 'mask': 'mask003.nii.gz'},
]

# Process all cases
results = []
for case in cases:
    try:
        features = extractor.execute(case['image'], case['mask'])
        features['id'] = case['id']
        results.append(features)
        print(f"✓ Processed {case['id']}")
    except Exception as e:
        print(f"✗ Error processing {case['id']}: {e}")

# Convert to DataFrame and save
df = pd.DataFrame(results)
df.to_csv('batch_features.csv', index=False)
print(f"\nProcessed {len(results)}/{len(cases)} cases successfully")

With Progress Bar

Add a progress bar using tqdm:

from tqdm import tqdm

results = []
for case in tqdm(cases, desc="Processing images"):
    try:
        features = extractor.execute(case['image'], case['mask'])
        features['id'] = case['id']
        results.append(features)
    except Exception as e:
        tqdm.write(f"Error processing {case['id']}: {e}")

Error Handling

Comprehensive error handling:

import traceback

results = []
errors = []

for case in tqdm(cases):
    try:
        features = extractor.execute(case['image'], case['mask'])
        features['id'] = case['id']
        features['status'] = 'success'
        results.append(features)

    except FileNotFoundError as e:
        errors.append({
            'id': case['id'],
            'error': 'File not found',
            'details': str(e)
        })

    except ValueError as e:
        errors.append({
            'id': case['id'],
            'error': 'Invalid input',
            'details': str(e)
        })

    except Exception as e:
        errors.append({
            'id': case['id'],
            'error': 'Processing failed',
            'details': str(e),
            'traceback': traceback.format_exc()
        })

# Save results and errors
if results:
    pd.DataFrame(results).to_csv('batch_features.csv', index=False)
if errors:
    pd.DataFrame(errors).to_csv('batch_errors.csv', index=False)

print(f"Success: {len(results)}, Errors: {len(errors)}")

Loading Cases from CSV

Read case list from a CSV file:

import pandas as pd

# CSV format:
# id,image_path,mask_path
# case001,/path/to/scan1.nii.gz,/path/to/mask1.nii.gz
# case002,/path/to/scan2.nii.gz,/path/to/mask2.nii.gz

cases_df = pd.read_csv('cases.csv')

results = []
for _, row in tqdm(cases_df.iterrows(), total=len(cases_df)):
    try:
        features = extractor.execute(
            image=row['image_path'],
            mask=row['mask_path'] if pd.notna(row['mask_path']) else None
        )
        features['id'] = row['id']
        results.append(features)
    except Exception as e:
        print(f"Error processing {row['id']}: {e}")

pd.DataFrame(results).to_csv('batch_features.csv', index=False)

Parallel Processing (Python)

Use multiprocessing for parallel execution:

from multiprocessing import Pool
from functools import partial

def process_case(case, extractor_params):
    """Worker function to process a single case."""
    from medtda import FeatureExtractor

    # Create extractor (can't pickle extractor object)
    extractor = FeatureExtractor(**extractor_params)

    try:
        features = extractor.execute(case['image'], case['mask'])
        features['id'] = case['id']
        features['status'] = 'success'
        return features
    except Exception as e:
        return {
            'id': case['id'],
            'status': 'error',
            'error': str(e)
        }

# Extractor parameters
extractor_params = {
    'normalize': True,
    'spacing': (1.0, 1.0, 1.0),
    'crop_to_roi': True,
    'vectorization_method': 'PersStats'
}

# Process in parallel with 4 workers
with Pool(processes=4) as pool:
    worker = partial(process_case, extractor_params=extractor_params)
    results = pool.map(worker, cases)

# Save results
pd.DataFrame(results).to_csv('batch_features.csv', index=False)

Command-Line Batch Processing

The CLI provides powerful batch processing capabilities with built-in parallelization.

CSV Input Format

Create a CSV file with your cases:

id,image_path,mask_path
case001,/data/scans/patient001.nii.gz,/data/masks/patient001.nii.gz
case002,/data/scans/patient002.nii.gz,/data/masks/patient002.nii.gz
case003,/data/scans/patient003.nii.gz,/data/masks/patient003.nii.gz
case004,/data/scans/patient004.nii.gz,

Requirements:

  • Must have id and image_path columns

  • mask_path is optional (can be empty)

  • Paths can be absolute or relative

  • CSV file can have additional columns (ignored)

Basic Batch Processing

Process all cases sequentially:

medtda cases.csv --output-dir ./results --normalize

This processes each case one by one and saves:

  • batch_features.csv - All features combined

  • config.yaml - Configuration used

Parallel Processing

Use multiple workers for faster processing:

# Use 4 parallel workers
medtda cases.csv --output-dir ./results --workers 4 --normalize

# Use all available CPU cores
medtda cases.csv --output-dir ./results --workers -1 --normalize

# Short form
medtda cases.csv -o ./results -j 4 --normalize

Performance:

  • Sequential (workers=1): ~100% single-core usage

  • Parallel (workers=4): ~400% cumulative CPU usage

  • Parallel (workers=-1): Uses all cores

With Preprocessing

Add preprocessing options:

medtda cases.csv --output-dir ./results --workers 4 \\
    --normalize --normalize-method minmax \\
    --spacing 1.0 1.0 1.0 \\
    --crop-roi --roi-padding 2 \\
    --verbose

With Multiple Vectorization Methods

Specify multiple methods:

medtda cases.csv --output-dir ./results --workers 4 \\
    --normalize \\
    --methods persistence_stats betti_curve entropy_summary

Save Barcodes

Save persistence barcodes for each case:

medtda cases.csv --output-dir ./results --workers 4 \\
    --normalize --save-barcodes

Creates files:

  • batch_features.csv - Features for all cases

  • case001_barcodes.pkl - Barcodes for case001

  • case002_barcodes.pkl - Barcodes for case002

  • etc.

Verbose Output

Show detailed progress:

medtda cases.csv --output-dir ./results --workers 4 --verbose

Displays:

  • Configuration summary

  • Progress bar

  • Per-case status

  • Error messages

  • Summary statistics

Using Configuration Files

Save processing parameters in a YAML file for reproducibility.

Create Configuration File

# config.yaml
preprocessing:
  normalize: true
  normalize_method: minmax
  spacing: [1.0, 1.0, 1.0]
  crop_roi: true
  roi_padding: 2
  window: null
  background_value: null
  label: 1

persistent_homology:
  filtration: sublevel
  construction: T
  max_dimension: 2

vectorization:
  methods:
    - persistence_stats
    - betti_curve
    - entropy_summary
  save_barcodes: false

parallel:
  workers: 4

output:
  output_dir: ./results
  verbose: true
  quiet: false

Use Configuration File

# Use config file
medtda cases.csv --config config.yaml

# Override specific parameters
medtda cases.csv --config config.yaml --workers 8 --verbose

Priority: CLI arguments > config file > defaults

Benefits of Config Files

  • Reproducibility - Same parameters across runs

  • Documentation - Self-documenting processing pipeline

  • Easy sharing - Share exact processing parameters

  • Version control - Track parameter changes

  • Batch consistency - Ensure all images processed identically

Output Organization

Batch Processing Outputs

After batch processing, the output directory contains:

results/
├── batch_features.csv      # All features (one row per case)
├── config.yaml              # Configuration used
├── case001_barcodes.pkl     # Barcodes (if --save-barcodes)
├── case002_barcodes.pkl
└── ...

batch_features.csv Format

The output CSV includes:

id,status,error_message,PersStats_H0_mean,PersStats_H0_std,...
case001,success,,0.523,0.142,...
case002,success,,0.487,0.156,...
case003,error,File not found,,,

Columns:

  • id - Case identifier

  • status - ‘success’ or ‘error’

  • error_message - Error details (if failed)

  • Feature columns - All extracted features

Loading Results

import pandas as pd

# Load results
df = pd.read_csv('results/batch_features.csv')

# Filter successful cases
df_success = df[df['status'] == 'success']

# Get feature columns only
feature_cols = [c for c in df.columns
                if c not in ['id', 'status', 'error_message']]
X = df_success[feature_cols]

print(f"Successfully processed: {len(df_success)}/{len(df)}")
print(f"Features per case: {len(feature_cols)}")

Large-Scale Batch Processing

For very large datasets (100s-1000s of images):

Chunking

Process in chunks to manage resources:

import pandas as pd
from tqdm import tqdm

# Read case list
all_cases = pd.read_csv('all_cases.csv')

# Process in chunks of 100
chunk_size = 100
for i in range(0, len(all_cases), chunk_size):
    chunk = all_cases.iloc[i:i+chunk_size]

    # Save chunk to temp CSV
    chunk_csv = f'temp_chunk_{i}.csv'
    chunk.to_csv(chunk_csv, index=False)

    # Process chunk with CLI
    import subprocess
    subprocess.run([
        'medtda', chunk_csv,
        '--output-dir', f'results_chunk_{i}',
        '--workers', '8',
        '--normalize'
    ])

# Combine all chunk results
all_results = []
for i in range(0, len(all_cases), chunk_size):
    result_file = f'results_chunk_{i}/batch_features.csv'
    all_results.append(pd.read_csv(result_file))

final_df = pd.concat(all_results, ignore_index=True)
final_df.to_csv('final_results.csv', index=False)

Resource Management

Optimize for large batches:

# Use moderate workers (avoid memory issues)
medtda large_cases.csv --workers 4 --output-dir ./results \\
    --max-dimension 1 \\         # Skip H2 to save memory
    --spacing 2.0 2.0 2.0 \\     # Downsample to save time
    --crop-roi --roi-padding 1   # Reduce image size

Monitoring Progress

For long-running batches:

# Run with verbose output and redirect to log
medtda cases.csv --workers 8 --output-dir ./results --verbose \\
    2>&1 | tee processing.log

# Monitor progress in another terminal
tail -f processing.log

Resume Failed Cases

Re-process only failed cases:

import pandas as pd

# Load previous results
results = pd.read_csv('results/batch_features.csv')

# Find failed cases
failed = results[results['status'] == 'error']['id'].tolist()

# Load original case list
all_cases = pd.read_csv('cases.csv')

# Create CSV with only failed cases
failed_cases = all_cases[all_cases['id'].isin(failed)]
failed_cases.to_csv('failed_cases.csv', index=False)

# Re-process
# medtda failed_cases.csv --output-dir ./results_retry --workers 4

Best Practices

  1. Test on subset first - Verify parameters on 5-10 cases

  2. Use configuration files - Document and reproduce parameters

  3. Enable –verbose - Monitor progress and catch errors early

  4. Save barcodes - For future re-vectorization

  5. Use parallel processing - Significant speedup for batches

  6. Handle errors gracefully - Continue processing after failures

  7. Organize outputs - Use clear output directory structure

  8. Document case selection - Save case list and exclusion criteria

  9. Version control configs - Track parameter changes over time

  10. Validate outputs - Check feature distributions and missing values

Troubleshooting

Some cases fail with “File not found”

  • Check CSV paths are correct (absolute or relative to CWD)

  • Verify all files exist before processing

  • Use absolute paths to avoid confusion

Parallel processing crashes

  • Reduce number of workers

  • Check available RAM (each worker needs memory)

  • Try sequential processing (workers=1) first

Different number of features per case

  • All cases should have same features

  • Check for errors in some cases

  • Verify same vectorization methods used

Output CSV is huge

  • Reduce number of vectorization methods

  • Use simpler methods (persistence_stats instead of persistence_image)

  • Consider if all features are needed

Processing is very slow

  • Use parallel processing (–workers 4 or more)

  • Enable ROI cropping (–crop-roi)

  • Set max-dimension to 1 instead of 2

  • Downsample images (–spacing 2.0 2.0 2.0)

Example Complete Workflow

# 1. Create case list CSV
cat > cases.csv << EOF
id,image_path,mask_path
patient001,/data/scans/001.nii.gz,/data/masks/001.nii.gz
patient002,/data/scans/002.nii.gz,/data/masks/002.nii.gz
patient003,/data/scans/003.nii.gz,/data/masks/003.nii.gz
EOF

# 2. Create configuration file
cat > config.yaml << EOF
preprocessing:
  normalize: true
  spacing: [1.0, 1.0, 1.0]
  crop_roi: true
persistent_homology:
  filtration: sublevel
  max_dimension: 2
vectorization:
  methods:
    - persistence_stats
    - betti_curve
EOF

# 3. Test on first case
head -2 cases.csv | tail -1 > test_case.csv
medtda test_case.csv --config config.yaml --output-dir ./test --verbose

# 4. Process full batch in parallel
medtda cases.csv --config config.yaml --output-dir ./results \\
    --workers 8 --verbose

# 5. Check results
python -c "
import pandas as pd
df = pd.read_csv('results/batch_features.csv')
print(f'Success: {(df.status==\"success\").sum()}/{len(df)}')
# Count feature columns (exclude metadata: id, status, error_message)
feature_cols = [c for c in df.columns if c not in ['id', 'status', 'error_message']]
print(f'Features: {len(feature_cols)}')
"

Next Steps