code/scripts/generate_synthetic_parallel.py

#!/usr/bin/env python3
"""
Fast Parallel Synthetic ECG Generator using ecg-image-kit

Generates synthetic ECG images from PTB-XL records in parallel.
Run with venv: source ecgkit_venv/bin/activate && python generate_synthetic_parallel.py

Usage:
    python generate_synthetic_parallel.py --n_samples 50000 --n_workers 16
"""

import os
import sys
import argparse
import subprocess
import random
import json
import shutil
import numpy as np
import wfdb
from scipy import signal as scipy_signal
from pathlib import Path
from multiprocessing import Pool, cpu_count
from tqdm import tqdm

# Constants for GT extraction
TARGET_HEIGHT = 1696
ZERO_MV = np.array([703.5, 987.5, 1271.5, 1531.5])
MV_TO_PIXEL = 78.5
T0, T1 = 235, 4161
OUTPUT_GT_WIDTH = T1 - T0  # 3926

# Paths
PROJECT_ROOT = Path(__file__).parent.parent
ECG_IMAGE_KIT = PROJECT_ROOT / 'ecg-image-kit' / 'codes' / 'ecg-image-generator'
PTBXL_DIR = Path('/data/ecg-digitization/ptbxl/physionet.org/files/ptb-xl/1.0.3')

# Grid color mappings (0001-0012 styles)
STYLE_CONFIGS = [
    {'grid_color': 5, 'augment': False, 'wrinkles': False},  # 0001: Clean red
    {'grid_color': 5, 'augment': True, 'wrinkles': True},    # 0002: Aged red
    {'grid_color': 2, 'augment': False, 'wrinkles': False},  # 0003: Green
    {'grid_color': 2, 'augment': True, 'wrinkles': True},    # 0004: Aged green
    {'grid_color': 1, 'augment': False, 'wrinkles': False},  # 0005: Blue
    {'grid_color': 3, 'augment': True, 'wrinkles': False},   # 0006: Grey
    {'grid_color': 4, 'augment': True, 'wrinkles': True},    # 0007: Yellow aged
    {'grid_color': 2, 'augment': True, 'wrinkles': False},   # 0008: Faded green
    {'grid_color': 6, 'augment': False, 'wrinkles': False},  # 0009: Orange
    {'grid_color': 5, 'augment': True, 'wrinkles': False},   # 0010: High contrast
    {'grid_color': 5, 'augment': True, 'wrinkles': True},    # 0011: Low contrast
    {'grid_color': 0, 'augment': False, 'wrinkles': False},  # 0012: No grid
]


def find_ptbxl_records():
    """Find all PTB-XL record files."""
    records = []
    
    for records_dir in ['records500', 'records100']:
        rec_path = PTBXL_DIR / records_dir
        if rec_path.exists():
            for subdir in rec_path.iterdir():
                if subdir.is_dir():
                    for hea in subdir.glob('*.hea'):
                        dat = hea.with_suffix('.dat')
                        if dat.exists():
                            records.append((str(hea), str(dat)))
            if records:
                break
    
    return records


def extract_gt_signal(hea_file):
    """Extract ground truth signal from PTB-XL record as pixel Y coordinates."""
    try:
        record_path = hea_file.replace('.hea', '')
        record = wfdb.rdrecord(record_path)
        
        signals = record.p_signal  # [samples, leads]
        sig_names = [n.upper() for n in record.sig_name]
        fs = record.fs
        
        # Resample to 500Hz if needed
        if fs != 500:
            num_samples = int(signals.shape[0] * 500 / fs)
            signals = scipy_signal.resample(signals, num_samples, axis=0)
        
        # Map lead names to indices
        lead_signals = {}
        for i, name in enumerate(sig_names):
            lead_signals[name] = signals[:, i]
        
        # 4-row layout matching Kaggle format
        # Row 0: I, aVR, V1, V4 (each 2.5s segment)
        # Row 1: II, aVL, V2, V5
        # Row 2: III, aVF, V3, V6
        # Row 3: Full Lead II (10s rhythm strip)
        row_leads = [
            ['I', 'AVR', 'V1', 'V4'],
            ['II', 'AVL', 'V2', 'V5'],
            ['III', 'AVF', 'V3', 'V6'],
        ]
        
        segment_width = OUTPUT_GT_WIDTH // 4  # ~981 pixels per segment
        gt_pixels = np.zeros((4, OUTPUT_GT_WIDTH), dtype=np.float32)
        
        # First 3 rows: 4 leads each, 2.5s segments
        for row_idx in range(3):
            for col_idx in range(4):
                lead_name = row_leads[row_idx][col_idx]
                if lead_name not in lead_signals:
                    continue
                    
                lead_data = lead_signals[lead_name]
                # 2.5s at 500Hz = 1250 samples
                start = col_idx * 1250
                end = min(start + 1250, len(lead_data))
                segment_mv = lead_data[start:end]
                
                if len(segment_mv) == 0:
                    continue
                
                col_start = col_idx * segment_width
                col_end = (col_idx + 1) * segment_width if col_idx < 3 else OUTPUT_GT_WIDTH
                
                # Resample to pixel width
                x_old = np.linspace(0, 1, len(segment_mv))
                x_new = np.linspace(0, 1, col_end - col_start)
                segment_resampled = np.interp(x_new, x_old, segment_mv)
                
                # Convert mV to pixel Y coordinates
                pixel_y = ZERO_MV[row_idx] - segment_resampled * MV_TO_PIXEL
                gt_pixels[row_idx, col_start:col_end] = pixel_y
        
        # Row 4: Full Lead II rhythm strip (10s)
        if 'II' in lead_signals:
            lead_ii = lead_signals['II']
            x_old = np.linspace(0, 1, len(lead_ii))
            x_new = np.linspace(0, 1, OUTPUT_GT_WIDTH)
            lead_ii_resampled = np.interp(x_new, x_old, lead_ii)
            gt_pixels[3, :] = ZERO_MV[3] - lead_ii_resampled * MV_TO_PIXEL
        
        return gt_pixels
        
    except Exception as e:
        return None


