training/SFT/sanity_check.py

#!/usr/bin/env python3
"""
SFT sanity checks — run BEFORE any long training run.

Checks
------
1. Manifest counts and category distribution
2. Dataset: sample counts, label ranges, no train/val video overlap
3. Model: HazardHead init (output ≈ 0.27), TTAHead init (output ≈ 5.0)
4. Single-step forward pass: shapes, loss components
5. Single-step backward pass: gradients flow through all heads

Usage
-----
python -m training.SFT.sanity_check \
    --manifest_dir PROJECT_ROOT/data/sft_manifests \
    --model_name   PROJECT_ROOT/models/Qwen2.5-VL-3B-Instruct \
    --pretrained_lora PROJECT_ROOT/checkpoints/pretrain_v2/stage_b/best_model
"""

from __future__ import annotations

import argparse
import json
import logging
import sys
from pathlib import Path
from typing import List

import torch
import torch.nn.functional as F
from torch.utils.data import DataLoader

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("SFT.sanity")

PASS = "✅"
FAIL = "❌"
WARN = "⚠️"


def check(cond: bool, msg: str, fatal: bool = True):
    if cond:
        logger.info(f"{PASS} {msg}")
    else:
        if fatal:
            logger.error(f"{FAIL} {msg}")
            sys.exit(1)
        else:
            logger.warning(f"{WARN} {msg}")


# ── 1. Manifest checks ────────────────────────────────────────────────────────

def check_manifests(manifest_dir: Path):
    logger.info("\n=== 1. Manifest Checks ===")
    manifest_dir = Path(manifest_dir)

    expected_train = ["nexar_train", "dada_pos_train", "dada_noneego_train", "dada_neg_train"]
    expected_val   = ["nexar_val",   "dada_pos_val",   "dada_noneego_val"]

    for name in expected_train + expected_val:
        p = manifest_dir / f"{name}.json"
        check(p.exists(), f"Manifest exists: {p.name}")
        if not p.exists():
            continue
        with open(p) as f:
            m = json.load(f)
        nv = m.get("num_videos", 0)
        cc = m.get("category_counts", {})
        logger.info(f"    {name}: {nv} videos  cats={cc}")
        check(nv > 0, f"{name} has > 0 videos")

    # Check no video_id overlap between train and val
    train_ids: set = set()
    val_ids:   set = set()
    for name in expected_train:
        p = manifest_dir / f"{name}.json"
        if not p.exists():
            continue
        with open(p) as f:
            m = json.load(f)
        for v in m.get("videos", []):
            train_ids.add(v["video_id"])
    for name in expected_val:
        p = manifest_dir / f"{name}.json"
        if not p.exists():
            continue
        with open(p) as f:
            m = json.load(f)
        for v in m.get("videos", []):
            val_ids.add(v["video_id"])

    overlap = train_ids & val_ids
    check(len(overlap) == 0,
          f"Zero train/val video_id overlap (found {len(overlap)})" if overlap else "Zero train/val video_id overlap",
          fatal=True)
    logger.info(f"    train videos: {len(train_ids)}  val videos: {len(val_ids)}")


# ── 2. Dataset checks ─────────────────────────────────────────────────────────

