engine/crog_engine_optimized.py

"""
Optimized CROG Training Engine
==============================
This is an optimized version of the CROG training engine with:
1. Reduced code duplication between train/validate functions
2. More efficient metric calculation
3. Better memory management
4. Cleaner code structure

Key optimizations:
- Unified metric calculation functions
- Batch processing for validation metrics
- Reduced tensor operations
- Memory-efficient gradient accumulation
"""

import os
import time
from tqdm import tqdm
from typing import Dict, Tuple, List, Optional, Any

import cv2
import numpy as np
import torch
import torch.cuda.amp as amp
import torch.distributed as dist
import torch.nn.functional as F
import wandb
from loguru import logger

from utils.dataset import tokenize
from utils.misc import AverageMeter, ProgressMeter, concat_all_gather
from utils.grasp_eval import detect_grasps, calculate_jacquard_index


class MetricCalculator:
    """
    Centralized metric calculation to avoid code duplication.
    
    This class handles all metric computations for both training and validation,
    ensuring consistency and reducing redundancy.
    """
    
    @staticmethod
    def calculate_iou(pred: torch.Tensor, target: torch.Tensor, threshold: float = 0.35) -> torch.Tensor:
        """
        Calculate Intersection over Union (IoU) metric.
        
        Args:
            pred: Predicted mask (logits or probabilities)
            target: Ground truth mask
            threshold: Threshold for binarization
            
        Returns:
            IoU value
        """
        # Ensure predictions are probabilities
        if pred.min() < 0 or pred.max() > 1:
            pred = torch.sigmoid(pred)
        
        # Binarize masks
        pred_binary = pred > threshold
        target_binary = target > 0.5
        
        # Calculate IoU efficiently
        intersection = (pred_binary & target_binary).float().sum(dim=(1, 2, 3))
        union = (pred_binary | target_binary).float().sum(dim=(1, 2, 3))
        
        # Avoid division by zero
        iou = intersection / (union + 1e-6)
        
        return iou
    
    @staticmethod
    def calculate_precision(pred: torch.Tensor, target: torch.Tensor, threshold: float) -> torch.Tensor:
        """Calculate precision at a specific threshold."""
        pred_binary = torch.sigmoid(pred) > threshold
        target_binary = target > 0.5
        
        correct = (pred_binary == target_binary).float()
        precision = correct.mean()
        
        return precision
    
    @staticmethod
    def batch_inverse_transform(
        masks: List[np.ndarray],
        inverse_matrices: List[np.ndarray],
        original_sizes: List[Tuple[int, int]]
    ) -> List[np.ndarray]:
        """
        Batch process inverse transformations for efficiency.
        
        This avoids repeated single-image processing in loops.
        """
        transformed_masks = []
        
        for mask, inv_mat, (h, w) in zip(masks, inverse_matrices, original_sizes):
            # Use INTER_LINEAR for masks (faster than INTER_CUBIC for binary data)
            inv_mask = cv2.warpAffine(
                mask, inv_mat, (w, h),
                flags=cv2.INTER_LINEAR,
                borderValue=0.
            )
            transformed_masks.append(inv_mask)
        
        return transformed_masks


