training/VLA/build_dota_cot.py

#!/usr/bin/env python3
"""Build per-frame CoT+Belief JSONL for DoTA clips.

DoTA structure:
  DoTA/frames/{clip}/images/000000.jpg, 000001.jpg, ...
  DoTA/annotations/{clip}.json         — has anomaly_start/end (frame idx), ego_involve,
                                         accident_name, night, per-frame object list
  DoTA/train_split.txt, val_split.txt

We emit one JSONL record per clip with:
  {
    "id":         str (clip name),
    "label":      int (1 = ego-involved anomaly, 0 = normal segment BEFORE anomaly),
    "video_path": str (absolute path to frames dir),
    "cot":        {"scene": ..., "critical_objects": [...], "threat_analysis": ...},
    "belief":     {"action": str,
                   "actions_per_frame": [T str],
                   "frame_times_sec":   [T float],
                   "time_of_event_sec": float,
                   "alert_s":           float,
                   "observe_s":         float}
  }

CoT content is a rule-based placeholder (no teacher model). Replace in a later
pass if the budget allows running GPT-4o over DoTA val.

Usage:
  python -m training.VLA.build_dota_cot \
      --dota_root DoTA \
      --split val \
      --n_frames 8 \
      --fps 10 \
      --alert_s 0.5 --observe_s 2.5 \
      --output data/vla_cot/dota_val_perframe.jsonl
"""
from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any, Dict, List, Optional

import numpy as np

from training.VLA.augment_cot_with_belief import (
    derive_clip_action, per_frame_actions,
)


# ── tiny helpers ──────────────────────────────────────────────────────────
NIGHT_STR = {True: "night", False: "daytime"}


def _threat_from_accident(accident_name: str, ego_involve: bool) -> str:
    nm = (accident_name or "").replace("_", " ").strip()
    if not nm:
        return "Possible traffic anomaly ahead; trajectory appears unstable."
    if ego_involve:
        return (f"Ego vehicle is at elevated collision risk: the scene shows "
                f"a '{nm}' pattern developing in the ego lane.")
    return (f"Nearby agents exhibit a '{nm}' pattern; ego not directly involved "
            "but secondary risk cannot be ruled out.")


def _scene_str(accident_name: str, night: bool) -> str:
    tag = "night" if night else "daytime"
    nm = (accident_name or "unknown").replace("_", " ")
    return f"Dashcam view, {tag}; anomaly pattern: {nm}."


def _critical_from_labels(labels: List[Dict[str, Any]], anomaly_start: int,
                          n_keep: int = 4) -> List[str]:
    """Pick up to n_keep distinct object classes that appear near anomaly_start."""
    names: Dict[str, int] = {}
    if not labels:
        return []
    lo = max(0, anomaly_start - 5)
    hi = min(len(labels), anomaly_start + 3)
    for lab in labels[lo:hi]:
        for obj in lab.get("objects", []) or []:
            cat = obj.get("category") or obj.get("label") or obj.get("accident_name")
            if cat and cat != "normal":
                names[cat] = names.get(cat, 0) + 1
    if not names:
        return []
    ranked = sorted(names.items(), key=lambda kv: -kv[1])
    return [k for k, _ in ranked[:n_keep]]


def _uniform_indices(total: int, n: int) -> List[int]:
    return np.linspace(0, total - 1, n).round().astype(int).tolist()


