feat(tiling): 自适应瓦片规划器 + VLM 分辨率离线标定

背景:逐图框审查固定「整体 + 2x2 象限」截图,大图幅(A0/A1)读不清细节。
改为按**实体分布**自适应决定截几张、截哪里。

- 新增 dwg_review/tiling.py(纯函数):
  SHEET_SIZES_MM / frame_world_size / frame_scale(sheet_scale>0 用之,否则由
  bbox/纸张反算,否则 default)/ effective_px(与 downsample 同规则)/
  world_per_px / is_text_type(标准名 + 自定义类名含 TEXT/LABEL/DIM/NOTE/TAG/
  ANNO/TITLE)/ cluster_text_flags / tiles_for_box / subdivide / plan_regions
  (文字簇优先、空白区不出图、max_tiles 截断、窗口夹到图框范围)
- 新增 verify_vlm_resolution_offline.py:PIL 合成类图纸图(精确像素字高 +
  可控线间距/线宽/虚线)→ 问 VLM 强约束 JSON → 定 R_text / R_line。
  实测:R_text=12px(黑底;白底 10px);R_line=6px。
  另:实测该模型**无法可靠计数**(6 条线在 2~10px 间距下答 4/10/4/5/4/4),
  故线条判据改用「1 条 vs 2 条」二选一。
- config 新增 TILING_ENABLED / TEXT_HEIGHT_MM / TEXT_READABLE_PX=12 /
  LINE_READABLE_PX=6 / TILE_MAX_COUNT / TILE_OVERLAP / HEATMAP_CELL_DIV /
  HEATMAP_CLUSTER_DILATE / HEATMAP_MAX_AREA_FRAC;EVIDENCE_DETAIL_TILES 与
  EVIDENCE_MAX_IMAGES 保留作回退路径参数
- llm_client 新增 image_size_b64(测量绘图区实际像素,决定单张覆盖范围)
- phases/evidence.capture_images 支持瓦片规划:先截整体图并测像素 →
  plan_regions 成功用规划窗口,否则退回 sub_windows(行为零变化);
  末尾始终复位整体视口。collect_frame 传实体/文字 bbox;任一信封
  total>returned(截断)→ 退回固定策略,避免误删有内容的瓦片
- 测试:tests/test_tiling.py(28 项)、tests/test_evidence.py 新增规划路径与
  降级用例、test_config / test_llm_client 补新常量与 image_size_b64;
  全量 283 项全绿
This commit is contained in:
gjm committed 2026-10-15 23:03:27 +08:00
1 parent 9b8a2329df
commit cba05b0c65
9 files changed
+1246 -10

No files matched your search

+26
View File
@@ -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())
+15
View File
@@ -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 调用入口。"""
+97 -9
View File
@@ -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),
+284
View File
@@ -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]
+22
View File
@@ -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
+151 -1
View File
@@ -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",
]
+28
View File
@@ -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)
+208
View File
@@ -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
+415
View File
@@ -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())