| """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 |
| 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('<name>')}\"；(b) \"{WORD.format('<name>')}\"，" |
| "真值 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() |
|
|