def build_record(
    clip_name: str,
    ann: Dict[str, Any],
    frames_root: Path,
    n_frames: int,
    fps: float,
    alert_s: float,
    observe_s: float,
    event_anchored: bool = True,
    lookback_s: float = 3.0,
    post_margin_s: float = 0.0,
    teacher_cot: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]:
    total = int(ann.get("num_frames", 0))
    if total <= 0:
        return None
    ego = bool(ann.get("ego_involve", False))
    night = bool(ann.get("night", False))
    acc_name = str(ann.get("accident_name") or ann.get("anomaly_class") or "")
    anomaly_start = int(ann.get("anomaly_start", -1))

    # label = 1 iff ego-involved anomaly clip (a positive sample for POMDP SFT)
    # non-ego anomaly clips are excluded (they add noise; ego_involve handles them).
    if not ego:
        return None
    label = 1

    # time of event = anomaly_start / fps (seconds from clip start)
    toe = float(anomaly_start) / float(fps) if anomaly_start >= 0 else None

    if event_anchored and toe is not None:
        end_s   = min(float(total - 1) / fps, toe + post_margin_s)
        start_s = max(0.0, end_s - (lookback_s + post_margin_s))
        times = np.linspace(start_s, end_s, n_frames)
        idx = [int(round(t * fps)) for t in times]
        sampling = "event_anchored"
    else:
        idx = _uniform_indices(total, n_frames)
        sampling = "uniform"
    idx = [max(0, min(total - 1, int(i))) for i in idx]
    frame_times = [float(i) / float(fps) for i in idx]
    actions_pf = per_frame_actions(label=label,
                                   time_of_event=toe,
                                   frame_times=frame_times,
                                   alert_s=alert_s,
                                   observe_s=observe_s)

    # clip-level action: if any ALERT → ALERT; elif any OBSERVE → OBSERVE; else SILENT
    if "ALERT" in actions_pf:
        clip_action = "ALERT"
    elif "OBSERVE" in actions_pf:
        clip_action = "OBSERVE"
    else:
        clip_action = "SILENT"

    if teacher_cot is not None and isinstance(teacher_cot, dict) \
            and teacher_cot.get("scene") and teacher_cot.get("threat_analysis"):
        cot = {
            "scene":            str(teacher_cot.get("scene", "")).strip(),
            "critical_objects": list(teacher_cot.get("critical_objects", []) or []),
            "threat_analysis":  str(teacher_cot.get("threat_analysis", "")).strip(),
            "source":           "gpt_teacher",
        }
    else:
        cot = {
            "scene":            _scene_str(acc_name, night),
            "critical_objects": _critical_from_labels(ann.get("labels", []) or [],
                                                     anomaly_start),
            "threat_analysis":  _threat_from_accident(acc_name, ego),
            "source":           "rule_template",
        }
    belief = {
        "action":            clip_action,
        "tta_sec":           round(toe, 3) if toe is not None else -1.0,
        "actions_per_frame": actions_pf,
        "frame_indices":     idx,
        "frame_times_sec":   [round(t, 3) for t in frame_times],
        "time_of_event_sec": round(toe, 3) if toe is not None else -1.0,
        "alert_s":           alert_s,
        "observe_s":         observe_s,
        "sampling":          sampling,
        "total_frames":      total,
        "fps":               round(float(fps), 3),
        "lookback_s":        lookback_s,
        "post_margin_s":     post_margin_s,
    }

    return {
        "id":         f"dota_{clip_name}",
        "label":      label,
        "video_path": str(frames_root / clip_name / "images"),
        "cot":        cot,
        "belief":     belief,
        "source":     "dota",
        "accident_name": acc_name,
    }


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--dota_root", default="DoTA")
    ap.add_argument("--split", choices=["train", "val"], default="val")
    ap.add_argument("--n_frames",  type=int,   default=8)
    ap.add_argument("--fps",       type=float, default=10.0)
    ap.add_argument("--alert_s",   type=float, default=0.5)
    ap.add_argument("--observe_s", type=float, default=2.5)
    ap.add_argument("--output",    required=True)
    ap.add_argument("--event_anchored", action="store_true", default=True,
                    help="Sample T frames in [event - lookback, event + post_margin]")
    ap.add_argument("--no_event_anchored", dest="event_anchored",
                    action="store_false")
    ap.add_argument("--lookback_s",    type=float, default=3.0)
    ap.add_argument("--post_margin_s", type=float, default=0.0)
    ap.add_argument("--teacher_json",  type=str,   default="",
                    help="Optional JSON mapping clip_name -> {cot, usage, ...} "
                         "from GPT distillation; overrides rule template when present")
    ap.add_argument("--limit",     type=int,   default=0,
                    help="If >0, only emit first N records (smoke test)")
    args = ap.parse_args()

    root = Path(args.dota_root).resolve()
    split_file = root / f"{args.split}_split.txt"
    ann_dir    = root / "annotations"
    frames_root = root / "frames"

    with open(split_file) as f:
        clips = [ln.strip() for ln in f if ln.strip()]
    print(f"[dota] {args.split}: {len(clips)} clip ids")

    teacher_map: Dict[str, Dict[str, Any]] = {}
    if args.teacher_json:
        tp = Path(args.teacher_json)
        if tp.exists():
            raw = json.loads(tp.read_text())
            for k, v in raw.items():
                if isinstance(v, dict) and v.get("cot"):
                    teacher_map[k] = v["cot"]
            print(f"[dota] teacher CoT loaded: {len(teacher_map)} clips from {tp}")
        else:
            print(f"[dota] WARN: teacher_json not found: {tp}")

    out_path = Path(args.output)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    n_emitted = 0
    n_missing = 0
    n_no_ego = 0
    with open(out_path, "w") as f_out:
        for clip_name in clips:
            ann_path = ann_dir / f"{clip_name}.json"
            if not ann_path.exists():
                n_missing += 1
                continue
            with open(ann_path) as f:
                ann = json.load(f)
            frames_dir = frames_root / clip_name / "images"
            if not frames_dir.exists():
                n_missing += 1
                continue
            rec = build_record(
                clip_name=clip_name,
                ann=ann,
                frames_root=frames_root,
                n_frames=args.n_frames,
                fps=args.fps,
                alert_s=args.alert_s,
                observe_s=args.observe_s,
                event_anchored=args.event_anchored,
                lookback_s=args.lookback_s,
                post_margin_s=args.post_margin_s,
                teacher_cot=teacher_map.get(clip_name),
            )
            if rec is None:
                n_no_ego += 1
                continue
            f_out.write(json.dumps(rec) + "\n")
            n_emitted += 1
            if args.limit and n_emitted >= args.limit:
                break

    print(f"[dota] emitted {n_emitted} ego-involved clips; "
          f"skipped non-ego={n_no_ego}  missing={n_missing}")
    print(f"[dota] saved -> {out_path}")


if __name__ == "__main__":
    main()