"""구조물 상세도면 ↔ 3D 빌드 결과 간 VLM(Gemini Vision) 피드백 루프. 흐름: 1. 상세 DXF를 평면 PNG로 렌더(ezdxf + matplotlib) 2. 빌드된 3D 메시를 top-down PNG로 렌더(PyVista off-screen) 3. 두 이미지 + 현재 파라미터 JSON을 Gemini Vision에 전달 4. 구조화된 JSON diff 수신 (missing/incorrect/excess) 5. diff를 파라미터에 머지(사용자 승인 후) 기존 Gemini 인프라(google.genai + Vertex AI gcp-key.json)를 그대로 재사용하며, 별도 API/SDK/결제선 없음. 모델은 기본 gemini-2.5-flash(저비용) → 필요 시 2.5-pro. """ from __future__ import annotations import json import os import re from dataclasses import asdict, is_dataclass from pathlib import Path from typing import Any from collections.abc import Callable import numpy as np # --------------------------------------------------------------------------- # 렌더링 # --------------------------------------------------------------------------- def render_dxf_to_png(dxf_paths: list[str] | str, output_path: str, size: int = 1400, dpi: int = 140, bg: str = "white", fg: str = "black") -> str: """상세 DXF를 matplotlib으로 렌더링해 PNG 저장. Args: dxf_paths: 단일 경로 또는 경로 리스트(첫 파일만 렌더) output_path: 저장 경로 size: 결과 이미지 한 변 픽셀 (정사각) dpi: matplotlib dpi bg/fg: 배경/선 색 Returns: output_path """ import matplotlib matplotlib.use("Agg", force=True) import matplotlib.pyplot as plt import ezdxf from ezdxf.addons.drawing import Frontend, RenderContext from ezdxf.addons.drawing.matplotlib import MatplotlibBackend from ezdxf.addons.drawing.config import Configuration dxf_path = dxf_paths[0] if isinstance(dxf_paths, (list, tuple)) else dxf_paths doc = ezdxf.readfile(dxf_path) msp = doc.modelspace() fig_in = size / dpi fig, ax = plt.subplots(figsize=(fig_in, fig_in), dpi=dpi) fig.patch.set_facecolor(bg) ax.set_facecolor(bg) ax.set_aspect("equal") ax.set_axis_off() ctx = RenderContext(doc) try: cfg = Configuration() except Exception: cfg = None backend = MatplotlibBackend(ax) frontend = Frontend(ctx, backend, config=cfg) if cfg else Frontend(ctx, backend) frontend.draw_layout(msp, finalize=True) fig.savefig(output_path, dpi=dpi, bbox_inches="tight", facecolor=bg, pad_inches=0.1) plt.close(fig) return output_path def render_meshes_topdown(meshes: list[tuple], output_path: str, size: int = 1400, bg: str = "white") -> str: """빌드된 메시 리스트를 top-down(평면) 뷰로 렌더. Args: meshes: [(pv.PolyData, color, opacity), ...] output_path: 저장 경로 size: 정사각 픽셀 bg: 배경색 Returns: output_path """ import pyvista as pv p = pv.Plotter(off_screen=True, window_size=(size, size)) p.set_background(bg) for item in meshes: try: if len(item) >= 3: mesh, color, opacity = item[0], item[1], item[2] else: mesh, color, opacity = item[0], item[1], 1.0 p.add_mesh(mesh, color=color, opacity=opacity, show_edges=True, edge_color="#888888", line_width=0.5, smooth_shading=False) except Exception: continue p.enable_parallel_projection() p.view_xy() # +Z에서 -Z 방향 내려다봄 p.camera.zoom(1.0) try: p.screenshot(output_path, transparent_background=False) finally: p.close() return output_path # --------------------------------------------------------------------------- # 파라미터 직렬화 # --------------------------------------------------------------------------- def params_to_dict(params: Any) -> dict: """dataclass / dict 객체를 JSON-직렬화 가능한 dict로 변환.""" if params is None: return {} if is_dataclass(params): d = asdict(params) elif isinstance(params, dict): d = dict(params) else: # 일반 객체 속성 덤프 d = {k: v for k, v in vars(params).items() if not k.startswith("_")} return _json_safe(d) def _json_safe(obj): if isinstance(obj, dict): return {k: _json_safe(v) for k, v in obj.items()} if isinstance(obj, (list, tuple)): return [_json_safe(v) for v in obj] if isinstance(obj, (np.floating, np.integer)): return float(obj) if isinstance(obj, np.floating) else int(obj) if isinstance(obj, np.ndarray): return obj.tolist() if isinstance(obj, (str, int, float, bool)) or obj is None: return obj return str(obj) # --------------------------------------------------------------------------- # Gemini 호출 # --------------------------------------------------------------------------- _DIFF_SCHEMA_PROMPT = """당신은 기계설비 설계도면을 검토하는 엔지니어링 검증 도구입니다. 첨부된 이미지 2장: image 1 = 원본 설계 도면 (DXF의 평면도 렌더) image 2 = 현재 파서·빌더가 생성한 3D 모델의 top-down 뷰 현재 추출된 파라미터 JSON: ```json {params_json} ``` 구조물 유형: {structure_type} 두 이미지를 비교해 도면의 의도가 3D 모델에 얼마나 정확히 반영됐는지 평가하고, 차이(missing/incorrect/excess)를 다음 JSON 스키마로만 반환: {{ "summary": "1-2문장 한국어 요약", "match_score": 0.0~1.0 사이 실수 (도면 반영률, 1.0=완벽), "param_updates": [ {{"path": "필드명 (예: chamber_width, bottom_el)", "current": "현재 값", "suggested": "도면에서 관찰한 값", "reason": "한국어 근거 1문장"}} ], "valves_missing": [ {{"name": "M-NNN 또는 설명", "x": float, "y": float, "diameter_mm": int, "valve_type": "GATE|BUTTERFLY|CHECK|BALL", "reason": "왜 누락으로 판단했는지"}} ], "valves_incorrect": [ {{"name": "M-NNN", "field": "수정할 필드", "current": "현재 값", "suggested": "제안 값", "reason": "..."}} ], "pipes_missing": [ {{"name": "식별명", "diameter_mm": int, "start": [x, y, z], "end": [x, y, z], "reason": "..."}} ], "pipes_incorrect": [ {{"name": "식별명", "field": "...", "current": "...", "suggested": "...", "reason": "..."}} ], "excess_notes": ["모델에 있지만 도면에 없는 요소 설명"] }} 주의: - 확실하지 않은 항목은 제안하지 마세요(false positive 최소화). - 좌표·직경은 chamber/구조물 로컬 좌표계 기준 meter 또는 mm를 파라미터 JSON 단위에 맞추세요. - 단순 렌더 품질 차이(색·조명)는 무시하세요. 도면 의도만 비교. - JSON 외 어떠한 텍스트도 반환하지 마세요. """ def _read_bytes(path: str) -> bytes: with open(path, "rb") as f: return f.read() def request_structure_diff(client, drawing_png_path: str, render_png_path: str, params_dict: dict, structure_type: str = "valve_chamber", model: str = "gemini-2.5-flash", log_fn: Callable[[str], None] = print, timeout_s: float = 60.0) -> dict: """Gemini Vision에 도면+렌더+파라미터 전달해 JSON diff 수신. Args: client: google.genai.Client 인스턴스 (caller가 인증 설정) drawing_png_path: 원본 도면 PNG render_png_path: 3D top-down PNG params_dict: 현재 파라미터 (JSON-safe) structure_type: 구조물 종류 (프롬프트 컨텍스트) model: Gemini 모델명 (기본 2.5-flash) log_fn: 로그 callback timeout_s: 호출 타임아웃 (실제로는 SDK 설정에 따름) Returns: diff dict (스키마는 _DIFF_SCHEMA_PROMPT 참고) Raises: RuntimeError: 호출 실패 또는 JSON 파싱 실패 """ try: from google.genai import types as gtypes except ImportError as e: raise RuntimeError(f"google.genai SDK 필요: {e}") from e params_json = json.dumps(params_dict, ensure_ascii=False, indent=2) prompt = _DIFF_SCHEMA_PROMPT.format( params_json=params_json, structure_type=structure_type, ) drawing_bytes = _read_bytes(drawing_png_path) render_bytes = _read_bytes(render_png_path) parts = [ gtypes.Part.from_bytes(data=drawing_bytes, mime_type="image/png"), gtypes.Part.from_bytes(data=render_bytes, mime_type="image/png"), gtypes.Part.from_text(text=prompt), ] log_fn(f" [VLM] Gemini 호출: model={model}, drawing={len(drawing_bytes)//1024}KB, render={len(render_bytes)//1024}KB") try: resp = client.models.generate_content( model=model, contents=parts, config=gtypes.GenerateContentConfig( response_mime_type="application/json", temperature=0.1, ), ) except Exception as e: raise RuntimeError(f"Gemini 호출 실패: {e}") from e text = getattr(resp, "text", None) or "" if not text: # 일부 SDK 버전은 candidates[0].content.parts[0].text 사용 try: text = resp.candidates[0].content.parts[0].text except Exception: text = "" if not text: raise RuntimeError("Gemini 응답이 비어있습니다.") # 혹시 모를 코드블록 제거 text = re.sub(r"^```(?:json)?\s*", "", text.strip()) text = re.sub(r"\s*```$", "", text) try: diff = json.loads(text) except json.JSONDecodeError as e: # 부분 복구 시도 (첫 {부터 마지막 }까지) m = re.search(r"\{.*\}", text, re.DOTALL) if m: try: diff = json.loads(m.group(0)) except Exception: raise RuntimeError(f"JSON 파싱 실패: {e}\n원문: {text[:300]}") from e else: raise RuntimeError(f"JSON 파싱 실패: {e}\n원문: {text[:300]}") from e log_fn(f" [VLM] 응답 수신: match_score={diff.get('match_score', '?')} " f"updates={len(diff.get('param_updates', []))} " f"v_missing={len(diff.get('valves_missing', []))} " f"p_missing={len(diff.get('pipes_missing', []))}") return diff # --------------------------------------------------------------------------- # diff 적용 # --------------------------------------------------------------------------- def apply_diff_to_params(params: Any, diff: dict, selections: dict | None = None, log_fn: Callable[[str], None] = print) -> dict: """diff를 params 객체에 in-place 적용. Args: params: dataclass 인스턴스 (ValveChamberParams 등) diff: request_structure_diff 반환값 selections: {"param_updates": [bool, ...], "valves_missing": [bool, ...], "pipes_missing": [bool, ...]} — 사용자 체크박스. None이면 모두 적용. log_fn: 로그 callback Returns: {"applied": int, "skipped": int, "errors": [str, ...]} """ sel = selections or {} applied = 0 errors: list[str] = [] # 1) 스칼라/벡터 파라미터 업데이트 for i, upd in enumerate(diff.get("param_updates", []) or []): if sel.get("param_updates") is not None and not sel["param_updates"][i]: continue path = upd.get("path", "").strip() suggested = upd.get("suggested") if not path: continue try: _set_by_path(params, path, suggested) applied += 1 log_fn(f" [VLM apply] {path} = {suggested!r}") except Exception as e: errors.append(f"{path}: {e}") # 2) Valve 추가 if hasattr(params, "valves") and isinstance(params.valves, list): try: from valve_chamber_parser import Valve except ImportError: Valve = None if Valve is not None: for i, v in enumerate(diff.get("valves_missing", []) or []): if sel.get("valves_missing") is not None and not sel["valves_missing"][i]: continue try: dia_mm = float(v.get("diameter_mm", 400)) params.valves.append(Valve( index=len(params.valves), name=v.get("name", f"V+{i+1}"), valve_type=v.get("valve_type", "GATE"), center_x=float(v.get("x", 0.0)), center_y=float(v.get("y", 0.0)), elevation=float(getattr(params, "bottom_el", 0.0)) + 1.5, diameter=dia_mm / 1000.0, label=(v.get("name", "") + " [VLM 추가]")[:60], )) applied += 1 log_fn(f" [VLM apply] +valve {v.get('name')}") except Exception as e: errors.append(f"valve_missing[{i}]: {e}") # 3) Pipe 추가 if hasattr(params, "pipes") and isinstance(params.pipes, list): try: from valve_chamber_parser import Pipe except ImportError: Pipe = None if Pipe is not None: for i, pp in enumerate(diff.get("pipes_missing", []) or []): if sel.get("pipes_missing") is not None and not sel["pipes_missing"][i]: continue try: dia_mm = float(pp.get("diameter_mm", 800)) start = tuple(pp.get("start", (0.0, 0.0, 0.0))) end = tuple(pp.get("end", (0.0, 0.0, 0.0))) params.pipes.append(Pipe( name=pp.get("name", f"P+{i+1}") + " [VLM]", diameter=dia_mm / 1000.0, start=start, end=end, elevation=start[2] if len(start) > 2 else 0.0, )) applied += 1 log_fn(f" [VLM apply] +pipe {pp.get('name')}") except Exception as e: errors.append(f"pipe_missing[{i}]: {e}") return {"applied": applied, "errors": errors} def _set_by_path(obj: Any, path: str, value: Any): """단순 속성 경로로 값 설정 (현재 PoC는 평면 필드만 지원).""" # a.b.c[0] 형식은 최소화 — 평면 필드만 if "." in path or "[" in path: # 차후 확장 포인트: 지금은 경고만 기록 raise ValueError(f"중첩 경로는 미지원: {path}") if not hasattr(obj, path): raise AttributeError(f"속성 없음: {path}") current = getattr(obj, path) # 타입 강제 변환 (숫자/문자열만) if isinstance(current, bool): new_val = bool(value) elif isinstance(current, int): new_val = int(float(value)) elif isinstance(current, float): new_val = float(value) elif isinstance(current, str): new_val = str(value) else: new_val = value setattr(obj, path, new_val) # --------------------------------------------------------------------------- # 클라이언트 생성 (scanvas_maker의 패턴 재사용) # --------------------------------------------------------------------------- def build_genai_client(project: str | None = None, location: str = "global", use_vertex: bool = True, api_key: str | None = None, log_fn: Callable[[str], None] = print): """Gemini 클라이언트 생성. Vertex AI 우선, 실패 시 API Key 폴백. scanvas_maker의 AI 렌더링 경로와 동일 인증(gcp-key.json 또는 API Key)을 사용. """ try: from google import genai except ImportError as e: raise RuntimeError(f"google-genai SDK 필요: pip install google-genai ({e})") from e if use_vertex: proj = project or os.environ.get("GCP_PROJECT_ID", "") if proj: try: client = genai.Client(vertexai=True, project=proj, location=location) log_fn(f" [VLM] Vertex AI client: project={proj}, location={location}") return client except Exception as e: log_fn(f" [VLM] Vertex AI 실패 → API Key 폴백: {e}") key = api_key or os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY", "") if not key: raise RuntimeError("Gemini 인증 정보 없음 (Vertex project 또는 API key 필요)") client = genai.Client(api_key=key) log_fn(" [VLM] API Key client") return client # --------------------------------------------------------------------------- # 편의 함수: 전체 루프 1회 실행 # --------------------------------------------------------------------------- def run_feedback_once(params: Any, meshes: list, dxf_paths: list[str], client, structure_type: str = "valve_chamber", model: str = "gemini-2.5-flash", work_dir: str | Path = "cache/vlm", log_fn: Callable[[str], None] = print) -> dict: """1회 피드백 사이클 실행: 렌더 2장 + Gemini 호출. diff 반환만. apply는 호출자가 사용자 승인 후 apply_diff_to_params 호출. """ work_dir = Path(work_dir) work_dir.mkdir(parents=True, exist_ok=True) drawing_png = str(work_dir / "drawing.png") render_png = str(work_dir / "render_topdown.png") log_fn(" [VLM] 도면 PNG 렌더링...") render_dxf_to_png(dxf_paths, drawing_png) log_fn(" [VLM] 3D top-down 렌더링...") render_meshes_topdown(meshes, render_png) params_dict = params_to_dict(params) diff = request_structure_diff( client, drawing_png, render_png, params_dict, structure_type=structure_type, model=model, log_fn=log_fn, ) diff["_artifacts"] = {"drawing_png": drawing_png, "render_png": render_png} return diff