code/bench/diag_perception_presence.py

"""Verification step 1b (2026-10-13): presence scores of the base Qwen3.5-9B (no LoRA) from token logprobs, plus the
box temporal check of PLAN_DECOUPLED_20261013.md 2.4 point 3. Thinking off, temperature 0, top 20 logprobs.

1b, on step 1's 500 items (runs/diag_20261013/perception_direction/items.jsonl: 300 geometric positives, 100 behind,
100 wrong-name), two scores:
- s_yes: B's question, max_tokens 1; P(yes) / (P(yes) + P(no)), case and leading-space variants summed;
- s_box: step 1's grounding question with max_tokens 48 (the controller's call). Answers either open a code fence
  ("```", "json", "\\n", "[", "\\n", "\\t", '{"', ...) or refuse ("There are no objects belonging to ..."); after the fence
  the list token is "[" with P("[]") ~ 0 (stored per item as p_list_empty), so empty vs non-empty is decided at the
  first token: s_box = P(first token "```").
AUC: 300 positives vs 200 negatives, and step 1's hand-labelled contact-sheet positives (37 visible vs 11 not).
Temporal check on GRPO v2 rounds 1-5: disjoint windows of 5 consecutive frames turning >= 5 deg at each step.
real = a target in the geometric field of view (|h| <= 51.2, |v| <= 35) 3-16 blocks away in all 5 frames;
wrong = the same windows asked about a wrong name (step 1's rule); behind = a target 3-16 blocks away with |h| >= 120
in all 5 frames (spread as step 1, up to 100). Per step, the first box centre's angle change is compared with
-delta yaw, the change a fixed world point shows.
Writes items.jsonl, windows.jsonl, summary.md.
Usage: diag_perception_presence.py --url http://localhost:9041/v1 --model NAME [--out DIR]
"""
import argparse
import base64
from concurrent.futures import ThreadPoolExecutor
import json
import math
from pathlib import Path
import random
import statistics

from openai import OpenAI

from diag_perception import BENCH, HALF_VFOV, ROOT, episodes, spread, view
from diag_perception_direction import BOX, GROUND, SIDES, TAN_HALF_HFOV, side

STEP1 = ROOT / "runs/diag_20261013/perception_direction/items.jsonl"
HIDDEN = {11, 17, 19, 20, 27, 29, 33, 36, 43, 45, 47}  # contact-sheet positives judged not visible (step 1 summary.md)
HALF_HFOV = math.degrees(math.atan(TAN_HALF_HFOV))
YESNO = "Is there a {0} in this image? Answer yes or no."


def windows(rng):
    """(real, wrong, behind) windows: start frame, the 4 yaw changes, geometric (dist, h, v) per frame."""
    real, behind, names, placed = [], [], {}, {}
    for ep, scene, _, poses, targets in episodes():
        placed[scene] = " ".join(json.loads((BENCH / scene / "multi-agent/metadata.json").read_text())["commands"])
        turn = [(b["yaw"] - a["yaw"] + 180) % 360 - 180 for a, b in zip(poses, poses[1:])]
        for mid, task, target in targets:
            if not task.startswith("find "):
                continue
            name = task[5:].removeprefix("a ").replace("_", " ")
            names.setdefault(scene, set()).add(name)
            free = {"real": 0, "behind": 0}
            for s in range(len(poses) - 4):
                if not all(abs(d) >= 5 for d in turn[s:s + 4]):
                    continue
                geo = [[round(a, 1) for a in view(poses[s + j], target)] for j in range(5)]
                kind = ("real" if all(3 <= d <= 16 and abs(h) <= HALF_HFOV and abs(v) <= HALF_VFOV for d, h, v in geo)
                        else "behind" if all(3 <= d <= 16 and abs(h) >= 120 for d, h, v in geo) else None)
                if kind and s >= free[kind]:
                    free[kind] = s + 5
                    (real if kind == "real" else behind).append(
                        {"kind": kind, "ep": str(ep.relative_to(ROOT)), "scene": scene, "mid": mid, "frame": s,
                         "name": name, "asked": name, "turn": [round(d, 2) for d in turn[s:s + 4]], "geo": geo})
    every = sorted(set().union(*names.values()))
    wrong = [{**w, "kind": "wrong", "asked": rng.choice([x for x in every if x not in names[w["scene"]]
                                                         and x.replace(" ", "_") not in placed[w["scene"]]])}
             for w in real]
    return real, wrong, spread(behind, 100, rng)


