From e1b410c391dc07acc39ec9f4d33b55f93651c0a5 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Fri, 31 Jul 2026 17:44:53 +0900 Subject: [PATCH] =?UTF-8?q?feat(B05):=201=EC=B0=A8=20=EB=B0=B0=EC=88=98?= =?UTF-8?q?=EC=9C=A0=EC=97=AD=EC=9D=84=20=EC=83=81=EB=A5=98=20=EC=84=B8?= =?UTF-8?q?=EB=A5=98=EB=A7=9D=20=EA=B8=B0=EC=A4=80=EC=9C=BC=EB=A1=9C=20?= =?UTF-8?q?=EC=9E=AC=EC=A0=95=EC=9D=98=20+=20=EB=8B=A8=EA=B3=84=20?= =?UTF-8?q?=EA=B2=80=EC=A6=9D=20=ED=99=94=EB=A9=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1차 영역이 도로 전체 버퍼까지 포함해 도로 아래(하류)로 퍼지고, 세류선을 피처 단위로만 갈라 상류망이 통째로 누락되던 문제를 고친다. - 1차 영역 = 도로 교차점 상류로 이어진 세류망만 반경 버퍼. 노선은 버퍼하지 않음. 노선이 영역 밖으로 나간 길이(road_outside_m)를 재서 반경 판단 근거로 노출. - 상류 판정을 연결망 기준으로 교체: unary_union 노딩 -> 도로에서 절단 -> 끝점 그래프 -> 도로 교차 노드를 통과하지 않는 확산. T자로 붙은 지류와 2단계 이상 이어진 지류까지 상류망으로 따라간다. - 상하류 판정 표고를 최근접 등고선 정점에서 TIN 선형보간으로 교체 (ElevationSampler). 최근접 정점은 오차가 등고선 간격만큼 나서 계곡 교차점이 한 등고선 위로 잡히고 상류 조각이 전부 하류로 오판됐다. - 격자 크기 자동 강등 삭제 - config 값을 그대로 쓴다. 셀 수가 많으면 경고만. 자동 강등이 도로 굽기 두께 전제를 조용히 깨뜨렸다. - TIN 삼각망을 격자 범위 + 여유로 클리핑. 결과 동일, 속도만 개선. - 기본 반경 300m -> 50m (단계 검증용). 단계 검증 수단 - GET /drainage/primary-region : TIN/흐름 계산 없이 상류망/하류망/1차영역/격자만 반환 - 프론트 1차영역 버튼 : 상류망(굵은 파랑), 하류망(회색 파선), 1차 영역(초록 채움), 해석 격자를 실제 셀 눈금으로 렌더. 켤 때마다 재요청한다. - 클릭 시 storage/{project}/B05_wf2_Route/drainage/primary_region.geojson 저장 합성 검증(T자 지류 + 2단계 지류 + 하류 지류 + 고아 세류): 상류망 4조각 채택 / 하류망 3조각 / 미연결 1개 제외 - 전부 기대대로. Co-Authored-By: Claude Fable 5 --- B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts | 38 ++ .../B05_wf2_Route_Engine_Watershed_Basin.py | 61 +++- .../B05_wf2_Route_Engine_Watershed_Grid.py | 335 ++++++++++++++---- .../B05_wf2_Route_Router_Drainage.py | 132 ++++++- .../B05_wf2_Route_UI_Drainage_Panel.ts | 185 +++++++++- config/config_system.py | 13 +- 6 files changed, 685 insertions(+), 79 deletions(-) diff --git a/B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts b/B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts index d28e705b..75e6841a 100644 --- a/B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts +++ b/B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts @@ -296,6 +296,44 @@ export interface DrainageBasinResponse { basins: DrainageBasin[]; } +/** 1차 배수유역 근거(단계 검증용). TIN·흐름 계산 없이 세류 상·하류 판정과 격자 범위만 준다. */ +export interface DrainagePrimaryRegion { + status: string; + project_id: string; + route_id: number; + radius_m: number; + /** 도로와 만난 세류선의 상류측 = 1차 영역의 기준선. */ + upstream_lines: Array>; + /** 교차했으나 하류로 판정해 제외한 조각. 판정이 맞는지 눈으로 대조하는 용도. */ + downstream_lines: Array>; + /** 상·하류 어느 망에도 이어지지 않아 제외한 세류 조각 수. */ + no_contact_count: number; + /** 노선이 1차 영역 밖으로 나간 길이(m). 크면 반경을 올려야 한다는 신호. */ + road_outside_m: number; + /** 1차 영역(상류 세류망 버퍼 합집합)의 외곽 링 목록. */ + region_rings: Array>; + grid: { + cell_m: number; + rows: number; + cols: number; + cells: number; + width_m: number; + height_m: number; + /** 격자 bbox 링. 화면은 여기에 rows×cols 간격으로 실제 셀을 그린다. */ + bbox_lonlat: Array<[number, number]>; + }; + /** 영구저장소에 남긴 검증용 GeoJSON 경로. */ + saved_to: string | null; +} + +export async function fetchDrainagePrimaryRegion( + projectId: string, +): Promise { + return requestJson(`/projects/${projectId}/drainage/primary-region`, { + method: "GET", + }); +} + export async function fetchDrainageCandidates( projectId: string, ): Promise { diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Basin.py b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Basin.py index b3a54450..d98a9fea 100644 --- a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Basin.py +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Basin.py @@ -45,12 +45,12 @@ from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Flow import ( ) from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Grid import ( GridSpec, + PrimaryRegion, build_contour_cloud, - build_grid_spec, + build_primary_region, build_terrain_grid, expand_grid_spec, route_elevation_floor, - select_upstream_streams, ) from config.config_system import ( DRAINAGE_DITCH_SAMPLE_M, @@ -164,7 +164,44 @@ def build_drainage_watershed( ) -# ── ①~④ 격자 해석 (캐시 대상) ─────────────────────────────────────────────── +# ── ①~② 1차 배수유역 (단계 검증 대상) ────────────────────────────────────── + + +def resolve_primary_region( + vertices: list[RouteVertex], + route_line: LineString, + contour_features: list[dict[str, Any]], + stream_features: list[dict[str, Any]], +) -> PrimaryRegion | None: + """도로 교차 세류선(상류측)과 노선을 반경 버퍼한 1차 배수유역과 격자 범위를 정한다. + + 상·하류 판정에 쓸 등고선은 노선 주변만 있으면 된다(교차점이 전부 노선 위이므로). + 도엽 전체를 읽으면 이 단계에서만 수십 초가 날아간다. + """ + floor = route_elevation_floor([vertex.z for vertex in vertices]) + near_bounds = route_line.buffer(DRAINAGE_INITIAL_RADIUS_M * 2.0).bounds + cloud = build_contour_cloud(contour_features, floor, near_bounds) + if cloud.is_empty: + logger.warning("배수유역: 노선 주변에 등고선이 없어 1차 영역을 정할 수 없습니다.") + return None + return build_primary_region( + route_line, stream_features, cloud, DRAINAGE_INITIAL_RADIUS_M, DRAINAGE_GRID_SIZE_M + ) + + +def preview_primary_region( + vertices: list[RouteVertex], + contour_features: list[dict[str, Any]], + stream_features: list[dict[str, Any]], +) -> PrimaryRegion | None: + """단계 검증용 — TIN·흐름 계산 없이 1차 배수유역 근거만 뽑는다.""" + if len(vertices) < 2: + return None + route_line = LineString([(vertex.x, vertex.y) for vertex in vertices]) + return resolve_primary_region(vertices, route_line, contour_features, stream_features) + + +# ── ③~④ 격자 해석 (캐시 대상) ─────────────────────────────────────────────── def _solve_grid( @@ -180,13 +217,21 @@ def _solve_grid( logger.info("배수유역: 격자 캐시 재사용 (%s)", cache_path) return cached - floor = route_elevation_floor([vertex.z for vertex in vertices]) - cloud = build_contour_cloud(contour_features, floor) + region = resolve_primary_region(vertices, route_line, contour_features, stream_features) + if region is None: + return None + spec = region.spec + # TIN용 등고선은 격자가 확장될 여지까지 한 번에 읽어 두고, 회차마다 범위 안쪽만 골라 쓴다. + reach = DRAINAGE_EXPAND_STEP_M * DRAINAGE_MAX_EXPAND_ROUNDS + x_min, y_min, x_max, y_max = region.area.bounds + cloud = build_contour_cloud( + contour_features, + route_elevation_floor([vertex.z for vertex in vertices]), + (x_min - reach, y_min - reach, x_max + reach, y_max + reach), + ) if cloud.is_empty: - logger.warning("배수유역: 등고선이 없어 격자 해석을 건너뜁니다.") + logger.warning("배수유역: 1차 영역 안에 등고선이 없어 격자 해석을 건너뜁니다.") return None - streams = select_upstream_streams(route_line, stream_features, cloud) - spec = build_grid_spec(route_line, streams, DRAINAGE_INITIAL_RADIUS_M, DRAINAGE_GRID_SIZE_M) terrain = road = flow = None for round_index in range(DRAINAGE_MAX_EXPAND_ROUNDS + 1): diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py index 54f4a7ad..b8c2c2e3 100644 --- a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py @@ -19,7 +19,7 @@ from __future__ import annotations import logging import math -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any import numpy as np @@ -27,11 +27,12 @@ from scipy.interpolate import LinearNDInterpolator from scipy.ndimage import distance_transform_edt from scipy.spatial import cKDTree from shapely import segmentize -from shapely.geometry import LineString, shape +from shapely.geometry import LineString, MultiPolygon, Polygon, shape from shapely.ops import substring, unary_union from skimage.morphology import reconstruction from config.config_system import ( + DRAINAGE_CONTOUR_CLIP_MARGIN_M, DRAINAGE_CONTOUR_MARGIN_M, DRAINAGE_CONTOUR_MIN_LENGTH_M, DRAINAGE_CONTOUR_RESAMPLE_M, @@ -144,18 +145,23 @@ def _iter_linestrings(geometry: Any) -> list[LineString]: def build_contour_cloud( contour_features: list[dict[str, Any]], elevation_floor_m: float | None = None, + clip_bounds: tuple[float, float, float, float] | None = None, ) -> ContourCloud: """등고선 피처를 표고가 붙은 정점 구름으로 바꾼다. `elevation_floor_m` 아래 등고선은 계획선 최저점보다 낮아 상류 기여가 불가능하므로 버린다. 길이가 짧은 파편도 노이즈로 보고 버리되, 임계 이상인 봉우리 폐합 등고선은 남긴다(봉우리 표고가 사라지면 그 일대 흐름 방향이 통째로 틀어진다). + + `clip_bounds`(x_min, y_min, x_max, y_max)를 주면 그 밖 등고선은 읽지 않는다. 도엽 + 전체 등고선을 다 물고 가면 TIN 삼각망 비용만 커지고 결과는 같다. """ xs: list[np.ndarray] = [] ys: list[np.ndarray] = [] zs: list[np.ndarray] = [] dropped_low = 0 dropped_short = 0 + dropped_outside = 0 for feature in contour_features: geometry = feature.get("geometry") if not geometry: @@ -170,6 +176,9 @@ def build_contour_cloud( parsed = shape(geometry) except Exception: # noqa: BLE001 - 손상된 피처는 건너뛴다 continue + if clip_bounds is not None and _outside_bounds(parsed.bounds, clip_bounds): + dropped_outside += 1 + continue for line in _iter_linestrings(parsed): if line.length < DRAINAGE_CONTOUR_MIN_LENGTH_M: dropped_short += 1 @@ -182,22 +191,30 @@ def build_contour_cloud( zs.append(np.full(coords.shape[0], elevation, dtype=np.float64)) if not xs: logger.warning( - "배수유역: 사용할 등고선이 없습니다(저지대 %d, 파편 %d 제외).", + "배수유역: 사용할 등고선이 없습니다(저지대 %d, 파편 %d, 범위밖 %d 제외).", dropped_low, dropped_short, + dropped_outside, ) return ContourCloud(np.zeros((0, 2)), np.zeros(0)) xy = np.column_stack((np.concatenate(xs), np.concatenate(ys))) z = np.concatenate(zs) logger.info( - "배수유역: 등고선 정점 %d개 (저지대 %d, 파편 %d 제외)", + "배수유역: 등고선 정점 %d개 (저지대 %d, 파편 %d, 범위밖 %d 제외)", xy.shape[0], dropped_low, dropped_short, + dropped_outside, ) return ContourCloud(xy, z) +def _outside_bounds( + bounds: tuple[float, float, float, float], clip: tuple[float, float, float, float] +) -> bool: + return bounds[2] < clip[0] or bounds[0] > clip[2] or bounds[3] < clip[1] or bounds[1] > clip[3] + + # ── ② 세류선 상류측만 남기기 ──────────────────────────────────────────────── @@ -206,14 +223,36 @@ def select_upstream_streams( stream_features: list[dict[str, Any]], cloud: ContourCloud, ) -> list[LineString]: - """노선과 교차하는 세류선을 교차점에서 잘라 상류(고지대)측만 돌려준다. + """도로 교차점 기준 상류측으로 이어진 세류 연결망 전체를 돌려준다.""" + return split_streams_at_road(route_line, stream_features, cloud).upstream - 노선과 만나지 않는 세류선은 판단 근거가 없으므로 그대로 남긴다 — 어차피 격자 해석에서 - 도로에 물이 닿지 않으면 비활성 처리된다. 상·하류 판정은 가장 가까운 등고선 정점의 - 표고 평균으로 한다(TIN은 이 시점에 아직 없다). + +@dataclass +class StreamSplit: + """세류망을 도로 교차점에서 상·하류로 가른 결과. 검증 화면이 그대로 그린다.""" + + upstream: list[LineString] = field(default_factory=list) # 채택 — 1차 영역의 기준 + downstream: list[LineString] = field(default_factory=list) # 도로 아래로 이어진 망 + no_contact: int = 0 # 어느 쪽에도 이어지지 않아 제외한 조각 수 + + +def split_streams_at_road( + route_line: LineString, + stream_features: list[dict[str, Any]], + cloud: ContourCloud, +) -> StreamSplit: + """세류망을 도로에서 끊고, 교차점 상류측으로 **이어진 망 전체**를 채택한다. + + 도엽 하천중심선은 피처가 잘게 쪼개져 있고 지류가 본류 중간에 T자로 붙는다. 그래서 + 피처 단위로 보면 상류망이 통째로 빠진다. 순서를 이렇게 잡는다: + + ① 세류선끼리 `unary_union`으로 노딩 — 중간에서 만나는 지류도 연결로 인식된다 + ② 도로 교차점에서 한 번 더 잘라 상·하류 조각을 물리적으로 분리한다 + ③ 끝점 그래프를 만들고, **도로 교차 노드는 통과하지 못하게** 막는다 + ④ 도로에 접한 조각을 교차점 표고와 비교해 상·하류 씨앗으로 정한다 + ⑤ 씨앗에서 퍼뜨려 이어진 망 전체를 채택 — 도로를 넘어가지 못하므로 상·하류가 섞이지 않는다 """ - tree = cKDTree(cloud.xy) if not cloud.is_empty else None - kept: list[LineString] = [] + lines: list[LineString] = [] for feature in stream_features: geometry = feature.get("geometry") if not geometry: @@ -222,46 +261,148 @@ def select_upstream_streams( parsed = shape(geometry) except Exception: # noqa: BLE001 continue - for line in _iter_linestrings(parsed): - if line.is_empty or line.length <= 0: - continue - if not line.intersects(route_line): - kept.append(line) - continue - kept.extend(_upstream_parts(line, route_line, tree, cloud)) - return kept + lines.extend(line for line in _iter_linestrings(parsed) if line.length > 0) + if not lines: + return StreamSplit() + pieces, crossing_nodes = _cut_network_at_road(lines, route_line) + if not pieces: + return StreamSplit() -def _upstream_parts( - line: LineString, - route_line: LineString, - tree: cKDTree | None, - cloud: ContourCloud, -) -> list[LineString]: - """세류선을 노선 교차점에서 잘라 평균 표고가 높은 조각만 남긴다.""" - cuts = sorted( - { - line.project(point) - for point in _intersection_points(line.intersection(route_line)) - if 0.0 < line.project(point) < line.length - } + node_edges: dict[tuple[float, float], list[int]] = {} + ends: list[tuple[tuple[float, float], tuple[float, float]]] = [] + for index, piece in enumerate(pieces): + head = _node_key(*piece.coords[0]) + tail = _node_key(*piece.coords[-1]) + ends.append((head, tail)) + node_edges.setdefault(head, []).append(index) + node_edges.setdefault(tail, []).append(index) + + sampler = ElevationSampler(cloud) + upper_seeds: set[int] = set() + lower_seeds: set[int] = set() + for index, piece in enumerate(pieces): + touching = [node for node in ends[index] if node in crossing_nodes] + if not touching: + continue + crossing_z = float(np.min(sampler.at(np.array(touching, dtype=np.float64)))) + if _mean_elevation(piece, sampler) > crossing_z: + upper_seeds.add(index) + else: + lower_seeds.add(index) + + upstream = _spread_network(upper_seeds, ends, node_edges, crossing_nodes) + downstream = _spread_network(lower_seeds, ends, node_edges, crossing_nodes) - upstream + logger.info( + "배수유역: 세류 조각 %d개 → 상류망 %d개 채택 / 하류망 %d개 · 미연결 %d개 제외", + len(pieces), + len(upstream), + len(downstream), + len(pieces) - len(upstream) - len(downstream), ) - if not cuts: - return [line] - bounds = [0.0, *cuts, line.length] - parts: list[tuple[float, LineString]] = [] - for start, end in zip(bounds, bounds[1:]): - if end - start < 1.0: + return StreamSplit( + upstream=[pieces[index] for index in sorted(upstream)], + downstream=[pieces[index] for index in sorted(downstream)], + no_contact=len(pieces) - len(upstream) - len(downstream), + ) + + +def _cut_network_at_road( + lines: list[LineString], route_line: LineString +) -> tuple[list[LineString], set[tuple[float, float]]]: + """세류망을 노딩한 뒤 도로 교차점에서 자르고, 그 교차 노드를 함께 돌려준다.""" + noded = unary_union(lines) + pieces: list[LineString] = [] + crossing_nodes: set[tuple[float, float]] = set() + for piece in _iter_linestrings(noded): + if not piece.intersects(route_line): + pieces.append(piece) continue - piece = _substring(line, start, end) - if piece is None: + hits = _intersection_points(piece.intersection(route_line)) + positions = sorted( + { + position + for position in (piece.project(point) for point in hits) + if 0.0 < position < piece.length + } + ) + for point in hits: + crossing_nodes.add(_node_key(point.x, point.y)) + if not positions: + # 끝점이 도로에 닿은 경우 — 자를 필요는 없고 그 끝점이 곧 교차 노드다. + pieces.append(piece) continue - parts.append((_mean_elevation(piece, tree, cloud), piece)) - if not parts: - return [] - highest = max(value for value, _ in parts) - # 최상류 조각과 표고가 비슷한(1m 이내) 조각까지 상류로 본다. 나머지는 하류이므로 버린다. - return [piece for value, piece in parts if highest - value <= 1.0] + bounds = [0.0, *positions, piece.length] + for start, end in zip(bounds, bounds[1:]): + if end - start <= 0: + continue + cut = _substring(piece, start, end) + if cut is not None: + pieces.append(cut) + return pieces, crossing_nodes + + +def _spread_network( + seeds: set[int], + ends: list[tuple[tuple[float, float], tuple[float, float]]], + node_edges: dict[tuple[float, float], list[int]], + blocked: set[tuple[float, float]], +) -> set[int]: + """씨앗 조각에서 끝점을 타고 퍼진다. 도로 교차 노드는 통과하지 않는다.""" + reached = set(seeds) + queue = list(seeds) + while queue: + index = queue.pop() + for node in ends[index]: + if node in blocked: + continue + for neighbour in node_edges.get(node, ()): + if neighbour not in reached: + reached.add(neighbour) + queue.append(neighbour) + return reached + + +def _node_key(x: float, y: float) -> tuple[float, float]: + """끝점 일치 판정용 좌표 키. 노딩 후에도 부동소수 오차가 남아 mm로 반올림한다.""" + return (round(float(x), 3), round(float(y), 3)) + + +class ElevationSampler: + """등고선 구름에서 임의 지점 표고를 읽는다 — 상·하류 판정 전용. + + 최근접 등고선 정점만 쓰면 오차가 등고선 간격(주곡선 5m)만큼 나서, 계곡 교차점 표고가 + 실제보다 한 등고선 위로 잡히고 상류 조각이 통째로 하류로 오판된다. TIN 선형보간을 + 1차로 쓰고, TIN 밖(볼록껍질 외부)만 최근접 정점으로 메운다. + """ + + def __init__(self, cloud: ContourCloud) -> None: + self._z = cloud.z + if cloud.is_empty: + self._interpolator = None + self._tree = None + return + self._interpolator = LinearNDInterpolator(cloud.xy, cloud.z) + self._tree = cKDTree(cloud.xy) + + def at(self, xy: np.ndarray) -> np.ndarray: + """(N, 2) 좌표의 표고 (N,).""" + if self._interpolator is None or self._tree is None: + return np.zeros(xy.shape[0]) + values = np.asarray(self._interpolator(xy), dtype=np.float64) + missing = ~np.isfinite(values) + if missing.any(): + _, indices = self._tree.query(xy[missing]) + values[missing] = self._z[indices] + return values + + +def _point_elevation(point: Any, tree: cKDTree | None, cloud: ContourCloud) -> float: + """가장 가까운 등고선 정점의 표고. TIN이 없는 단계의 근사 표고다.""" + if tree is None: + return 0.0 + _, index = tree.query([[point.x, point.y]]) + return float(cloud.z[index[0]]) def _intersection_points(geometry: Any) -> list[Any]: @@ -287,46 +428,95 @@ def _substring(line: LineString, start: float, end: float) -> LineString | None: return piece -def _mean_elevation(line: LineString, tree: cKDTree | None, cloud: ContourCloud) -> float: - if tree is None: - return 0.0 +def _mean_elevation(line: LineString, sampler: ElevationSampler) -> float: + """선을 10m 간격으로 훑은 평균 표고.""" samples = max(2, int(line.length // 10.0) + 1) positions = np.linspace(0.0, line.length, samples) points = np.array([list(line.interpolate(position).coords)[0] for position in positions]) - _, indices = tree.query(points) - return float(np.mean(cloud.z[indices])) + return float(np.mean(sampler.at(points))) # ── ③ 격자 범위 ───────────────────────────────────────────────────────────── -def build_grid_spec( +@dataclass +class PrimaryRegion: + """1차 배수유역 — 격자 범위의 근거. 검증 화면이 이 내용을 그대로 그린다.""" + + split: StreamSplit + # 상류 세류망을 반경 버퍼해 합친 영역. 노선은 버퍼하지 않는다. + area: Polygon | MultiPolygon | None + spec: GridSpec + radius_m: float + # 1차 영역 밖으로 나간 노선 길이(m). 그 구간 사면은 해석에서 빠진다는 경고 지표. + road_outside_m: float = 0.0 + + +def build_primary_region( route_line: LineString, - streams: list[LineString], + stream_features: list[dict[str, Any]], + cloud: ContourCloud, radius_m: float, cell_m: float = DRAINAGE_GRID_SIZE_M, -) -> GridSpec: - """노선과 상류 세류선을 반경 버퍼한 범위의 bbox로 격자를 잡는다. +) -> PrimaryRegion: + """**상류 세류망만** 반경 버퍼한 범위 = 1차 배수유역, 그 bbox = 해석 격자. - 셀 수가 상한을 넘으면 셀 크기를 자동으로 키워 맞춘다(메모리 보호). 실제 유역 모양은 - 격자가 아니라 흐름 해석이 정한다 — 여기서는 넉넉한 사각 범위만 확보하면 된다. + 노선은 버퍼하지 않는다(2026-07-31 사용자 지시). 1차 영역의 기준은 도로 교차점 상류로 + 이어진 세류선 그 자체이며, 도로를 버퍼하면 도로 아래쪽(하류)까지 영역이 퍼져 의미가 없다. + + 노선이 이 영역 밖으로 나가는 길이는 따로 재서 남긴다 — 그 구간은 도로 셀이 격자에 + 없어 유역이 잡히지 않으므로 반경을 올릴지 판단하는 근거가 된다. """ - geometries = [route_line.buffer(radius_m)] - geometries.extend(line.buffer(radius_m) for line in streams) - x_min, y_min, x_max, y_max = unary_union(geometries).bounds - return _spec_from_bounds(x_min, y_min, x_max, y_max, cell_m) + split = split_streams_at_road(route_line, stream_features, cloud) + geometries = [line.buffer(radius_m) for line in split.upstream] + if not geometries: + logger.warning("배수유역: 상류 세류망이 없어 노선 버퍼로 대체합니다.") + geometries = [route_line.buffer(radius_m)] + area = unary_union(geometries) + x_min, y_min, x_max, y_max = area.bounds + spec = _spec_from_bounds(x_min, y_min, x_max, y_max, cell_m) + outside = route_line.difference(area) + road_outside_m = float(outside.length) if not outside.is_empty else 0.0 + logger.info( + "배수유역: 1차 영역 %.0f㎡, 범위 %.0fm × %.0fm, 격자 %d×%d (%.2fm), 노선 이탈 %.0fm/%.0fm", + area.area, + x_max - x_min, + y_max - y_min, + spec.n_rows, + spec.n_cols, + spec.cell_m, + road_outside_m, + route_line.length, + ) + return PrimaryRegion( + split=split, area=area, spec=spec, radius_m=radius_m, road_outside_m=road_outside_m + ) def _spec_from_bounds( x_min: float, y_min: float, x_max: float, y_max: float, cell_m: float ) -> GridSpec: + """격자 크기는 절대 자동으로 바꾸지 않는다 — config 값이 그대로 쓰인다(사용자 지시). + + 셀 수가 많으면 경고만 남기고 그대로 진행한다. 느리면 `DRAINAGE_GRID_SIZE_M`을 사용자가 + 직접 올린다. 자동 강등은 도로 굽기 두께·정밀도 전제를 조용히 깨뜨려서 금지한다. + """ width = max(x_max - x_min, cell_m) height = max(y_max - y_min, cell_m) - while (width / cell_m) * (height / cell_m) > DRAINAGE_MAX_GRID_CELLS: - cell_m *= 2.0 - logger.warning("배수유역: 셀 수 상한 초과 — 격자 크기를 %.1fm로 키웁니다.", cell_m) n_cols = int(math.ceil(width / cell_m)) n_rows = int(math.ceil(height / cell_m)) + if n_cols * n_rows > DRAINAGE_MAX_GRID_CELLS: + logger.warning( + "배수유역: 격자 %d×%d = %d셀 (%.0fm × %.0fm, 셀 %.1fm) — 권장 상한 %d셀 초과. " + "그대로 진행합니다. 느리면 DRAINAGE_GRID_SIZE_M을 올리세요.", + n_rows, + n_cols, + n_rows * n_cols, + width, + height, + cell_m, + DRAINAGE_MAX_GRID_CELLS, + ) return GridSpec( x_min=x_min, y_max=y_min + n_rows * cell_m, cell_m=cell_m, n_rows=n_rows, n_cols=n_cols ) @@ -345,11 +535,26 @@ def expand_grid_spec(spec: GridSpec, sides: dict[str, bool], step_m: float) -> G def interpolate_elevation(spec: GridSpec, cloud: ContourCloud) -> np.ndarray: - """등고선 정점 Delaunay TIN으로 셀 표고를 선형보간한다. 외부는 NaN.""" + """등고선 정점 Delaunay TIN으로 셀 표고를 선형보간한다. 외부는 NaN. + + 삼각망 비용은 정점 수에 비례한다. 격자 밖 정점으로 만든 삼각형은 어차피 쓰이지 않으므로 + 격자 범위 + 여유만큼만 남기고 잘라낸다 — 결과 표고는 그대로고 속도만 는다. + """ surface = np.full((spec.n_rows, spec.n_cols), np.nan, dtype=np.float32) if cloud.is_empty: return surface - interpolator = LinearNDInterpolator(cloud.xy, cloud.z) + margin = DRAINAGE_CONTOUR_CLIP_MARGIN_M + inside = ( + (cloud.xy[:, 0] >= spec.x_min - margin) + & (cloud.xy[:, 0] <= spec.x_min + spec.n_cols * spec.cell_m + margin) + & (cloud.xy[:, 1] >= spec.y_max - spec.n_rows * spec.cell_m - margin) + & (cloud.xy[:, 1] <= spec.y_max + margin) + ) + if inside.sum() < 3: + logger.warning("배수유역: 격자 범위 안에 등고선 정점이 없습니다.") + return surface + logger.info("배수유역: TIN 정점 %d개 사용 (전체 %d개)", int(inside.sum()), cloud.xy.shape[0]) + interpolator = LinearNDInterpolator(cloud.xy[inside], cloud.z[inside]) xs = spec.cell_centers_x() ys = spec.cell_centers_y() # 행 묶음 단위로 평가해 (행×열) 좌표 배열을 한 번에 들고 있지 않게 한다. diff --git a/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py b/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py index 01b8b369..acb8ab9c 100644 --- a/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py +++ b/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py @@ -22,7 +22,10 @@ from B05_wf2_Route.B05_wf2_Route_Engine_Drainage import ( build_route_vertices, propose_structure_stations, ) -from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Basin import build_drainage_watershed +from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Basin import ( + build_drainage_watershed, + preview_primary_region, +) from B05_wf2_Route.B05_wf2_Route_Repository import ( get_latest_route, get_route_points, @@ -30,7 +33,11 @@ from B05_wf2_Route.B05_wf2_Route_Repository import ( ) from common_util.common_util_storage import resolve_stored_project_path from config.config_db import get_db_pool -from config.config_system import DRAINAGE_CACHE_DIRNAME, DRAINAGE_CACHE_FILENAME +from config.config_system import ( + DRAINAGE_CACHE_DIRNAME, + DRAINAGE_CACHE_FILENAME, + DRAINAGE_REGION_FILENAME, +) logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/projects", tags=["B05 Route Drainage"]) @@ -156,6 +163,7 @@ async def _prepare(project_id: UUID) -> dict[str, Any] | JSONResponse: "vertices": vertices, "streams": streams, "contours": contour_features, + "stored_path": stored_path, "cache_path": _cache_path(stored_path), "to_lonlat": lambda x, y: to_lonlat_transformer.transform(x, y), } @@ -177,6 +185,126 @@ async def get_structure_candidates(project_id: UUID) -> dict[str, Any] | JSONRes } +@router.get("/{project_id}/drainage/primary-region", response_model=None) +async def get_primary_region(project_id: UUID) -> dict[str, Any] | JSONResponse: + """1차 배수유역 근거를 돌려준다 — 단계 검증용, TIN·흐름 계산은 하지 않는다. + + 도로 교차점 상류로 이어진 세류망, 제외된 하류망, 그 상류망을 반경 버퍼한 1차 영역, + 그 bbox로 잡은 격자 정보를 함께 준다. 같은 내용을 영구저장소에 GeoJSON으로도 남겨 + QGIS 등으로 직접 열어 대조할 수 있게 한다. + """ + prepared = await _prepare(project_id) + if isinstance(prepared, JSONResponse): + return prepared + region = await asyncio.to_thread( + preview_primary_region, + prepared["vertices"], + prepared["contours"], + prepared["streams"], + ) + if region is None: + return JSONResponse( + status_code=400, + content={"status": "error", "message": "1차 배수유역을 정할 등고선이 없습니다."}, + ) + to_lonlat = prepared["to_lonlat"] + spec = region.spec + payload = { + "status": "success", + "project_id": str(project_id), + "route_id": prepared["route_id"], + "radius_m": region.radius_m, + # 채택된 상류 세류망 = 1차 영역의 기준선. + "upstream_lines": [_line_lonlat(line, to_lonlat) for line in region.split.upstream], + # 도로 아래로 이어진 하류망 — 판정이 맞는지 눈으로 대조하기 위해 함께 준다. + "downstream_lines": [_line_lonlat(line, to_lonlat) for line in region.split.downstream], + "no_contact_count": region.split.no_contact, + # 노선이 1차 영역 밖으로 나간 길이(m). 크면 반경을 올려야 한다는 신호. + "road_outside_m": round(region.road_outside_m, 1), + # 1차 영역(버퍼 합집합) 외곽 링 목록. + "region_rings": _polygon_rings(region.area, to_lonlat), + "grid": { + "cell_m": spec.cell_m, + "rows": spec.n_rows, + "cols": spec.n_cols, + "cells": spec.size, + "width_m": round(spec.n_cols * spec.cell_m, 1), + "height_m": round(spec.n_rows * spec.cell_m, 1), + # 격자 bbox 링(닫힌 사각형). 프론트가 여기에 cell_m 간격으로 실제 셀을 그린다. + "bbox_lonlat": _grid_bbox_lonlat(spec, to_lonlat), + }, + } + payload["saved_to"] = _save_region_geojson(prepared["stored_path"], payload) + return payload + + +def _save_region_geojson(stored_path: str, payload: dict[str, Any]) -> str | None: + """1차 영역 검증 산출물을 영구저장소에 GeoJSON(WGS84)으로 남긴다.""" + features: list[dict[str, Any]] = [] + for index, ring in enumerate(payload["region_rings"]): + features.append(_geojson_feature("primary_region", index, "Polygon", [ring])) + for index, line in enumerate(payload["upstream_lines"]): + features.append(_geojson_feature("upstream", index, "LineString", line)) + for index, line in enumerate(payload["downstream_lines"]): + features.append(_geojson_feature("downstream", index, "LineString", line)) + features.append(_geojson_feature("grid_bbox", 0, "Polygon", [payload["grid"]["bbox_lonlat"]])) + document = { + "type": "FeatureCollection", + "crs": {"type": "name", "properties": {"name": "urn:ogc:def:crs:OGC:1.3:CRS84"}}, + "properties": { + "radius_m": payload["radius_m"], + "road_outside_m": payload["road_outside_m"], + "no_contact_count": payload["no_contact_count"], + "grid": {key: value for key, value in payload["grid"].items() if key != "bbox_lonlat"}, + }, + "features": features, + } + target = ( + Path(resolve_stored_project_path(stored_path)) + / "B05_wf2_Route" + / DRAINAGE_CACHE_DIRNAME + / DRAINAGE_REGION_FILENAME + ) + try: + target.parent.mkdir(parents=True, exist_ok=True) + with target.open("w", encoding="utf-8") as file: + json.dump(document, file, ensure_ascii=False) + except OSError: + logger.warning("배수유역: 1차 영역 GeoJSON을 저장하지 못했습니다 (%s).", target) + return None + logger.info("배수유역: 1차 영역 GeoJSON 저장 — %s (피처 %d개)", target, len(features)) + return str(target) + + +def _geojson_feature(kind: str, index: int, geom_type: str, coordinates: Any) -> dict[str, Any]: + return { + "type": "Feature", + "properties": {"kind": kind, "index": index}, + "geometry": {"type": geom_type, "coordinates": coordinates}, + } + + +def _line_lonlat(line: Any, to_lonlat: Any) -> list[list[float]]: + return [list(to_lonlat(x, y)) for x, y in line.coords] + + +def _polygon_rings(geometry: Any, to_lonlat: Any) -> list[list[list[float]]]: + """폴리곤/멀티폴리곤의 외곽 링만 뽑아 lonlat으로 바꾼다.""" + if geometry is None or geometry.is_empty: + return [] + parts = geometry.geoms if geometry.geom_type == "MultiPolygon" else [geometry] + return [[list(to_lonlat(x, y)) for x, y in part.exterior.coords] for part in parts] + + +def _grid_bbox_lonlat(spec: Any, to_lonlat: Any) -> list[list[float]]: + x_min = spec.x_min + x_max = spec.x_min + spec.n_cols * spec.cell_m + y_max = spec.y_max + y_min = spec.y_max - spec.n_rows * spec.cell_m + corners = ((x_min, y_min), (x_min, y_max), (x_max, y_max), (x_max, y_min), (x_min, y_min)) + return [list(to_lonlat(x, y)) for x, y in corners] + + @router.post("/{project_id}/drainage/basins", response_model=None) async def post_drainage_basins( project_id: UUID, diff --git a/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Panel.ts b/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Panel.ts index eba48bd3..c127a8ed 100644 --- a/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Panel.ts +++ b/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Panel.ts @@ -21,7 +21,9 @@ import { } from "../B04_wf1_Surface/B04_wf1_Surface_UI_MapRender"; import { fetchDrainageBasins, + fetchDrainagePrimaryRegion, type DrainageBasin, + type DrainagePrimaryRegion, type RoutePoint, } from "./B05_wf2_Route_Api_Fetch"; import { createPipeEditor } from "./B05_wf2_Route_UI_Drainage_Pipes"; @@ -107,7 +109,16 @@ export function createDrainagePanel(): DrainagePanel { autoButton.type = "button"; autoButton.className = "b05-drainage__analyze b05-drainage__tool"; autoButton.textContent = "자동 제안"; - header.append(analyzeButton, editButton, deleteButton, autoButton); + // 1차 영역 확인 — TIN·흐름 계산 없이 세류 상·하류 판정과 격자 범위만 그려 눈으로 검증한다. + const regionButton = document.createElement("button"); + regionButton.type = "button"; + regionButton.className = "b05-drainage__analyze b05-drainage__tool"; + regionButton.textContent = "1차 영역"; + regionButton.title = + "도로와 만나는 세류선의 상류측(파랑 굵은 선)·하류측(회색 파선)과 " + + "그 반경 버퍼로 잡은 1차 배수유역, 해석 격자 범위를 표시합니다."; + regionButton.setAttribute("aria-pressed", "false"); + header.append(analyzeButton, editButton, deleteButton, autoButton, regionButton); const viewport = document.createElement("div"); viewport.className = "b05-drainage__viewport"; @@ -145,6 +156,9 @@ export function createDrainagePanel(): DrainagePanel { // 2차 전체 배수유역 외곽선 = 분수령. 별도 "능선" 레이어를 두지 않고 이 선 하나로 표시한다 // (전체 유역 외곽선과 능선이 같은 선이므로 — 2026-07-31 사용자 지시). let mainBoundary: Array<[number, number]> = []; + // 1차 영역 검증 오버레이. null이면 표시하지 않는다. + let primaryRegion: DrainagePrimaryRegion | null = null; + let showRegion = false; let scale = 1; let offsetX = 0; let offsetY = 0; @@ -213,6 +227,8 @@ export function createDrainagePanel(): DrainagePanel { }); // 전체 유역 외곽선 = 분수령(능선). 세부유역 경계와 구분되게 파선 한 겹만 얹는다. if (mainBoundary.length > 2) drawRidgeRing(context, mainBoundary, normalizer, view); + // 1차 영역 검증 오버레이는 채움 위·등고선 아래에 깐다. + if (showRegion && primaryRegion) drawPrimaryRegion(context, normalizer, view); } // 등고선을 얇게 깔고 세류를 그 위에, 노선을 맨 위에 둔다. DRAINAGE_LAYERS.forEach((layer) => { @@ -234,6 +250,121 @@ export function createDrainagePanel(): DrainagePanel { updateImageTransform(); } + /** lon/lat 폴리라인을 화면 좌표로 옮겨 한 줄 그린다(1차 영역 오버레이 전용). */ + function strokeLonLat( + context: CanvasRenderingContext2D, + line: ReadonlyArray, + map: Normalizer, + view: ViewState, + ): void { + if (line.length < 2) return; + const ax = view.mapRect.width * view.scale; + const bx = view.mapRect.x * view.scale + (view.width / 2) * (1 - view.scale) + view.offsetX; + const ay = view.mapRect.height * view.scale; + const by = view.mapRect.y * view.scale + (view.height / 2) * (1 - view.scale) + view.offsetY; + context.beginPath(); + line.forEach(([lon, lat], index) => { + const x = ((lon - map.lonMin) / map.lonRange) * ax + bx; + const y = (1 - (lat - map.latMin) / map.latRange) * ay + by; + if (index === 0) context.moveTo(x, y); + else context.lineTo(x, y); + }); + context.stroke(); + } + + /** 해석 격자를 실제 셀 눈금으로 그린다. + * + * 셀 간격이 화면에서 너무 촘촘하면(2px 미만) 눈금이 뭉개져 회색 덩어리가 되므로, + * 그때는 테두리만 남기고 "확대하면 셀이 보인다"는 상태를 유지한다. */ + function drawGridCells( + context: CanvasRenderingContext2D, + map: Normalizer, + view: ViewState, + region: DrainagePrimaryRegion, + ): void { + const ring = region.grid.bbox_lonlat; + if (ring.length < 4) return; + const lons = ring.map(([lon]) => lon); + const lats = ring.map(([, lat]) => lat); + const lonMin = Math.min(...lons); + const lonMax = Math.max(...lons); + const latMin = Math.min(...lats); + const latMax = Math.max(...lats); + const ax = view.mapRect.width * view.scale; + const bx = view.mapRect.x * view.scale + (view.width / 2) * (1 - view.scale) + view.offsetX; + const ay = view.mapRect.height * view.scale; + const by = view.mapRect.y * view.scale + (view.height / 2) * (1 - view.scale) + view.offsetY; + const toX = (lon: number): number => ((lon - map.lonMin) / map.lonRange) * ax + bx; + const toY = (lat: number): number => (1 - (lat - map.latMin) / map.latRange) * ay + by; + + const left = toX(lonMin); + const right = toX(lonMax); + const top = toY(latMax); + const bottom = toY(latMin); + context.save(); + context.setLineDash([]); + // 셀 눈금 — 열/행 수로 나눠 실제 셀 경계를 그대로 찍는다. + const cellWidthPx = Math.abs(right - left) / Math.max(region.grid.cols, 1); + const cellHeightPx = Math.abs(bottom - top) / Math.max(region.grid.rows, 1); + if (Math.min(cellWidthPx, cellHeightPx) >= 2) { + context.lineWidth = 0.5; + context.strokeStyle = "rgba(120, 113, 108, 0.35)"; + context.beginPath(); + for (let col = 0; col <= region.grid.cols; col += 1) { + const x = left + (right - left) * (col / region.grid.cols); + if (x < -50 || x > view.width + 50) continue; + context.moveTo(x, top); + context.lineTo(x, bottom); + } + for (let row = 0; row <= region.grid.rows; row += 1) { + const y = top + (bottom - top) * (row / region.grid.rows); + if (y < -50 || y > view.height + 50) continue; + context.moveTo(left, y); + context.lineTo(right, y); + } + context.stroke(); + } + // 격자 전체 테두리는 항상 그린다. + context.setLineDash([10, 6]); + context.lineWidth = 1.5; + context.strokeStyle = "rgba(120, 113, 108, 0.9)"; + context.strokeRect(left, top, right - left, bottom - top); + context.restore(); + } + + /** 1차 배수유역 근거를 겹쳐 그린다 — 단계 검증용. */ + function drawPrimaryRegion( + context: CanvasRenderingContext2D, + map: Normalizer, + view: ViewState, + ): void { + const region = primaryRegion; + if (!region) return; + context.save(); + // ① 해석 격자 — bbox 테두리 + 실제 셀 눈금. + drawGridCells(context, map, view, region); + // ② 1차 배수유역 = 상류 세류망의 반경 버퍼 합집합. + context.setLineDash([]); + context.lineWidth = 2; + context.strokeStyle = "rgba(5, 150, 105, 0.95)"; + context.fillStyle = "rgba(16, 185, 129, 0.12)"; + region.region_rings.forEach((ring) => { + strokeLonLat(context, ring, map, view); + context.fill(); + }); + // ③ 도로 아래로 이어진 하류망 — 판정이 맞는지 대조하도록 회색 파선으로 남긴다. + context.setLineDash([6, 5]); + context.lineWidth = 2; + context.strokeStyle = "rgba(120, 113, 108, 0.85)"; + region.downstream_lines.forEach((line) => strokeLonLat(context, line, map, view)); + // ④ 채택된 상류망 = 1차 영역의 기준선. 가장 굵게, 맨 위에. + context.setLineDash([]); + context.lineWidth = 4; + context.strokeStyle = "rgba(29, 78, 216, 0.95)"; + region.upstream_lines.forEach((line) => strokeLonLat(context, line, map, view)); + context.restore(); + } + /** 현재 프레임 뷰 상태 (draw()와 동일 계산 — 포인터 히트 판정용). */ function currentView(): ViewState { const rect = viewport.getBoundingClientRect(); @@ -351,6 +482,53 @@ export function createDrainagePanel(): DrainagePanel { void analyze(true); }); + /** 1차 영역 근거를 불러와 겹쳐 그린다. + * + * 켤 때는 **항상 다시 요청한다** — config(반경·격자)를 바꾸고 서버를 재시작한 뒤 + * 눌렀는데 캐시된 예전 결과가 나오면 검증이 성립하지 않는다. 끌 때만 요청 없이 숨긴다. */ + async function toggleRegion(): Promise { + if (!projectId) return; + if (showRegion) { + showRegion = false; + regionButton.classList.remove("is-active"); + regionButton.setAttribute("aria-pressed", "false"); + status.hidden = true; + scheduleDraw(); + return; + } + regionButton.disabled = true; + status.hidden = false; + status.textContent = "1차 배수유역을 확인하는 중…"; + try { + primaryRegion = await fetchDrainagePrimaryRegion(projectId); + showRegion = true; + regionButton.classList.add("is-active"); + regionButton.setAttribute("aria-pressed", "true"); + status.textContent = regionSummary(primaryRegion); + } catch (error) { + status.textContent = + error instanceof Error ? error.message : "1차 배수유역을 확인하지 못했습니다."; + } finally { + regionButton.disabled = false; + scheduleDraw(); + } + } + + /** 상태줄에 띄울 1차 영역 요약 — 격자 셀 수를 보고 격자 크기를 조정할 근거가 된다. */ + function regionSummary(region: DrainagePrimaryRegion): string { + const cells = region.grid.cells.toLocaleString(); + const outside = + region.road_outside_m > 0 ? ` · 노선 이탈 ${Math.round(region.road_outside_m)}m` : ""; + return ( + `1차 영역(반경 ${region.radius_m}m): 상류망 ${region.upstream_lines.length}조각 채택 / ` + + `하류망 ${region.downstream_lines.length}조각·미연결 ${region.no_contact_count}개 제외 · ` + + `격자 ${region.grid.width_m}×${region.grid.height_m}m, ` + + `${region.grid.cell_m}m 셀 ${cells}개${outside}` + ); + } + + regionButton.addEventListener("click", () => void toggleRegion()); + /** 노선 전체가 보이도록 배율·중심을 맞춘다. 노선이 없으면 도엽 전체를 그대로 보여준다. */ function fitToRoute(): void { scale = 1; @@ -388,6 +566,11 @@ export function createDrainagePanel(): DrainagePanel { const sequence = ++loadSequence; meta = null; preparedLayers.clear(); + // 1차 영역은 프로젝트·노선에 종속이므로 새로 불러올 때 버린다. + primaryRegion = null; + showRegion = false; + regionButton.classList.remove("is-active"); + regionButton.setAttribute("aria-pressed", "false"); routeLayer = null; backgroundImage.removeAttribute("src"); status.hidden = false; diff --git a/config/config_system.py b/config/config_system.py index 8710eb23..29852717 100644 --- a/config/config_system.py +++ b/config/config_system.py @@ -242,13 +242,15 @@ SKELETON_NODE_SPACING_M = float(os.getenv("SKELETON_NODE_SPACING_M", "10.0")) # ───────────────────────────────────────────────────────────────────────── # 해석 격자 한 변(m). 작을수록 정밀하나 셀 수가 제곱으로 늘어난다. DRAINAGE_GRID_SIZE_M = float(os.getenv("DRAINAGE_GRID_SIZE_M", "1.0")) -# 1차 배수유역 반경(m). 정리된 세류선과 노선을 이 반경으로 버퍼해 초기 해석 범위를 잡는다. -DRAINAGE_INITIAL_RADIUS_M = float(os.getenv("DRAINAGE_INITIAL_RADIUS_M", "300.0")) +# 1차 배수유역 반경(m). 도로 교차점 상류로 이어진 세류망을 이 반경으로 버퍼한 범위가 +# 1차 영역이며 그 bbox가 해석 격자다. 노선은 버퍼하지 않는다(2026-07-31 사용자 지시). +DRAINAGE_INITIAL_RADIUS_M = float(os.getenv("DRAINAGE_INITIAL_RADIUS_M", "50.0")) # 활성 셀이 격자 최외곽에 닿았을 때 한 번에 넓히는 폭(m). DRAINAGE_EXPAND_STEP_M = float(os.getenv("DRAINAGE_EXPAND_STEP_M", "200.0")) # 확장 반복 상한. 경계 링이 전부 비활성이 되면 그 전에 스스로 멈춘다(안전핀). DRAINAGE_MAX_EXPAND_ROUNDS = int(os.getenv("DRAINAGE_MAX_EXPAND_ROUNDS", "6")) -# 격자 셀 수 상한. 초과하면 셀 크기를 자동으로 키워 맞춘다(메모리 보호). +# 격자 셀 수 권장 상한. 넘으면 **경고만** 남기고 그대로 계산한다 — 격자 크기 자동 조절은 +# 하지 않는다(2026-07-31 사용자 지시). 느리면 위 DRAINAGE_GRID_SIZE_M을 직접 올린다. DRAINAGE_MAX_GRID_CELLS = int(os.getenv("DRAINAGE_MAX_GRID_CELLS", "16000000")) # 도로 폭(m). 이 폭으로 노선을 격자에 구워 D8 흐름이 도로를 대각선으로 건너뛰지 못하게 한다. DRAINAGE_ROAD_WIDTH_M = float(os.getenv("DRAINAGE_ROAD_WIDTH_M", "4.0")) @@ -258,6 +260,9 @@ DRAINAGE_CONTOUR_MARGIN_M = float(os.getenv("DRAINAGE_CONTOUR_MARGIN_M", "10.0") DRAINAGE_CONTOUR_MIN_LENGTH_M = float(os.getenv("DRAINAGE_CONTOUR_MIN_LENGTH_M", "20.0")) # 등고선 정점 재샘플 간격(m). 조밀할수록 TIN이 정확하나 Delaunay 비용이 커진다. DRAINAGE_CONTOUR_RESAMPLE_M = float(os.getenv("DRAINAGE_CONTOUR_RESAMPLE_M", "5.0")) +# TIN 삼각망을 만들 때 격자 범위 밖으로 남길 여유(m). 도엽 전체 등고선을 다 물면 삼각망 +# 비용만 커지고 결과는 같다. 여유가 0이면 격자 가장자리가 TIN 밖으로 나가 NaN이 된다. +DRAINAGE_CONTOUR_CLIP_MARGIN_M = float(os.getenv("DRAINAGE_CONTOUR_CLIP_MARGIN_M", "100.0")) # 평탄면 해소용 미세 경사(m/셀). 채움 후 흐름 방향이 없는 셀에 출구 쪽 경사를 만들어 준다. DRAINAGE_FLAT_EPSILON_M = float(os.getenv("DRAINAGE_FLAT_EPSILON_M", "0.001")) # 관 매설 최대 간격(m). 이 간격을 넘으면 흐름 강도가 가장 큰 지점에 관을 보충한다. @@ -273,6 +278,8 @@ DRAINAGE_MIN_BASIN_AREA_M2 = float(os.getenv("DRAINAGE_MIN_BASIN_AREA_M2", "100. # 격자 해석 결과 캐시 파일명. 프로젝트 저장소의 B05_wf2_Route/drainage/ 아래에 놓인다. DRAINAGE_CACHE_DIRNAME = "drainage" DRAINAGE_CACHE_FILENAME = "watershed_grid.npz" +# 1차 영역 검증 산출물. 버튼을 누를 때마다 덮어써서 사람이 QGIS 등으로 직접 열어볼 수 있게 한다. +DRAINAGE_REGION_FILENAME = "primary_region.geojson" # ─────────────────────────────────────────────────────────────────────────