"""CAD 基础构件候选识别模块 —— 从 DXF 数据中识别柱子、文字和墙体候选。 设计原则： - 面向对象，每个识别模块独立 - 所有规则可配置 - 保留原始实体 handle 方便追溯 - 不假设图层命名规范，仅做启发式匹配 """ import math import json from pathlib import Path from typing import Optional import ezdxf # ============================================================ # 配置常量（可修改） # ============================================================ # 柱子识别：INSERT 块名中包含以下关键词之一 COLUMN_KEYWORDS = ["柱", "column", "COL"] # 墙体候选：最小线段长度（图纸单位，通常为毫米） WALL_MIN_LENGTH = 500 # 墙体候选：排除的图层关键词（标注、文字等非建筑图层） WALL_EXCLUDE_LAYERS = ["DIM", "TEXT", "DEFPOINTS", "标注", "NOTE", "PUB_DIM", "AXIS"] # 墙体候选：优先包含的图层关键词（空列表 = 不限制） WALL_INCLUDE_LAYERS = [] # ============================================================ # 柱子识别器 # ============================================================ class ColumnDetector: """从 INSERT 实体中识别候选柱子。 规则：块名包含柱/column/COL 等关键词。 """ def __init__(self, keywords: Optional[list[str]] = None): """ Args: keywords: 块名匹配关键词列表，默认使用 COLUMN_KEYWORDS """ self.keywords = keywords or COLUMN_KEYWORDS def detect(self, msp) -> list[dict]: """遍历模型空间，检测柱子候选。 Args: msp: ezdxf 模型空间对象 Returns: list[dict]: 柱子候选列表 """ columns = [] for entity in msp: if entity.dxftype() != "INSERT": continue block_name = entity.dxf.name if self._match(block_name): insert = entity.dxf.insert columns.append({ "name": block_name, "position": { "x": round(insert[0], 4), "y": round(insert[1], 4), }, "source": "INSERT", "handle": entity.dxf.handle, "layer": entity.dxf.layer, "rotation": round(entity.dxf.rotation, 2), "scale": [ round(entity.dxf.xscale, 2), round(entity.dxf.yscale, 2), round(entity.dxf.zscale, 2), ], }) return columns def _match(self, block_name: str) -> bool: """检查块名是否匹配柱子关键词。""" upper = block_name.upper() for kw in self.keywords: if kw.upper() in upper or kw in block_name: return True return False # ============================================================ # 文字提取器 # ============================================================ class TextExtractor: """提取所有 TEXT / MTEXT 实体的文本内容及其坐标。""" def extract(self, msp) -> list[dict]: """提取模型空间中所有文字实体。 Args: msp: ezdxf 模型空间对象 Returns: list[dict]: 文字信息列表 """ texts = [] for entity in msp: dxftype = entity.dxftype() if dxftype not in ("TEXT", "MTEXT"): continue content = entity.plain_text() if dxftype == "MTEXT" else entity.dxf.text insert = entity.dxf.insert if hasattr(entity.dxf, "insert") else None item = { "content": content, "type": dxftype, "handle": entity.dxf.handle, "layer": entity.dxf.layer, } if insert is not None: item["x"] = round(insert[0], 4) item["y"] = round(insert[1], 4) if hasattr(entity.dxf, "height"): item["height"] = round(entity.dxf.height, 2) texts.append(item) return texts # ============================================================ # 墙体候选识别器 # ============================================================ class WallCandidateDetector: """从 LINE / LWPOLYLINE 中识别墙体候选线段。 规则： 1. 线段长度 >= min_length 2. 所在图层不是标注类图层 3. 可选：限制在主要建筑图层 """ def __init__( self, min_length: float = WALL_MIN_LENGTH, exclude_layers: Optional[list[str]] = None, include_layers: Optional[list[str]] = None, ): """ Args: min_length: 最小线段长度阈值 exclude_layers: 要排除的图层关键词列表 include_layers: 仅包含的图层关键词列表（空 = 不限制） """ self.min_length = min_length self.exclude_layers = exclude_layers or WALL_EXCLUDE_LAYERS self.include_layers = include_layers or WALL_INCLUDE_LAYERS def detect(self, msp) -> list[dict]: """检测墙体候选线段。""" candidates = [] for entity in msp: dxftype = entity.dxftype() layer = entity.dxf.layer if self._is_excluded_layer(layer): continue if dxftype == "LINE": start = entity.dxf.start end = entity.dxf.end length = math.dist([start[0], start[1]], [end[0], end[1]]) if length >= self.min_length: candidates.append({ "entity_type": "LINE", "handle": entity.dxf.handle, "layer": layer, "length": round(length, 2), "start": [round(start[0], 4), round(start[1], 4)], "end": [round(end[0], 4), round(end[1], 4)], }) elif dxftype == "LWPOLYLINE": try: pts = entity.get_points() except Exception: continue if len(pts) < 2: continue # 计算总长度和最长段 total_length = 0.0 max_seg_length = 0.0 segments = [] for i in range(len(pts) - 1): seg_len = math.dist([pts[i][0], pts[i][1]], [pts[i + 1][0], pts[i + 1][1]]) total_length += seg_len if seg_len > max_seg_length: max_seg_length = seg_len segments.append({ "start": [round(pts[i][0], 4), round(pts[i][1], 4)], "end": [round(pts[i + 1][0], 4), round(pts[i + 1][1], 4)], "length": round(seg_len, 2), }) # 至少最长段满足阈值才算候选 if max_seg_length >= self.min_length: candidates.append({ "entity_type": "LWPOLYLINE", "handle": entity.dxf.handle, "layer": layer, "vertex_count": len(pts), "closed": entity.closed, "total_length": round(total_length, 2), "max_segment_length": round(max_seg_length, 2), "segments": [s for s in segments if s["length"] >= self.min_length], "points": [[round(p[0], 4), round(p[1], 4)] for p in pts], }) return candidates def _is_excluded_layer(self, layer_name: str) -> bool: """检查图层是否属于排除范围（标注类图层）。""" upper = layer_name.upper() for kw in self.exclude_layers: if kw.upper() in upper or kw in layer_name: return True # 如果配置了 include 列表，则仅包含匹配的图层 if self.include_layers: for kw in self.include_layers: if kw.upper() in upper or kw in layer_name: return False return True # 不在 include 列表中则排除 return False # ============================================================ # 主检测器（编排） # ============================================================ class ElementDetector: """编排各识别器，输出统一的检测结果。""" def __init__( self, column_keywords: Optional[list[str]] = None, wall_min_length: float = WALL_MIN_LENGTH, wall_exclude_layers: Optional[list[str]] = None, wall_include_layers: Optional[list[str]] = None, ): self.column_detector = ColumnDetector(keywords=column_keywords) self.text_extractor = TextExtractor() self.wall_detector = WallCandidateDetector( min_length=wall_min_length, exclude_layers=wall_exclude_layers, include_layers=wall_include_layers, ) def detect_all(self, doc) -> dict: """执行全部识别，返回结构化结果。 Args: doc: ezdxf Drawing 对象 Returns: dict: {"columns": [...], "texts": [...], "wall_candidates": [...]} """ msp = doc.modelspace() return { "columns": self.column_detector.detect(msp), "texts": self.text_extractor.extract(msp), "wall_candidates": self.wall_detector.detect(msp), } # ============================================================ # 报告生成 # ============================================================ def generate_report(result: dict) -> str: """根据检测结果生成纯文本统计报告。 Args: result: detect_all() 的返回值 Returns: str: 格式化报告文本 """ lines = [] lines.append("=" * 50) lines.append(" CAD 元素分析报告") lines.append("=" * 50) lines.append("") # 柱子 columns = result["columns"] lines.append(f"【柱子候选】") lines.append(f" 数量: {len(columns)}") if columns: # 按块名分组统计 block_counts = {} for col in columns: name = col["name"] block_counts[name] = block_counts.get(name, 0) + 1 for name, count in sorted(block_counts.items()): lines.append(f" - {name}: {count} 个") lines.append("") # 文字 texts = result["texts"] lines.append(f"【文字】") lines.append(f" 数量: {len(texts)}") if texts: # 按图层分组统计 layer_counts = {} for t in texts: ly = t["layer"] layer_counts[ly] = layer_counts.get(ly, 0) + 1 for ly, count in sorted(layer_counts.items()): lines.append(f" - 图层 {ly}: {count} 条") lines.append("") # 墙体候选 walls = result["wall_candidates"] lines.append(f"【墙体候选】") lines.append(f" 数量: {len(walls)}") if walls: # 按类型统计 type_counts = {} layer_counts = {} lengths = [] for w in walls: etype = w["entity_type"] type_counts[etype] = type_counts.get(etype, 0) + 1 ly = w["layer"] layer_counts[ly] = layer_counts.get(ly, 0) + 1 # 获取长度 if etype == "LINE": lengths.append(w["length"]) elif etype == "LWPOLYLINE": lengths.append(w["max_segment_length"]) lines.append(" 按类型:") for etype, count in sorted(type_counts.items()): lines.append(f" - {etype}: {count} 条") lines.append(" 按图层:") for ly, count in sorted(layer_counts.items(), key=lambda x: -x[1]): lines.append(f" - 图层 {ly}: {count} 条") if lengths: lines.append(f" 最长段: {max(lengths):.2f}") lines.append(f" 最短段: {min(lengths):.2f}") lines.append(f" 平均长: {sum(lengths) / len(lengths):.2f}") lines.append("") lines.append("=" * 50) lines.append(" 报告结束") lines.append("=" * 50) return "\n".join(lines) # ============================================================ # 入口 # ============================================================ def run_detection( dxf_path: str, output_dir: str, column_keywords: Optional[list[str]] = None, wall_min_length: float = WALL_MIN_LENGTH, wall_exclude_layers: Optional[list[str]] = None, wall_include_layers: Optional[list[str]] = None, ) -> dict: """执行元素检测并输出 JSON 和报告文件。 Args: dxf_path: DXF 文件路径 output_dir: 输出目录路径 column_keywords: 柱子识别关键词 wall_min_length: 墙体最小长度阈值 wall_exclude_layers: 墙体排除图层关键词 wall_include_layers: 墙体限定图层关键词 Returns: dict: 检测结果 """ doc = ezdxf.readfile(dxf_path) filename = Path(dxf_path).stem detector = ElementDetector( column_keywords=column_keywords, wall_min_length=wall_min_length, wall_exclude_layers=wall_exclude_layers, wall_include_layers=wall_include_layers, ) result = detector.detect_all(doc) out_dir = Path(output_dir) out_dir.mkdir(parents=True, exist_ok=True) # 输出 JSON json_path = out_dir / f"{filename}_element_detection.json" json_path.write_text( json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8", ) print(f"检测结果已保存到: {json_path}") # 输出报告 report_path = out_dir / f"{filename}_element_report.txt" report = generate_report(result) report_path.write_text(report, encoding="utf-8") print(f"分析报告已保存到: {report_path}") # 打印摘要 print(f"\n 柱子候选 : {len(result['columns'])} 个") print(f" 文字 : {len(result['texts'])} 条") print(f" 墙体候选 : {len(result['wall_candidates'])} 条") print(report) return result # ============================================================ # 独立运行入口 # ============================================================ if __name__ == "__main__": import sys BASE_DIR = Path(__file__).parent.parent.parent SAMPLE_DIR = BASE_DIR / "samples" OUTPUT_DIR = Path(__file__).parent / "output" dxf_file = SAMPLE_DIR / "洗浴中心C-48.dxf" if not dxf_file.exists(): print(f"错误：文件不存在 -> {dxf_file}") sys.exit(1) run_detection(str(dxf_file), str(OUTPUT_DIR))