"""도엽등고선 3D 서피스 — 1:5,000 수치지형도 등고선으로 DTM 격자를 만든다. LAS 없는 설계(2026-08-30 사용자 확정)의 지형 원천이자, LAS가 있어도 참고용으로 같이 만들어 두는 서피스다. 산출 형식은 LAS 파이프라인의 DTM과 완전히 같게 맞춘다 (`dtm_sheet.npz`: x/y/z/valid_mask) — 종·횡단·배수 세부설계가 쓰는 `build_surface_sampler(models_dir, "sheet", "dtm", smooth=False)`가 무수정으로 돈다. 절취 범위는 노선 XY bbox + `SHEET_SURFACE_MARGIN_M`(300m) 직사각형(사용자 확정), 격자는 `SHEET_SURFACE_GRID_M`(1m). 순서는 **2D 먼저, 메시는 맨 마지막**이다(2026-08-30 사용자 지시): ① 등고 라인을 격자에 굽고 ② 등고선 사이 거리 비례 보간으로 표고 격자를 만든 뒤 ③ 폐합 등고선 안쪽(마루·웅덩이)을 바깥 사면 경사로 연장하고 ④ 라인을 고정한 채 완화(라플라스)해 등고 간격을 고르게 한다. 격자에서 뽑는 1m 등고선이 곧 2D 보간선이며, 메시(glb)는 그 격자의 표현일 뿐이다. """ import json import logging import time from pathlib import Path from typing import Any import numpy as np from pyproj import Transformer from B04_PreProcess.B04_PreProcess_Engine_ModelContext import ( atomic_npz, clip_and_compact_mesh, grid_faces, grid_vertices, write_glb, ) from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import grid_spec_from_bounds from config.config_system import ( SHEET_SURFACE_GRID_M, SHEET_SURFACE_MARGIN_M, SHEET_SURFACE_RELAX_ITERATIONS, SURFACE_MAX_PREVIEW_VERTICES, ) logger = logging.getLogger(__name__) # 도엽 병합 산출물 파일명 (B04_PreProcess_Router_Watershed와 같은 값) _CONTOUR_FILE = "도엽_등고선.geojson" # 산출 모델 식별자 — surface_models.generation_params.source_filter 및 파일 stem에 쓴다. SHEET_SOURCE_FILTER = "sheet" def _load_features_metric( processed_dir: Path, epsg: int, filename: str = _CONTOUR_FILE ) -> list[dict[str, Any]]: """병합 도엽 레이어(WGS84)를 읽어 사업지 CRS(m)로 재투영한다.""" path = processed_dir / filename if not path.exists(): logger.warning("도엽 서피스: 도엽 레이어 파일이 없습니다: %s", path) return [] try: with path.open("r", encoding="utf-8") as file: data = json.load(file) except (OSError, json.JSONDecodeError): logger.warning("도엽 서피스: 등고선 GeoJSON을 읽지 못했습니다: %s", path) return [] features = data.get("features") if not isinstance(features, list): return [] transformer = Transformer.from_crs("EPSG:4326", f"EPSG:{epsg}", always_xy=True) def _map(coords: Any) -> Any: if not isinstance(coords, list): return coords if coords and isinstance(coords[0], (int, float)): x, y = transformer.transform(coords[0], coords[1]) return [x, y, *coords[2:]] return [_map(item) for item in coords] converted: list[dict[str, Any]] = [] for feature in features: geometry = feature.get("geometry") or {} coordinates = _map(geometry.get("coordinates")) if coordinates is None: continue converted.append( { "type": "Feature", "properties": feature.get("properties") or {}, "geometry": {"type": geometry.get("type"), "coordinates": coordinates}, } ) return converted def _preview_mesh( x: np.ndarray, y: np.ndarray, z: np.ndarray, valid: np.ndarray ) -> tuple[np.ndarray, np.ndarray]: """프리뷰용 정점·면 — 정점 수가 상한을 넘으면 격자를 성기게 딴다.""" stride = 1 while (len(x) // stride + 1) * (len(y) // stride + 1) > SURFACE_MAX_PREVIEW_VERTICES: stride += 1 px, py = x[::stride], y[::stride] pz, pv = z[::stride, ::stride], valid[::stride, ::stride] vertices = grid_vertices(px, py, pz.astype(np.float64)) faces = grid_faces(len(py), len(px)) return clip_and_compact_mesh(vertices, faces, pv.reshape(-1)) def _rasterize_contour_levels(spec: Any, features: list[dict[str, Any]]) -> np.ndarray: """등고 라인을 격자에 굽는다 — 셀 = 그 위를 지나는 라인의 표고, 그 외 NaN. 배수유역의 `rasterize_contours()`와 두 가지가 다르다(둘 다 서피스 품질 때문이다): · `all_touched=False` — 스치는 셀까지 칠하면 라인이 2px 두께가 되고, 그 폭만큼 정확히 등고 표고인 **평탄 띠**가 생겨 사이 1m 등고선 간격이 찌그러진다 (2026-08-30 사용자 지적: 보간선이 등간격이 아님). · 길이 필터 없음 — 봉우리 폐합 링 같은 짧은 등고선을 버리면 그 일대가 통째로 평평해진다. 배수유역은 노이즈를 버려야 하지만 지형면은 다 있어야 한다. 같은 셀을 두 표고가 지나면 낮은 쪽을 남긴다(배수유역과 같은 규칙). """ from rasterio.features import rasterize from shapely.geometry import shape from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import ( ELEVATION_KEYS, grid_transform, iter_linestrings, ) by_level: dict[float, list[Any]] = {} for feature in features: geometry = feature.get("geometry") if not geometry: continue properties = feature.get("properties") or {} elevation = next( (float(properties[key]) for key in ELEVATION_KEYS if properties.get(key) is not None), None, ) if elevation is None: continue try: parsed = shape(geometry) except Exception: # noqa: BLE001 continue for line in iter_linestrings(parsed): by_level.setdefault(elevation, []).append(line) burned = np.full((spec.n_rows, spec.n_cols), np.nan, dtype=np.float32) transform = grid_transform(spec) for elevation in sorted(by_level, reverse=True): stamp = rasterize( [(line, 1) for line in by_level[elevation]], out_shape=(spec.n_rows, spec.n_cols), transform=transform, fill=0, dtype="uint8", all_touched=False, ).astype(bool) burned[stamp] = elevation logger.info( "도엽 서피스: 등고 라인 %d단을 격자에 굽어 %d셀", len(by_level), int(np.isfinite(burned).sum()), ) return burned def _relax_surface(surface: np.ndarray, fixed: np.ndarray, iterations: int) -> None: """등고 라인을 고정한 채 이웃 평균으로 표고를 **살짝만** 다듬는다 (in-place). 거리 보간은 두 라인의 굽은 정도가 다르면 능선·계곡 중심축에 각진 자국이 남는다. 이웃 평균 몇 회로 그 자국만 지운다. **끝까지 수렴시키면 안 된다.** 수렴한 해는 라플라스 방정식의 해(harmonic)인데, 원뿔·능선 형상 z=r은 harmonic이 아니라 biharmonic이라 harmonic 해로 끌고 가면 마루가 눌리고 평탄 셀이 늘어난다. 등고선 DEM의 표준인 ANUDEM/Topo to Raster가 라플라스가 아니라 **thin plate spline(biharmonic)** 을 쓰는 이유가 이것이다 (Hutchinson 1988/89). 거리 보간은 원뿔을 정확히 재현하므로 그것을 바탕으로 두고 다듬기만 한다. 합성 원뿔 검증(2026-08-30): 거리보간만 오차 0.121m·간격 CV 2.18 → 5회 다듬기 0.118m·1.43(개선) → 20회 이상 CV 50↑·평탄 셀 증가(악화). """ if iterations <= 0: return free = ~fixed & np.isfinite(surface) if not free.any(): return # 체커보드(red-black) 순서 — 한 색을 갱신할 때 이웃(다른 색)은 그대로라 제자리 # 갱신이 안전하다. 과완화는 쓰지 않는다(omega=1): 빨리 수렴시킬수록 harmonic 해에 # 가까워져 능선이 눌린다. 여기서 원하는 건 수렴이 아니라 자국 제거다. omega = np.float32(1.0) rows, cols = np.indices(surface.shape) red = free & (((rows + cols) & 1) == 0) black = free & ~red padded = np.zeros((surface.shape[0] + 2, surface.shape[1] + 2), dtype=np.float32) for _ in range(iterations): for colour in (red, black): padded[1:-1, 1:-1] = surface padded[0, 1:-1] = surface[0] padded[-1, 1:-1] = surface[-1] padded[1:-1, 0] = surface[:, 0] padded[1:-1, -1] = surface[:, -1] neighbours = ( padded[:-2, 1:-1] + padded[2:, 1:-1] + padded[1:-1, :-2] + padded[1:-1, 2:] ) * np.float32(0.25) surface[colour] += omega * (neighbours[colour] - surface[colour]) logger.info("도엽 서피스: 등고선 고정 완화 %d회 (자유 셀 %d개)", iterations, int(free.sum())) def _interpolate_between_contours(burned: np.ndarray, cell_m: float) -> np.ndarray: """2D 거리 보간 — 셀마다 가장 가까운 서로 다른 표고 두 라인 사이를 선형 보간한다. 등고선 정점 Delaunay TIN은 같은 표고 정점 3개짜리 평탄 삼각형이 계단을 만든다. 여기서는 삼각망을 쓰지 않고, 지도 제작의 표준인 **등고선 사이 비례 보간**을 2D에서 직접 한다(2026-08-30 사용자 지시 — 메시는 맨 마지막): z = (L1·d2 + L2·d1) / (d1 + d2) d1·L1 = 가장 가까운 등고 라인까지의 거리·표고, d2·L2 = **표고가 다른 것 중** 가장 가까운 라인. 두 라인 사이에서 z가 거리에 비례해 연속으로 변하므로 계단이 원리적으로 생기지 않는다. 라인 위(d1=0)에서는 그 표고 그대로다. 표고별 거리장을 한 번씩 구하며 **가장 작은 두 값**을 추적한다 — 표고가 서로 다른 것끼리 비교하므로 L1≠L2가 보장된다. (KD-tree로 k개 이웃을 뽑는 방식은 이웃이 전부 같은 라인의 셀이라 다른 표고를 못 찾는다.) `burned`: 라인 셀 = 표고, 그 외 NaN. 계곡 구조선 앵커도 같은 격자에 구워 두면 같은 규칙으로 제약이 된다. """ from scipy.ndimage import distance_transform_edt levels = np.unique(burned[np.isfinite(burned)]) if len(levels) < 2: return np.full(burned.shape, levels[0] if len(levels) else np.nan, dtype=np.float32) infinity = np.float32(np.inf) first_d = np.full(burned.shape, infinity, dtype=np.float32) first_z = np.zeros(burned.shape, dtype=np.float32) second_d = np.full(burned.shape, infinity, dtype=np.float32) second_z = np.zeros(burned.shape, dtype=np.float32) for level in levels: distance = distance_transform_edt(burned != level, sampling=cell_m).astype(np.float32) beats_first = distance < first_d # 1등이 2등으로 밀린다. second_d = np.where(beats_first, first_d, second_d) second_z = np.where(beats_first, first_z, second_z) first_d = np.where(beats_first, distance, first_d) first_z = np.where(beats_first, np.float32(level), first_z) beats_second = ~beats_first & (distance < second_d) second_d = np.where(beats_second, distance, second_d) second_z = np.where(beats_second, np.float32(level), second_z) total = first_d + second_d usable = np.isfinite(second_d) & (total > 1e-9) surface = first_z.astype(np.float32) surface[usable] = ( ( first_z[usable].astype(np.float64) * second_d[usable].astype(np.float64) + second_z[usable].astype(np.float64) * first_d[usable].astype(np.float64) ) / total[usable].astype(np.float64) ).astype(np.float32) logger.info("도엽 서피스: 2D 거리보간 %d셀 (제약 표고 %d단)", surface.size, len(levels)) return surface def _resolve_enclosed_interiors( burned: np.ndarray, present: list[float], surface: np.ndarray, cell_m: float, interval_m: float ) -> np.ndarray: """폐합 등고선 안쪽(봉우리·웅덩이)을 바깥 사면 경사로 연장한다 (in-place). 거리 보간은 "가장 가까운 서로 다른 두 라인 사이"를 채우므로, 마지막 등고선 안쪽에는 더 높은 라인이 없어 아래쪽 라인 쪽으로 끌려 **분화구처럼 파인다**. 등고선이 'ㅜ'에서 점점 짧아지다 사라지는 마루가 바로 이 자리다(2026-08-30 사용자 지적). 폐합 라인 내부에 다른 표고 제약이 하나도 없으면 그 안이 마루(바깥이 낮을 때) 또는 웅덩이(바깥이 높을 때)다. 바깥 사면의 국소 경사를 안쪽으로 연장하되 ±(간격−0.5m)로 제한한다 — 다음 등고선이 없다는 사실과 모순되지 않는 범위다. 처리한 영역의 마스크를 반환한다 — 이어지는 완화에서 함께 고정해야 한다. """ from scipy.ndimage import binary_dilation, binary_fill_holes, distance_transform_edt, label constrained = np.isfinite(burned) band_cells = 8 limit = max(interval_m - 0.5, 0.5) resolved = 0 handled = np.zeros(burned.shape, dtype=bool) for level in present: mask = burned == level interior = binary_fill_holes(mask) & ~mask if not interior.any(): continue components, count = label(interior) for component_id in range(1, count + 1): component = components == component_id if (constrained & component).any(): continue # 안에 다른 제약이 있으면 마루가 아니다(보통의 감싸는 링) ring = binary_dilation(binary_fill_holes(mask) | mask) & ~component & ~mask ring &= np.isfinite(surface) if not ring.any(): continue outside_mean = float(surface[ring].mean()) direction = 1.0 if outside_mean < level else -1.0 inner = distance_transform_edt(component, sampling=cell_m) # 바깥 사면 경사 — 라인 밖 band_cells 이내 유효 셀의 (낙차 / 거리) 평균. outer_distance = distance_transform_edt(~(mask | component), sampling=cell_m) band = (outer_distance > 0) & (outer_distance <= band_cells * cell_m) band &= np.isfinite(surface) & ~component & ~mask if band.any(): slope = float( np.mean(np.abs(level - surface[band].astype(np.float64)) / outer_distance[band]) ) else: slope = 0.0 if slope > 1e-3: offset = np.minimum(slope * inner[component], limit) else: peak = float(inner.max()) offset = (limit / 2.0) * (inner[component] / peak) if peak > 0 else 0.0 surface[component] = level + direction * offset handled |= component resolved += 1 if resolved: logger.info("도엽 서피스: 폐합 등고선 내부 %d곳을 사면 경사로 연장", resolved) return handled def build_sheet_surface_model( project_root: Path, processed_dir: Path, models_dir: Path, route_xy: np.ndarray, epsg: int, ) -> dict[str, Any] | None: """도엽등고선으로 DTM npz·프리뷰 glb를 만들고 surface_models 등록용 dict를 돌려준다. `route_xy`: (N, 2) 노선 정점 XY(사업지 CRS, m). 실패하면 None — 호출측은 분석을 계속한다(도엽 미확보 지역 폴백). """ started = time.monotonic() features = _load_features_metric(processed_dir, epsg) if not features: return None x_min = float(np.min(route_xy[:, 0])) - SHEET_SURFACE_MARGIN_M x_max = float(np.max(route_xy[:, 0])) + SHEET_SURFACE_MARGIN_M y_min = float(np.min(route_xy[:, 1])) - SHEET_SURFACE_MARGIN_M y_max = float(np.max(route_xy[:, 1])) + SHEET_SURFACE_MARGIN_M spec = grid_spec_from_bounds(x_min, y_min, x_max, y_max, SHEET_SURFACE_GRID_M) # ① 2D — 등고 라인을 격자에 굽는다(라인 셀 = 표고, 그 외 NaN). burned = _rasterize_contour_levels(spec, features) present = sorted(np.unique(burned[np.isfinite(burned)]).tolist()) if len(present) < 2: logger.warning("도엽 서피스: 절취 범위 안에 등고선이 부족합니다.") return None # 등고 간격(m) — 마루 연장 상한의 근거. 레벨이 하나뿐이면 5m(주곡선) 폴백. interval_m = float(np.diff(np.array(present)).min()) if len(present) > 1 else 5.0 # ② 2D — 등고선 사이 거리 비례 보간(메시 없음). 여기서 나온 격자에서 1m 등고선을 # 뽑으므로 화면 등고선이 곧 2D 보간선이다(2026-08-30 사용자 지시). # 제약은 **등고선만** 쓴다 — 계곡 구조선 앵커를 1m로 양자화해 섞었더니 제약 # 표고가 29단→121단이 되어 계곡 주변만 1m 간격이 되고 보간선이 등간격을 # 잃었다(2026-08-30 사용자 지적). V자 등고선이 계곡 하강을 이미 담고 있다. surface = _interpolate_between_contours(burned, spec.cell_m) # ③ 폐합 등고선 안쪽(마루·웅덩이)을 바깥 사면 경사로 연장 (2026-08-30 사용자 확정). summits = _resolve_enclosed_interiors(burned, present, surface, spec.cell_m, interval_m) # ④ 등고 라인(과 마루)을 고정한 채 완화 — 중심축 뭉침을 풀어 간격을 고르게 한다. _relax_surface(surface, np.isfinite(burned) | summits, SHEET_SURFACE_RELAX_ITERATIONS) # DtmGridSampler 규약에 맞춰 y 오름차순으로 뒤집어 저장한다. x_coords = spec.cell_centers_x() y_coords = spec.cell_centers_y()[::-1] z_grid = surface[::-1, :].astype(np.float32) valid_grid = np.isfinite(z_grid) if not valid_grid.any(): logger.warning("도엽 서피스: 유효 표고 셀이 없습니다.") return None stem = f"dtm_{SHEET_SOURCE_FILTER}" model_path = models_dir / f"{stem}.npz" preview_path = models_dir / f"{stem}_preview.glb" finite_z = z_grid[valid_grid] # bounds를 npz에 같이 넣는다 — 등고선 API가 이 값을 화면 원점으로 쓴다. 없으면 # LAS structured.npz로 폴백해 메시(glb) 원점과 어긋난다(LAS 없는 설계는 아예 실패). bounds = np.array( [ [x_coords[0], x_coords[-1]], [y_coords[0], y_coords[-1]], [float(finite_z.min()), float(finite_z.max())], ] ) atomic_npz( model_path, x=x_coords, y=y_coords, z=z_grid, valid_mask=valid_grid, bounds=bounds, resolution=np.array([SHEET_SURFACE_GRID_M], np.float32), ) vertices, faces = _preview_mesh(x_coords, y_coords, z_grid, valid_grid) write_glb(preview_path, vertices, faces, bounds) logger.info( "도엽 서피스 생성 완료: %d×%d 격자, 등고 %d단 (%.1fs)", spec.n_rows, spec.n_cols, len(present), time.monotonic() - started, ) return { "model_type": "dtm", "source_filter": SHEET_SOURCE_FILTER, "representation": "regular_grid", "model_file_path": str(model_path.relative_to(project_root)).replace("\\", "/"), "resolution_m": SHEET_SURFACE_GRID_M, "generation_params": { "source_filter": SHEET_SOURCE_FILTER, "representation": "regular_grid", "source": "map_sheet_contours", "margin_m": SHEET_SURFACE_MARGIN_M, }, "layers": [ { "layer_name": f"dtm_{SHEET_SOURCE_FILTER}_preview", "geometry_type": "MESH", "file_path": str(preview_path.relative_to(project_root)).replace("\\", "/"), "file_format": "glb", } ], } def build_sheet_surface_from_route( project_root: Path, processed_dir: Path, models_dir: Path ) -> dict[str, Any] | None: """B03 업로드 계획노선 CSV를 찾아 도엽 서피스를 만든다. 없거나 실패하면 None.""" from common_util.common_util_route_geometry import ( find_planned_route_file, read_planned_route_csv, ) route_file = find_planned_route_file(project_root / "B03_FileInput" / "input") if route_file is None: logger.warning("도엽 서피스: 계획 노선 파일이 없습니다.") return None planned = read_planned_route_csv(route_file) if planned is None or len(planned.vertices) < 2: logger.warning("도엽 서피스: 계획 노선 파일을 읽지 못했습니다: %s", route_file.name) return None route_xy = np.array([(v.x, v.y) for v in planned.vertices], dtype=np.float64) return build_sheet_surface_model( project_root, processed_dir, models_dir, route_xy, planned.epsg or 5186 ) def run_sheet_surface_analysis( project_root: Path, route_csv_path: Path, *, on_progress: Any = None, ) -> dict[str, Any]: """LAS 없는 WF1 — 도엽 확보 후 도엽등고선 서피스만으로 분석 결과를 만든다. 반환 형식은 `run_surface_analysis()`와 같다(save_surface_analysis_to_db 호환). """ from common_util.common_util_route_geometry import read_planned_route_csv def _report(percent: int, stage: str, message: str) -> None: if on_progress is not None: on_progress(percent, stage, message) stage_root = project_root / "B04_PreProcess" processed_dir = stage_root / "processed" models_dir = stage_root / "models" processed_dir.mkdir(parents=True, exist_ok=True) models_dir.mkdir(parents=True, exist_ok=True) planned = read_planned_route_csv(route_csv_path) if planned is None or len(planned.vertices) < 2: raise ValueError(f"계획 노선 파일을 읽지 못했습니다: {route_csv_path.name}") epsg = planned.epsg or 5186 route_xy = np.array([(v.x, v.y) for v in planned.vertices], dtype=np.float64) bounds_dict = { "x": [float(route_xy[:, 0].min()), float(route_xy[:, 0].max())], "y": [float(route_xy[:, 1].min()), float(route_xy[:, 1].max())], "z": [ float(min(v.z for v in planned.vertices)), float(max(v.z for v in planned.vertices)), ], } # VWorld 지도·GIS 벡터·도엽 확보 — LAS 경로와 같은 공용 블록 (지연 import로 순환 회피) _report(30, "download_maps", "VWorld 지도 및 수치지형도 도엽 확보 중") from B04_PreProcess.B04_PreProcess_Engine import download_geodata download_geodata( project_root, processed_dir, bounds_dict, route_csv_path.parent, rebuild=False, default_epsg=f"EPSG:{epsg}", report=_report, ) _report(70, "surface_model", "도엽등고선 3D 서피스 생성 중") model = build_sheet_surface_model(project_root, processed_dir, models_dir, route_xy, epsg) if model is None: raise ValueError("도엽등고선으로 지표면을 만들지 못했습니다 — 도엽 확보를 확인하세요.") _report(95, "saving", "결과 저장 중") return { "processed": { "processed_file_path": str( (processed_dir / _CONTOUR_FILE).relative_to(project_root) ).replace("\\", "/"), "converted_file_path": None, "point_count": int(len(route_xy)), "bounds": { "x_min": bounds_dict["x"][0], "x_max": bounds_dict["x"][1], "y_min": bounds_dict["y"][0], "y_max": bounds_dict["y"][1], }, "statistics": { "min_z": bounds_dict["z"][0], "max_z": bounds_dict["z"][1], "mean_z": None, }, }, "ground_summary": {}, "manifest": {"status": "sheet_only"}, "models": [model], }