def auc(pos, neg):
    return sum((a > b) + 0.5 * (a == b) for a in pos for b in neg) / (len(pos) * len(neg))


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--url", required=True)
    ap.add_argument("--model", required=True)
    ap.add_argument("--out", type=Path, default=ROOT / "runs/diag_20261013/perception_presence")
    args = ap.parse_args()
    args.out.mkdir(parents=True, exist_ok=True)
    step1 = [json.loads(line) for line in STEP1.read_text().splitlines()]
    real, wrong, behind = windows(random.Random(0))
    print(f"windows: real {len(real)} ({len({w['scene'] for w in real})} scenes, {len({w['ep'] for w in real})} "
          f"episodes), wrong {len(wrong)}, behind {len(behind)} ({len({w['scene'] for w in behind})} scenes)", flush=True)

    client = OpenAI(api_key="EMPTY", base_url=args.url)

    def chat(key, prompt, max_tokens):
        ep, frame, asked = key
        png = (ROOT / ep / "frames" / f"{frame:04d}.png").read_bytes()
        r = client.chat.completions.create(
            model=args.model, temperature=0, max_tokens=max_tokens, logprobs=True, top_logprobs=20,
            messages=[{"role": "user", "content": [
                {"type": "image_url", "image_url": {"url": "data:image/png;base64," + base64.b64encode(png).decode()}},
                {"type": "text", "text": prompt.format(asked)}]}],
            extra_body={"chat_template_kwargs": {"enable_thinking": False}})
        return r.choices[0].message.content, r.choices[0].logprobs.content

    def ground(key):
        text, lp = chat(key, GROUND, 48)
        boxes = [[float(c) for c in b] for b in BOX.findall(text)]
        out = {"ground": text, "boxes": boxes,
               "s_box": sum(math.exp(t.logprob) for t in lp[0].top_logprobs if t.token.strip() == "```"),
               "p_list_empty": sum(math.exp(t.logprob) for t in lp[3].top_logprobs if t.token == "[]")
               if text.startswith("```json\n[") else None}
        if boxes:
            x1, _, x2, _ = boxes[0]
            out["h_box"] = round(math.degrees(math.atan(((x1 + x2) / 1000 - 1) * TAN_HALF_HFOV)), 1)
        return out

    def yes(key):
        text, lp = chat(key, YESNO, 1)
        p = {w: sum(math.exp(t.logprob) for t in lp[0].top_logprobs if t.token.strip().lower() == w)
             for w in ("yes", "no")}
        return {"answer": text, "s_yes": p["yes"] / (p["yes"] + p["no"])}

    keys1 = [(x["ep"], x["frame"], x["asked"]) for x in step1]
    keys = list(dict.fromkeys(keys1 + [(w["ep"], w["frame"] + j, w["asked"]) for w in real + wrong + behind
                                       for j in range(5)]))
    with ThreadPoolExecutor(32) as pool:
        g = dict(zip(keys, pool.map(ground, keys)))
        y = dict(zip(keys1, pool.map(yes, keys1)))
    items = [{**{k: x[k] for k in ("kind", "ep", "scene", "mid", "frame", "name", "asked", "dist", "h", "v", "pitch")},
              **y[key], **g[key]} for x, key in zip(step1, keys1)]
    wins = [{**w, "ground": [g[(w["ep"], w["frame"] + j, w["asked"])] for j in range(5)]}
            for w in real + wrong + behind]
    real, wrong, behind = ([w for w in wins if w["kind"] == k] for k in ("real", "wrong", "behind"))
    with (args.out / "items.jsonl").open("w") as f:
        f.writelines(json.dumps(x, ensure_ascii=False) + "\n" for x in items)
    with (args.out / "windows.jsonl").open("w") as f:
        f.writelines(json.dumps(w, ensure_ascii=False) + "\n" for w in wins)

    by = {k: [x for x in items if x["kind"] == k] for k in ("pos", "behind", "wrong")}
    shown = [x for s in SIDES for x in [x for x in by["pos"] if side(x["h"]) == s][:16]]
    vis = [x for k, x in enumerate(shown) if k not in HIDDEN]
    hid = [x for k, x in enumerate(shown) if k in HIDDEN]
    neg = by["behind"] + by["wrong"]
    aucs = ["| 比较 | n | s_yes AUC | s_box AUC |", "|---|---|---:|---:|"]
    aucs += [f"| {label} | {len(a)} 对 {len(b)} | {auc([x['s_yes'] for x in a], [x['s_yes'] for x in b]):.3f} | "
             f"{auc([x['s_box'] for x in a], [x['s_box'] for x in b]):.3f} |"
             for label, a, b in (("正例对负例（背后 + 错名）", by["pos"], neg), ("正例对背后", by["pos"], by["behind"]),
                                 ("正例对错名", by["pos"], by["wrong"]), ("人工：可见对看不见", vis, hid),
                                 ("人工可见对负例", vis, neg))]
    med = lambda xs: f"{statistics.median(xs):.3f}"
    scores = ["| 组 | n | s_yes 中位数 | 答 yes | s_box 中位数 | 给框（48 token） |", "|---|---:|---:|---:|---:|---:|"]
    scores += [f"| {label} | {len(xs)} | {med([x['s_yes'] for x in xs])} | "
               f"{sum(x['answer'].strip().lower().startswith('yes') for x in xs)} | {med([x['s_box'] for x in xs])} | "
               f"{sum(bool(x['boxes']) for x in xs)} |"
               for label, xs in (("正例", by["pos"]), ("背后", by["behind"]), ("错名", by["wrong"]),
                                 ("人工可见", vis), ("人工看不见", hid))]
    fence = [x["p_list_empty"] for x in items if x["p_list_empty"] is not None]
    starts = sorted({x["ground"][:12] for x in items if x["p_list_empty"] is None})
    scores += ["", f"代码块开头的回答 {len(fence)} 条，列表 token 上 P(\"[]\") 最大 {max(fence):.1e}；"
                   f"其余 {len(items) - len(fence)} 条开头为 {starts}。"]

    def row(label, seqs):
        """seqs: (4 yaw changes, 5 box centre angles, None = no box)."""
        had = [(-d, a, b) for turns, hs in seqs for d, a, b in zip(turns, hs, hs[1:]) if a is not None]
        both = [(x, b - a) for x, a, b in had if b is not None]
        slope, icpt = statistics.linear_regression(*zip(*both))
        rms = math.sqrt(statistics.fmean((y - slope * x - icpt) ** 2 for x, y in both))
        frames = [h for _, hs in seqs for h in hs]
        big = [y for x, y in both if abs(x) > 10]
        pct = lambda k, n: f"{100 * k / n:.0f}%"
        return (f"| {label} | {len(seqs)} | {pct(sum(h is not None for h in frames), len(frames))} | "
                f"{pct(sum(b is None for _, _, b in had), len(had))} | {len(both)} | "
                f"{statistics.median(abs(x) for x, _ in both):.1f} | {slope:.2f} | {icpt:+.1f} | {rms:.1f} | "
                f"{pct(sum(abs(y - x) <= 5 for x, y in both), len(both))} | {pct(sum(abs(y) <= 5 for y in big), len(big))} | "
                f"{pct(sum(min(abs(y - x), abs(y)) > 10 for x, y in both), len(both))} | "
                f"{statistics.median(abs(y) for _, y in both):.1f} |")

    boxes = lambda ws: [(w["turn"], [f.get("h_box") for f in w["ground"]]) for w in ws]
    rng = random.Random(0)
    pool = [f["h_box"] for w in wrong + behind for f in w["ground"] if "h_box" in f]
    errs = [x["h_box"] - x["h"] for x in by["pos"] if "h_box" in x]

    def fake_sim(redraw, world):
        """Fake box angles over the fake windows' turns (20 repeats): first from the observed fake boxes; each step
        redrawn with probability redraw, else kept (world = 1: shifted by -delta yaw like a fixed world point)."""
        seqs = []
        for w in (wrong + behind) * 20:
            hs = [rng.choice(pool)]
            for d in w["turn"]:
                hs.append(rng.choice(pool) if rng.random() < redraw else hs[-1] - world * d)
            seqs.append((w["turn"], hs))
        return seqs

    def real_sim(redraw):
        """True target angle + an error from the 300 positives (plan 2.4 point 1), redrawn with probability redraw."""
        seqs = []
        for w in real * 20:
            e, hs = rng.choice(errs), []
            for _, h, _ in w["geo"]:
                e = rng.choice(errs) if hs and rng.random() < redraw else e
                hs.append(h + e)
            seqs.append((w["turn"], hs))
        return seqs

    temporal = ["| 窗口 | 窗口数 | 帧给框 | 消失（上一帧有框，这一帧没有） | 相邻两帧都有框的步 | \\|Δyaw\\| 中位数 | 斜率 | 截距 | "
                "残差 RMS | 随世界点（\\|Δh_box + Δyaw\\| ≤ 5°） | 画面不动（\\|Δh_box\\| ≤ 5°，\\|Δyaw\\| > 10° 的步） | "
                "跳（两者都偏 >10°） | \\|Δh_box\\| 中位数 |",
                "|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|",
                row("真目标：几何真值 h（参照）", [(w["turn"], [g[1] for g in w["geo"]]) for w in real]),
                row("真目标：框", boxes(real)), row("假框：错名（同一批窗口）", boxes(wrong)),
                row("假框：背后", boxes(behind)), row("假框合计", boxes(wrong + behind)),
                row("模拟假框 (1) 每步独立", fake_sim(1, 0)), row("模拟假框 (2) 固定世界点", fake_sim(0, 1)),
                row("模拟假框 (3) 画面里不动", fake_sim(0, 0)), row("模拟假框 (3)，每步以 0.5 概率重抽", fake_sim(0.5, 0)),
                row("模拟真框：每步独立误差（2.4 第 1 点原样）", real_sim(1)), row("模拟真框：误差保持，每步以 0.5 概率重抽", real_sim(0.5))]
    (args.out / "summary.md").write_text(
        "基座 Qwen3.5-9B（无 LoRA），思考关闭，温度 0，取前 20 个 token 的 logprob。\n\n"
        "## 1b 存在性分数\n\n"
        "同第 1 步的 500 条（runs/diag_20261013/perception_direction/items.jsonl：几何正例 300，背后负例 100，错名负例 100）。"
        f"s_yes：B 的问题 \"{YESNO.format('<name>')}\"，max_tokens 1，P(yes)/(P(yes)+P(no))，大小写和前导空格变体合计。"
        "s_box：第 1 步的定位问题，max_tokens 48（2.2 里控制器的调用）。回答只有两种开头：代码块（```、json、换行、[、换行、制表符、{\"…）"
        "或拒答（There are no objects belonging to …）；进了代码块后列表 token 是 [，P(\"[]\") 约为 0，"
        "所以空和非空在第一个 token 上决定：s_box = P(第一个 token 是 ```)。人工标注取第 1 步 summary.md 接触表核对："
        "48 个正例里序号 11、17、19、20、27、29、33、36、43、45、47 看不见，其余 37 个可见。\n\n"
        + "\n".join(aucs) + "\n\n" + "\n".join(scores) + "\n\n"
        "## 框的时间行为\n\n"
        "GRPO v2 第 1 到 5 轮 rollout，5 帧连续窗口，4 步每步 |Δyaw| ≥ 5°，同一回合同一目标的窗口不重叠。"
        f"真目标：目标在全部 5 帧都在几何视场内（|h| ≤ {HALF_HFOV:.1f}°，|v| ≤ 35°）、距离 3 到 16 格，全部取用 {len(real)} 个"
        f"（{len({w['scene'] for w in real})} 个场景、{len({w['ep'] for w in real})} 个回合，这批 rollout 里满足条件的只有这么多）；"
        "错名：同一批窗口问别的场景的目标名（第 1 步的规则）；背后：目标在全部 5 帧都在 3 到 16 格、|h| ≥ 120°，按第 1 步的 spread 取"
        f" {len(behind)} 个（{len({w['scene'] for w in behind})} 个场景）。定位问题同 1b（max_tokens 48），取第一个框中心换算水平角 h_box。"
        "每步把 Δh_box 对 −Δyaw（固定世界点在画面里的水平角变化）做最小二乘：固定世界点斜率 1，画面里不动斜率 0。"
        "参照行用目标点的几何 h 代替框，差别来自走路的视差。跳：框中心变化离固定世界点的预测和画面不动都超过 10°。"
        "模拟行在观测窗口的转头序列上各重复 20 次（seed 0）：假框的位置从观测到的假框 h_box 里抽，真框 = 目标几何 h + 从 1b 的 300 个正例里抽的"
        "框误差；“重抽”指这一步换一个新抽样，否则沿用上一步（假框 (2) 再按 −Δyaw 平移）。\n\n"
        + "\n".join(temporal) + "\n\n"
        "逐条结果：items.jsonl（1b 的 500 条），windows.jsonl（每个窗口 5 帧的回答、框、h_box、s_box）。\n")
    print((args.out / "summary.md").read_text(), flush=True)


if __name__ == "__main__":
    main()