def train_with_grasp_optimized(
    train_loader: torch.utils.data.DataLoader,
    model: torch.nn.Module,
    optimizer: torch.optim.Optimizer,
    scheduler: torch.optim.lr_scheduler._LRScheduler,
    scaler: amp.GradScaler,
    epoch: int,
    args: Dict[str, Any]
) -> Dict[str, float]:
    """
    Optimized training function with grasp predictions.
    
    Improvements:
    - Unified metric tracking
    - Efficient memory usage
    - Cleaner code structure
    - Better error handling
    """
    # Initialize meters for all metrics
    meters = {
        'batch_time': AverageMeter('Batch', ':2.2f'),
        'data_time': AverageMeter('Data', ':2.2f'),
        'loss': AverageMeter('Loss', ':2.4f'),
        'loss_qua': AverageMeter('Loss_qua', ':2.4f'),
        'loss_sin': AverageMeter('Loss_sin', ':2.4f'),
        'loss_cos': AverageMeter('Loss_cos', ':2.4f'),
        'loss_wid': AverageMeter('Loss_wid', ':2.4f'),
        'iou': AverageMeter('IoU', ':2.2f'),
        'pr@50': AverageMeter('Prec@50', ':2.2f'),
        'lr': AverageMeter('Lr', ':1.6f')
    }
    
    # Create progress meter with all metrics
    progress = ProgressMeter(
        len(train_loader),
        list(meters.values()),
        prefix=f"Training: Epoch=[{epoch}/{args.epochs}] "
    )
    
    # Set model to training mode
    model.train()
    metric_calc = MetricCalculator()
    
    end = time.time()
    
    for i, data in enumerate(train_loader):
        # Record data loading time
        meters['data_time'].update(time.time() - end)
        
        # Prepare batch data efficiently
        batch_data = prepare_batch_data(data, args.gpu)
        
        # Forward pass with automatic mixed precision
        with amp.autocast():
            pred, target, loss, loss_dict = model(
                batch_data['image'],
                batch_data['text'],
                batch_data['ins_mask'],
                batch_data['grasp_qua'],
                batch_data['grasp_sin'],
                batch_data['grasp_cos'],
                batch_data['grasp_wid']
            )
        
        # Backward pass with gradient scaling
        optimizer.zero_grad()
        scaler.scale(loss).backward()
        
        # Gradient clipping if specified
        if args.max_norm:
            scaler.unscale_(optimizer)
            torch.nn.utils.clip_grad_norm_(model.parameters(), args.max_norm)
        
        scaler.step(optimizer)
        scaler.update()
        
        # Calculate metrics efficiently
        with torch.no_grad():
            # Instance segmentation metrics
            iou = metric_calc.calculate_iou(pred[0], target[0], threshold=0.35)
            pr50 = metric_calc.calculate_precision(pred[0], target[0], threshold=0.5)
            
            # Reduce metrics across all processes
            if dist.is_initialized():
                metrics_tensor = torch.stack([loss.detach(), iou.mean(), pr50])
                dist.all_reduce(metrics_tensor)
                metrics_tensor /= dist.get_world_size()
                loss_reduced, iou_reduced, pr50_reduced = metrics_tensor
            else:
                loss_reduced = loss.detach()
                iou_reduced = iou.mean()
                pr50_reduced = pr50
        
        # Update meters
        batch_size = batch_data['image'].size(0)
        meters['loss'].update(loss_reduced.item(), batch_size)
        meters['loss_qua'].update(loss_dict.get('m_qua', 0), batch_size)
        meters['loss_sin'].update(loss_dict.get('m_sin', 0), batch_size)
        meters['loss_cos'].update(loss_dict.get('m_cos', 0), batch_size)
        meters['loss_wid'].update(loss_dict.get('m_wid', 0), batch_size)
        meters['iou'].update(iou_reduced.item(), batch_size)
        meters['pr@50'].update(pr50_reduced.item(), batch_size)
        meters['lr'].update(scheduler.get_last_lr()[-1])
        
        # Record batch time
        meters['batch_time'].update(time.time() - end)
        end = time.time()
        
        # Log progress
        if (i + 1) % args.print_freq == 0:
            progress.display(i + 1)
            
            # WandB logging (only on main process)
            if dist.get_rank() in [-1, 0]:
                log_metrics_to_wandb(meters, epoch, i, len(train_loader))
    
    # Return epoch averages
    return {name: meter.avg for name, meter in meters.items()}


