diff --git a/dwg_review/config.py b/dwg_review/config.py index db1d863..6e69023 100644 --- a/dwg_review/config.py +++ b/dwg_review/config.py @@ -122,10 +122,36 @@ IMAGE_FORMAT = os.environ.get("IMAGE_FORMAT", "png").strip().lower() IMAGE_JPEG_QUALITY = int(os.environ.get("IMAGE_JPEG_QUALITY", "80")) # 取证阶段每图框的局部截图数（0=仅整体；4=整体+2x2 象限） +# —— 仅在自适应瓦片规划不可用时作为**回退**路径参数 EVIDENCE_DETAIL_TILES = int(os.environ.get("EVIDENCE_DETAIL_TILES", "4")) # 每图框图片硬上限（整体 + 局部总张数） EVIDENCE_MAX_IMAGES = int(os.environ.get("EVIDENCE_MAX_IMAGES", "5")) +# --------------------------------------------------------------------------- +# 自适应瓦片规划（按实体分布决定截几张、截哪里） +# --------------------------------------------------------------------------- +# 是否启用瓦片规划；False = 退回固定 EVIDENCE_DETAIL_TILES（行为与旧版一致） +TILING_ENABLED = _read_bool("TILING_ENABLED", True) +# 纸面字高（mm）。国标标注字高 2~3mm，取保守下限 +TEXT_HEIGHT_MM = float(os.environ.get("TEXT_HEIGHT_MM", "2.0")) +# VLM 可读的**最小像素字高**。 +# 离线标定（verify_vlm_resolution_offline.py，repeat=2）：黑底 12px / 白底 10px 全行正确。 +# CAD 模型空间默认黑底 → 取 12；离线合成图偏干净，故以 B2 CAD 实测为最终依据。 +TEXT_READABLE_PX = int(os.environ.get("TEXT_READABLE_PX", "12")) +# VLM 可分辨的**最小线间距像素**。 +# 同上标定：4~5px 有噪声、≥6px 稳定（1 条 vs 2 条二选一）。取 6 保守。 +LINE_READABLE_PX = int(os.environ.get("LINE_READABLE_PX", "6")) +# 单图框瓦片数硬上界（成本兜底） +TILE_MAX_COUNT = int(os.environ.get("TILE_MAX_COUNT", "36")) +# 相邻瓦片重叠比例（防边界切字） +TILE_OVERLAP = float(os.environ.get("TILE_OVERLAP", "0.0")) +# 热力图网格划分（长边切几份） +HEATMAP_CELL_DIV = int(os.environ.get("HEATMAP_CELL_DIV", "32")) +# 热力图聚类前的膨胀格数（合并相近内容；0=不膨胀） +HEATMAP_CLUSTER_DILATE = int(os.environ.get("HEATMAP_CLUSTER_DILATE", "0")) +# 丢弃 bbox 面积超过图框该比例的实体（去图框边框/图签块） +HEATMAP_MAX_AREA_FRAC = float(os.environ.get("HEATMAP_MAX_AREA_FRAC", "0.5")) + def has_api_key() -> bool: return bool(API_KEY.strip()) diff --git a/dwg_review/llm_client.py b/dwg_review/llm_client.py index 7505aa0..a41fa1e 100644 --- a/dwg_review/llm_client.py +++ b/dwg_review/llm_client.py @@ -67,6 +67,21 @@ def downsample_image_b64(b64, mime="image/png", max_edge=None, fmt=None, quality return b64, mime +def image_size_b64(b64, mime="image/png"): + """解码图片并返回 `(宽, 高)`；缺 Pillow / 解码失败返回 None。 + + 用于瓦片规划前测量**绘图区实际像素**（决定单张截图能覆盖多大世界范围）。 + """ + if not HAS_PIL or not b64: + return None + try: + img = Image.open(io.BytesIO(base64.b64decode(b64))) + w, h = img.size + return (int(w), int(h)) + except Exception: + return None + + class LlmClient: """统一的 LLM 调用入口。""" diff --git a/dwg_review/phases/evidence.py b/dwg_review/phases/evidence.py index 89cc7f8..ec3020b 100644 --- a/dwg_review/phases/evidence.py +++ b/dwg_review/phases/evidence.py @@ -10,7 +10,8 @@ import math import time -from .. import config, evidence_model +from .. import config, evidence_model, heatmap, tiling +from ..llm_client import image_size_b64 from ..mcp_client import result_images, result_json _LABEL_OVERALL = "overall" @@ -66,10 +67,80 @@ def _zoom_and_shot(ctx, window, margin=0.02): return images[0] -def capture_images(ctx, window, *, tiles=0, max_images=None): - """采集视觉证据：整体一张 + 局部 tiles 张。 +def _is_truncated(*envelopes): + """任一信封 `total > len(items)` 即视为**被截断**（数据不全）。""" + for env in envelopes: + if not isinstance(env, dict): + continue + items = env.get("items") + if not isinstance(items, list): + continue + try: + total = int(env.get("total") or len(items)) + except (TypeError, ValueError): + total = len(items) + if total > len(items): + return True + return False + + +def _text_boxes(entities, texts, dimensions): + """收集「文字类」实体 bbox。 + + 标准文字 / 标注信封（`get_text_content` / `get_dimensions`）全收； + 再加**类名含 text/label/dim 等关键词**的自定义实体（`tiling.is_text_type`）。 + """ + boxes = list(heatmap.entity_boxes(texts)) + boxes.extend(heatmap.entity_boxes(dimensions)) + items = entities.get("items") if isinstance(entities, dict) else None + for it in items or []: + if isinstance(it, dict) and tiling.is_text_type(it.get("type")): + b = heatmap.norm_box(it.get("bbox")) + if b is not None: + boxes.append(b) + return boxes + + +def _plan_tiles(window, frame, boxes, text_boxes, truncated, view_px): + """尝试按实体分布规划瓦片窗口；任何条件不满足都返回 []（调用方退回固定策略）。 + + 降级条件：规划被关闭 / 实体数据被截断 / 缺图框 / 无实体 / 拿不到绘图区像素 / + 比例未知（`frame_scale` 返 0）。 + """ + if not config.TILING_ENABLED or truncated or not frame or not boxes or not view_px: + return [] + scale = tiling.frame_scale(frame) + s_text = tiling.world_per_px(scale, config.TEXT_HEIGHT_MM, config.TEXT_READABLE_PX) + # 线条区只需看清走向：同以「最小关注特征尺寸」（纸面 TEXT_HEIGHT_MM）为基准， + # 但像素门槛更低（R_line < R_text）→ 覆盖范围更大、瓦片更少。 + s_line = tiling.world_per_px(scale, config.TEXT_HEIGHT_MM, config.LINE_READABLE_PX) + if s_text <= 0 or s_line <= 0: + return [] + eff = tiling.effective_px(view_px, config.IMAGE_MAX_EDGE) + if eff[0] <= 0 or eff[1] <= 0: + return [] + + cell = heatmap.div_to_cell(window, config.HEATMAP_CELL_DIV) + grid = heatmap.build_grid(window, boxes, cell) + clusters = heatmap.cluster_cells(grid, dilate=config.HEATMAP_CLUSTER_DILATE) + if not clusters: + return [] + flags = tiling.cluster_text_flags(clusters, text_boxes) + text_idx = {i for i, is_text in enumerate(flags) if is_text} + return tiling.plan_regions( + frame, clusters, + text_clusters=text_idx, s_text=s_text, s_line=s_line, view_px=eff, + max_tiles=config.TILE_MAX_COUNT, overlap=config.TILE_OVERLAP, + ) + + +def capture_images(ctx, window, *, tiles=0, max_images=None, frame=None, + boxes=None, text_boxes=None, truncated=False): + """采集视觉证据：整体一张 + 局部若干张（自适应瓦片优先，否则等分象限回退）。 - 整体 label="overall"，局部 label="tile-1"..；行优先。 + - 局部窗口优先由 `_plan_tiles` 按**实体分布**给出（空白区不出图）； + 不可用时退回 `sub_windows(window, tiles)`（行为与旧版完全一致）。 - 结束时**复位到整体视口**，保证后续流程从整幅视图起手。 - 受 max_images（默认 config.EVIDENCE_MAX_IMAGES）截断。 """ @@ -81,11 +152,17 @@ def capture_images(ctx, window, *, tiles=0, max_images=None): if shot and len(images) < max_images: images.append(evidence_model.make_image(shot[0], shot[1], label)) - # 整体 - _add(_zoom_and_shot(ctx, window), _LABEL_OVERALL) + # 整体图：既是证据，也用于测量绘图区**实际像素**（决定单张截图的世界覆盖范围） + overall = _zoom_and_shot(ctx, window) + view_px = image_size_b64(overall[0]) if overall else None + _add(overall, _LABEL_OVERALL) - # 局部（等分象限） - for i, sub in enumerate(sub_windows(window, tiles), 1): + # 局部：优先自适应瓦片，失败退回等分象限 + regions = _plan_tiles(window, frame, boxes, text_boxes, truncated, view_px) + if not regions: + regions = sub_windows(window, tiles) + + for i, sub in enumerate(regions, 1): if len(images) >= max_images: break _add(_zoom_and_shot(ctx, sub), f"tile-{i}") @@ -115,8 +192,19 @@ def collect_frame(ctx, index, frame): ctx.tools.call("get_polylines_info", {"window": win, "limit": limit}), {} ) - # b. 视觉：整体 + 可选局部（多图） - images = capture_images(ctx, win, tiles=config.EVIDENCE_DETAIL_TILES) + # b. 视觉：整体 + 局部（优先按实体分布自适应瓦片；数据截断则退回固定策略） + truncated = _is_truncated(entities, texts, dimensions, blocks, polylines) + boxes = heatmap.drop_oversized( + heatmap.entity_boxes(entities), win, config.HEATMAP_MAX_AREA_FRAC + ) + images = capture_images( + ctx, win, + tiles=config.EVIDENCE_DETAIL_TILES, + frame=frame, + boxes=boxes, + text_boxes=_text_boxes(entities, texts, dimensions), + truncated=truncated, + ) return { "source_type": evidence_model.source_type_of(frame), diff --git a/dwg_review/tiling.py b/dwg_review/tiling.py new file mode 100644 index 0000000..011b9b4 --- /dev/null +++ b/dwg_review/tiling.py @@ -0,0 +1,284 @@ +"""自适应瓦片规划器：按实体分布把图框切成若干 zoom 窗口，使 VLM 能看清细节。 + +纯函数模块（仅依赖 `math`），不依赖 Pillow / CAD。 + +核心思路（用户方案）： + 1. 由图框纸张 + 比例换算「CAD 图形单位 / 像素」s； + 2. 由 VLM 可读下限（文字 `R_text` px、线条 `R_line` px）定每张截图的**世界覆盖范围**； + 3. 复用热力图（`heatmap.py`）的**占用网格 + 聚类**得到簇包围盒；簇内是否含**文字类实体** + 决定用细（`s_text`）/ 粗（`s_line`）尺度； + 4. 簇 ≤ 单屏 → 1 张；超出 → 行优先细分；**空白区不出图**（整体图已覆盖）； + 受 `max_tiles` 截断，**文字簇优先保留**。 + +坐标约定：与 `heatmap.py` / CAD 一致——x 向右、y **向上**；bbox=(min_x,min_y,max_x,max_y)； +瓦片行优先且 row 0 在**最上方**（与 `phases/evidence.py::sub_windows` 一致）。 +""" + +import math + +# 纸张尺寸（mm，长边在前）：GB/T 14689 基本幅面 +SHEET_SIZES_MM = { + "A0": (1189.0, 841.0), + "A1": (841.0, 594.0), + "A2": (594.0, 420.0), + "A3": (420.0, 297.0), + "A4": (297.0, 210.0), +} + +# 标准实体名：本身就是文字/标注类 +_STD_TEXT_TYPES = frozenset({"TEXT", "MTEXT", "ATTRIB", "ATTDEF", "DIMENSION"}) + +# 自定义实体类名关键词（用户经验：类名常含 text/label/dim 等） +_TEXT_KEYWORDS = ("TEXT", "LABEL", "DIM", "NOTE", "TAG", "ANNO", "TITLE") + +# 瓦片重叠上限（防边界切字；过大会显著增加张数） +_MAX_OVERLAP = 0.5 + + +def norm_box(b): + """归一 bbox 为 (minx,miny,maxx,maxy)；非法返回 None。""" + if not isinstance(b, (list, tuple)) or len(b) != 4: + return None + try: + x1, y1, x2, y2 = (float(v) for v in b) + except (TypeError, ValueError): + return None + if not all(math.isfinite(v) for v in (x1, y1, x2, y2)): + return None + if x1 > x2: + x1, x2 = x2, x1 + if y1 > y2: + y1, y2 = y2, y1 + return (x1, y1, x2, y2) + + +def frame_bbox(frame): + """图框 bbox 元组；非法返回 None。""" + if not isinstance(frame, dict): + return None + box = norm_box(frame.get("bbox")) + if box is None or box[2] <= box[0] or box[3] <= box[1]: + return None + return box + + +def frame_world_size(frame): + """图框在 CAD 中的世界尺寸 (W, H)；bbox 非法返回 (0.0, 0.0)。""" + box = frame_bbox(frame) + if box is None: + return (0.0, 0.0) + return (box[2] - box[0], box[3] - box[1]) + + +def frame_scale(frame, default=0.0): + """图框出图比例（1:100 → 100.0）。 + + 优先级： + 1. `sheet_scale`（工具已识别，>0 直接用）； + 2. 由「图框世界尺寸 / 纸张尺寸」反算（需 `sheet_size` 命中纸张表）； + 3. `default`（拿不到 → 调用方退回固定策略）。 + """ + if not isinstance(frame, dict): + return default + try: + s = float(frame.get("sheet_scale") or 0.0) + except (TypeError, ValueError): + s = 0.0 + if s > 0: + return s + + w, h = frame_world_size(frame) + if w <= 0 or h <= 0: + return default + sheet = SHEET_SIZES_MM.get(str(frame.get("sheet_size") or "").strip().upper()) + if not sheet: + return default + sw, sh = sheet + # 判断横放 / 纵放：取长宽比更接近的朝向 + ar = w / h + if abs(ar - sw / sh) <= abs(ar - sh / sw): + return w / sw # 横放：世界宽 ↔ 纸宽 + return w / sh # 纵放：世界宽 ↔ 纸高 + + +def effective_px(view_px, max_edge): + """按长边下采样后的**有效像素** (ew, eh)。 + + 与 `llm_client.downsample_image_b64` 的缩放规则一致（长边缩到 max_edge）。 + 非法输入返回 (0, 0)；max_edge<=0 表示不缩放。 + """ + if not view_px or len(view_px) != 2: + return (0, 0) + try: + w, h = int(view_px[0]), int(view_px[1]) + except (TypeError, ValueError): + return (0, 0) + if w <= 0 or h <= 0: + return (0, 0) + if max_edge and max_edge > 0 and max(w, h) > max_edge: + k = max_edge / float(max(w, h)) + return (max(1, int(round(w * k))), max(1, int(round(h * k)))) + return (w, h) + + +def world_per_px(scale, text_mm, readable_px): + """每像素代表的 CAD 图形单位：s = (text_mm * scale) / readable_px。 + + `text_mm` 为纸面字高（国标 2~3mm），`scale` 为出图比例（1:100 → 100）， + `readable_px` 为 VLM 可读的**最小像素字高**。任一非正 → 返回 0.0（表示无法规划）。 + """ + try: + scale, text_mm, readable_px = float(scale), float(text_mm), float(readable_px) + except (TypeError, ValueError): + return 0.0 + if scale <= 0 or text_mm <= 0 or readable_px <= 0: + return 0.0 + return text_mm * scale / readable_px + + +def is_text_type(type_name): + """实体类型名是否为「文字/标注类」。 + + 标准实体给 DXF 名（TEXT/MTEXT/ATTRIB/ATTDEF/DIMENSION）→ 精确命中； + 自定义实体兜底为**类名去 AcDb 前缀并大写**，故按**关键词包含**判定 + （TEXT/LABEL/DIM/NOTE/TAG/ANNO/TITLE）。非字符串/空 → False。 + """ + if not isinstance(type_name, str): + return False + t = type_name.strip().upper() + if not t: + return False + if t in _STD_TEXT_TYPES: + return True + return any(k in t for k in _TEXT_KEYWORDS) + + +def boxes_intersect(a, b, eps=0.0): + """两个 bbox 是否相交（接触视为相交；eps 可放宽）。""" + if a is None or b is None: + return False + return not ( + a[2] < b[0] - eps or b[2] < a[0] - eps + or a[3] < b[1] - eps or b[3] < a[1] - eps + ) + + +def cluster_text_flags(clusters, text_boxes): + """标记每个簇是否含文字类实体（bbox 相交即算，保守）。 + + 返回与 `clusters` 等长的 bool 列表。 + """ + boxes = [b for b in (norm_box(x) for x in (text_boxes or [])) if b] + flags = [] + for cl in clusters or []: + b = norm_box(cl.get("bbox")) if isinstance(cl, dict) else None + flags.append(bool(b) and any(boxes_intersect(b, tb) for tb in boxes)) + return flags + + +def tiles_for_box(box, cap_w, cap_h, overlap=0.0): + """覆盖 box 所需的 (cols, rows)。 + + `cap_w`/`cap_h` 为单张截图的世界覆盖范围；`overlap` 为相邻瓦片重叠比例 + （>0 时有效步长缩短，张数增加）。非法输入返回 (1, 1)。 + """ + if box is None: + return (1, 1) + w, h = box[2] - box[0], box[3] - box[1] + if w <= 0 or h <= 0: + return (1, 1) + ov = min(max(float(overlap or 0.0), 0.0), _MAX_OVERLAP) + step_w = float(cap_w) * (1.0 - ov) + step_h = float(cap_h) * (1.0 - ov) + if step_w <= 0 or step_h <= 0: + return (1, 1) + return (max(1, int(math.ceil(w / step_w))), max(1, int(math.ceil(h / step_h)))) + + +def _clamp_window(win, frame_win): + """把窗口夹到图框范围内；夹后退化（宽/高<=0）返回 None。""" + if frame_win is None: + return win + fx1, fy1, fx2, fy2 = frame_win + x1 = max(win["min_x"], fx1) + y1 = max(win["min_y"], fy1) + x2 = min(win["max_x"], fx2) + y2 = min(win["max_y"], fy2) + if x2 <= x1 or y2 <= y1: + return None + return {"min_x": x1, "min_y": y1, "max_x": x2, "max_y": y2} + + +def subdivide(box, cols, rows, overlap=0.0, frame_win=None): + """把 box 等分为 cols×rows 个 zoom 窗口（行优先，row 0 在最上方）。 + + 每张按 `overlap` 比例向外扩张（防边界切字），再夹到 `frame_win`。 + """ + x1, y1, x2, y2 = box + cols, rows = max(1, int(cols)), max(1, int(rows)) + w = (x2 - x1) / cols + h = (y2 - y1) / rows + ov = min(max(float(overlap or 0.0), 0.0), _MAX_OVERLAP) + mx, my = w * ov, h * ov + out = [] + for r in range(rows): + for c in range(cols): + sx1 = x1 + c * w + sy2 = y2 - r * h # row 0 在最上方 + win = { + "min_x": sx1 - mx, "min_y": sy2 - h - my, + "max_x": sx1 + w + mx, "max_y": sy2 + my, + } + clamped = _clamp_window(win, frame_win) + if clamped: + out.append(clamped) + return out + + +def plan_regions(frame, clusters, *, text_clusters=(), s_text=0.0, s_line=0.0, + view_px=(0, 0), max_tiles=36, overlap=0.0): + """把图框内的簇规划成一组 zoom 窗口（自适应瓦片）。 + + 参数： + - `frame` ：图框（含 bbox / sheet_size / sheet_scale）； + - `clusters` ：`heatmap.cluster_cells` 的输出（已按实体计数降序）； + - `text_clusters`：含文字类实体的**簇下标**集合（决定用 s_text 还是 s_line）； + - `s_text`/`s_line`：CAD 图形单位 / 像素（`world_per_px` 的输出）； + - `view_px` ：单张截图的**有效像素** (ew, eh)（`effective_px` 的输出）； + - `max_tiles` ：返回窗口数硬上界（文字簇优先保留）； + - `overlap` ：相邻瓦片重叠比例。 + + 返回行优先的 window 列表；**无法规划时返回 []**（调用方退回固定策略）。 + """ + frame_win = frame_bbox(frame) + if frame_win is None: + return [] + ew, eh = view_px[0] if view_px else 0, view_px[1] if view_px and len(view_px) > 1 else 0 + if ew <= 0 or eh <= 0: + return [] + + text_idx = set(text_clusters or ()) + # 文字簇优先（同级内保持输入的计数降序） + ordered = sorted( + enumerate(clusters or []), + key=lambda p: (0 if p[0] in text_idx else 1, p[0]), + ) + + regions = [] + for i, cl in ordered: + if len(regions) >= max_tiles: + break + box = norm_box(cl.get("bbox")) if isinstance(cl, dict) else None + if box is None: + continue + s = s_text if i in text_idx else s_line + if s <= 0: + continue + cap_w, cap_h = s * ew, s * eh + cols, rows = tiles_for_box(box, cap_w, cap_h, overlap) + # 单屏可覆盖 → 1 张（subdivide 仍会按 overlap 外扩并夹到图框） + for win in subdivide(box, cols, rows, overlap, frame_win): + regions.append(win) + if len(regions) >= max_tiles: + break + return regions[:max_tiles] diff --git a/tests/test_config.py b/tests/test_config.py index 5b63875..755f9b5 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -79,6 +79,28 @@ def test_credibility_constants_have_expected_types(): assert isinstance(config.DUP_GROUP_LIMIT, int) +def test_tiling_constants_have_expected_types(): + assert isinstance(config.TILING_ENABLED, bool) + assert isinstance(config.TEXT_HEIGHT_MM, float) + assert isinstance(config.TEXT_READABLE_PX, int) + assert isinstance(config.LINE_READABLE_PX, int) + assert isinstance(config.TILE_MAX_COUNT, int) + assert isinstance(config.TILE_OVERLAP, float) + assert isinstance(config.HEATMAP_CELL_DIV, int) + assert isinstance(config.HEATMAP_CLUSTER_DILATE, int) + assert isinstance(config.HEATMAP_MAX_AREA_FRAC, float) + + +def test_tiling_defaults_are_sane(): + # 默认启用瓦片规划；可读下限为正；成本上界为正 + assert config.TILING_ENABLED is True + assert config.TEXT_HEIGHT_MM > 0 + assert config.TEXT_READABLE_PX > 0 + assert config.LINE_READABLE_PX > 0 + assert config.TILE_MAX_COUNT > 0 + assert config.HEATMAP_CELL_DIV > 0 + + def test_credibility_defaults_are_on(): # 默认开启，保证默认拿到可信度收益 assert config.DETERMINISTIC_CHECKS is True diff --git a/tests/test_evidence.py b/tests/test_evidence.py index 47af502..0c57abc 100644 --- a/tests/test_evidence.py +++ b/tests/test_evidence.py @@ -1,8 +1,17 @@ """`dwg_review.phases.evidence` 单元测试。""" +import base64 +import io + import pytest -from dwg_review.phases.evidence import bbox_to_window, collect_frame +from dwg_review.phases.evidence import ( + _is_truncated, + _plan_tiles, + _text_boxes, + bbox_to_window, + collect_frame, +) from tests.fakes import FakeMcpClient, frame, make_ctx, ok_image, ok_text @@ -141,3 +150,144 @@ def test_collect_returns_none_for_unsupported_source(): ctx = make_ctx(client=FakeMcpClient()) assert collect(ctx, 1, {"source_type": "pdf", "bbox": [0, 0, 1, 1]}) is None + + +# --------------------------------------------------------------------------- +# 自适应瓦片规划 +# --------------------------------------------------------------------------- +@pytest.fixture +def tiling_env(monkeypatch): + """把瓦片相关配置钉死，使用例不受外部环境变量影响。""" + from dwg_review import config + + for name, value in { + "TILING_ENABLED": True, + "TEXT_HEIGHT_MM": 2.0, + "TEXT_READABLE_PX": 12, + "LINE_READABLE_PX": 6, + "TILE_MAX_COUNT": 36, + "TILE_OVERLAP": 0.0, + "HEATMAP_CELL_DIV": 32, + "HEATMAP_CLUSTER_DILATE": 0, + "HEATMAP_MAX_AREA_FRAC": 0.5, + "IMAGE_MAX_EDGE": 1280, + "EVIDENCE_MAX_IMAGES": 5, + "EVIDENCE_DETAIL_TILES": 4, + }.items(): + monkeypatch.setattr(config, name, value) + return config + + +def _png_b64(w=2000, h=1414): + from PIL import Image + + buf = io.BytesIO() + Image.new("RGB", (w, h), "black").save(buf, format="PNG") + return base64.b64encode(buf.getvalue()).decode("ascii") + + +def _tiling_client(b64, entity_type="LINE"): + return FakeMcpClient( + responses={ + "query_entities": ok_text( + {"total": 1, "items": [{"handle": "A", "type": entity_type, + "bbox": [0, 0, 60000, 20000]}]} + ), + "get_text_content": ok_text({"total": 0, "items": []}), + "get_dimensions": ok_text({"total": 0, "items": []}), + "get_block_references": ok_text({"total": 0, "items": []}), + "get_polylines_info": ok_text({"total": 0, "items": []}), + "get_viewport_screenshot": ok_image(b64), + } + ) + + +def test_is_truncated(): + assert _is_truncated({"total": 5, "items": [1]}) is True + assert _is_truncated({"total": 1, "items": [1]}) is False + assert _is_truncated({"items": [1, 2]}) is False + assert _is_truncated(None, {}, {"total": 3, "items": []}) is True + assert _is_truncated() is False + + +def test_text_boxes_includes_std_and_custom(): + entities = {"items": [ + {"type": "LINE", "bbox": [0, 0, 1, 1]}, + {"type": "AECDLABEL", "bbox": [2, 2, 3, 3]}, + {"type": "TEXT", "bbox": [4, 4, 5, 5]}, + ]} + texts = {"items": [{"bbox": [6, 6, 7, 7]}]} + dimensions = {"items": [{"bbox": [8, 8, 9, 9]}]} + boxes = _text_boxes(entities, texts, dimensions) + for expected in ((2, 2, 3, 3), (4, 4, 5, 5), (6, 6, 7, 7), (8, 8, 9, 9)): + assert tuple(float(v) for v in expected) in boxes + assert (0.0, 0.0, 1.0, 1.0) not in boxes # 普通线不算文字 + + +def test_plan_tiles_degrades(tiling_env): + win = {"min_x": 0.0, "min_y": 0.0, "max_x": 100.0, "max_y": 50.0} + f = frame([0, 0, 100, 50], sheet_size="A0", sheet_scale=100.0) + boxes = [(0.0, 0.0, 10.0, 10.0)] + assert _plan_tiles(win, f, boxes, [], False, (100, 50)) != [] # 正常路径非空 + assert _plan_tiles(win, None, boxes, [], False, (100, 50)) == [] # 缺图框 + assert _plan_tiles(win, f, [], [], False, (100, 50)) == [] # 无实体 + assert _plan_tiles(win, f, boxes, [], False, None) == [] # 无像素 + assert _plan_tiles(win, f, boxes, [], True, (100, 50)) == [] # 数据截断 + # 比例未知（sheet_scale=0 且 sheet_size 不在纸张表）→ 无法规划 + unknown = frame([0, 0, 100, 50], sheet_size="ZZ", sheet_scale=0.0) + assert _plan_tiles(win, unknown, boxes, [], False, (100, 50)) == [] + + +def test_plan_tiles_disabled_by_config(tiling_env, monkeypatch): + tiling_env.TILING_ENABLED = False + win = {"min_x": 0.0, "min_y": 0.0, "max_x": 100.0, "max_y": 50.0} + f = frame([0, 0, 100, 50], sheet_size="A0", sheet_scale=100.0) + assert _plan_tiles(win, f, [(0.0, 0.0, 10.0, 10.0)], [], False, (100, 50)) == [] + + +def test_capture_images_tiles_line_cluster(tiling_env): + """线条簇：单张可覆盖 → 2 张瓦片（行优先）。""" + pytest.importorskip("PIL") + ctx = make_ctx(client=_tiling_client(_png_b64())) + ev = collect_frame( + ctx, 1, frame([0, 0, 120000, 40000], sheet_size="A0", sheet_scale=100.0) + ) + assert [i["label"] for i in ev["images"]] == ["overall", "tile-1", "tile-2"] + + +def test_capture_images_tiles_text_cluster_is_finer(tiling_env): + """文字簇用更细尺度 → 瓦片更多，并受 EVIDENCE_MAX_IMAGES 截断。""" + pytest.importorskip("PIL") + ctx = make_ctx(client=_tiling_client(_png_b64(), entity_type="AECDLABEL")) + ev = collect_frame( + ctx, 1, frame([0, 0, 120000, 40000], sheet_size="A0", sheet_scale=100.0) + ) + labels = [i["label"] for i in ev["images"]] + assert labels == ["overall", "tile-1", "tile-2", "tile-3", "tile-4"] # 上限 5 张 + + +def test_capture_images_falls_back_without_scale(tiling_env): + """缺 sheet_size/sheet_scale → 退回固定 2x2 象限。""" + pytest.importorskip("PIL") + ctx = make_ctx(client=_tiling_client(_png_b64())) + ev = collect_frame(ctx, 1, frame([0, 0, 120000, 40000], sheet_size="ZZ")) + assert [i["label"] for i in ev["images"]] == [ + "overall", "tile-1", "tile-2", "tile-3", "tile-4", + ] + + +def test_capture_images_falls_back_when_truncated(tiling_env): + """实体被截断 → 禁用内容过滤，退回固定策略。""" + pytest.importorskip("PIL") + client = _tiling_client(_png_b64()) + client.responses["query_entities"] = ok_text( + {"total": 500, "items": [{"handle": "A", "type": "LINE", + "bbox": [0, 0, 60000, 20000]}]} + ) + ctx = make_ctx(client=client) + ev = collect_frame( + ctx, 1, frame([0, 0, 120000, 40000], sheet_size="A0", sheet_scale=100.0) + ) + assert [i["label"] for i in ev["images"]] == [ + "overall", "tile-1", "tile-2", "tile-3", "tile-4", + ] diff --git a/tests/test_llm_client.py b/tests/test_llm_client.py index f4e554c..73886d5 100644 --- a/tests/test_llm_client.py +++ b/tests/test_llm_client.py @@ -186,3 +186,31 @@ def test_downsample_bad_base64_returns_original(): pytest.importorskip("PIL") bad = base64.b64encode(b"this is not an image").decode("ascii") assert downsample_image_b64(bad, "image/png", max_edge=10) == (bad, "image/png") + + +# --------------------------------------------------------------------------- +# 图片尺寸探测（瓦片规划用） +# --------------------------------------------------------------------------- +def test_image_size_returns_none_without_pil(monkeypatch): + monkeypatch.setattr(llm_module, "HAS_PIL", False) + assert llm_module.image_size_b64("QUJD") is None + + +def test_image_size_returns_none_on_empty(): + assert llm_module.image_size_b64("") is None + + +def test_image_size_returns_none_on_bad_data(): + pytest.importorskip("PIL") + bad = base64.b64encode(b"not an image").decode("ascii") + assert llm_module.image_size_b64(bad) is None + + +def test_image_size_reads_dimensions(): + pytest.importorskip("PIL") + from PIL import Image + + buf = io.BytesIO() + Image.new("RGB", (321, 123), "white").save(buf, format="PNG") + b64 = base64.b64encode(buf.getvalue()).decode("ascii") + assert llm_module.image_size_b64(b64) == (321, 123) diff --git a/tests/test_tiling.py b/tests/test_tiling.py new file mode 100644 index 0000000..c24d9fb --- /dev/null +++ b/tests/test_tiling.py @@ -0,0 +1,208 @@ +"""`dwg_review.tiling` 单元测试（纯函数，不依赖 CAD / LLM / Pillow）。""" + +from dwg_review import tiling as tl + + +def _frame(bbox, sheet_size="A0", sheet_scale=0.0): + return {"bbox": list(bbox), "sheet_size": sheet_size, "sheet_scale": sheet_scale} + + +def _cluster(bbox, count=10): + return {"bbox": tuple(float(v) for v in bbox), "cells": [], "count": count} + + +# --- norm_box / frame_bbox -------------------------------------------------- + +def test_norm_box_valid_and_invalid(): + assert tl.norm_box([0, 0, 10, 20]) == (0.0, 0.0, 10.0, 20.0) + assert tl.norm_box([10, 20, 0, 0]) == (0.0, 0.0, 10.0, 20.0) + assert tl.norm_box([1, 2, 3]) is None + assert tl.norm_box(None) is None + assert tl.norm_box([0, 0, float("nan"), 1]) is None + + +def test_frame_bbox_rejects_degenerate_and_non_dict(): + assert tl.frame_bbox(_frame((0, 0, 100, 50))) == (0.0, 0.0, 100.0, 50.0) + assert tl.frame_bbox(_frame((0, 0, 0, 50))) is None # 零宽 + assert tl.frame_bbox(_frame((0, 0, 100, 0))) is None # 零高 + assert tl.frame_bbox({"bbox": [1, 2, 3]}) is None + assert tl.frame_bbox(None) is None + + +def test_frame_world_size(): + assert tl.frame_world_size(_frame((0, 0, 100, 50))) == (100.0, 50.0) + assert tl.frame_world_size(_frame((0, 0, 0, 0))) == (0.0, 0.0) + + +# --- frame_scale：三条路径 -------------------------------------------------- + +def test_frame_scale_prefers_sheet_scale(): + assert tl.frame_scale(_frame((0, 0, 100, 50), sheet_scale=100.0)) == 100.0 + + +def test_frame_scale_derived_landscape(): + # A0 横放：世界 118900 x 84100 → 118900/1189 = 100 + f = _frame((0, 0, 118900, 84100), sheet_size="A0") + assert tl.frame_scale(f) == 100.0 + + +def test_frame_scale_derived_portrait(): + # A0 纵放：世界宽 ↔ 纸高 841 + f = _frame((0, 0, 84100, 118900), sheet_size="A0") + assert tl.frame_scale(f) == 100.0 + + +def test_frame_scale_falls_back_to_default(): + assert tl.frame_scale(_frame((0, 0, 100, 50), sheet_size="ZZ"), default=7.0) == 7.0 + assert tl.frame_scale(_frame((0, 0, 0, 0)), default=7.0) == 7.0 + assert tl.frame_scale(None, default=7.0) == 7.0 + + +# --- effective_px / world_per_px ------------------------------------------- + +def test_effective_px_no_resize_when_small(): + assert tl.effective_px((800, 600), 1280) == (800, 600) + + +def test_effective_px_downsamples_long_edge(): + # 实测 CAD 绘图区 2110x462 → 长边缩到 1280 + assert tl.effective_px((2110, 462), 1280) == (1280, 280) + + +def test_effective_px_invalid(): + assert tl.effective_px(None, 1280) == (0, 0) + assert tl.effective_px((0, 100), 1280) == (0, 0) + assert tl.effective_px((100, 100), 0) == (100, 100) # 0 = 不缩放 + + +def test_world_per_px(): + # 1:100、纸面 2mm、VLM 需 10px → 每像素 20 图形单位 + assert tl.world_per_px(100.0, 2.0, 10.0) == 20.0 + assert tl.world_per_px(0, 2.0, 10.0) == 0.0 + assert tl.world_per_px(100.0, 2.0, 0) == 0.0 + + +# --- is_text_type ----------------------------------------------------------- + +def test_is_text_type_standard(): + for t in ("TEXT", "MTEXT", "ATTRIB", "ATTDEF", "DIMENSION", "text"): + assert tl.is_text_type(t) is True + + +def test_is_text_type_custom_keywords(): + for t in ("AECDLABEL", "AECDDIMTEXT", "MYNOTE", "PARTTAG", "ANNOTATION", "TITLEBLK"): + assert tl.is_text_type(t) is True + + +def test_is_text_type_negative(): + for t in ("AECDBEWPIPE", "LINE", "LWPOLYLINE", "INSERT", "HATCH", "", None, 123): + assert tl.is_text_type(t) is False + + +# --- 相交 / 文字簇标记 ------------------------------------------------------- + +def test_boxes_intersect(): + assert tl.boxes_intersect((0, 0, 10, 10), (5, 5, 15, 15)) is True + assert tl.boxes_intersect((0, 0, 10, 10), (10, 10, 20, 20)) is True # 接触算相交 + assert tl.boxes_intersect((0, 0, 10, 10), (11, 11, 20, 20)) is False + assert tl.boxes_intersect(None, (0, 0, 1, 1)) is False + + +def test_cluster_text_flags(): + clusters = [_cluster((0, 0, 10, 10)), _cluster((100, 100, 110, 110))] + flags = tl.cluster_text_flags(clusters, [(5, 5, 6, 6)]) + assert flags == [True, False] + assert tl.cluster_text_flags(clusters, []) == [False, False] + assert tl.cluster_text_flags([], [(0, 0, 1, 1)]) == [] + + +# --- tiles_for_box / subdivide --------------------------------------------- + +def test_tiles_for_box_fits_single(): + assert tl.tiles_for_box((0, 0, 1000, 500), 1000, 500) == (1, 1) + + +def test_tiles_for_box_dense(): + assert tl.tiles_for_box((0, 0, 2000, 1000), 1000, 500) == (2, 2) + + +def test_tiles_for_box_overlap_increases_count(): + assert tl.tiles_for_box((0, 0, 1000, 500), 1000, 500, overlap=0.1) == (2, 2) + + +def test_tiles_for_box_invalid(): + assert tl.tiles_for_box(None, 10, 10) == (1, 1) + assert tl.tiles_for_box((0, 0, 0, 0), 10, 10) == (1, 1) + assert tl.tiles_for_box((0, 0, 10, 10), 0, 10) == (1, 1) + + +def test_subdivide_row_major_top_first(): + wins = tl.subdivide((0, 0, 100, 100), 2, 2) + assert len(wins) == 4 + assert wins[0] == {"min_x": 0.0, "min_y": 50.0, "max_x": 50.0, "max_y": 100.0} + assert wins[1] == {"min_x": 50.0, "min_y": 50.0, "max_x": 100.0, "max_y": 100.0} + assert wins[2] == {"min_x": 0.0, "min_y": 0.0, "max_x": 50.0, "max_y": 50.0} + assert wins[3] == {"min_x": 50.0, "min_y": 0.0, "max_x": 100.0, "max_y": 50.0} + + +def test_subdivide_overlap_expands_then_clamps(): + wins = tl.subdivide((0, 0, 100, 100), 2, 2, overlap=0.1, frame_win=(0, 0, 100, 100)) + # 每格 50x50，外扩 5 → 左上格：x 0..55（左被夹到 0），y 45..100（上被夹到 100） + assert wins[0] == {"min_x": 0.0, "min_y": 45.0, "max_x": 55.0, "max_y": 100.0} + + +# --- plan_regions ----------------------------------------------------------- + +def test_plan_regions_sparse_single_tile(): + frame = _frame((0, 0, 1000, 500)) + clusters = [_cluster((100, 100, 400, 300))] + wins = tl.plan_regions(frame, clusters, s_line=10.0, view_px=(100, 50)) + assert len(wins) == 1 + assert wins[0]["min_x"] <= 100 and wins[0]["max_x"] >= 400 + + +def test_plan_regions_dense_splits(): + frame = _frame((0, 0, 2000, 1000)) + clusters = [_cluster((0, 0, 2000, 1000))] + wins = tl.plan_regions(frame, clusters, s_line=10.0, view_px=(100, 50)) + assert len(wins) == 4 # 2x2 + + +def test_plan_regions_text_uses_finer_scale(): + frame = _frame((0, 0, 1000, 500)) + clusters = [_cluster((0, 0, 1000, 500))] + line_only = tl.plan_regions(frame, clusters, s_line=10.0, view_px=(100, 50)) + text = tl.plan_regions(frame, clusters, text_clusters={0}, s_text=5.0, + s_line=10.0, view_px=(100, 50)) + assert len(line_only) == 1 + assert len(text) == 4 # 半径减半 → 2x2 + + +def test_plan_regions_text_cluster_prioritized_on_truncation(): + frame = _frame((0, 0, 4000, 2000)) + clusters = [_cluster((0, 0, 2000, 1000), count=99), # 非文字，排前面（输入序） + _cluster((2000, 1000, 4000, 2000), count=1)] # 文字 + wins = tl.plan_regions(frame, clusters, text_clusters={1}, + s_text=10.0, s_line=10.0, view_px=(100, 50), max_tiles=2) + assert len(wins) == 2 + # 文字簇（下标 1，位于右上）应优先出图 + assert wins[0]["min_x"] >= 2000 and wins[0]["min_y"] >= 1000 + + +def test_plan_regions_degrades(): + frame = _frame((0, 0, 1000, 500)) + clusters = [_cluster((0, 0, 1000, 500))] + assert tl.plan_regions(frame, clusters, s_line=0.0, view_px=(100, 50)) == [] + assert tl.plan_regions(frame, clusters, s_line=10.0, view_px=(0, 0)) == [] + assert tl.plan_regions({"bbox": [0, 0, 0, 0]}, clusters, s_line=10.0, + view_px=(100, 50)) == [] + assert tl.plan_regions(frame, [], s_line=10.0, view_px=(100, 50)) == [] + + +def test_plan_regions_clamps_to_frame(): + frame = _frame((0, 0, 1000, 500)) + clusters = [_cluster((0, 0, 2000, 1000))] # 超出图框 + wins = tl.plan_regions(frame, clusters, s_line=10.0, view_px=(100, 50), max_tiles=4) + for w in wins: + assert w["min_x"] >= 0 and w["max_x"] <= 1000 + assert w["min_y"] >= 0 and w["max_y"] <= 500 diff --git a/verify_vlm_resolution_offline.py b/verify_vlm_resolution_offline.py new file mode 100644 index 0000000..cba4549 --- /dev/null +++ b/verify_vlm_resolution_offline.py @@ -0,0 +1,415 @@ +"""离线标定 VLM 读图下限（不碰 CAD，纯 PIL 合成）。 + +目的：在接入自适应瓦片规划前，先离线测出视觉模型（deepseek-flash）的 +「最小可读文字像素高 `R_text`」与「最小可分辨线间距 `R_line`」，为 `tiling.py` +的 `world_per_px` 提供**区间参考**。 + +做法：PIL 合成"类图纸"图（可控背景极性），**精确控制像素字高**（先用 +`font.getbbox` 二分出目标 ink 高度对应的字号），加可控间距/线宽的平行线， +以及实线 / 虚线对比。每档发 VLM，提示词强约束 JSON，与真值比对。 + +重要前提：**合成图比真实 CAD 渲染干净 → 结论偏乐观**，只作区间；最终以 +`verify_text_resolution.py`（CAD 临时加字实测）为准。 + +用法： + python verify_vlm_resolution_offline.py --dry-run # 只出图，不调 VLM + python verify_vlm_resolution_offline.py --repeat 2 # 每档重复 2 次 + python verify_vlm_resolution_offline.py --heights 4,6,8,10,12 --spacings 2,3,4,6,8 + python verify_vlm_resolution_offline.py --polarity light # 白底黑线 + +前置：需要 Pillow；调 VLM 需要 DEEPSEEK_API_KEY（--dry-run 不需要密钥）。 +""" + +import argparse +import base64 +import io +import json +import math +import random +import re +from pathlib import Path + +from dwg_review import config +from dwg_review.llm_client import HAS_PIL, LlmClient, downsample_image_b64 + +# 画布宽固定为 1280（= IMAGE_MAX_EDGE），使发给 VLM 前**不发生下采样**， +# 与真实管线「截图长边缩到 1280」后的条件一致。 +_CANVAS_W = 1280 +_MIN_CANVAS_H = 280 + +# 易混字符集：去掉 I / O，避免"真值本身歧义"污染标定 +_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" + +_DARK = {"bg": (0, 0, 0), "fg": (255, 255, 255)} +_LIGHT = {"bg": (255, 255, 255), "fg": (0, 0, 0)} +_POLARITY = {"dark": _DARK, "light": _LIGHT} + +_FONT_CANDIDATES = ( + r"C:\Windows\Fonts\consola.ttf", # 等宽，字高最稳 + r"C:\Windows\Fonts\cour.ttf", + r"C:\Windows\Fonts\lucon.ttf", + r"C:\Windows\Fonts\arial.ttf", + r"/usr/share/fonts/truetype/dejavu/DejaVuSansMono.ttf", + r"/System/Library/Fonts/Menlo.ttc", +) + +_PROBE = "K7Q2M9" # 全为大写/数字，ink 高度一致，用于二分字号 + +_SYS_TEXT = ( + "你是图像文字识别器。只输出一个 JSON 对象，不要任何解释、不要 markdown 代码块。" +) +_SYS_LINE = ( + "你是工程图纸线型识别器。只输出一个 JSON 对象，不要任何解释、不要 markdown 代码块。" +) + + +# --------------------------------------------------------------------------- +# 字体与像素字高 +# --------------------------------------------------------------------------- +def find_font_path(explicit=None): + if explicit: + p = Path(explicit) + if p.exists(): + return str(p) + raise SystemExit(f"指定的字体不存在：{explicit}") + for cand in _FONT_CANDIDATES: + if Path(cand).exists(): + return cand + raise SystemExit("未找到可用 TTF 字体，请用 --font 指定") + + +def font_for_ink_height(target_h, font_path, probe=_PROBE): + """二分出使 `probe` 的 ink 高度最接近 target_h 的字体。 + + 返回 (font, size, actual_h)。ink 高度 = getbbox 的 (bottom - top)。 + """ + from PIL import ImageFont + + target_h = max(1, int(target_h)) + lo, hi = 1, 400 + best = None # (font, size, h) + while lo <= hi: + mid = (lo + hi) // 2 + font = ImageFont.truetype(font_path, mid) + bb = font.getbbox(probe) + h = bb[3] - bb[1] + if best is None or abs(h - target_h) < abs(best[2] - target_h): + best = (font, mid, h) + if h == target_h: + break + if h < target_h: + lo = mid + 1 + else: + hi = mid - 1 + return best + + +# --------------------------------------------------------------------------- +# 合成图 +# --------------------------------------------------------------------------- +def _canvas(h, polarity): + from PIL import Image + + pal = _POLARITY[polarity] + return Image.new("RGB", (_CANVAS_W, max(_MIN_CANVAS_H, int(h))), pal["bg"]) + + +def synth_text_image(height_px, tokens, font_path, *, polarity="dark", left=90): + """合成 N 行等高文字（每行一个随机串）。返回 (image, actual_h)。""" + from PIL import ImageDraw + + font, _size, actual_h = font_for_ink_height(height_px, font_path) + n = len(tokens) + pitch = max(actual_h + 3, int(round(actual_h * 3))) + block_h = pitch * (n - 1) + actual_h + canvas_h = block_h + 4 * actual_h + 60 + img = _canvas(canvas_h, polarity) + draw = ImageDraw.Draw(img) + fg = _POLARITY[polarity]["fg"] + + y_top = (img.height - block_h) // 2 + for i, tok in enumerate(tokens): + bb = font.getbbox(tok) + # getbbox 的 top 可能为负（字体 ascent 内留白），补偿使 ink 顶对齐 y_top + draw.text((left, y_top - bb[1]), tok, font=font, fill=fg) + y_top += pitch + return img, actual_h + + +def synth_lines_image(spacing_px, count, *, width=1, polarity="dark", + x0=200, x1=1080, dashed=False, dash=(6, 4)): + """合成 count 条水平平行线，间距 spacing_px，线宽 width。返回 image。""" + from PIL import ImageDraw + + spacing_px = max(1, int(spacing_px)) + block_h = (count - 1) * spacing_px + canvas_h = max(block_h + 8 * max(spacing_px, 4), 2 * block_h + 60) + img = _canvas(canvas_h, polarity) + draw = ImageDraw.Draw(img) + fg = _POLARITY[polarity]["fg"] + + y0 = (img.height - block_h) // 2 + for i in range(count): + y = y0 + i * spacing_px + if dashed: + _draw_dashed(draw, x0, x1, y, dash[0], dash[1], fg, width) + else: + draw.line([(x0, y), (x1, y)], fill=fg, width=width) + return img + + +def _draw_dashed(draw, x0, x1, y, on, off, color, width): + x = x0 + while x < x1: + draw.line([(x, y), (min(x + on, x1), y)], fill=color, width=width) + x += on + off + + +def _to_b64(img, max_edge=None): + buf = io.BytesIO() + img.save(buf, format="PNG") + b64 = base64.b64encode(buf.getvalue()).decode("ascii") + return downsample_image_b64(b64, "image/png", max_edge=max_edge) + + +# --------------------------------------------------------------------------- +# VLM 交互 +# --------------------------------------------------------------------------- +def _extract_json(text): + """从模型输出里抠出第一个 JSON 对象（容忍 ```json 包裹与前后噪声）。""" + if not text: + return None + m = re.search(r"\{.*\}", text, re.S) + if not m: + return None + try: + return json.loads(m.group(0)) + except json.JSONDecodeError: + return None + + +def ask_vlm(client, img, system, prompt, *, max_edge=None): + """发一张图问 VLM，返回解析后的 dict（失败返回 None）。""" + b64, mime = _to_b64(img, max_edge) + resp = client.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 _norm_token(s): + return re.sub(r"[^A-Z0-9]", "", str(s).upper()) + + +def _lenient(s): + """宽松归一：常见混淆 0<->O、1<->I<->L。""" + s = _norm_token(s) + return s.replace("0", "O").replace("1", "I").replace("L", "I") + + +# --------------------------------------------------------------------------- +# 各子测试 +# --------------------------------------------------------------------------- +def run_text_level(client, height_px, font_path, polarity, repeat, out_dir, dry_run): + rng = random.Random(1000 + height_px) + tokens = ["".join(rng.choice(_ALPHABET) for _ in range(6)) for _ in range(5)] + img, actual_h = synth_text_image(height_px, tokens, font_path, polarity=polarity) + png = out_dir / f"text_h{height_px}_{polarity}.png" + img.save(str(png)) + + prompt = ( + "图中是若干行**等高的白色字符**（黑底）。请逐行读出，按从上到下顺序输出 JSON：" + '{"texts":["第一行","第二行",...]}。无法辨认的行填空字符串。' + ) + if polarity == "light": + prompt = prompt.replace("白色字符**（黑底）", "黑色字符**（白底）") + + trials = [] + if not dry_run: + for _ in range(max(1, repeat)): + data = ask_vlm(client, img, _SYS_TEXT, prompt) + got = list(data.get("texts") or []) if isinstance(data, dict) else [] + got = [str(x) for x in got] + exact = sum(1 for i, t in enumerate(tokens) if i < len(got) and _norm_token(got[i]) == t) + lenient = sum(1 for i, t in enumerate(tokens) if i < len(got) and _lenient(got[i]) == _lenient(t)) + trials.append({"got": got, "exact": exact, "lenient": lenient}) + + n = len(tokens) + best_exact = max((t["exact"] for t in trials), default=0) + return { + "height_px": height_px, "actual_h": actual_h, "tokens": tokens, + "trials": trials, "n": n, "best_exact": best_exact, + "readable": best_exact == n, # 全部行都对 → 可读 + "mostly": best_exact >= n - 1, # 容错 1 行 + "png": png.name, + } + + +def _ask_count(client, img, prompt): + data = ask_vlm(client, img, _SYS_LINE, prompt) + if not isinstance(data, dict): + return None + try: + return int(data.get("count")) + except (TypeError, ValueError): + return None + + +def run_line_level(client, spacing, polarity, repeat, out_dir, dry_run): + """1 vs 2 **二选一**（比计数稳健得多）：间距 spacing 的两条细线 vs 一条等宽粗线。 + + 实测该模型**无法可靠计数**（6 条线在 2~10px 间距下答 4/10/4/5/4/4）， + 故线条判据只测「相邻线能否被分辨开」这一最核心的细节能力。 + """ + pair = synth_lines_image(spacing, 2, width=1, polarity=polarity) + single = synth_lines_image(20, 1, width=max(1, int(spacing)), polarity=polarity) + pair_png = out_dir / f"pair_s{spacing}_{polarity}.png" + single_png = out_dir / f"single_s{spacing}_{polarity}.png" + pair.save(str(pair_png)) + single.save(str(single_png)) + + prompt = ( + "图中是**水平直线**（黑底白线）。只判断线条数量是 1 还是 2，" + '输出 JSON：{"count": 1 或 2}。' + ) + if polarity == "light": + prompt = prompt.replace("黑底白线", "白底黑线") + + trials = [] + if not dry_run: + for _ in range(max(1, repeat)): + g_pair = _ask_count(client, pair, prompt) + g_single = _ask_count(client, single, prompt) + trials.append({"pair": g_pair, "single": g_single, + "ok": g_pair == 2 and g_single == 1}) + return { + "spacing": spacing, "truth_pair": 2, "truth_single": 1, + "trials": trials, "resolved": any(t["ok"] for t in trials), + "png_pair": pair_png.name, "png_single": single_png.name, + } + + +def run_dashed_test(client, polarity, repeat, out_dir, dry_run): + """虚线 vs 实线识别（两条线，避开计数干扰）。""" + out = [] + for dashed in (False, True): + img = synth_lines_image(12, 2, width=1, polarity=polarity, + dashed=dashed, dash=(8, 6)) + png = out_dir / f"dashed_{'on' if dashed else 'off'}_{polarity}.png" + img.save(str(png)) + prompt = ( + "图中是两条**水平直线**（黑底白线）。判断它们是否为虚线，" + '输出 JSON：{"dashed": true 或 false}。' + ) + if polarity == "light": + prompt = prompt.replace("黑底白线", "白底黑线") + trials = [] + if not dry_run: + for _ in range(max(1, repeat)): + data = ask_vlm(client, img, _SYS_LINE, prompt) + got = data.get("dashed") if isinstance(data, dict) else None + if isinstance(got, str): + got = got.strip().lower() in ("true", "1", "yes", "是") + trials.append({"got_dashed": got, "ok": got is dashed}) + out.append({"truth_dashed": dashed, "trials": trials, + "correct": any(t["ok"] for t in trials), "png": png.name}) + return out + + +# --------------------------------------------------------------------------- +# 主流程 +# --------------------------------------------------------------------------- +def _parse_ints(s): + return [int(x) for x in str(s).split(",") if x.strip()] + + +def main(argv=None): + ap = argparse.ArgumentParser() + ap.add_argument("--heights", default="4,6,8,10,12,16,20,24", help="文字像素高扫描档") + ap.add_argument("--spacings", default="2,3,4,5,6,8,10", help="线间距(px)扫描档") + ap.add_argument("--polarity", choices=("dark", "light", "both"), default="dark", + help="背景极性：dark=黑底白线（CAD 模型空间默认），light=白底黑线") + ap.add_argument("--repeat", type=int, default=1, help="每档重复次数（取多数/任一成功）") + ap.add_argument("--font", default=None, help="TTF 字体路径（默认自动查找等宽字体）") + ap.add_argument("--max-edge", type=int, default=None, help="发给 VLM 前长边上限（默认 config）") + ap.add_argument("--dry-run", action="store_true", help="只合成图并落盘，不调用 VLM") + ap.add_argument("--out", default="reports/_probe/vlm_calib") + args = ap.parse_args(argv) + + if not HAS_PIL: + raise SystemExit("离线标定需要 Pillow：pip install pillow") + + font_path = find_font_path(args.font) + out_dir = Path(config.PROJECT_ROOT) / args.out + out_dir.mkdir(parents=True, exist_ok=True) + + polarities = ("dark", "light") if args.polarity == "both" else (args.polarity,) + client = None + if not args.dry_run: + client = LlmClient() + print(f"模型：{client.model} 字体：{font_path}") + + heights = _parse_ints(args.heights) + spacings = _parse_ints(args.spacings) + results = {"polarity": list(polarities), "text": {}, "line": {}, "dashed": {}} + + for pol in polarities: + print(f"\n===== 极性 {pol} =====") + + print("-- 文字可读性 --") + text_rows = [] + for h in heights: + row = run_text_level(client, h, font_path, pol, args.repeat, out_dir, args.dry_run) + text_rows.append(row) + if args.dry_run: + print(f" 字高 {h:>3}px (实际 {row['actual_h']}px) -> {row['png']}") + else: + print(f" 字高 {h:>3}px (实际 {row['actual_h']}px) 命中 " + f"{row['best_exact']}/{row['n']} 可读={row['readable']}") + results["text"][pol] = text_rows + + print("-- 线间距可分辨性（1 vs 2 二选一） --") + line_rows = [] + for s in spacings: + row = run_line_level(client, s, pol, args.repeat, out_dir, args.dry_run) + line_rows.append(row) + if args.dry_run: + print(f" 间距 {s:>3}px -> {row['png_pair']} / {row['png_single']}") + else: + print(f" 间距 {s:>3}px 可分辨={row['resolved']} {row['trials']}") + results["line"][pol] = line_rows + + print("-- 虚线识别 --") + d = run_dashed_test(client, pol, args.repeat, out_dir, args.dry_run) + results["dashed"][pol] = d + if not args.dry_run: + for item in d: + print(f" 虚线={item['truth_dashed']} 识别正确={item['correct']}") + + # 汇总（取两极性中更优者作为"区间参考"） + summary = {} + for pol in polarities: + text_rows = results["text"].get(pol, []) + line_rows = results["line"].get(pol, []) + r_text = next((r["height_px"] for r in text_rows if r["readable"]), None) + r_text_mostly = next((r["height_px"] for r in text_rows if r["mostly"]), None) + r_line = next((r["spacing"] for r in line_rows if r["resolved"]), None) + summary[pol] = {"R_text": r_text, "R_text_mostly": r_text_mostly, "R_line": r_line} + print(f"\n[{pol}] R_text(全部行正确)={r_text} R_text(容错1行)={r_text_mostly} " + f"R_line={r_line}") + + results["summary"] = summary + (out_dir / "summary.json").write_text( + json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8" + ) + print(f"\n输出目录: {out_dir}") + print("注意：合成图偏干净，结论偏乐观，仅作区间；以 CAD 实测 (verify_text_resolution.py) 为准。") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())