scripts/run_inference_benchmark.py

#!/usr/bin/env python3
"""Batch-1 inference latency benchmark for revised RAVEL checkpoints."""

from __future__ import annotations

import argparse
import json
import os
import platform
import sys
import time
from pathlib import Path
from typing import Any, Dict, List, Sequence, Tuple

import numpy as np
import pandas as pd
import torch
import torch.nn as nn
from PIL import Image
from transformers import DebertaV2Tokenizer
from transformers.utils import logging as hf_logging

PROJECT_ROOT = Path(__file__).resolve().parents[1]
SCRIPT_DIR = Path(__file__).resolve().parent
for path in [PROJECT_ROOT, SCRIPT_DIR]:
    if str(path) not in sys.path:
        sys.path.insert(0, str(path))

from run_revised_experiments import (  # noqa: E402
    METHODS,
    apply_method_trainability,
    freeze_non_lora,
    load_dataset,
    logits_for_target,
    set_seed,
)

hf_logging.set_verbosity_error()

DATASETS = ["mvsa_multiple", "hfm_deleak"]
MODELS = ["lora_concat", "legacy_global", "token_coattn", "token_aux", "param_mlp", "full_revised"]
CLIP_MEAN = np.array([0.48145466, 0.4578275, 0.40821073], dtype=np.float32)
CLIP_STD = np.array([0.26862954, 0.26130258, 0.27577711], dtype=np.float32)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run E10 inference benchmark.")
    parser.add_argument("--output-root", default="ravel_revision_results")
    parser.add_argument("--datasets", nargs="+", default=DATASETS, choices=DATASETS)
    parser.add_argument("--models", nargs="+", default=MODELS, choices=MODELS)
    parser.add_argument("--seed", type=int, default=7)
    parser.add_argument("--device", default="cuda")
    parser.add_argument("--precision", choices=["fp32", "fp16", "bf16"], default="fp16")
    parser.add_argument("--batch-size", type=int, default=1)
    parser.add_argument("--warmup-iterations", type=int, default=50)
    parser.add_argument("--measurement-iterations", type=int, default=500)
    parser.add_argument("--max-length", type=int, default=None)
    parser.add_argument("--hfm-deleak-manifest", default="ravel_revision_results/data_audit/hfm_split_manifest_deleaked.csv")
    parser.add_argument("--overwrite", action="store_true")
    return parser.parse_args()


def image_to_tensor(image_path: str) -> torch.Tensor:
    try:
        image = Image.open(image_path).convert("RGB").resize((224, 224))
    except Exception:
        image = Image.new("RGB", (224, 224), (0, 0, 0))
    arr = np.asarray(image, dtype=np.float32) / 255.0
    arr = (arr - CLIP_MEAN) / CLIP_STD
    return torch.from_numpy(np.transpose(arr, (2, 0, 1))).float()


def sample_label(dataset: str, sample: Any) -> int:
    if dataset.startswith("mvsa"):
        return {"positive": 0, "neutral": 1, "negative": 2}[str(sample.combined_majority).lower()]
    return int(sample.label)


def sample_text(sample: Any) -> str:
    return str(getattr(sample, "text", ""))


def make_batch(sample: Any, dataset: str, tokenizer: DebertaV2Tokenizer, max_length: int, device: torch.device) -> Dict[str, torch.Tensor]:
    text_inputs = tokenizer(
        [sample_text(sample)],
        max_length=max_length,
        padding=True,
        truncation=True,
        return_tensors="pt",
    )
    batch = {
        "pixel_values": image_to_tensor(sample.image_path).unsqueeze(0),
        "input_ids": text_inputs["input_ids"],
        "attention_mask": text_inputs["attention_mask"],
        "labels": torch.tensor([sample_label(dataset, sample)], dtype=torch.long),
    }
    return {key: value.to(device, non_blocking=True) for key, value in batch.items()}


def cuda_sync(device: torch.device) -> None:
    if device.type == "cuda":
        torch.cuda.synchronize(device)