def check_datasets(manifest_dir: Path, n_samples: int = 200):
    logger.info("\n=== 2. Dataset Checks ===")
    from .dataset import SFTDataset, sft_collate_fn, MAX_TTA, MIN_TTA

    manifest_dir = Path(manifest_dir)
    train_manifests = [
        manifest_dir / "nexar_train.json",
        manifest_dir / "dada_pos_train.json",
        manifest_dir / "dada_noneego_train.json",
        manifest_dir / "dada_neg_train.json",
    ]
    val_manifests = [
        manifest_dir / "nexar_val.json",
        manifest_dir / "dada_pos_val.json",
        manifest_dir / "dada_noneego_val.json",
    ]
    train_manifests = [m for m in train_manifests if m.exists()]
    val_manifests   = [m for m in val_manifests   if m.exists()]

    logger.info("  Loading train dataset (debug mode)...")
    train_ds = SFTDataset(manifests=train_manifests, split="train", debug=True, debug_samples=n_samples)
    logger.info("  Loading val dataset (debug mode)...")
    val_ds   = SFTDataset(manifests=val_manifests,   split="val",   debug=True, debug_samples=n_samples // 2)

    check(len(train_ds) > 0, f"Train dataset non-empty ({len(train_ds)} samples)")
    check(len(val_ds)   > 0, f"Val dataset non-empty ({len(val_ds)} samples)")

    # Inspect a few samples
    loader = DataLoader(train_ds, batch_size=4, shuffle=False, collate_fn=sft_collate_fn, num_workers=0)
    batch = next(iter(loader))

    tta_labels  = batch["tta_labels"]
    haz_labels  = batch["hazard_labels"]
    haz_weights = batch["hazard_weights"]
    is_ego      = batch["is_ego_positive"]
    is_cen      = batch["is_censored"]

    check(tta_labels.shape[0] == 4,        f"Batch size correct: {tta_labels.shape[0]}")
    check((haz_labels >= 0).all() and (haz_labels <= 1).all(),
          f"hazard_labels in [0,1]: min={haz_labels.min():.2f} max={haz_labels.max():.2f}")
    check((haz_weights > 0).all() and (haz_weights <= 1.01).all(),
          f"hazard_weights in (0,1]: min={haz_weights.min():.2f} max={haz_weights.max():.2f}")
    # TTA labels for censored can be > 10 (will be clamped in loss)
    check((tta_labels >= 0).all(),
          f"tta_labels >= 0: min={tta_labels.min():.2f}")

    # ego_positive must have hazard_label=1
    if is_ego.any():
        ego_haz = haz_labels[is_ego]
        check((ego_haz == 1).all(),
              f"ego_positive samples have hazard_label=1 (all {int(is_ego.sum())} checked)")

    # non-ego and safe-neg must always be censored (no TTA supervision)
    is_ne    = batch.get("is_non_ego", torch.zeros(tta_labels.shape[0], dtype=torch.bool))
    non_pos  = ~is_ego
    if non_pos.any():
        check(is_cen[non_pos].all(),
              f"All non-ego/safe-neg samples are censored ({int(non_pos.sum())} non-pos, {int(is_cen[non_pos].sum())} censored)")

    logger.info(
        f"    Sample breakdown — ego_pos={int(is_ego.sum())}  "
        f"censored={int(is_cen.sum())}  "
        f"is_non_ego={int(batch.get('is_non_ego', torch.zeros(4)).sum())}"
    )

    # Check images shape
    imgs = batch["images"]
    check(len(imgs) == 4, f"images list length == batch_size ({len(imgs)})")
    check(len(imgs[0]) > 0, f"Each sample has ≥1 frame ({len(imgs[0])} frames)")
    logger.info(f"    Image size: {imgs[0][0].size}")

    return train_ds, val_ds


# ── 3. Model checks ───────────────────────────────────────────────────────────

def check_model(model_name: str, pretrained_lora: str | None):
    logger.info("\n=== 3. Model Checks ===")
    from .trainer import SFTModel

    logger.info("  Loading model (this may take a while)...")
    model = SFTModel(
        model_name=model_name,
        pretrained_lora_path=pretrained_lora,
        belief_strategy="mean_pool",
        use_lora=True,
        use_bf16=True,
        device="auto",
    )

    # HazardHead init: sigmoid(-1) ≈ 0.269
    dummy = torch.zeros(1, model.hidden_dim, device=model.device, dtype=model.dtype)
    with torch.no_grad():
        h_logit = model.hazard_head(dummy).float().item()
        h_prob  = torch.sigmoid(torch.tensor(h_logit)).item()
        t_mean, _ = model.tta_head(dummy)
        t_mean = t_mean.float().item()

    check(0.20 <= h_prob <= 0.35,
          f"HazardHead init: hazard_logit={h_logit:.3f}  hazard_prob={h_prob:.3f} (expected ≈0.27, range [0.20,0.35])")
    check(3.0 <= t_mean <= 8.0,
          f"TTAHead init: tta_mean={t_mean:.3f} (expected ≈5.0, range [3,8])")

    logger.info(f"    Hidden dim: {model.hidden_dim}")
    logger.info(f"    Device: {model.device}")
    logger.info(f"    Dtype: {model.dtype}")

    # Trainable params
    lora_params = [(n, p) for n, p in model.vlm.named_parameters() if p.requires_grad and "lora_" in n.lower()]
    head_params = list(model.hazard_head.parameters()) + list(model.tta_head.parameters()) + list(model.belief_aggregator.parameters())
    check(len(lora_params) > 0, f"LoRA has trainable params ({len(lora_params)} tensors)")
    check(len(head_params) > 0, f"Head params exist ({len(head_params)} tensors)")

    return model


# ── 4 & 5. Forward/backward checks ───────────────────────────────────────────

def check_forward_backward(model, train_ds, val_ds):
    logger.info("\n=== 4. Forward Pass Check ===")
    from .dataset import sft_collate_fn
    from .trainer import compute_sft_loss, SFTTrainer

    loader = DataLoader(train_ds, batch_size=2, shuffle=False, collate_fn=sft_collate_fn, num_workers=0)
    batch  = next(iter(loader))

    proc = model.processor
    apply_chat = (
        proc.apply_chat_template
        if hasattr(proc, "apply_chat_template")
        else proc.tokenizer.apply_chat_template
    )
    SYSTEM = "You are a driving safety AI analyzing dashcam footage for collision risk."
    images = batch["images"]
    texts = []
    for i in range(len(batch["video_ids"])):
        frames  = images[i]
        content = [{"type": "image"} for _ in range(len(frames))]
        content.append({"type": "text", "text": "Estimate time to collision. Output a single number."})
        msgs = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": content}]
        texts.append(apply_chat(msgs, tokenize=False, add_generation_prompt=False))

    inputs = proc(text=texts, images=images, return_tensors="pt", padding=True, truncation=True)

    dev = model.device
    t = {k: batch[k].to(dev) for k in ["tta_labels", "hazard_labels", "hazard_weights", "is_ego_positive", "is_censored"]}

    model.train()
    from torch.amp import autocast
    with autocast(device_type="cuda", dtype=torch.bfloat16, enabled=True):
        out = model(inputs)

    check("hazard_logit" in out, "Output has hazard_logit")
    check("hazard_prob"  in out, "Output has hazard_prob")
    check("tta_mean"     in out, "Output has tta_mean")
    check("tta_logvar"   in out, "Output has tta_logvar")
    check(out["hazard_logit"].shape == (2,), f"hazard_logit shape == (2,): {tuple(out['hazard_logit'].shape)}")
    check(out["tta_mean"].shape     == (2,), f"tta_mean shape == (2,): {tuple(out['tta_mean'].shape)}")

    hp = out["hazard_prob"].float()
    check((hp >= 0).all() and (hp <= 1).all(),
          f"hazard_prob in [0,1]: [{hp.min().item():.3f}, {hp.max().item():.3f}]")
    check((out["tta_mean"] > 0).all(),
          f"tta_mean > 0: [{out['tta_mean'].min().item():.3f}, {out['tta_mean'].max().item():.3f}]")

    with autocast(device_type="cuda", dtype=torch.bfloat16, enabled=True):
        loss, metrics = compute_sft_loss(
            hazard_logit=out["hazard_logit"],
            tta_mean=out["tta_mean"],
            tta_logvar=out["tta_logvar"],
            hazard_label=t["hazard_labels"],
            hazard_weight=t["hazard_weights"],
            is_ego_positive=t["is_ego_positive"],
            is_censored=t["is_censored"],
            tta_label=t["tta_labels"],
            nll_weight=0.5,
        )

    check(torch.isfinite(loss), f"Loss is finite: {loss.item():.4f}")
    check(loss.item() > 0,     f"Loss > 0: {loss.item():.4f}")
    check("loss_hazard" in metrics, "metrics has loss_hazard")
    logger.info(f"    loss={loss.item():.4f}  loss_hazard={metrics['loss_hazard']:.4f}  "
                f"loss_tta_obs={metrics['loss_tta_obs']:.4f}  hazard_acc={metrics['hazard_acc']:.3f}")

    logger.info("\n=== 5. Backward Pass Check ===")
    loss.backward()

    # Check hazard_head gets gradient
    hh_grad = model.hazard_head.fc.weight.grad
    check(hh_grad is not None,               "hazard_head.fc.weight has gradient")
    check(hh_grad is not None and hh_grad.abs().sum() > 0, "hazard_head gradient is non-zero")

    # Check tta_head gets gradient
    tta_last = model.tta_head.net[-1]
    th_grad  = tta_last.weight.grad
    if t["is_ego_positive"].any():
        check(th_grad is not None,               "tta_head last layer has gradient")
        check(th_grad is not None and th_grad.abs().sum() > 0, "tta_head gradient is non-zero")
    else:
        logger.info(f"    {WARN} No ego_positive in batch; tta_head gradient may be zero (OK)")

    # Check LoRA params have gradients (may be zero at init if hazard_head.weight=zeros)
    lora_total    = [(n, p) for n, p in model.vlm.named_parameters() if p.requires_grad and "lora_" in n.lower()]
    lora_with_grad = [(n, p) for n, p in lora_total if p.grad is not None and p.grad.abs().sum() > 0]
    if len(lora_with_grad) > 0:
        check(True, f"LoRA params have non-zero gradient ({len(lora_with_grad)} tensors)")
    else:
        # At init, HazardHead.fc.weight==0 → d(loss)/d(belief)=0 → LoRA grad=0.
        # This is expected; gradient will flow once weights are non-zero.
        logger.info(
            f"    {WARN} LoRA gradient is zero at init (HazardHead.fc.weight=0 → no VLM grad path from hazard loss alone). "
            f"Expected: LoRA grads appear after first optimizer step. "
            f"LoRA params with requires_grad: {len(lora_total)}"
        )
        check(len(lora_total) > 0, f"LoRA params exist ({len(lora_total)} tensors with requires_grad)")


# ── main ──────────────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser("SFT sanity check")
    parser.add_argument("--manifest_dir",   type=str, default="PROJECT_ROOT/data/sft_manifests")
    parser.add_argument("--model_name",     type=str, default="PROJECT_ROOT/models/Qwen2.5-VL-3B-Instruct")
    parser.add_argument("--pretrained_lora", type=str, default="PROJECT_ROOT/checkpoints/pretrain_v2/stage_b/best_model")
    parser.add_argument("--n_samples",      type=int, default=200, help="Debug samples for dataset check")
    parser.add_argument("--skip_model",     action="store_true", help="Skip model load (much faster)")
    args = parser.parse_args()

    logger.info("=" * 60)
    logger.info("LKAlert SFT Sanity Check")
    logger.info("=" * 60)

    check_manifests(Path(args.manifest_dir))
    train_ds, val_ds = check_datasets(Path(args.manifest_dir), n_samples=args.n_samples)

    if args.skip_model:
        logger.info("\n⚠️  Skipping model checks (--skip_model)")
    else:
        model = check_model(args.model_name, args.pretrained_lora)
        check_forward_backward(model, train_ds, val_ds)

    logger.info("\n" + "=" * 60)
    logger.info("✅ All sanity checks passed!")
    logger.info("=" * 60)


if __name__ == "__main__":
    main()