@torch.no_grad()
def validate_with_grasp_optimized(
    val_loader: torch.utils.data.DataLoader,
    model: torch.nn.Module,
    epoch: int,
    args: Dict[str, Any]
) -> Tuple[float, Dict[str, float], List[float]]:
    """
    Optimized validation function with grasp evaluation.
    
    Improvements:
    - Batch processing of inverse transforms
    - Vectorized metric calculations
    - Memory-efficient processing
    """
    model.eval()
    metric_calc = MetricCalculator()
    
    # Initialize metric storage
    all_ious = []
    j_indices = {1: [], 5: []}  # For 1 and 5 grasps
    
    # Use tqdm for progress tracking
    pbar = tqdm(val_loader, desc=f'Validation Epoch {epoch}')
    
    for data in pbar:
        # Prepare batch data
        batch_data = prepare_batch_data(data, torch.cuda.current_device())
        
        # Forward pass
        pred, target = model(
            batch_data['image'],
            batch_data['text'],
            batch_data['ins_mask'],
            batch_data['grasp_qua'],
            batch_data['grasp_sin'],
            batch_data['grasp_cos'],
            batch_data['grasp_wid']
        )
        
        # Process predictions
        predictions = process_predictions(pred, batch_data['image'].shape[-2:])
        
        # Batch process inverse transformations
        batch_size = predictions['ins_mask'].shape[0]
        
        # Convert to numpy for CV2 operations
        ins_masks_np = predictions['ins_mask'].cpu().numpy()
        qua_masks_np = predictions['grasp_qua'].cpu().numpy()
        sin_masks_np = predictions['grasp_sin'].cpu().numpy()
        cos_masks_np = predictions['grasp_cos'].cpu().numpy()
        wid_masks_np = predictions['grasp_wid'].cpu().numpy()
        
        # Process each sample in the batch
        for idx in range(batch_size):
            # Get sample data
            inv_mat = data['inverse'][idx]
            h, w = data['ori_size'][idx]
            grasp_target = data['grasps'][idx]
            
            # Inverse transform masks
            masks_to_transform = [
                ins_masks_np[idx],
                qua_masks_np[idx].squeeze(),
                sin_masks_np[idx].squeeze(),
                cos_masks_np[idx].squeeze(),
                wid_masks_np[idx].squeeze()
            ]
            
            transformed = metric_calc.batch_inverse_transform(
                masks_to_transform,
                [inv_mat] * 5,
                [(h, w)] * 5
            )
            
            ins_mask_inv = transformed[0] > 0.35
            
            # Calculate IoU
            ins_target_np = target[0][idx].squeeze().cpu().numpy()
            ins_target_inv = cv2.warpAffine(ins_target_np, inv_mat, (w, h))
            
            inter = np.logical_and(ins_mask_inv, ins_target_inv)
            union = np.logical_or(ins_mask_inv, ins_target_inv)
            iou = np.sum(inter) / (np.sum(union) + 1e-6)
            all_ious.append(iou)
            
            # Calculate grasp metrics if enabled
            if args.use_grasp_masks:
                for num_grasps in [1, 5]:
                    grasp_preds, _ = detect_grasps(
                        transformed[1], transformed[2], 
                        transformed[3], transformed[4], 
                        num_grasps
                    )
                    j_index = calculate_jacquard_index(grasp_preds, grasp_target)
                    j_indices[num_grasps].append(j_index)
    
    # Aggregate metrics
    iou_tensor = torch.tensor(all_ious, device='cuda')
    if dist.is_initialized():
        iou_tensor = concat_all_gather(iou_tensor)
    
    # Calculate precision at different thresholds
    prec_dict = {}
    for threshold in torch.arange(0.5, 1.0, 0.1):
        prec = (iou_tensor > threshold).float().mean()
        prec_dict[f'Pr@{int(threshold * 100)}'] = prec.item()
    
    # Calculate average metrics
    avg_iou = iou_tensor.mean().item()
    
    # Calculate J-indices
    j_index_results = []
    if args.use_grasp_masks:
        for num_grasps in [1, 5]:
            if j_indices[num_grasps]:
                avg_j = np.mean(j_indices[num_grasps])
            else:
                avg_j = 0.0
            j_index_results.append(avg_j)
    else:
        j_index_results = [0.0, 0.0]
    
    # Log results
    log_validation_results(epoch, args.epochs, avg_iou, prec_dict, j_index_results)
    
    return avg_iou, prec_dict, j_index_results