def autocast_context(device: torch.device, precision: str):
    if device.type != "cuda" or precision == "fp32":
        return torch.autocast(device_type="cpu", enabled=False)
    dtype = torch.float16 if precision == "fp16" else torch.bfloat16
    return torch.autocast(device_type="cuda", dtype=dtype)


def run_forward(model: nn.Module, method_key: str, batch: Dict[str, torch.Tensor], precision: str, device: torch.device) -> None:
    criterion = nn.CrossEntropyLoss()
    labels = batch["labels"].long()
    with torch.no_grad(), autocast_context(device, precision):
        logits_for_target(model, batch, METHODS[method_key].train_target, criterion, labels)


def load_model_and_sample(args: argparse.Namespace, dataset: str, method_key: str) -> Tuple[nn.Module, Any, int, torch.device, Dict[str, Any]]:
    root = Path(args.output_root)
    run_dir = root / "runs" / dataset / method_key / f"seed_{args.seed}"
    ckpt_path = run_dir / "checkpoint.pt"
    if not ckpt_path.exists():
        raise FileNotFoundError(ckpt_path)
    start = time.perf_counter()
    ckpt = torch.load(ckpt_path, map_location="cpu")
    ckpt_cfg = ckpt.get("cfg", {}) if isinstance(ckpt, dict) else {}
    max_length = int(args.max_length or ckpt_cfg.get("max_length") or 96)
    method = METHODS[method_key]
    set_seed(args.seed)
    device = torch.device(args.device if torch.cuda.is_available() or args.device == "cpu" else "cpu")
    (
        model_cls,
        cfg,
        _train_loader,
        _val_loader,
        _test_loader,
        _train_samples,
        _val_samples,
        test_samples,
        _num_classes,
        _label_names,
    ) = load_dataset(
        dataset_key=dataset,
        seed=args.seed,
        batch_size=1,
        max_length=max_length,
        num_workers=0,
        method=method,
        hfm_deleak_manifest=args.hfm_deleak_manifest,
        limits=(None, None, None),
    )
    cfg.update(ckpt_cfg)
    cfg.update(
        {
            "architecture": method.architecture,
            "enable_clip_lora": method.enable_lora,
            "enable_text_lora": method.enable_lora,
            "seed": args.seed,
            "batch_size": 1,
            "max_length": max_length,
        }
    )
    model = model_cls(cfg).to(device)
    if hasattr(model, "vision_lora"):
        freeze_non_lora(model.vision_lora)
    if hasattr(model, "text"):
        freeze_non_lora(model.text)
    apply_method_trainability(model, method)
    model.load_state_dict(ckpt.get("model_state", {}), strict=False)
    model.eval()
    load_time = time.perf_counter() - start
    stats = model.parameter_stats() if hasattr(model, "parameter_stats") else {
        "total": sum(p.numel() for p in model.parameters()),
        "trainable": sum(p.numel() for p in model.parameters() if p.requires_grad),
    }
    meta = {
        "checkpoint_size_mb": ckpt_path.stat().st_size / (1024 * 1024),
        "model_load_time_seconds": load_time,
        "total_parameters": int(stats.get("total", 0)),
        "trainable_parameters": int(stats.get("trainable", 0)),
        "max_length": max_length,
    }
    return model, test_samples[0], max_length, device, meta


def latency_stats(values_ms: Sequence[float]) -> Dict[str, float]:
    arr = np.array(values_ms, dtype=float)
    return {
        "mean": float(arr.mean()),
        "std": float(arr.std(ddof=1)) if arr.size > 1 else 0.0,
        "p50": float(np.percentile(arr, 50)),
        "p95": float(np.percentile(arr, 95)),
    }


