zio
0702b414bf
- Extract trench channels from original 1F site plan DWG (same origin as hall frames) via streaming DXF tag parser that walks xref BLOCK contents and resolves INSERT-chain transforms (missed by top-level iteration, which caused the earlier 'no trench layer in K2' misjudgment). - Power trunk plan (E3-59~69) used as cross-validation only: its util-line xref is a cropped copy (x<=509.2m) with ~3.5m phase offset; angle clusters (0/+-11/-12 deg) confirm frame rotation alignment. - Generator tools/cad/extract_trench_seed_k2.py: frame rotation, horizontal clip (Liang-Barsky), >=30m runs, 1.5m twin-line merge, 6m point sampling (V61 convention). H6 9 / H7 H8 H9 H10 14 channels each, is_assumed=false. - All halls H1-H10 now have measured trench (PLANNING R4 resolved). - Docs: cad-extraction.md V66 section, OWNER_FEEDBACK log. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
315 lines
12 KiB
Python
315 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""K2(제2전시장 H6~H10) 트렌치 실측 시드 생성기 — V66 (PLANNING R4 실측 대체, K1 V61의 K2판).
|
|
|
|
소스: 원본-1층 평면도.dwg → LibreDWG dwg2dxf 변환본(예: scratchpad k2_plan1f.dxf).
|
|
홀 프레임(FRAMES)을 실측한 파일과 동일 도면이라 좌표계가 자동 정합된다.
|
|
※ "2전시장 E3-59~69 지상1층 전력간선 평면도(분할).dwg"는 교차검증용으로만 사용 —
|
|
그 안의 '전시장 유틸라인' xref는 x=509.2m에서 잘린 크롭본 + 채널 위상이 ~3.5m 어긋나
|
|
시드 소스로 부적합(docs/analysis/cad-extraction.md §K2 트렌치 참조).
|
|
|
|
방식: 111MB DXF를 ezdxf 없이 스트리밍 태그 파싱(4GB RAM 노트북 안전) —
|
|
BLOCKS 내부까지 트렌치 레이어(LINE/LWPOLYLINE) 수집, INSERT 체인 변환으로 월드 좌표 복원,
|
|
홀 프레임(±11~12° 회전) 로컬로 변환·클리핑 후 가로 채널 행을 병합(쌍선 스냅 1.5m),
|
|
채널을 따라 6m 간격 공급 포인트 생성(K1과 동일 샘플링 규약).
|
|
|
|
사용:
|
|
python tools/cad/extract_trench_seed_k2.py <k2_plan1f.dxf> <out.sql>
|
|
|
|
좌표 규약: CAD mm → 홀 로컬 m. 홀 로컬 y=0은 도면 상단(v=v1), y = v1 - v (상하 반전).
|
|
"""
|
|
import io
|
|
import math
|
|
import sys
|
|
from collections import defaultdict
|
|
|
|
MM = 1000.0
|
|
MIN_RUN_M = 30.0 # 채널 인정 최소 연장(짧은 스퍼·피더 배제 — K1 동일)
|
|
POINT_STEP_M = 6.0 # 공급 포인트 샘플링 간격(K1 동일 규약)
|
|
ROW_SNAP_M = 1.5 # 쌍선(채널 양변) 병합 허용(K1 동일)
|
|
HORIZ_TOL_M = 1.0 # 프레임 로컬 수평 판정 허용 dv
|
|
|
|
# 홀 프레임(fork 확정, docs/analysis/cad-extraction.md §K2): (회전deg, 중심, u0,u1, v0,v1)
|
|
FRAMES = {
|
|
"H6": (-12, (530, 405), 482, 542, 350.5, 443.5),
|
|
"H7": (0, (0, 0), 439, 529, 215.5, 341.5),
|
|
"H8": (0, (0, 0), 439, 529, 89.5, 215.5),
|
|
"H9": (-11, (245, 155), 196, 295, 95.5, 227.5),
|
|
"H10": (11, (255, 330), 194.5, 293.5, 250, 382),
|
|
}
|
|
|
|
TRENCH_KEYS = [b"TRENCH", "트렌치".encode("cp949"), "트랜치".encode("cp949")]
|
|
|
|
|
|
def is_trench_layer(layer_bytes):
|
|
u = layer_bytes.upper()
|
|
return any(k in u for k in TRENCH_KEYS)
|
|
|
|
|
|
# ---------- 1) DXF 스트리밍 파싱 (BLOCKS + ENTITIES) ----------
|
|
|
|
def parse(src):
|
|
segs_by_container = defaultdict(list) # container(bytes) -> [(x1,y1,x2,y2)] mm
|
|
inserts_by_container = defaultdict(list) # container -> [(name,x,y,sx,sy,rot)]
|
|
section = None
|
|
container = None
|
|
ent = None
|
|
|
|
def flush(ent):
|
|
if ent is None:
|
|
return
|
|
c = ent["container"]
|
|
if c is None or ent.get("paper"):
|
|
return
|
|
t = ent["type"]
|
|
if t == b"INSERT" and ent.get("name"):
|
|
inserts_by_container[c].append(
|
|
(ent["name"], ent.get("x", 0.0), ent.get("y", 0.0),
|
|
ent.get("sx", 1.0), ent.get("sy", 1.0), ent.get("rot", 0.0)))
|
|
return
|
|
if not is_trench_layer(ent.get("layer", b"")):
|
|
return
|
|
if t == b"LINE":
|
|
if all(k in ent for k in ("x", "y", "x2", "y2")):
|
|
segs_by_container[c].append((ent["x"], ent["y"], ent["x2"], ent["y2"]))
|
|
elif t == b"LWPOLYLINE":
|
|
pts = ent.get("pts", [])
|
|
for a, b in zip(pts, pts[1:]):
|
|
segs_by_container[c].append((a[0], a[1], b[0], b[1]))
|
|
|
|
f = open(src, "rb", buffering=1024 * 1024 * 8)
|
|
expect_section_name = False
|
|
while True:
|
|
code_raw = f.readline()
|
|
if not code_raw:
|
|
break
|
|
val = f.readline()
|
|
if not val:
|
|
break
|
|
try:
|
|
code = int(code_raw)
|
|
except ValueError:
|
|
continue
|
|
v = val.strip()
|
|
if code == 0:
|
|
flush(ent)
|
|
ent = None
|
|
if v == b"SECTION":
|
|
expect_section_name = True
|
|
elif v == b"ENDSEC":
|
|
section = None
|
|
container = None
|
|
elif v == b"BLOCK" and section == b"BLOCKS":
|
|
ent = {"type": b"BLOCK", "container": None}
|
|
elif v == b"ENDBLK":
|
|
container = None
|
|
elif section in (b"BLOCKS", b"ENTITIES") and v in (b"LINE", b"LWPOLYLINE", b"INSERT"):
|
|
ent = {"type": v, "container": container}
|
|
continue
|
|
if expect_section_name and code == 2:
|
|
section = v
|
|
expect_section_name = False
|
|
if section == b"ENTITIES":
|
|
container = b"*MS"
|
|
continue
|
|
if ent is None:
|
|
continue
|
|
if ent["type"] == b"BLOCK":
|
|
if code == 2:
|
|
container = v
|
|
ent = None
|
|
continue
|
|
if code == 8:
|
|
ent["layer"] = v
|
|
elif code == 67:
|
|
ent["paper"] = v == b"1"
|
|
elif code == 2 and ent["type"] == b"INSERT":
|
|
ent["name"] = v
|
|
elif code == 10:
|
|
x = float(v)
|
|
if ent["type"] == b"LWPOLYLINE":
|
|
ent.setdefault("pts", []).append([x, 0.0])
|
|
else:
|
|
ent["x"] = x
|
|
elif code == 20:
|
|
y = float(v)
|
|
if ent["type"] == b"LWPOLYLINE":
|
|
pts = ent.get("pts")
|
|
if pts:
|
|
pts[-1][1] = y
|
|
else:
|
|
ent["y"] = y
|
|
elif code == 11:
|
|
ent["x2"] = float(v)
|
|
elif code == 21:
|
|
ent["y2"] = float(v)
|
|
elif code == 41:
|
|
ent["sx"] = float(v)
|
|
elif code == 42:
|
|
ent["sy"] = float(v)
|
|
elif code == 50 and ent["type"] == b"INSERT":
|
|
ent["rot"] = float(v)
|
|
flush(ent)
|
|
f.close()
|
|
return segs_by_container, inserts_by_container
|
|
|
|
|
|
# ---------- 2) INSERT 체인 해석 → 월드 세그(m) ----------
|
|
|
|
def _transform(px, py, ins):
|
|
x0, y0, sx, sy, rot = ins
|
|
px, py = px * sx, py * sy
|
|
if rot:
|
|
t = math.radians(rot)
|
|
px, py = px * math.cos(t) - py * math.sin(t), px * math.sin(t) + py * math.cos(t)
|
|
return px + x0, py + y0
|
|
|
|
|
|
def world_segs(segs_by_container, inserts_by_container):
|
|
cache = {}
|
|
|
|
def placements(cname, depth=0):
|
|
if cname == b"*MS":
|
|
return [lambda p: p]
|
|
if depth > 6:
|
|
return []
|
|
if cname in cache:
|
|
return cache[cname]
|
|
out = []
|
|
for parent, inss in inserts_by_container.items():
|
|
for ins in inss:
|
|
if ins[0] != cname:
|
|
continue
|
|
geo = ins[1:]
|
|
for up in placements(parent, depth + 1):
|
|
out.append(lambda p, geo=geo, up=up: up(_transform(p[0], p[1], geo)))
|
|
cache[cname] = out
|
|
return out
|
|
|
|
ws = []
|
|
for c, ss in segs_by_container.items():
|
|
for pl in placements(c):
|
|
for (x1, y1, x2, y2) in ss:
|
|
wx1, wy1 = pl((x1, y1))
|
|
wx2, wy2 = pl((x2, y2))
|
|
ws.append((wx1 / MM, wy1 / MM, wx2 / MM, wy2 / MM))
|
|
return ws
|
|
|
|
|
|
# ---------- 3) 홀 프레임 변환·클리핑·채널 병합 ----------
|
|
|
|
def _rot_frame(x, y, deg, c):
|
|
if deg == 0:
|
|
return x, y
|
|
t = math.radians(-deg)
|
|
dx, dy = x - c[0], y - c[1]
|
|
return c[0] + dx * math.cos(t) - dy * math.sin(t), c[1] + dx * math.sin(t) + dy * math.cos(t)
|
|
|
|
|
|
def _clip(ua, va, ub, vb, u0, u1, v0, v1):
|
|
"""Liang-Barsky 세그 사각 클리핑."""
|
|
du, dv = ub - ua, vb - va
|
|
t0, t1 = 0.0, 1.0
|
|
for p, q in ((-du, ua - u0), (du, u1 - ua), (-dv, va - v0), (dv, v1 - va)):
|
|
if p == 0:
|
|
if q < 0:
|
|
return None
|
|
continue
|
|
r = q / p
|
|
if p < 0:
|
|
if r > t1:
|
|
return None
|
|
t0 = max(t0, r)
|
|
else:
|
|
if r < t0:
|
|
return None
|
|
t1 = min(t1, r)
|
|
if t0 > t1:
|
|
return None
|
|
return (ua + t0 * du, ua + t1 * du, (va + vb) / 2 + 0 * t0) # (cu0, cu1, vm) — 수평 전제
|
|
|
|
|
|
def channels_of(ws):
|
|
"""홀별 채널 [(center_y, x0, x1)] — 로컬 y=0 도면 상단(상하 반전)."""
|
|
out = {}
|
|
for hall, (deg, c, u0, u1, v0, v1) in FRAMES.items():
|
|
rows = defaultdict(list)
|
|
for (x1, y1, x2, y2) in ws:
|
|
ua, va = _rot_frame(x1, y1, deg, c)
|
|
ub, vb = _rot_frame(x2, y2, deg, c)
|
|
if abs(vb - va) > HORIZ_TOL_M:
|
|
continue
|
|
cl = _clip(ua, va, ub, vb, u0, u1, v0, v1)
|
|
if cl is None:
|
|
continue
|
|
cu0, cu1, vm = cl
|
|
lx0, lx1 = min(cu0, cu1) - u0, max(cu0, cu1) - u0
|
|
if lx1 - lx0 < MIN_RUN_M:
|
|
continue
|
|
rows[round((v1 - vm) * 2) / 2].append((lx0, lx1))
|
|
merged = [] # (center_y, x0, x1)
|
|
for y, spans in sorted(rows.items()):
|
|
x0 = min(s[0] for s in spans)
|
|
x1 = max(s[1] for s in spans)
|
|
if merged and y - merged[-1][0] <= ROW_SNAP_M:
|
|
py, px0, px1 = merged[-1]
|
|
merged[-1] = ((py + y) / 2, min(px0, x0), max(px1, x1))
|
|
else:
|
|
merged.append((y, x0, x1))
|
|
out[hall] = merged
|
|
return out
|
|
|
|
|
|
# ---------- 4) SQL 산출 ----------
|
|
|
|
def main():
|
|
src, out_sql = sys.argv[1], sys.argv[2]
|
|
segs_by_container, inserts_by_container = parse(src)
|
|
ws = world_segs(segs_by_container, inserts_by_container)
|
|
chans = channels_of(ws)
|
|
|
|
sql = io.StringIO()
|
|
sql.write(
|
|
"-- V66: K2 홀6~10 트렌치 실측 시드 — CAD(원본-1층 평면도.dwg) TRENCH 레이어 추출(V61의 K2판).\n"
|
|
"-- 생성기: tools/cad/extract_trench_seed_k2.py (재생성 가능·결정적). 근거: docs/analysis/cad-extraction.md §K2 트렌치.\n"
|
|
"-- 실측: 홀 프레임(±11~12° 회전) 로컬 가로 채널(쌍선 병합 중심선)·연장 실좌표. 포인트 6m 간격은 샘플링 규약.\n"
|
|
"-- 교차검증: 전력간선 평면도(E3-59~69)의 채널 구조(9m 피치·쌍선·홀 배치각)와 정합 —\n"
|
|
"-- 단, 그 도면의 유틸라인 xref는 크롭본(x<=509.2m)이라 시드 소스는 홀 프레임과 동일 원점인 원본 평면도 사용.\n"
|
|
"-- 정책: 실측 확보 홀(H6~H10)의 가정(is_assumed) 포인트를 실측 포인트로 대체(K1 V61과 동일).\n"
|
|
"-- 공급 플래그는 V5 정책 유지(공동구 — 전 서비스 true·gas는 홀 보유 여부), 위치만 실측 대체.\n\n"
|
|
)
|
|
report = []
|
|
total = 0
|
|
for hall, merged in chans.items():
|
|
report.append(f"{hall}: 채널 {len(merged)}개 " + ", ".join(f"y={m[0]:.1f} x[{m[1]:.1f}..{m[2]:.1f}]" for m in merged))
|
|
if not merged:
|
|
continue
|
|
total += len(merged)
|
|
sql.write(f"-- {hall}: 실측 채널 {len(merged)}개\n")
|
|
sql.write(f"DELETE FROM trench WHERE hall_id = '{hall}' AND is_assumed;\n")
|
|
values = []
|
|
for ri, (cy, x0, x1) in enumerate(merged, 1):
|
|
k = 0
|
|
x = x0
|
|
while x <= x1 + 1e-6:
|
|
k += 1
|
|
values.append(
|
|
f"('{hall}-RT-{ri}-{k}', '{hall}', ST_SetSRID(ST_MakePoint({x:.2f}, {cy:.2f}), 0), "
|
|
f"true, true, true, true, (SELECT has_gas FROM hall WHERE id='{hall}'), false)"
|
|
)
|
|
x += POINT_STEP_M
|
|
sql.write(
|
|
"INSERT INTO trench (id, hall_id, geom, supply_power, supply_water, supply_air, supply_network, supply_gas, is_assumed)\nVALUES\n "
|
|
+ ",\n ".join(values)
|
|
+ "\nON CONFLICT (id) DO UPDATE SET geom = EXCLUDED.geom, is_assumed = EXCLUDED.is_assumed;\n\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} (채널 {total}개)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|