kintex/tools/cad/extract_obstacle_seed.py

# -*- coding: utf-8 -*-
"""벽·계단·엘리베이터 장애물 실측 시드 생성기 — V63 (hall_obstacle).

방식: 홀 바닥면(로컬 m)을 1m 셀로 래스터화, 장애물 레이어(WALL/STAIR/ELEVATOR/DRYWALL 계열)
지오메트리가 지나는 셀을 마킹 → 연결 성분 → bbox 를 장애 영역으로 시드.
자동배치는 이 영역(+여유)을 예약 영역으로 회피한다.

필터: 면적 3m² 미만(파편)·최소변 1.2m 미만(단순 벽선)·폭/깊이 80% 이상 관통 박판(외곽 벽선) 제외.

사용:
  python tools/cad/extract_obstacle_seed.py <k1_trench.dxf> <out.sql>

좌표 규약: 홀 로컬 m, y=0 도면 상단(= CAD y 상하 반전, V62와 동일).
"""
import io
import math
import sys
from collections import deque

import ezdxf

MM = 1000.0
FLOOR_X0, FLOOR_W = 293.5, 171.0
BAND_Y0, BAND_H = 80.1, 63.0
K1_HALLS = [1, 2, 3, 4, 5]

KIND_LAYERS = {
    "stair": ("STAIR", "계단"),
    "elevator": ("ELEVATOR", "ELV", "승강"),
    "door": ("DOOR", "출입구", "4-창호"),
    "wall": ("WALL", "DRYWALL", "조적", "석고", "판넬"),
}
CELL = 1.0  # m
MIN_AREA = 3.0   # wall/stair/elevator — 파편 제거
MIN_DIM = 1.2
DOOR_MIN_AREA = 0.5  # 문 심볼은 소형 — 별도 기준(문 전면 클리어 확보 목적)


def kind_of(layer):
    u = layer.upper()
    for kind, keys in KIND_LAYERS.items():
        if any(k.upper() in u for k in keys):
            return kind
    return None


def sample_points(e):
    t = e.dxftype()
    try:
        if t == "LINE":
            a, b = e.dxf.start, e.dxf.end
            ln = math.hypot(b[0] - a[0], b[1] - a[1]) / MM
            n = max(2, int(ln / 0.5))
            return [((a[0] + (b[0] - a[0]) * i / n) / MM, (a[1] + (b[1] - a[1]) * i / n) / MM) for i in range(n + 1)]
        if t == "LWPOLYLINE":
            pts = [(p[0] / MM, p[1] / MM) for p in e.get_points()]
            out = []
            for a, b in zip(pts, pts[1:]):
                ln = math.hypot(b[0] - a[0], b[1] - a[1])
                n = max(1, int(ln / 0.5))
                out += [(a[0] + (b[0] - a[0]) * i / n, a[1] + (b[1] - a[1]) * i / n) for i in range(n + 1)]
            return out
        if t in ("ARC", "CIRCLE"):
            return [(p[0] / MM, p[1] / MM) for p in e.flattening(200)]
    except Exception:
        pass
    return []


