"""Diagnosis B2 (2026-10-13): can the base Qwen3.5-9B (no LoRA) tell a controller WHERE the named target is in our 320x180 frames, not only whether it is there? Frames, poses, angles and spreading come from bench/diag_perception.py (B), with wider bands: - positives: target 3-16 blocks away, |h| <= 45, |v| <= 30, 100 from each h bin [-45,-15), [-15,15], (15,45]; - behind negatives: same distance band, |h| >= 120, 100; - wrong-name negatives: every third positive frame, asked about a target name of another scene that is neither a target of this scene nor a block its build commands place (B's rule). Two questions per frame, thinking off, temperature 0: (a) grounding in Qwen's native JSON format. Qwen3.5 answers [{"bbox_2d": [x1, y1, x2, y2], "label": ...}] in relative 0-1000 coordinates (checked on six clearly visible frames of B: values up to 959 and 825 on a 320x180 frame that vLLM resizes to 352x192; scaled by W/1000 and H/1000 the boxes sit on the objects). The first box counts; its center gives a horizontal angle with B's pinhole model (exact at pitch 0); (b) one word: left, center, right or none (truth for positives: center = |h| <= 15). Writes items.jsonl, summary.md and contact_sheet.png (48 positives, 16 per h bin: green = boxes, the first one thick, red = projected target point as in B). Usage: diag_perception_direction.py --url http://localhost:9045/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 re import statistics from openai import OpenAI from PIL import Image, ImageDraw from diag_perception import BENCH, H, HALF_VFOV, ROOT, W, episodes, spread, view TAN_HALF_HFOV = math.tan(math.radians(HALF_VFOV)) * W / H # horizontal half fov ~51.2 deg SIDES = ("left", "center", "right") GROUND = ("Locate the {0} in this image and report its bbox coordinates in JSON format. " "If there is no {0} in the image, output an empty list [].") WORD = ("Where is the {0} in this image? Answer with one word: left, center, or right. " "Answer none if there is no {0} in the image.") BOX = re.compile(r'"bbox_2d"\s*:\s*\[\s*(-?[\d.]+)\s*,\s*(-?[\d.]+)\s*,\s*(-?[\d.]+)\s*,\s*(-?[\d.]+)\s*\]') def side(h): return "left" if h < -15 else "center" if h <= 15 else "right" def select(): """(positives, behind negatives, wrong-name negatives).""" rng = random.Random(0) by_side, behind, names, placed = {s: [] for s in SIDES}, [], {}, {} for ep, scene, _, poses, targets in episodes(): placed[scene] = " ".join(json.loads((BENCH / scene / "multi-agent/metadata.json").read_text())["commands"]) for mid, task, target in targets: if not task.startswith("find "): continue name = task[5:].removeprefix("a ").replace("_", " ") names.setdefault(scene, set()).add(name) for i, pose in enumerate(poses): dist, h, v = (round(a, 1) for a in view(pose, target)) item = {"ep": str(ep.relative_to(ROOT)), "scene": scene, "mid": mid, "frame": i, "name": name, "asked": name, "dist": dist, "h": h, "v": v, "pitch": round(pose["pitch"], 1)} if 3 <= dist <= 16 and abs(h) <= 45 and abs(v) <= 30: by_side[side(h)].append(item) elif 3 <= dist <= 16 and abs(h) >= 120: behind.append(item) pos = [x for s in SIDES for x in spread(by_side[s], 100, rng)] behind = spread(behind, 100, rng) every = sorted(set().union(*names.values())) wrong = [{**p, "asked": rng.choice([x for x in every if x not in names[p["scene"]] and x.replace(" ", "_") not in placed[p["scene"]]])} for p in pos[::3]] return pos, behind, wrong def spearman(a, b): rank = lambda xs: [sorted(xs).index(x) + (sorted(xs).count(x) - 1) / 2 for x in xs] return statistics.correlation(rank(a), rank(b)) 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_direction") args = ap.parse_args() args.out.mkdir(parents=True, exist_ok=True) pos, behind, wrong = select() counts = ", ".join(f"{s} {sum(side(p['h']) == s for p in pos)}" for s in SIDES) print(f"positives {len(pos)} ({counts}) from " f"{len({p['scene'] for p in pos})} scenes, behind {len(behind)} from {len({b['scene'] for b in behind})} " f"scenes, wrong-name {len(wrong)}", flush=True) client = OpenAI(api_key="EMPTY", base_url=args.url) def ask(kind_item): kind, item = kind_item png = (ROOT / item["ep"] / "frames" / f"{item['frame']:04d}.png").read_bytes() image = {"type": "image_url", "image_url": {"url": "data:image/png;base64," + base64.b64encode(png).decode()}} def chat(prompt, max_tokens): r = client.chat.completions.create( model=args.model, temperature=0, max_tokens=max_tokens, messages=[{"role": "user", "content": [image, {"type": "text", "text": prompt.format(item["asked"])}]}], extra_body={"chat_template_kwargs": {"enable_thinking": False}}) return r.choices[0].message.content ground, word = chat(GROUND, 256), chat(WORD, 5) boxes = [[float(c) for c in b] for b in BOX.findall(ground)] out = {"kind": kind, **item, "ground": ground, "boxes": boxes, "word_raw": word, "word": next((w for w in SIDES + ("none",) if word.strip().lower().startswith(w)), "other")} if boxes: x1, _, x2, _ = boxes[0] out["h_box"] = round(math.degrees(math.atan(((x1 + x2) / 1000 - 1) * TAN_HALF_HFOV)), 1) return out with ThreadPoolExecutor(32) as pool: items = list(pool.map(ask, [("pos", p) for p in pos] + [("behind", b) for b in behind] + [("wrong", w) for w in wrong])) with (args.out / "items.jsonl").open("w") as f: f.writelines(json.dumps(item, ensure_ascii=False) + "\n" for item in items) by_kind = {k: [x for x in items if x["kind"] == k] for k in ("pos", "behind", "wrong")} pct = lambda k, n: f"{100 * k / n:.1f}%" if n else "-" med = lambda xs: f"{statistics.median(xs):+.1f}" if xs else "-" def row(label, xs): det = [x for x in xs if x["boxes"]] err = [x["h_box"] - x["h"] for x in det] said = [x for x in xs if x["word"] in SIDES] return (f"| {label} | {len(xs)} | {pct(len(det), len(xs))} | {med([abs(e) for e in err]).lstrip('+')} | " f"{med(err)} | {pct(sum(abs(e) <= 10 for e in err), len(det))} | " f"{pct(sum(abs(e) <= 15 for e in err), len(det))} | " f"{pct(sum(side(x['h_box']) == side(x['h']) for x in det), len(xs))} | {pct(len(said), len(xs))} | " f"{pct(sum(x['word'] == side(x['h']) for x in said), len(xs))} | " f"{pct(sum(x['word'] == side(x['h']) for x in said), len(said))} |") p = by_kind["pos"] lines = ["| 正例分组 | n | (a) 给框率 | (a) 角度误差中位数（度） | (a) 有符号误差中位数 | (a) 误差 ≤10° | (a) 误差 ≤15° | " "(a) 框推出的方向对（占全部） | (b) 非 none 率 | (b) 方向对（占全部） | (b) 方向对（占作答） |", "|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|", row("全部", p)] lines += [row(f"距离 {lo}-{hi} 格", [x for x in p if lo <= x["dist"] < hi or (hi == 16 and x["dist"] == 16)]) for lo, hi in ((3, 6), (6, 10), (10, 16))] lines += [row(f"h {label}", [x for x in p if side(x["h"]) == s]) for s, label in zip(SIDES, ("左 [-45,-15)", "中 [-15,15]", "右 (15,45]"))] lines += ["", "(a) 的误差只算给了框的正例；“占全部”把没给框或答 none 算作错。", "", "(b) 的混淆表（行为真值，列为回答）：", "", "| 真值 | left | center | right | none | 其他 |", "|---|---:|---:|---:|---:|---:|"] lines += [f"| {s} | " + " | ".join(str(sum(x["word"] == w for x in p if side(x["h"]) == s)) for w in SIDES + ("none", "other")) + " |" for s in SIDES] lines += ["", "负例（误报率）：", "", "| 负例类型 | n | (a) 给框率 | (b) 非 none 率 |", "|---|---:|---:|---:|"] lines += [f"| {label} | {len(by_kind[k])} | {pct(sum(bool(x['boxes']) for x in by_kind[k]), len(by_kind[k]))} | " f"{pct(sum(x['word'] in SIDES for x in by_kind[k]), len(by_kind[k]))} |" for k, label in (("behind", "背后（目标在身后）"), ("wrong", "错名（正例帧，问别的场景的目标名）"))] det = [x for x in p if x["boxes"]] height = [x["boxes"][0][3] - x["boxes"][0][1] for x in det] area = [(x["boxes"][0][2] - x["boxes"][0][0]) * (x["boxes"][0][3] - x["boxes"][0][1]) for x in det] dist = [x["dist"] for x in det] lines += ["", f"距离线索（给了框的 {len(det)} 个正例，第一个框）：框高与几何距离的 Spearman ρ = " f"{spearman(height, dist):+.2f}，框面积与几何距离 ρ = {spearman(area, dist):+.2f}。"] (args.out / "summary.md").write_text( "帧和几何同 B（bench/diag_perception.py）：GRPO v2 第 1 到 5 轮 rollout 中策略实际看到的 320×180 PNG。" "正例：目标距离 3 到 16 格，水平角 |h| ≤ 45°、竖直角 |v| ≤ 30°，按 h 分三段 [-45,-15)、[-15,15]、(15,45] 各抽 100；" "背后负例：同距离段，|h| ≥ 120°，100 个；错名负例：每三个正例取一帧，问另一个场景的目标名（B 的规则），" f"{len(by_kind['wrong'])} 个。每个回合每个目标在每段最多 2 帧、相隔 ≥20 帧，按场景轮流抽：正例来自 " f"{len({x['scene'] for x in p})} 个场景、{len({x['ep'] for x in p})} 个回合，背后负例来自 " f"{len({x['scene'] for x in by_kind['behind']})} 个场景。只按几何选帧，不判断遮挡。\n\n" "坐标系核对：Qwen3.5 的原生定位输出是 JSON 列表 [{\"bbox_2d\": [x1, y1, x2, y2], \"label\": ...}]，坐标是相对 0 到 1000，" "不是像素：在 B 的 6 张目标清楚可见的帧上，y 最大到 825、x 最大到 959（原图 320×180，vLLM 送进模型时缩放为 352×192，" "两种像素坐标都不可能出现这些值），按 x·W/1000、y·H/1000 换算后框落在物体上（品红/紫色玻璃、箱子、淡蓝混凝土准确，" "矢车菊和盆栽竹子这种很小的物体偏约 15 像素）。下面取第一个框的中心，用针孔模型（水平半视场 " f"{math.degrees(math.atan(TAN_HALF_HFOV)):.1f}°，俯仰为 0 时精确）换算成水平角，和几何 h 比较。\n\n" f"问题（思考关闭，温度 0）：(a) \"{GROUND.format('')}\"；(b) \"{WORD.format('')}\"，" "真值 center 为 |h| ≤ 15°。\n\n" + "\n".join(lines) + "\n\n" "接触表 contact_sheet.png：每段前 16 个正例（左、中、右依次排列），绿框为 (a) 的框（第一个加粗），红圈为目标点按视角投影的位置" "（同 B），标注为序号、目标名、距离、h、(b) 的回答；全部逐条结果见 items.jsonl。\n") sheet = Image.new("RGB", (6 * W, 8 * (H + 14)), "white") shown = [x for s in SIDES for x in [x for x in p if side(x["h"]) == s][:16]] for k, x in enumerate(shown): img = Image.open(ROOT / x["ep"] / "frames" / f"{x['frame']:04d}.png").convert("RGB") draw = ImageDraw.Draw(img) u = W / 2 + W / 2 * math.tan(math.radians(x["h"])) / TAN_HALF_HFOV v = H / 2 - H / 2 * math.tan(math.radians(x["v"])) / math.tan(math.radians(HALF_VFOV)) draw.ellipse((u - 6, v - 6, u + 6, v + 6), outline="red", width=2) for j, (x1, y1, x2, y2) in enumerate(x["boxes"]): draw.rectangle((x1 * W / 1000, y1 * H / 1000, x2 * W / 1000, y2 * H / 1000), outline=(0, 255, 0), width=2 if j == 0 else 1) cell = (k % 6 * W, k // 6 * (H + 14)) sheet.paste(img, (cell[0], cell[1] + 14)) ImageDraw.Draw(sheet).text((cell[0] + 2, cell[1] + 1), f"{k} {x['name']} | {x['dist']:.0f}m | h{x['h']:+.0f} | " f"{x['word_raw'].strip()[:7]}", fill="black") sheet.save(args.out / "contact_sheet.png") print((args.out / "summary.md").read_text(), flush=True) if __name__ == "__main__": main()