"""CAD 实测标定 VLM 读图下限（临时加字 + 平行线，测完自动清理、不保存图纸）。 与 `verify_vlm_resolution_offline.py` 配套：离线预扫给**区间**，本脚本在**真实 CAD 渲染**下确认最终 `R_text` / `R_line`（离线合成图偏干净，结论偏乐观）。 流程： 1. 在空白区临时插入标定物（图层 `AI_CALIB`）： - 3 个单行文字，字高 T（默认 200 图形单位；1:100 时 = 纸面 2mm），内容为易错随机串； - 若干「1 条 / 2 条」平行线对（世界间距按目标像素间距换算）。 2. 量出绘图区**原生像素** → 有效像素 eff（长边 ≤ `IMAGE_MAX_EDGE`）+ 视口宽高比。 3. 目标像素字高 target：窗口宽 = `T*eff_w/target`、高 = 宽/视口宽高比 → zoom(margin=0) → 截图 → 下采样 → 问 VLM 读字。 4. 线对同理：世界间距 `L = g*W_REF/eff_w`，问「1 条还是 2 条」。 5. 汇总 R_text / R_line；无论成败都在 finally 里 `delete_entities` 清理。 用法： python verify_text_resolution.py # 自动选空白锚点 python verify_text_resolution.py --anchor 500000,0 python verify_text_resolution.py --targets 6,8,10,12,16 --gaps 2,3,4,6,8 python verify_text_resolution.py --keep # 保留标定物（调试） 前置：CAD 在线（127.0.0.1:8080）+ `DEEPSEEK_API_KEY`。需要 Pillow。 """ import argparse import json import random import time from pathlib import Path from dwg_review import config, tiling from dwg_review.context import ReviewContext from dwg_review.llm_client import LlmClient, downsample_image_b64, image_size_b64 from dwg_review.mcp_client import result_images, result_json from dwg_review.phases import planning # 复用离线脚本的随机串与解析工具（避免重复实现） from verify_vlm_resolution_offline import _ALPHABET, _extract_json, _lenient, _norm_token CALIB_LAYER = "AI_CALIB" _TEXT_GAP = 2000.0 # 文字之间的水平间距（世界单位） _LINE_LEN_FRAC = 0.6 # 线长占测线窗口宽的比例 _W_REF = 20000.0 # 测线用的固定窗口宽（世界单位） _SINGLE_DX = 30000.0 # 「单条线」相对线对的水平偏移（避免同框） _ROW_DY = 6000.0 # 各档线对的行距 _FIRST_ROW_DY = 10000.0 # 第一行线对相对文字的下移量 _SYS_TEXT = "你是图像文字识别器。只输出一个 JSON 对象，不要任何解释、不要 markdown 代码块。" _SYS_LINE = "你是工程图纸线型识别器。只输出一个 JSON 对象，不要任何解释、不要 markdown 代码块。" # --------------------------------------------------------------------------- # 基础交互 # --------------------------------------------------------------------------- def _shot(ctx): imgs = result_images(ctx.tools.call("get_viewport_screenshot", {})) return imgs[0] if imgs else None def _zoom_and_shot(ctx, win): ctx.tools.call("zoom_to_window", dict(win, margin=0.0)) time.sleep(config.RENDER_WAIT) return _shot(ctx) def _ask(ctx, shot, system, prompt, out_png=None): """截图（先下采样）→ 问 VLM → 返回解析后的 dict（失败 None）。""" b64, mime = downsample_image_b64(shot[0], shot[1]) if out_png is not None: import base64 out_png.write_bytes(base64.b64decode(b64)) resp = ctx.llm.chat([ {"role": "system", "content": system}, {"role": "user", "content": [ {"type": "text", "text": prompt}, LlmClient.image_content(b64, mime), ]}, ]) return _extract_json(resp.choices[0].message.content) def _center(win): return ((win["min_x"] + win["max_x"]) / 2.0, (win["min_y"] + win["max_y"]) / 2.0) def _window(cx, cy, w, h): return {"min_x": cx - w / 2.0, "min_y": cy - h / 2.0, "max_x": cx + w / 2.0, "max_y": cy + h / 2.0} # --------------------------------------------------------------------------- # 标定物 # --------------------------------------------------------------------------- def _auto_anchor(ctx): """把锚点放在所有图框的右侧远处（空白区）。""" plan = planning.plan(ctx) boxes = [f.get("bbox") for f in (plan.get("frames") or []) if isinstance(f.get("bbox"), list) and len(f["bbox"]) == 4] if not boxes: return (0.0, 0.0) return (max(b[2] for b in boxes) + 50000.0, min(b[1] for b in boxes)) def _calib_window(ax, ay, n_gaps): """覆盖全部标定物的查询窗口。""" return { "min_x": ax - 2 * _TEXT_GAP, "min_y": ay - _FIRST_ROW_DY - n_gaps * _ROW_DY - 2 * _W_REF, "max_x": ax + _SINGLE_DX + 2 * _W_REF, "max_y": ay + 2 * _TEXT_GAP, } def _add_calibration(ctx, ax, ay, gaps, text_h, eff_w, rng, color=7): """插入标定物，返回 (文字 token 列表, 各档线对中心 y 列表)。 `color` 默认 7（白/黑自适应），保证**黑底白字**的高对比——标定的是**分辨率下限**， 不能把「颜色对比度不足」混进来（实测红色 ACI 1 在 12px 时完全读不出）。 """ tokens = ["".join(rng.choice(_ALPHABET) for _ in range(6)) for _ in range(3)] for i, tok in enumerate(tokens): ctx.tools.call("add_text", { "text": tok, "position": [ax + i * _TEXT_GAP, ay], "height": text_h, "layer": CALIB_LAYER, "color": color, }) line_len = _LINE_LEN_FRAC * _W_REF ys = [] for k, g in enumerate(gaps): spacing = g * _W_REF / eff_w # 世界间距 ↔ 目标像素间距 yk = ay - _FIRST_ROW_DY - k * _ROW_DY ys.append(yk) # 线对（2 条） for dy in (-spacing / 2.0, spacing / 2.0): ctx.tools.call("add_line", { "start": [ax - line_len / 2.0, yk + dy], "end": [ax + line_len / 2.0, yk + dy], "layer": CALIB_LAYER, "color": color, }) # 对照（1 条） sx = ax + _SINGLE_DX ctx.tools.call("add_line", { "start": [sx - line_len / 2.0, yk], "end": [sx + line_len / 2.0, yk], "layer": CALIB_LAYER, "color": color, }) return tokens, ys def _collect_handles(ctx, window): env = result_json( ctx.tools.call("query_entities", { "window": window, "layer_filter": CALIB_LAYER, "limit": 1000, }), {}, ) return [it.get("handle") for it in (env.get("items") or []) if it.get("handle")] # --------------------------------------------------------------------------- # 各子测试 # --------------------------------------------------------------------------- def _text_center(ctx, window, fallback): """取文字类实体的 bbox 中心（拿不到则用 fallback）。 注意：`query_entities` 的 `type_filter` **必须是列表**（传字符串会被静默忽略）。 这里同时按 `tiling.is_text_type` 在 Python 侧再过滤一次，双保险。 """ env = result_json( ctx.tools.call("query_entities", { "window": window, "layer_filter": CALIB_LAYER, "type_filter": ["TEXT"], "limit": 1000, }), {}, ) boxes = [] for it in env.get("items") or []: if not tiling.is_text_type(it.get("type")): continue b = tiling.norm_box(it.get("bbox")) if b: boxes.append(b) if not boxes: return fallback return ( (min(b[0] for b in boxes) + max(b[2] for b in boxes)) / 2.0, (min(b[1] for b in boxes) + max(b[3] for b in boxes)) / 2.0, ) def run_text_levels(ctx, tokens, center, targets, text_h, eff_w, aspect, out_dir, repeat): rows = [] for target in targets: w = text_h * eff_w / float(target) win = _window(center[0], center[1], w, w / aspect) trials = [] for _ in range(max(1, repeat)): shot = _zoom_and_shot(ctx, win) if not shot: trials.append({"got": [], "exact": 0}) continue data = _ask( ctx, shot, _SYS_TEXT, "图中是若干处**白色字符**（黑底）。请把能看清的字符串按从左到右顺序读出，" '输出 JSON：{"texts":["串1","串2",...]}。看不清的串填空字符串。', out_png=out_dir / f"text_target{target}.png", ) got = [str(x) for x in (data or {}).get("texts") or []] exact = sum(1 for i, t in enumerate(tokens) if i < len(got) and _norm_token(got[i]) == t) trials.append({"got": got, "exact": exact}) best = max(t["exact"] for t in trials) row = {"target_px": target, "tokens": tokens, "trials": trials, "n": len(tokens), "best_exact": best, "readable": best == len(tokens)} rows.append(row) print(f" 目标字高 {target:>3}px 命中 {best}/{len(tokens)} 可读={row['readable']}") return rows def run_line_levels(ctx, ax, ys, gaps, aspect, out_dir, repeat): """对每档像素间距，分别看「2 条线对」与「1 条对照」是否被正确区分。""" rows = [] prompt = ('图中是**水平直线**（黑底白线）。只判断线条数量是 1 还是 2，' '输出 JSON：{"count": 1 或 2}。') for g, yk in zip(gaps, ys): res = {} for kind, cx in (("pair", ax), ("single", ax + _SINGLE_DX)): got = None for _ in range(max(1, repeat)): shot = _zoom_and_shot(ctx, _window(cx, yk, _W_REF, _W_REF / aspect)) if not shot: continue data = _ask(ctx, shot, _SYS_LINE, prompt, out_png=out_dir / f"line_g{g}_{kind}.png") try: got = int(data.get("count")) except (TypeError, ValueError, AttributeError): got = None if got == (2 if kind == "pair" else 1): break res[kind] = got ok = res.get("pair") == 2 and res.get("single") == 1 rows.append({"gap_px": g, "pair": res.get("pair"), "single": res.get("single"), "resolved": ok}) print(f" 目标间距 {g:>3}px 线对={res.get('pair')} 单条={res.get('single')} " f"可分辨={ok}") return rows def _parse_floats(s): return [float(x) for x in str(s).split(",") if x.strip()] def main(argv=None): ap = argparse.ArgumentParser() ap.add_argument("--anchor", default="auto", help='"auto" 或 "x,y"') ap.add_argument("--text-height", type=float, default=200.0, help="标定字高（图形单位；1:100 时 200 = 纸面 2mm）") ap.add_argument("--targets", default="6,8,10,12,16,20", help="目标像素字高扫描档") ap.add_argument("--gaps", default="2,3,4,5,6,8,12", help="目标像素线间距扫描档") ap.add_argument("--repeat", type=int, default=1, help="每档重复次数") ap.add_argument("--seed", type=int, default=7, help="随机串种子") ap.add_argument("--keep", action="store_true", help="保留标定物（调试用）") ap.add_argument("--out", default="reports/_probe/vlm_calib_cad") args = ap.parse_args(argv) targets = [int(x) for x in _parse_floats(args.targets)] gaps = [int(x) for x in _parse_floats(args.gaps)] ctx = ReviewContext.create(need_llm=True) out_dir = Path(config.PROJECT_ROOT) / args.out out_dir.mkdir(parents=True, exist_ok=True) if args.anchor.strip().lower() == "auto": ax, ay = _auto_anchor(ctx) else: ax, ay = (float(v) for v in args.anchor.split(",")) print(f"锚点：({ax:.0f}, {ay:.0f}) 模型：{ctx.llm.model}") # 量视口像素（决定世界↔像素换算） base = _shot(ctx) if not base: raise SystemExit("无法截图：get_viewport_screenshot 无返回") vp = image_size_b64(base[0]) if not vp: raise SystemExit("无法解析截图尺寸（缺 Pillow 或图片非法）") eff = tiling.effective_px(vp, config.IMAGE_MAX_EDGE) aspect = vp[0] / float(vp[1]) print(f"绘图区原生 {vp[0]}x{vp[1]} → 有效 {eff[0]}x{eff[1]} 视口宽高比 {aspect:.3f}") handles = [] results = {"viewport": list(vp), "effective": list(eff), "text_height": args.text_height, "anchor": [ax, ay]} try: rng = random.Random(args.seed) tokens, ys = _add_calibration(ctx, ax, ay, gaps, args.text_height, eff[0], rng) win = _calib_window(ax, ay, len(gaps)) handles = _collect_handles(ctx, win) print(f"已插入标定物：{len(handles)} 个实体（图层 {CALIB_LAYER}）") center = _text_center(ctx, win, (ax + _TEXT_GAP, ay)) print("\n-- 文字可读性（真实 CAD 渲染） --") text_rows = run_text_levels(ctx, tokens, center, targets, args.text_height, eff[0], aspect, out_dir, args.repeat) print("\n-- 线间距可分辨性（1 vs 2） --") line_rows = run_line_levels(ctx, ax, ys, gaps, aspect, out_dir, args.repeat) r_text = next((r["target_px"] for r in text_rows if r["readable"]), None) r_line = next((r["gap_px"] for r in line_rows if r["resolved"]), None) results.update({"text": text_rows, "line": line_rows, "R_text": r_text, "R_line": r_line}) print(f"\n实测：R_text={r_text}px R_line={r_line}px") print("回填：dwg_review/config.py 的 TEXT_READABLE_PX / LINE_READABLE_PX") finally: if handles and not args.keep: out = result_json(ctx.tools.call("delete_entities", {"handles": handles}), {}) print(f"\n清理标定物：删除 {out.get('count')} 个，失败 {len(out.get('failed') or [])} 个") elif handles: print(f"\n--keep：保留 {len(handles)} 个标定物（图层 {CALIB_LAYER}）") ctx.tools.call("zoom_extents", {}) (out_dir / "summary.json").write_text( json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8" ) print(f"输出目录：{out_dir}") return 0 if __name__ == "__main__": raise SystemExit(main())