支持直接审查 PDF 与位图图纸,全链路离线(不连 CAD、不连 8080 端口)。 渲染在内存内完成,复用既有墨迹热力图 + 自适应瓦片规划。 - 新增 dwg_review/sources.py - detect_source / plan_file / collect 统一入口 - PdfReader:pypdfium2 惰性导入,页序即图框序,处理 /Rotate 与 crop 边距 - ImageReader:世界单位 = 源像素,snap_box 整数对齐保证窗口与像素严格一致 - ink_grid:纯 PIL 灰度阈值 → 占用网格(不引入 numpy) - ReviewContext.create(connect_cad=False):离线时 client/tools 均为 None - 全链路 ctx.tools is None 守卫:review 只读子循环 / inspect_region / annotator / checks 的窗口类检查 / 绘图区最大化 - checklist.build_system_prompt(allow_tools=False, offline=True):离线提示词 - __main__:位置参数 source,按扩展名自动判定源类型并决定是否连 CAD - config 新增 8 个常量(PDF_MAX_PAGES / 文本最小字数 / 各源缩放上限 / 墨迹阈值等) - tiling.content_predicate 与 heatmap.cell_box 提升为公开函数 验证: - pytest 457 passed - verify_pdf_source.py:PDF 23/23、图片 9/9(不需 CAD),含实测瓦片字高像素与 旋转映射自检(文本区墨迹 0.311 vs 忽略 /Rotate 反向对照 0.145) - 真实 LLM 端到端:20 页 HVAC 图纸前 3 页跑通,结论具体可核验且未编造 handle 已知局限:页边线在墨迹网格里会把整页连成单簇,瓦片常退化为整页象限(CAD 侧 靠 drop_oversized 丢图框块,离线侧暂无等价手段),留待后续打磨。
369 lines
13 KiB
Python
369 lines
13 KiB
Python
"""自适应瓦片规划器:按实体分布把图框切成若干 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 box_to_window(box):
|
||
"""bbox 元组 → window 对象(`zoom_to_window` 的入参);非法返回 None。"""
|
||
b = norm_box(box)
|
||
if b is None:
|
||
return None
|
||
return {"min_x": b[0], "min_y": b[1], "max_x": b[2], "max_y": b[3]}
|
||
|
||
|
||
def window_to_box(win):
|
||
"""window 对象 → bbox 元组;非法返回 None。"""
|
||
if not isinstance(win, dict):
|
||
return None
|
||
return norm_box(
|
||
[win.get("min_x"), win.get("min_y"), win.get("max_x"), win.get("max_y")]
|
||
)
|
||
|
||
|
||
def expand_box(box, frac):
|
||
"""把 bbox 四周按 `frac` 比例外扩;非法输入返回 None。"""
|
||
b = norm_box(box)
|
||
if b is None:
|
||
return None
|
||
try:
|
||
f = float(frac or 0.0)
|
||
except (TypeError, ValueError):
|
||
f = 0.0
|
||
mx, my = (b[2] - b[0]) * f, (b[3] - b[1]) * f
|
||
return (b[0] - mx, b[1] - my, b[2] + mx, b[3] + my)
|
||
|
||
|
||
def visible_box(box, view_px, margin=0.0):
|
||
"""`zoom_to_window(box, margin)` 之后**实际可见**的世界范围。
|
||
|
||
CAD 的 `_.ZOOM _W` 是**等比适配**(不拉伸):指定窗口按视口宽高比放大到刚好装下,
|
||
多余的一维露出更多内容。故可见范围 = 以窗口中心为中心、按 `max(W/ew, H/eh)`
|
||
放大后的矩形。`view_px` 为截图的**原始像素**(未下采样);拿不到时退回
|
||
「窗口外扩 margin」的近似值。
|
||
|
||
用途:审查阶段把「截图上的归一化坐标」准确映射回 WCS(逐级下钻)。
|
||
"""
|
||
b = expand_box(box, margin)
|
||
if b is None:
|
||
return None
|
||
if not view_px or len(view_px) != 2:
|
||
return b
|
||
try:
|
||
ew, eh = int(view_px[0]), int(view_px[1])
|
||
except (TypeError, ValueError):
|
||
return b
|
||
if ew <= 0 or eh <= 0:
|
||
return b
|
||
w, h = b[2] - b[0], b[3] - b[1]
|
||
if w <= 0 or h <= 0:
|
||
return b
|
||
k = max(w / ew, h / eh)
|
||
vw, vh = ew * k, eh * k
|
||
cx, cy = (b[0] + b[2]) / 2.0, (b[1] + b[3]) / 2.0
|
||
return (cx - vw / 2.0, cy - vh / 2.0, cx + vw / 2.0, cy + vh / 2.0)
|
||
|
||
|
||
def content_predicate(boxes):
|
||
"""构造「该窗口内是否有内容」谓词;无 box 返回 None。
|
||
|
||
用于剔除空白瓦片:簇 / 象限的 bbox 只是**包围盒**,L 形或斜向分布会含大片空白。
|
||
这里直接按**内容 bbox** 判断——比先量化成占用格更精确(格子的跨界膨胀会把
|
||
相邻空白瓦片也判成有内容)。接触即算有内容(保守,宁可多拍不漏)。
|
||
|
||
CAD 侧传实体 bbox;PDF / 图片侧传**占用墨迹格**的 bbox(见 `sources.ink_grid`)。
|
||
"""
|
||
boxes = [b for b in (norm_box(x) for x in (boxes or [])) if b]
|
||
if not boxes:
|
||
return None
|
||
|
||
def _has_content(win):
|
||
box = window_to_box(win)
|
||
return box is not None and any(boxes_intersect(box, b) for b in boxes)
|
||
|
||
return _has_content
|
||
|
||
|
||
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, keep=None):
|
||
"""把图框内的簇规划成一组 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` :相邻瓦片重叠比例;
|
||
- `keep` :可选谓词 `f(window) -> bool`,返回 False 的瓦片被丢弃。
|
||
用于按**占用格**剔除簇 bbox 内的空白瓦片(簇 bbox 对 L 形/斜向分布含大片空白)。
|
||
|
||
返回行优先的 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):
|
||
if keep is not None and not keep(win):
|
||
continue # 该瓦片落在空白区,不出图
|
||
regions.append(win)
|
||
if len(regions) >= max_tiles:
|
||
break
|
||
return regions[:max_tiles]
|