def benchmark_one(args: argparse.Namespace, dataset: str, method_key: str) -> Dict[str, Any]:
    model, sample, max_length, device, meta = load_model_and_sample(args, dataset, method_key)
    tokenizer = DebertaV2Tokenizer.from_pretrained("microsoft/deberta-v3-base")
    batch = make_batch(sample, dataset, tokenizer, max_length, device)
    if device.type == "cuda":
        torch.cuda.reset_peak_memory_stats(device)
    for _ in range(args.warmup_iterations):
        run_forward(model, method_key, batch, args.precision, device)
    cuda_sync(device)

    model_only: List[float] = []
    for _ in range(args.measurement_iterations):
        cuda_sync(device)
        start = time.perf_counter()
        run_forward(model, method_key, batch, args.precision, device)
        cuda_sync(device)
        model_only.append((time.perf_counter() - start) * 1000.0)

    end_to_end: List[float] = []
    for _ in range(args.measurement_iterations):
        cuda_sync(device)
        start = time.perf_counter()
        fresh_batch = make_batch(sample, dataset, tokenizer, max_length, device)
        run_forward(model, method_key, fresh_batch, args.precision, device)
        cuda_sync(device)
        end_to_end.append((time.perf_counter() - start) * 1000.0)

    model_stats = latency_stats(model_only)
    e2e_stats = latency_stats(end_to_end)
    peak_vram = torch.cuda.max_memory_allocated(device) / (1024 * 1024) if device.type == "cuda" else float("nan")
    del model
    if device.type == "cuda":
        torch.cuda.empty_cache()
    return {
        "model": method_key,
        "dataset": dataset,
        "device": torch.cuda.get_device_name(device) if device.type == "cuda" else platform.processor(),
        "precision": args.precision.upper(),
        "batch_size": args.batch_size,
        **meta,
        "model_only_latency_mean_ms": model_stats["mean"],
        "model_only_latency_std_ms": model_stats["std"],
        "end_to_end_latency_mean_ms": e2e_stats["mean"],
        "end_to_end_latency_std_ms": e2e_stats["std"],
        "latency_mean_ms": e2e_stats["mean"],
        "latency_std_ms": e2e_stats["std"],
        "latency_p50_ms": e2e_stats["p50"],
        "latency_p95_ms": e2e_stats["p95"],
        "throughput_samples_per_second": 1000.0 / max(e2e_stats["mean"], 1e-9),
        "peak_inference_vram_mb": peak_vram,
        "warmup_iterations": args.warmup_iterations,
        "measurement_iterations": args.measurement_iterations,
        "preprocessing_included": "end_to_end includes PIL image decode, CLIP normalization, and tokenization for one sample; model_only excludes preprocessing",
    }


def main() -> None:
    args = parse_args()
    out_path = Path(args.output_root) / "aggregate_results" / "efficiency_results.csv"
    out_path.parent.mkdir(parents=True, exist_ok=True)
    existing = pd.read_csv(out_path) if out_path.exists() and out_path.stat().st_size else pd.DataFrame()
    rows: List[Dict[str, Any]] = [] if args.overwrite or existing.empty else existing.to_dict("records")
    done = {(str(row.get("model")), str(row.get("dataset")), str(row.get("precision"))) for row in rows if row.get("model_only_latency_mean_ms") == row.get("model_only_latency_mean_ms")}
    for dataset in args.datasets:
        for model in args.models:
            key = (model, dataset, args.precision.upper())
            if key in done and not args.overwrite:
                print(f"SKIP E10 {dataset} {model} {args.precision}", flush=True)
                continue
            print(f"RUN E10 dataset={dataset} model={model} precision={args.precision}", flush=True)
            row = benchmark_one(args, dataset, model)
            rows = [r for r in rows if not (str(r.get("model")) == model and str(r.get("dataset")) == dataset and str(r.get("precision")) == args.precision.upper())]
            rows.append(row)
            pd.DataFrame(rows).to_csv(out_path, index=False)
            print(
                f"DONE E10 {dataset} {model} p50={row['latency_p50_ms']:.2f}ms "
                f"p95={row['latency_p95_ms']:.2f}ms vram={row['peak_inference_vram_mb']:.1f}MB",
                flush=True,
            )
    pd.DataFrame(rows).to_csv(out_path, index=False)


if __name__ == "__main__":
    main()