def generate_single(args):
    """Generate a single synthetic image."""
    idx, hea_file, dat_file, output_dir, style_idx = args
    
    try:
        sample_id = f'syn_{idx:08d}'
        style_str = f'{(style_idx % 12) + 1:04d}'
        style_config = STYLE_CONFIGS[style_idx % 12]
        
        temp_dir = Path(f'/tmp/ecg_gen_{idx}_{os.getpid()}')
        temp_dir.mkdir(parents=True, exist_ok=True)
        
        # Build command
        # Note: -st (start_index) should be 0, -se (seed) controls randomization
        cmd = [
            'python', 'gen_ecg_image_from_data.py',
            '-i', dat_file,
            '-hea', hea_file,
            '-o', str(temp_dir),
            '-st', '0',
            '-se', str(idx),
            '-r', '200',
            '--num_columns', '4',
            '--full_mode', 'II',
        ]
        
        if style_config['grid_color'] > 0:
            cmd.extend(['--standard_grid_color', str(style_config['grid_color'])])
        else:
            cmd.extend(['--random_grid_present', '0'])
        
        if style_config['augment']:
            cmd.append('--augment')
        
        if style_config['wrinkles']:
            cmd.append('--wrinkles')
        
        # Set MPLCONFIGDIR to avoid lock conflicts
        env = os.environ.copy()
        env['MPLCONFIGDIR'] = f'/tmp/mpl_{idx}_{os.getpid()}'
        
        result = subprocess.run(
            cmd,
            cwd=str(ECG_IMAGE_KIT),
            capture_output=True,
            text=True,
            timeout=120,
            env=env
        )
        
        # Find and move generated image
        gen_images = list(temp_dir.glob('*.png'))
        output_path = None
        gt_path = None
        
        if gen_images:
            output_path = Path(output_dir) / 'raw' / f'{sample_id}-{style_str}.png'
            shutil.move(str(gen_images[0]), str(output_path))
            
            # Extract and save GT
            gt_pixels = extract_gt_signal(hea_file)
            if gt_pixels is not None:
                gt_path = Path(output_dir) / 'gt' / f'{sample_id}-{style_str}.csv'
                gt_path.parent.mkdir(parents=True, exist_ok=True)
                # Save as CSV with shape [N, 4] (pixel Y coordinates)
                np.savetxt(gt_path, gt_pixels.T, delimiter=',', fmt='%.2f')
        
        # Clean up
        shutil.rmtree(temp_dir, ignore_errors=True)
        mpl_dir = Path(f'/tmp/mpl_{idx}_{os.getpid()}')
        if mpl_dir.exists():
            shutil.rmtree(mpl_dir, ignore_errors=True)
        
        if output_path and output_path.exists() and gt_path and gt_path.exists():
            return {
                'sample_id': sample_id,
                'style': style_str,
                'path': str(output_path),
                'gt_path': str(gt_path),
                'source': hea_file,
            }
        return None
        
    except Exception as e:
        return None


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--output_dir', type=str, default='/data/ecg-digitization/synthetic_ecgkit')
    parser.add_argument('--n_samples', type=int, default=50000)
    parser.add_argument('--n_workers', type=int, default=16)
    parser.add_argument('--resume', action='store_true', help='Resume from existing progress')
    args = parser.parse_args()
    
    output_dir = Path(args.output_dir)
    (output_dir / 'raw').mkdir(parents=True, exist_ok=True)
    (output_dir / 'gt').mkdir(parents=True, exist_ok=True)
    
    # Find PTB-XL records
    print("Finding PTB-XL records...")
    records = find_ptbxl_records()
    print(f"Found {len(records)} records")
    
    if not records:
        print("No PTB-XL records found!")
        return
    
    # Check existing files for resume
    existing_indices = set()
    if args.resume:
        raw_dir = output_dir / 'raw'
        for f in raw_dir.glob('syn_*.png'):
            try:
                # Extract index from syn_00000123-0001.png
                idx = int(f.stem.split('_')[1].split('-')[0])
                existing_indices.add(idx)
            except:
                pass
        print(f"Found {len(existing_indices)} existing samples, resuming...")
    
    # Prepare tasks (skip existing if resuming)
    print(f"Preparing generation tasks...")
    tasks = []
    random.seed(42)  # Deterministic for resume
    for i in range(args.n_samples):
        hea, dat = random.choice(records)
        style_idx = 0  # Always style 0001 (clean red grid)
        if i not in existing_indices:
            tasks.append((i, hea, dat, str(output_dir), style_idx))
    
    print(f"Tasks to generate: {len(tasks)} (skipped {len(existing_indices)} existing)")
    
    if not tasks:
        print("All samples already generated!")
        return
    
    # Generate in parallel
    print(f"Generating with {args.n_workers} workers...")
    results = []
    
    with Pool(args.n_workers) as pool:
        for result in tqdm(pool.imap_unordered(generate_single, tasks), 
                          total=len(tasks), desc="Generating"):
            if result:
                results.append(result)
    
    print(f"\nSuccessfully generated {len(results)} images")
    
    # Save manifest
    manifest_path = output_dir / 'manifest.json'
    with open(manifest_path, 'w') as f:
        json.dump(results, f, indent=2)
    
    print(f"Manifest saved to {manifest_path}")
    print(f"Raw images in {output_dir / 'raw'}")


if __name__ == '__main__':
    main()