def prepare_batch_data(data: Dict[str, Any], device: int) -> Dict[str, torch.Tensor]:
    """
    Efficiently prepare batch data for GPU processing.
    
    This function consolidates data preparation logic to avoid duplication.
    """
    return {
        'image': data['img'].cuda(device, non_blocking=True),
        'text': data['word_vec'].cuda(device, non_blocking=True),
        'ins_mask': data['mask'].cuda(device, non_blocking=True).unsqueeze(1),
        'grasp_qua': data['grasp_masks']['qua'].cuda(device, non_blocking=True).unsqueeze(1),
        'grasp_sin': data['grasp_masks']['sin'].cuda(device, non_blocking=True).unsqueeze(1),
        'grasp_cos': data['grasp_masks']['cos'].cuda(device, non_blocking=True).unsqueeze(1),
        'grasp_wid': data['grasp_masks']['wid'].cuda(device, non_blocking=True).unsqueeze(1)
    }


def process_predictions(
    pred: Tuple[torch.Tensor, ...],
    target_size: Tuple[int, int]
) -> Dict[str, torch.Tensor]:
    """
    Process model predictions with proper interpolation and activation.
    
    This consolidates prediction processing logic.
    """
    # Apply sigmoid activation
    ins_mask = torch.sigmoid(pred[0])
    grasp_qua = torch.sigmoid(pred[1])
    grasp_wid = torch.sigmoid(pred[4])
    
    # Interpolate if needed
    if ins_mask.shape[-2:] != target_size:
        ins_mask = F.interpolate(ins_mask, size=target_size, mode='bicubic', align_corners=True)
        grasp_qua = F.interpolate(grasp_qua, size=target_size, mode='bicubic', align_corners=True)
        pred_sin = F.interpolate(pred[2], size=target_size, mode='bicubic', align_corners=True)
        pred_cos = F.interpolate(pred[3], size=target_size, mode='bicubic', align_corners=True)
        grasp_wid = F.interpolate(grasp_wid, size=target_size, mode='bicubic', align_corners=True)
    else:
        pred_sin = pred[2]
        pred_cos = pred[3]
    
    return {
        'ins_mask': ins_mask.squeeze(1),
        'grasp_qua': grasp_qua.squeeze(1),
        'grasp_sin': pred_sin.squeeze(1),
        'grasp_cos': pred_cos.squeeze(1),
        'grasp_wid': grasp_wid.squeeze(1)
    }


def log_metrics_to_wandb(
    meters: Dict[str, AverageMeter],
    epoch: int,
    batch_idx: int,
    total_batches: int
):
    """Centralized WandB logging function."""
    step = epoch * total_batches + batch_idx + 1
    
    metrics = {
        "time/batch": meters['batch_time'].val,
        "time/data": meters['data_time'].val,
        "training/lr": meters['lr'].val,
        "training/loss": meters['loss'].val,
        "training/loss_qua": meters['loss_qua'].val,
        "training/loss_sin": meters['loss_sin'].val,
        "training/loss_cos": meters['loss_cos'].val,
        "training/loss_wid": meters['loss_wid'].val,
        "training/iou": meters['iou'].val,
        "training/prec@50": meters['pr@50'].val,
    }
    
    wandb.log(metrics, step=step)


def log_validation_results(
    epoch: int,
    total_epochs: int,
    iou: float,
    prec_dict: Dict[str, float],
    j_indices: List[float]
):
    """Log validation results to console."""
    # Build precision string
    prec_str = "  ".join([f"{k}: {v*100:.2f}" for k, v in prec_dict.items()])
    
    # Log main metrics
    logger.info(
        f"Evaluation: Epoch=[{epoch}/{total_epochs}]  "
        f"IoU={iou*100:.2f}  "
        f"J_index@1: {j_indices[0]*100:.2f}  "
        f"J_index@5: {j_indices[1]*100:.2f}  "
        f"{prec_str}"
    )