def main():
    src, out_sql = sys.argv[1], sys.argv[2]
    doc = ezdxf.readfile(src)
    msp = doc.modelspace()

    W, H = int(FLOOR_W / CELL), int(BAND_H / CELL)
    grids = {n: [[None] * W for _ in range(H)] for n in K1_HALLS}  # kind 마킹

    for e in msp:
        kind = kind_of(e.dxf.layer)
        if not kind:
            continue
        for x, y in sample_points(e):
            lx = x - FLOOR_X0
            if not (0 <= lx < FLOOR_W):
                continue
            for n in K1_HALLS:
                y0 = BAND_Y0 + (n - 1) * BAND_H
                if y0 <= y < y0 + BAND_H:
                    ly = BAND_H - (y - y0)  # 상하 반전
                    ci, cj = int(lx / CELL), min(H - 1, int(ly / CELL))
                    cur = grids[n][cj][ci]
                    # stair/elevator 가 wall 보다 우선(용도 표기)
                    if cur is None or (cur == "wall" and kind != "wall"):
                        grids[n][cj][ci] = kind

    sql = io.StringIO()
    sql.write(
        "-- V63: 벽·계단·엘리베이터 장애물 실측 시드(hall_obstacle) — 자동배치 회피 근거(소유자 지시 2026-07-15).\n"
        "-- 생성기: tools/cad/extract_obstacle_seed.py (1m 셀 래스터 연결성분 bbox, 결정적 재생성).\n"
        "-- K1 H1~H5 = CAD 추출. K2 = 장애물 레이어 신뢰 추출 불가(도면 구조 상이) — 미시드(한계, cad-extraction.md).\n\n"
        "CREATE TABLE IF NOT EXISTS hall_obstacle (\n"
        "    id      varchar(60)  PRIMARY KEY,   -- 예: H1-OB-1\n"
        "    hall_id varchar(20)  NOT NULL REFERENCES hall(id) ON DELETE CASCADE,\n"
        "    kind    varchar(20)  NOT NULL,      -- stair | elevator | door | wall(부속실·DRYWALL 포함)\n"
        "    x0 numeric(8,2) NOT NULL, y0 numeric(8,2) NOT NULL,\n"
        "    x1 numeric(8,2) NOT NULL, y1 numeric(8,2) NOT NULL   -- 홀 로컬 m(y=0 도면 상단)\n"
        ");\n"
        "CREATE INDEX IF NOT EXISTS idx_hall_obstacle_hall ON hall_obstacle (hall_id);\n\n"
    )

    rows = []
    report = []
    for n in K1_HALLS:
        g = grids[n]
        seen = [[False] * W for _ in range(H)]
        comps = []
        for j in range(H):
            for i in range(W):
                if g[j][i] is None or seen[j][i]:
                    continue
                q = deque([(i, j)])
                seen[j][i] = True
                cells = []
                kinds = {}
                while q:
                    ci, cj = q.popleft()
                    cells.append((ci, cj))
                    k = g[cj][ci]
                    kinds[k] = kinds.get(k, 0) + 1
                    for ni, nj in ((ci + 1, cj), (ci - 1, cj), (ci, cj + 1), (ci, cj - 1),
                                   (ci + 1, cj + 1), (ci - 1, cj - 1), (ci + 1, cj - 1), (ci - 1, cj + 1)):
                        if 0 <= ni < W and 0 <= nj < H and g[nj][ni] is not None and not seen[nj][ni]:
                            seen[nj][ni] = True
                            q.append((ni, nj))
                area = len(cells) * CELL * CELL
                # 용도 우선순위: stair > elevator > door > wall
                kind = ("stair" if kinds.get("stair")
                        else "elevator" if kinds.get("elevator")
                        else "door" if kinds.get("door")
                        else "wall")
                min_area = DOOR_MIN_AREA if kind == "door" else MIN_AREA
                if area < min_area:
                    continue
                # ★bbox 뻥튀기 방지(L자형 얇은 벽 런): 행 스트립 분해 — 행별 연속 구간을
                #   rect 로 만들고 동일 x구간 인접 행을 세로 병합(실 커버리지 근사).
                by_row = {}
                for ci, cj in cells:
                    by_row.setdefault(cj, []).append(ci)
                strips = []  # (j0, j1, i0, i1)
                for j in sorted(by_row):
                    cols = sorted(by_row[j])
                    seg = [cols[0], cols[0]]
                    segs = []
                    for c in cols[1:]:
                        if c == seg[1] + 1:
                            seg[1] = c
                        else:
                            segs.append(tuple(seg))
                            seg = [c, c]
                    segs.append(tuple(seg))
                    for i0, i1 in segs:
                        merged_flag = False
                        for s in strips:
                            if s[1] == j - 1 and s[2] == i0 and s[3] == i1:
                                s[1] = j
                                merged_flag = True
                                break
                        if not merged_flag:
                            strips.append([j, j, i0, i1])
                for j0, j1, i0, i1 in strips:
                    x0, x1 = i0 * CELL, (i1 + 1) * CELL
                    y0, y1 = j0 * CELL, (j1 + 1) * CELL
                    w, h = x1 - x0, y1 - y0
                    if w * h < min_area and not (kind == "door" and w * h >= DOOR_MIN_AREA):
                        continue
                    if (w >= FLOOR_W * 0.8 and h <= 2.5) or (h >= BAND_H * 0.8 and w <= 2.5):
                        continue  # 외곽 벽선 관통 박판
                    comps.append((kind, x0, y0, x1, y1, w * h))
        comps.sort(key=lambda c: (c[1], c[2]))
        report.append(f"H{n}: {len(comps)}개 " + ", ".join(f"{c[0]}({c[1]:.0f},{c[2]:.0f})-({c[3]:.0f},{c[4]:.0f})" for c in comps[:8]))
        for i, (kind, x0, y0, x1, y1, _a) in enumerate(comps, 1):
            rows.append(f"('H{n}-OB-{i}', 'H{n}', '{kind}', {x0:.1f}, {y0:.1f}, {x1:.1f}, {y1:.1f})")

    if rows:
        sql.write("DELETE FROM hall_obstacle WHERE id LIKE 'H%-OB-%';\n")
        sql.write(
            "INSERT INTO hall_obstacle (id, hall_id, kind, x0, y0, x1, y1)\nVALUES\n    "
            + ",\n    ".join(rows)
            + "\nON CONFLICT (id) DO UPDATE SET kind = EXCLUDED.kind,\n"
            "    x0 = EXCLUDED.x0, y0 = EXCLUDED.y0, x1 = EXCLUDED.x1, y1 = EXCLUDED.y1;\n"
        )
    with open(out_sql, "w", encoding="utf-8") as f:
        f.write(sql.getvalue())
    for line in report:
        print(line)
    print(f"OK -> {out_sql} (장애물 {len(rows)}개)")


if __name__ == "__main__":
    main()