From 7f5c0ab069c9eb0d78844ac27ee994cefbd86265 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Tue, 28 Jul 2026 19:32:17 +0900 Subject: [PATCH] auto: 2026-07-28 19:32 (EOMSANGDON-HOME) --- .../B04_wf1_Surface_UI_MapRender.ts | 57 ++++ B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts | 57 ++++ .../B05_wf2_Route_Engine_Drainage.py | 279 ++++++++++++++++++ .../B05_wf2_Route_Engine_Drainage_Basin.py | 257 ++++++++++++++++ .../B05_wf2_Route_Router_Drainage.py | 224 ++++++++++++++ .../B05_wf2_Route_UI_Drainage_Panel.ts | 116 +++++++- B05_wf2_Route/B05_wf2_Route_UI_Style.css | 71 +++++ main.py | 2 + 8 files changed, 1058 insertions(+), 5 deletions(-) create mode 100644 B05_wf2_Route/B05_wf2_Route_Engine_Drainage.py create mode 100644 B05_wf2_Route/B05_wf2_Route_Engine_Drainage_Basin.py create mode 100644 B05_wf2_Route/B05_wf2_Route_Router_Drainage.py diff --git a/B04_wf1_Surface/B04_wf1_Surface_UI_MapRender.ts b/B04_wf1_Surface/B04_wf1_Surface_UI_MapRender.ts index 061e3f85..3892c42e 100644 --- a/B04_wf1_Surface/B04_wf1_Surface_UI_MapRender.ts +++ b/B04_wf1_Surface/B04_wf1_Surface_UI_MapRender.ts @@ -464,3 +464,60 @@ export function drawPreparedLabels( context.fillText(feature.labelText, x, y); } } + +/** 채움 폴리곤 오버레이(배수유역 등). 좌표는 lon/lat 링 1개. */ +export type FilledRing = { + ring: ReadonlyArray; + /** 면적 중심에 얹을 번호. 없으면 라벨을 그리지 않는다. */ + label?: string; +}; + +/** + * lon/lat 폴리곤 링을 파스텔 채움 + 테두리 + 중심 번호로 그린다. + * 사전 투영 캐시를 쓰지 않는 소량(유역 수 개) 오버레이 전용이라 매 프레임 변환해도 부담이 없다. + */ +export function drawFilledRing( + context: CanvasRenderingContext2D, + entry: FilledRing, + normalizer: Normalizer, + view: ViewState, + color: string, +): void { + if (entry.ring.length < 3) return; + const affine = affineOf(view); + let sumX = 0; + let sumY = 0; + context.beginPath(); + entry.ring.forEach(([lon, lat], index) => { + const nx = (lon - normalizer.lonMin) / normalizer.lonRange; + const ny = 1 - (lat - normalizer.latMin) / normalizer.latRange; + const x = nx * affine.ax + affine.bx; + const y = ny * affine.ay + affine.by; + sumX += x; + sumY += y; + if (index === 0) context.moveTo(x, y); + else context.lineTo(x, y); + }); + context.closePath(); + context.fillStyle = color; + context.fill(); + context.strokeStyle = color; + context.lineWidth = 1.6; + context.stroke(); + if (!entry.label) return; + // 면적 중심(정점 평균)에 번호를 원형 배지로 얹는다. + const centerX = sumX / entry.ring.length; + const centerY = sumY / entry.ring.length; + context.beginPath(); + context.arc(centerX, centerY, 11, 0, Math.PI * 2); + context.fillStyle = color; + context.fill(); + context.strokeStyle = "rgba(255, 255, 255, 0.9)"; + context.lineWidth = 1.5; + context.stroke(); + context.font = "600 12px sans-serif"; + context.textAlign = "center"; + context.textBaseline = "middle"; + context.fillStyle = "#1f2937"; + context.fillText(entry.label, centerX, centerY); +} diff --git a/B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts b/B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts index 60637049..e6c23a42 100644 --- a/B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts +++ b/B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts @@ -247,3 +247,60 @@ export async function fetchLatestRoute(projectId: string): Promise; + area_m2: number; + relief_m: number; + flow_length_m: number; + pipe_diameter_mm: number | null; +} + +export interface DrainageBasinResponse { + status: string; + project_id: string; + route_id: number; + basins: DrainageBasin[]; +} + +export async function fetchDrainageCandidates( + projectId: string, +): Promise { + return requestJson(`/projects/${projectId}/drainage/candidates`, { + method: "GET", + }); +} + +/** chainages를 주면 그 위치로 확정 산정하고, 비우면 자동 제안분으로 산정한다. */ +export async function fetchDrainageBasins( + projectId: string, + chainages?: number[], +): Promise { + return requestJson(`/projects/${projectId}/drainage/basins`, { + method: "POST", + body: JSON.stringify({ chainages: chainages ?? [] }), + }); +} diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Drainage.py b/B05_wf2_Route/B05_wf2_Route_Engine_Drainage.py new file mode 100644 index 00000000..99be8d85 --- /dev/null +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Drainage.py @@ -0,0 +1,279 @@ +"""배수유역 산정 엔진. + +관 매설 구조물 측점 후보를 제안하고, 각 측점이 받는 배수유역 경계를 산정한다. +지형 판단은 **도엽 등고선·세류선(하천중심선)·표고점**만 사용한다 — 3D 포인트클라우드나 +지형 메시는 쓰지 않는다(2026-07-28 사용자 지시). + +유역을 나누는 최종 목적은 각 지점의 파이프 관경 결정이다. 유역 경사면에 100년 강우빈도를 +적용해 모이는 물의 양을 산정하고 그 유량으로 관경을 정한다. 관경 수식은 아직 미확정이라 +`estimate_pipe_diameter_mm()`은 골격만 두고 비워 둔다. +""" + +from __future__ import annotations + +import logging +import math +from dataclasses import dataclass, field +from typing import Any + +from shapely.geometry import LineString, Point, shape + +logger = logging.getLogger(__name__) + +# 구조물 측점 사이 최대 허용 간격(m). 세류 교차가 없어도 이 간격을 넘으면 절토부에 추가 배치한다. +MAX_STRUCTURE_SPACING_M = 300.0 +# 같은 세류 교차로 볼 최소 이격(m). 이보다 가까운 교차점은 하나로 묶는다. +MIN_STRUCTURE_SPACING_M = 20.0 +# 유역 경계 탐색 반경(m). 측점에서 이 거리를 넘는 지형은 해당 유역으로 보지 않는다. +MAX_BASIN_RADIUS_M = 800.0 + + +@dataclass +class RouteVertex: + """노선 폴리라인의 한 점. chainage는 시점 기준 누가거리(m).""" + + x: float + y: float + z: float + chainage_m: float + + +@dataclass +class StructureCandidate: + """관 매설 구조물 측점 후보.""" + + chainage_m: float + x: float + y: float + # "stream"=세류 교차, "spacing"=300m 규칙에 따른 보충 배치 + reason: str + stream_name: str | None = None + + +@dataclass +class DrainageBasin: + """한 구조물 측점이 받는 배수유역.""" + + index: int + chainage_m: float + outlet_x: float + outlet_y: float + polygon_lonlat: list[list[float]] = field(default_factory=list) + area_m2: float = 0.0 + # 유역 최고 표고 − 측점 표고(m). 경사면 낙차. + relief_m: float = 0.0 + # 유하거리: 측점에서 유역 최상단까지 물길 길이(m). + flow_length_m: float = 0.0 + pipe_diameter_mm: float | None = None + + +def build_route_vertices(points: list[dict[str, Any]]) -> list[RouteVertex]: + """DB route_points 행을 누가거리가 채워진 정점 목록으로 바꾼다.""" + vertices: list[RouteVertex] = [] + cumulative = 0.0 + previous: tuple[float, float] | None = None + for row in points: + x = float(row["x"]) + y = float(row["y"]) + z = float(row.get("z") or 0.0) + if previous is not None: + cumulative += math.dist(previous, (x, y)) + chainage = row.get("chainage_m") + vertices.append( + RouteVertex( + x=x, + y=y, + z=z, + chainage_m=float(chainage) if chainage is not None else cumulative, + ) + ) + previous = (x, y) + return vertices + + +def _interpolate_vertex( + vertices: list[RouteVertex], chainage_m: float +) -> tuple[float, float, float]: + """누가거리 위치의 (x, y, z)를 선형 보간한다.""" + if not vertices: + return (0.0, 0.0, 0.0) + if chainage_m <= vertices[0].chainage_m: + return (vertices[0].x, vertices[0].y, vertices[0].z) + for previous, current in zip(vertices, vertices[1:]): + if chainage_m <= current.chainage_m: + span = current.chainage_m - previous.chainage_m + ratio = 0.0 if span <= 0 else (chainage_m - previous.chainage_m) / span + return ( + previous.x + (current.x - previous.x) * ratio, + previous.y + (current.y - previous.y) * ratio, + previous.z + (current.z - previous.z) * ratio, + ) + last = vertices[-1] + return (last.x, last.y, last.z) + + +def is_uphill_at(vertices: list[RouteVertex], chainage_m: float, window_m: float = 20.0) -> bool: + """해당 위치가 오르막(절토부)인지 판정한다. + + 내리막(성토부)은 물이 노선 바깥으로 흘러나가므로 배수유역을 만들지 않는다 + (2026-07-28 사용자 지시). 판정은 종단 계획선의 국소 기울기 부호로 한다. + """ + _, _, back_z = _interpolate_vertex(vertices, max(0.0, chainage_m - window_m)) + _, _, forward_z = _interpolate_vertex(vertices, chainage_m + window_m) + return forward_z >= back_z + + +def find_stream_crossings( + vertices: list[RouteVertex], + stream_features: list[dict[str, Any]], +) -> list[StructureCandidate]: + """노선 평면 선형과 세류선의 교차 지점을 찾는다.""" + if len(vertices) < 2: + return [] + route_line = LineString([(vertex.x, vertex.y) for vertex in vertices]) + candidates: list[StructureCandidate] = [] + for feature in stream_features: + geometry = feature.get("geometry") + if not geometry: + continue + try: + stream = shape(geometry) + except Exception: # noqa: BLE001 - 손상된 피처는 건너뛴다 + continue + if stream.is_empty: + continue + intersection = route_line.intersection(stream) + if intersection.is_empty: + continue + name = _stream_name(feature) + for point in _collect_points(intersection): + candidates.append( + StructureCandidate( + chainage_m=route_line.project(point), + x=point.x, + y=point.y, + reason="stream", + stream_name=name, + ) + ) + candidates.sort(key=lambda item: item.chainage_m) + return candidates + + +def _stream_name(feature: dict[str, Any]) -> str | None: + properties = feature.get("properties") or {} + for key in ("명칭", "하천명", "NAME", "name"): + value = properties.get(key) + if value: + return str(value) + return None + + +def _collect_points(geometry: Any) -> list[Point]: + """교차 결과(Point/MultiPoint/LineString 등)에서 대표 점들을 뽑는다.""" + if geometry.geom_type == "Point": + return [geometry] + if geometry.geom_type in {"MultiPoint", "GeometryCollection"}: + points: list[Point] = [] + for part in geometry.geoms: + points.extend(_collect_points(part)) + return points + # 선분끼리 겹쳐 선으로 나온 경우는 중점을 대표로 쓴다. + if geometry.geom_type in {"LineString", "MultiLineString"}: + return [geometry.interpolate(0.5, normalized=True)] + return [] + + +def propose_structure_stations( + vertices: list[RouteVertex], + stream_features: list[dict[str, Any]], +) -> list[StructureCandidate]: + """구조물 측점 후보를 제안한다. + + ① 세류 교차 지점 ② 내리막(성토부) 제외 ③ 직전 측점에서 300m 초과 시 절토부에 보충 배치. + """ + if len(vertices) < 2: + return [] + total_length = vertices[-1].chainage_m + crossings = [ + candidate + for candidate in find_stream_crossings(vertices, stream_features) + if is_uphill_at(vertices, candidate.chainage_m) + ] + + # 너무 가까운 교차는 하나로 본다(같은 계곡을 여러 선분이 지나는 경우). + merged: list[StructureCandidate] = [] + for candidate in crossings: + if merged and candidate.chainage_m - merged[-1].chainage_m < MIN_STRUCTURE_SPACING_M: + continue + merged.append(candidate) + + # 300m 규칙: 빈 구간에 절토부 지점을 찾아 보충한다. + filled: list[StructureCandidate] = [] + previous_chainage = 0.0 + for candidate in [*merged, None]: + boundary = candidate.chainage_m if candidate else total_length + filled.extend(_fill_spacing(vertices, previous_chainage, boundary)) + if candidate: + filled.append(candidate) + previous_chainage = candidate.chainage_m + else: + previous_chainage = boundary + filled.sort(key=lambda item: item.chainage_m) + return filled + + +def _fill_spacing( + vertices: list[RouteVertex], + start_m: float, + end_m: float, +) -> list[StructureCandidate]: + """[start, end] 구간이 300m를 넘으면 절토부 지점에 보충 측점을 만든다.""" + added: list[StructureCandidate] = [] + cursor = start_m + while end_m - cursor > MAX_STRUCTURE_SPACING_M: + target = cursor + MAX_STRUCTURE_SPACING_M + placed = _nearest_uphill(vertices, target, end_m) + if placed is None: + break + x, y, _ = _interpolate_vertex(vertices, placed) + added.append(StructureCandidate(chainage_m=placed, x=x, y=y, reason="spacing")) + cursor = placed + return added + + +def _nearest_uphill( + vertices: list[RouteVertex], + target_m: float, + limit_m: float, + step_m: float = 10.0, +) -> float | None: + """목표 위치에서 가장 가까운 절토부(오르막) 지점을 찾는다. 없으면 None.""" + if is_uphill_at(vertices, target_m): + return target_m + offset = step_m + while offset <= MAX_STRUCTURE_SPACING_M / 2: + for probe in (target_m - offset, target_m + offset): + if probe <= 0 or probe >= limit_m: + continue + if is_uphill_at(vertices, probe): + return probe + offset += step_m + return None + + +def estimate_pipe_diameter_mm( + area_m2: float, + relief_m: float, + flow_length_m: float, + rainfall_mm_per_hour: float | None = None, +) -> float | None: + """유역 제원으로 배수 파이프 관경(mm)을 산정한다. + + 100년 강우빈도와 유역 경사면을 곱해 유출량을 구하고, 그 유량으로 관경을 정하는 것이 + 목적이다. **수식은 아직 확정되지 않았다** — 사용자가 로직을 제공하면 여기를 채운다. + 그때까지는 None을 돌려 호출부가 "미정"으로 표기하게 한다. + """ + # TODO(사용자 로직 대기): 100년 강우강도 × 유역면적 × 유출계수 → 유량 Q → 관경 D 산정. + _ = (area_m2, relief_m, flow_length_m, rainfall_mm_per_hour) + return None diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Drainage_Basin.py b/B05_wf2_Route/B05_wf2_Route_Engine_Drainage_Basin.py new file mode 100644 index 00000000..854dd34f --- /dev/null +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Drainage_Basin.py @@ -0,0 +1,257 @@ +"""배수유역 경계 산정. + +구조물 측점(관 매설 지점)에서 산정상부까지 역추적해 밀폐된 유역 경계를 만든다. +지형 판단 근거는 도엽 등고선·세류선·표고점뿐이다(3D 미사용, 2026-07-28 사용자 지시). + +정상부 판정 규칙(사용자 지시): + 표고점 데이터는 산 정상부가 아닌 경우가 많다. 따라서 **등고선의 동심 폐합 패턴** + (안쪽으로 갈수록 표고가 높아지는 폐합 등고선의 최내곽)으로 정상부를 먼저 판단하고, + 표고점은 그 판정을 보조·검증하는 용도로만 쓴다. +""" + +from __future__ import annotations + +import logging +import math +from typing import Any + +from shapely.geometry import LineString, Point, Polygon, shape +from shapely.ops import unary_union + +from B05_wf2_Route.B05_wf2_Route_Engine_Drainage import ( + MAX_BASIN_RADIUS_M, + DrainageBasin, + RouteVertex, + StructureCandidate, + estimate_pipe_diameter_mm, +) + +logger = logging.getLogger(__name__) + +# 등고선을 폐합으로 볼 때 허용하는 시종점 이격(m). 도엽 경계에서 잘린 선을 걸러낸다. +CLOSED_TOLERANCE_M = 1.0 +# 폐합 등고선이 정상부 후보가 되는 최대 둘레(m). 이보다 크면 산체 전체라 정상부로 보지 않는다. +MAX_SUMMIT_PERIMETER_M = 1200.0 + + +class ContourField: + """도엽 등고선 피처 모음을 표고 조회·정상부 판정에 쓸 수 있게 감싼 것.""" + + def __init__(self, features: list[dict[str, Any]], elevation_keys: tuple[str, ...]): + self.lines: list[tuple[LineString, float]] = [] + self.closed: list[tuple[Polygon, float]] = [] + for feature in features: + elevation = _read_elevation(feature.get("properties") or {}, elevation_keys) + if elevation is None: + continue + geometry = feature.get("geometry") + if not geometry: + continue + try: + geom = shape(geometry) + except Exception: # noqa: BLE001 - 손상된 피처는 건너뛴다 + continue + for line in _iter_lines(geom): + self.lines.append((line, elevation)) + polygon = _as_closed_polygon(line) + if polygon is not None: + self.closed.append((polygon, elevation)) + + def elevation_at(self, x: float, y: float, radius_m: float = 200.0) -> float | None: + """가장 가까운 등고선의 표고를 그 지점의 표고로 본다.""" + point = Point(x, y) + best: tuple[float, float] | None = None + for line, elevation in self.lines: + distance = line.distance(point) + if distance > radius_m: + continue + if best is None or distance < best[0]: + best = (distance, elevation) + return None if best is None else best[1] + + def find_summits(self, near: Point, radius_m: float) -> list[tuple[Polygon, float]]: + """주변의 정상부 후보를 찾는다. + + 동심 폐합 등고선 중 **자기보다 높은 폐합 등고선을 안에 품지 않은 것**이 최내곽, + 즉 정상부다. 도엽 경계에서 잘린 선과 산체 전체를 감싸는 큰 폐합은 제외한다. + """ + nearby = [ + (polygon, elevation) + for polygon, elevation in self.closed + if polygon.length <= MAX_SUMMIT_PERIMETER_M and polygon.distance(near) <= radius_m + ] + summits: list[tuple[Polygon, float]] = [] + for polygon, elevation in nearby: + has_higher_inside = any( + other_elevation > elevation and polygon.contains(other.representative_point()) + for other, other_elevation in nearby + if other is not polygon + ) + if not has_higher_inside: + summits.append((polygon, elevation)) + return summits + + +def _read_elevation(properties: dict[str, Any], keys: tuple[str, ...]) -> float | None: + for key in keys: + value = properties.get(key) + if value is None: + continue + try: + return float(value) + except (TypeError, ValueError): + continue + return None + + +def _iter_lines(geometry: Any) -> list[LineString]: + if geometry.geom_type == "LineString": + return [geometry] + if geometry.geom_type == "MultiLineString": + return list(geometry.geoms) + return [] + + +def _as_closed_polygon(line: LineString) -> Polygon | None: + coords = list(line.coords) + if len(coords) < 4: + return None + if math.dist(coords[0], coords[-1]) > CLOSED_TOLERANCE_M: + return None + try: + polygon = Polygon(coords) + except Exception: # noqa: BLE001 + return None + return polygon if polygon.is_valid and polygon.area > 0 else None + + +def build_basins( + vertices: list[RouteVertex], + candidates: list[StructureCandidate], + contours: ContourField, + stream_features: list[dict[str, Any]], + to_lonlat: Any, +) -> list[DrainageBasin]: + """확정된 구조물 측점별 배수유역을 만든다. + + 측점에 물을 보내는 세류 가지를 따라 위로 올라가 정상부 폐합 등고선까지 닿는 범위를 + 유역으로 본다. 정상부는 ContourField.find_summits가 등고선 폐합 패턴으로 판정한다. + 번호는 노선 시점에 가까운 순서(측점 누가거리 오름차순)로 1부터 매긴다. + """ + route_line = ( + LineString([(vertex.x, vertex.y) for vertex in vertices]) if len(vertices) > 1 else None + ) + streams = _stream_lines(stream_features) + basins: list[DrainageBasin] = [] + for index, candidate in enumerate( + sorted(candidates, key=lambda item: item.chainage_m), start=1 + ): + outlet = Point(candidate.x, candidate.y) + uphill = _uphill_streams(outlet, streams, contours) + summits = contours.find_summits(outlet, MAX_BASIN_RADIUS_M) + boundary = _basin_polygon(outlet, uphill, summits, route_line) + if boundary is None or boundary.is_empty: + continue + outlet_elevation = contours.elevation_at(candidate.x, candidate.y) or 0.0 + top_elevation = max((elevation for _, elevation in summits), default=outlet_elevation) + basin = DrainageBasin( + index=index, + chainage_m=candidate.chainage_m, + outlet_x=candidate.x, + outlet_y=candidate.y, + polygon_lonlat=[list(to_lonlat(x, y)) for x, y in boundary.exterior.coords], + area_m2=float(boundary.area), + relief_m=float(max(0.0, top_elevation - outlet_elevation)), + flow_length_m=_flow_length(outlet, uphill, boundary), + ) + basin.pipe_diameter_mm = estimate_pipe_diameter_mm( + basin.area_m2, basin.relief_m, basin.flow_length_m + ) + basins.append(basin) + return basins + + +def _stream_lines(features: list[dict[str, Any]]) -> list[LineString]: + lines: list[LineString] = [] + for feature in features: + geometry = feature.get("geometry") + if not geometry: + continue + try: + lines.extend(_iter_lines(shape(geometry))) + except Exception: # noqa: BLE001 + continue + return lines + + +def _uphill_streams( + outlet: Point, + streams: list[LineString], + contours: ContourField, + tolerance_m: float = 30.0, +) -> list[LineString]: + """측점에 연결된 세류 가지 중 위쪽(표고가 높아지는 방향)으로 뻗은 것만 모은다.""" + connected = [line for line in streams if line.distance(outlet) <= tolerance_m] + uphill: list[LineString] = [] + outlet_elevation = contours.elevation_at(outlet.x, outlet.y) + for line in connected: + far = _far_end(line, outlet) + far_elevation = contours.elevation_at(far.x, far.y) + if outlet_elevation is None or far_elevation is None or far_elevation >= outlet_elevation: + uphill.append(line) + return uphill + + +def _far_end(line: LineString, outlet: Point) -> Point: + start = Point(line.coords[0]) + end = Point(line.coords[-1]) + return end if start.distance(outlet) <= end.distance(outlet) else start + + +def _basin_polygon( + outlet: Point, + uphill: list[LineString], + summits: list[tuple[Polygon, float]], + route_line: LineString | None, +) -> Polygon | None: + """유역 경계를 만든다. + + 측점 + 상류 세류 + 정상부 폐합 등고선을 함께 감싸는 볼록 껍질을 1차 경계로 삼고, + 노선 아래쪽(성토부 방향)은 노선을 경계로 잘라낸다. 세류가 없으면 정상부까지의 + 반경 안에서 만들어지는 범위만 남는다. + """ + parts: list[Any] = [outlet.buffer(5.0)] + parts.extend(uphill) + parts.extend(polygon for polygon, _ in summits) + if len(parts) <= 1: + return None + hull = unary_union(parts).convex_hull + if hull.geom_type != "Polygon": + return None + if route_line is not None: + hull = _clip_downhill(hull, route_line, outlet) + return hull if hull is not None and hull.geom_type == "Polygon" else None + + +def _clip_downhill(hull: Polygon, route_line: LineString, outlet: Point) -> Polygon | None: + """노선을 경계로 유역을 잘라 산 쪽(상류) 조각만 남긴다.""" + try: + pieces = hull.difference(route_line.buffer(0.5)) + except Exception: # noqa: BLE001 + return hull + if pieces.is_empty: + return hull + parts = list(pieces.geoms) if pieces.geom_type == "MultiPolygon" else [pieces] + # 상류 조각 판별이 애매할 때를 대비해 면적이 가장 큰 조각을 채택한다. + best = max(parts, key=lambda part: part.area, default=None) + return best if best is not None and best.geom_type == "Polygon" else hull + + +def _flow_length(outlet: Point, uphill: list[LineString], boundary: Polygon) -> float: + """유하거리: 측점에서 유역 최상단까지의 물길 길이(m). + + 상류 세류가 있으면 그 물길 길이의 최댓값을, 없으면 유역 안 최원점까지의 직선거리를 쓴다. + """ + if uphill: + return float(max(line.length for line in uphill)) + return float(max((outlet.distance(Point(xy)) for xy in boundary.exterior.coords), default=0.0)) diff --git a/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py b/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py new file mode 100644 index 00000000..fabb361a --- /dev/null +++ b/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py @@ -0,0 +1,224 @@ +"""배수유역도 API 라우터. + +구조물 측점(관 매설) 후보 제안과 배수유역 산정을 제공한다. 지형 근거는 도엽 등고선·세류선· +표고점 GeoJSON뿐이며, 좌표는 사업지 CRS(m)에서 계산하고 응답만 WGS84로 바꿔 내보낸다. +""" + +import json +import logging +from pathlib import Path +from typing import Any +from uuid import UUID + +from fastapi import APIRouter +from fastapi.responses import JSONResponse +from pyproj import Transformer + +from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path +from B05_wf2_Route.B05_wf2_Route_Engine_Drainage import ( + StructureCandidate, + build_route_vertices, + propose_structure_stations, +) +from B05_wf2_Route.B05_wf2_Route_Engine_Drainage_Basin import ContourField, build_basins +from B05_wf2_Route.B05_wf2_Route_Repository import ( + get_latest_route, + get_route_points, + get_surface_crs_epsg, +) +from common_util.common_util_storage import resolve_stored_project_path +from config.config_db import get_db_pool + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/projects", tags=["B05 Route Drainage"]) + +# 도엽 레이어 파일명 (B04 전처리 산출물과 동일 위치) +_CONTOUR_FILE = "도엽_등고선.geojson" +_STREAM_FILE = "도엽_하천중심선.geojson" +# 도엽 등고선의 표고 속성 키. gpkg 등고선(CTRLN_HG)도 함께 본다. +_ELEVATION_KEYS = ("등고수치", "CTRLN_HG", "elevation", "ELEV") + + +def _sheet_dir(stored_path: str) -> Path: + return Path(resolve_stored_project_path(stored_path)) / "B04_wf1_Surface" / "processed" + + +def _load_features(directory: Path, filename: str) -> list[dict[str, Any]]: + """도엽 GeoJSON을 읽어 피처 목록만 돌려준다. 없으면 빈 목록.""" + path = directory / filename + if not path.exists(): + 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") + return features if isinstance(features, list) else [] + + +def _reproject_features( + features: list[dict[str, Any]], + transformer: Transformer | None, +) -> list[dict[str, Any]]: + """WGS84 도엽 좌표를 사업지 CRS(m)로 바꾼다. 거리·면적을 미터로 계산하기 위함.""" + if transformer is None: + return features + converted: list[dict[str, Any]] = [] + for feature in features: + geometry = feature.get("geometry") + if not geometry: + continue + coordinates = _map_coordinates(geometry.get("coordinates"), transformer) + if coordinates is None: + continue + converted.append( + { + "type": "Feature", + "properties": feature.get("properties") or {}, + "geometry": {"type": geometry.get("type"), "coordinates": coordinates}, + } + ) + return converted + + +def _map_coordinates(coordinates: Any, transformer: Transformer) -> Any: + """중첩 좌표 배열을 재귀적으로 변환한다.""" + if not isinstance(coordinates, list) or not coordinates: + return None + first = coordinates[0] + if isinstance(first, (int, float)): + x, y = transformer.transform(float(coordinates[0]), float(coordinates[1])) + return [x, y] + mapped = [_map_coordinates(item, transformer) for item in coordinates] + return [item for item in mapped if item is not None] + + +def _candidate_payload( + candidate: StructureCandidate, + to_lonlat: Any, +) -> dict[str, Any]: + lon, lat = to_lonlat(candidate.x, candidate.y) + return { + "chainage_m": round(candidate.chainage_m, 2), + "x": candidate.x, + "y": candidate.y, + "lon": lon, + "lat": lat, + "reason": candidate.reason, + "stream_name": candidate.stream_name, + } + + +async def _prepare(project_id: UUID) -> dict[str, Any] | JSONResponse: + """노선 정점·도엽 피처·좌표 변환기를 한 번에 준비한다.""" + pool = get_db_pool() + async with pool.acquire() as connection: + stored_path = await get_project_storage_relative_path(connection, project_id) + route = await get_latest_route(connection, project_id) + if not route: + return JSONResponse( + status_code=404, + content={"status": "error", "message": "확정된 경로가 없습니다."}, + ) + points = await get_route_points(connection, int(route["id"])) + epsg = await get_surface_crs_epsg(connection, project_id) + + vertices = build_route_vertices(points) + if len(vertices) < 2: + return JSONResponse( + status_code=400, + content={"status": "error", "message": "노선 좌표가 부족합니다."}, + ) + + source_crs = epsg or "EPSG:5186" + to_lonlat_transformer = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True) + to_metric_transformer = Transformer.from_crs("EPSG:4326", source_crs, always_xy=True) + + directory = _sheet_dir(stored_path) + streams = _reproject_features(_load_features(directory, _STREAM_FILE), to_metric_transformer) + contour_features = _reproject_features( + _load_features(directory, _CONTOUR_FILE), to_metric_transformer + ) + return { + "route_id": int(route["id"]), + "vertices": vertices, + "streams": streams, + "contours": contour_features, + "to_lonlat": lambda x, y: to_lonlat_transformer.transform(x, y), + } + + +@router.get("/{project_id}/drainage/candidates", response_model=None) +async def get_structure_candidates(project_id: UUID) -> dict[str, Any] | JSONResponse: + """관 매설 구조물 측점 후보를 제안한다(세류 교차 + 300m 보충, 성토부 제외).""" + prepared = await _prepare(project_id) + if isinstance(prepared, JSONResponse): + return prepared + candidates = propose_structure_stations(prepared["vertices"], prepared["streams"]) + to_lonlat = prepared["to_lonlat"] + return { + "status": "success", + "project_id": str(project_id), + "route_id": prepared["route_id"], + "candidates": [_candidate_payload(candidate, to_lonlat) for candidate in candidates], + } + + +@router.post("/{project_id}/drainage/basins", response_model=None) +async def post_drainage_basins( + project_id: UUID, + payload: dict[str, Any] | None = None, +) -> dict[str, Any] | JSONResponse: + """확정된 구조물 측점별 배수유역을 산정한다. + + payload에 `chainages`(누가거리 목록)를 주면 그 위치로 확정하고, 없으면 자동 제안분을 쓴다. + """ + prepared = await _prepare(project_id) + if isinstance(prepared, JSONResponse): + return prepared + vertices = prepared["vertices"] + chainages = (payload or {}).get("chainages") + if isinstance(chainages, list) and chainages: + candidates = _candidates_from_chainages(vertices, chainages) + else: + candidates = propose_structure_stations(vertices, prepared["streams"]) + + contours = ContourField(prepared["contours"], _ELEVATION_KEYS) + basins = build_basins( + vertices, candidates, contours, prepared["streams"], prepared["to_lonlat"] + ) + return { + "status": "success", + "project_id": str(project_id), + "route_id": prepared["route_id"], + "basins": [ + { + "index": basin.index, + "chainage_m": round(basin.chainage_m, 2), + "polygon_lonlat": basin.polygon_lonlat, + "area_m2": round(basin.area_m2, 1), + "relief_m": round(basin.relief_m, 2), + "flow_length_m": round(basin.flow_length_m, 1), + # 관경 수식 미확정 — 산정 함수가 None을 돌려주면 프론트가 "미정"으로 표기한다. + "pipe_diameter_mm": basin.pipe_diameter_mm, + } + for basin in basins + ], + } + + +def _candidates_from_chainages(vertices: Any, chainages: list[Any]) -> list[StructureCandidate]: + """사용자가 확정한 누가거리 목록을 후보 구조로 되돌린다.""" + from B05_wf2_Route.B05_wf2_Route_Engine_Drainage import _interpolate_vertex + + candidates: list[StructureCandidate] = [] + for value in chainages: + try: + chainage = float(value) + except (TypeError, ValueError): + continue + x, y, _ = _interpolate_vertex(vertices, chainage) + candidates.append(StructureCandidate(chainage_m=chainage, x=x, y=y, reason="confirmed")) + return candidates 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 5f3ad08d..5cf7fd33 100644 --- a/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Panel.ts +++ b/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Panel.ts @@ -8,15 +8,21 @@ import { import { computeMapRect, createNormalizer, + drawFilledRing, drawPreparedLayer, prepareLayer, prepareMetricPolyline, type GeoJsonCollection, type MapRect, + type Normalizer, type PreparedLayer, type ViewState, } from "../B04_wf1_Surface/B04_wf1_Surface_UI_MapRender"; -import type { RoutePoint } from "./B05_wf2_Route_Api_Fetch"; +import { + fetchDrainageBasins, + type DrainageBasin, + type RoutePoint, +} from "./B05_wf2_Route_Api_Fetch"; // 배수유역도 패널 — 하단 종단 패널 안쪽 우측에 도킹되는 2단 사이드 패널. // 하단 패널이 닫히면 이 패널도 함께 화면에서 사라진다(부모 안에 들어 있으므로 자동). @@ -42,6 +48,18 @@ const LAYER_LABELS: Record = { const ROUTE_COLOR = "#f97316"; const COLLAPSED_KEY = "b05-route-drainage-collapsed"; +/** 유역 오버레이 파스텔 색상. 번호 순으로 돌려쓴다(사용자 지시: 파스텔톤). */ +const BASIN_COLORS = [ + "rgba(167, 216, 199, 0.45)", + "rgba(247, 208, 168, 0.45)", + "rgba(186, 199, 240, 0.45)", + "rgba(241, 183, 199, 0.45)", + "rgba(214, 226, 168, 0.45)", + "rgba(202, 186, 227, 0.45)", + "rgba(168, 214, 232, 0.45)", + "rgba(240, 219, 168, 0.45)", +] as const; + export interface DrainagePanel { root: HTMLElement; /** 프로젝트가 정해지면 배경지도·도엽 레이어를 불러온다. */ @@ -64,6 +82,13 @@ export function createDrainagePanel(): DrainagePanel { layerButtons.className = "b05-drainage__layers"; header.append(title, layerButtons); + // 유역 산정 실행 버튼 — 후보 제안·유역 산정을 한 번에 돌린다(자동 제안 + 사용자 확인 흐름). + const analyzeButton = document.createElement("button"); + analyzeButton.type = "button"; + analyzeButton.className = "b05-drainage__analyze"; + analyzeButton.textContent = "유역 산정"; + header.append(analyzeButton); + const viewport = document.createElement("div"); viewport.className = "b05-drainage__viewport"; const backgroundImage = document.createElement("img"); @@ -76,7 +101,11 @@ export function createDrainagePanel(): DrainagePanel { status.className = "b05-drainage__status"; status.textContent = "노선을 확정하면 배수유역 배경도가 표시됩니다."; viewport.append(backgroundImage, canvas, status); - root.append(panelHandle.root, header, viewport); + // 유역 제원 목록(면적·표고·유하거리·관경). 관경 수식 미확정이라 당분간 "미정"으로 나온다. + const basinList = document.createElement("div"); + basinList.className = "b05-drainage__basins"; + basinList.hidden = true; + root.append(panelHandle.root, header, viewport, basinList); let projectId: string | null = null; let meta: VWorldMeta | null = null; @@ -84,6 +113,9 @@ export function createDrainagePanel(): DrainagePanel { const activeLayers = new Set(DRAINAGE_LAYERS); let routeLayer: PreparedLayer | null = null; let routePoints: ReadonlyArray = []; + let normalizer: Normalizer | null = null; + let basins: DrainageBasin[] = []; + let selectedBasin: number | null = null; let scale = 1; let offsetX = 0; let offsetY = 0; @@ -136,7 +168,22 @@ export function createDrainagePanel(): DrainagePanel { context.clearRect(0, 0, width, height); const mapRect: MapRect = computeMapRect(meta, width, height); const view: ViewState = { width, height, scale, offsetX, offsetY, mapRect }; - // 등고선을 가장 아래에 얇게 깔고 세류·표고점을 그 위에, 노선을 맨 위에 둔다. + // 유역 채움을 가장 아래에 깔아 등고선·세류 판독을 가리지 않게 한다. + if (normalizer) { + basins.forEach((basin) => { + const color = BASIN_COLORS[(basin.index - 1) % BASIN_COLORS.length]; + drawFilledRing( + context, + { ring: basin.polygon_lonlat, label: String(basin.index) }, + normalizer!, + view, + selectedBasin === null || selectedBasin === basin.index + ? color + : color.replace(/0\.45\)$/, "0.18)"), + ); + }); + } + // 등고선을 얇게 깔고 세류·표고점을 그 위에, 노선을 맨 위에 둔다. DRAINAGE_LAYERS.forEach((layer) => { if (!activeLayers.has(layer)) return; const prepared = preparedLayers.get(layer); @@ -161,6 +208,65 @@ export function createDrainagePanel(): DrainagePanel { }); } + /** 유역 제원 목록을 다시 그린다. 항목을 누르면 해당 유역만 진하게 강조한다. */ + function renderBasinList(): void { + basinList.textContent = ""; + basinList.hidden = basins.length === 0; + basins.forEach((basin) => { + const row = document.createElement("button"); + row.type = "button"; + row.className = "b05-drainage__basin" + (selectedBasin === basin.index ? " is-selected" : ""); + const badge = document.createElement("span"); + badge.className = "b05-drainage__basin-index"; + badge.textContent = String(basin.index); + badge.style.background = BASIN_COLORS[(basin.index - 1) % BASIN_COLORS.length]; + const metrics = document.createElement("span"); + metrics.className = "b05-drainage__basin-metrics"; + // 관경은 수식 미확정이라 백엔드가 null을 주며, 확정 전까지 "미정"으로 표기한다. + const pipe = + basin.pipe_diameter_mm === null ? "미정" : `Ø${Math.round(basin.pipe_diameter_mm)}mm`; + metrics.textContent = + `면적 ${formatArea(basin.area_m2)} · 표고 ${basin.relief_m.toFixed(1)}m · ` + + `유하 ${Math.round(basin.flow_length_m)}m · 관경 ${pipe}`; + row.title = `측점 누가거리 ${basin.chainage_m.toFixed(1)}m`; + row.append(badge, metrics); + row.addEventListener("click", () => { + selectedBasin = selectedBasin === basin.index ? null : basin.index; + renderBasinList(); + scheduleDraw(); + }); + basinList.append(row); + }); + } + + function formatArea(areaM2: number): string { + return areaM2 >= 10000 ? `${(areaM2 / 10000).toFixed(2)}ha` : `${Math.round(areaM2)}㎡`; + } + + /** 구조물 측점 후보 제안 + 유역 산정을 백엔드에 요청한다(계산은 전부 백엔드). */ + async function analyze(): Promise { + if (!projectId) return; + analyzeButton.disabled = true; + status.hidden = false; + status.textContent = "배수유역을 산정하는 중…"; + try { + const response = await fetchDrainageBasins(projectId); + basins = response.basins; + selectedBasin = null; + renderBasinList(); + status.hidden = basins.length > 0; + if (basins.length === 0) status.textContent = "산정된 배수유역이 없습니다."; + scheduleDraw(); + } catch (error) { + status.hidden = false; + status.textContent = error instanceof Error ? error.message : "배수유역 산정에 실패했습니다."; + } finally { + analyzeButton.disabled = false; + } + } + + analyzeButton.addEventListener("click", () => void analyze()); + /** 노선 전체가 보이도록 배율·중심을 맞춘다. 노선이 없으면 도엽 전체를 그대로 보여준다. */ function fitToRoute(): void { scale = 1; @@ -216,12 +322,12 @@ export function createDrainagePanel(): DrainagePanel { ); if (sequence !== loadSequence) return; meta = nextMeta; - const normalizer = createNormalizer(nextMeta); + normalizer = createNormalizer(nextMeta); let featureCount = 0; loaded.forEach(([layer, data]) => { if (!data) return; featureCount += data.features?.length ?? 0; - preparedLayers.set(layer, prepareLayer(data, normalizer)); + preparedLayers.set(layer, prepareLayer(data, normalizer!)); }); backgroundImage.src = `${getVWorldMapUrl(activeProjectId, "satellite")}&_t=${Date.now()}`; if (routePoints.length > 1) routeLayer = prepareMetricPolyline(routePoints, nextMeta); diff --git a/B05_wf2_Route/B05_wf2_Route_UI_Style.css b/B05_wf2_Route/B05_wf2_Route_UI_Style.css index d9d16d5b..ad2e6289 100644 --- a/B05_wf2_Route/B05_wf2_Route_UI_Style.css +++ b/B05_wf2_Route/B05_wf2_Route_UI_Style.css @@ -930,3 +930,74 @@ font-size: var(--text-caption); pointer-events: none; } + +.b05-drainage__analyze { + margin-left: auto; + padding: 2px var(--spacing-8); + border: 1px solid + color-mix(in srgb, var(--color-royal-amethyst, rgb(109 40 217)) 55%, transparent); + border-radius: var(--radius-inputs); + background: var(--color-surface); + color: var(--color-text-body); + font-size: var(--text-caption); + cursor: pointer; +} + +.b05-drainage__analyze:disabled { + opacity: 0.45; + cursor: default; +} + +/* 유역 제원 목록 — 면적·유역표고·유하거리·관경(수식 확정 전까지 "미정"). */ +.b05-drainage__basins { + display: flex; + max-height: 34%; + flex: 0 0 auto; + flex-direction: column; + gap: 2px; + overflow-y: auto; + padding: var(--spacing-8); + border-top: 1px solid var(--color-border); +} + +.b05-drainage__basin { + display: flex; + align-items: center; + gap: var(--spacing-8); + padding: var(--spacing-4) var(--spacing-8); + border: 1px solid transparent; + border-radius: var(--radius-inputs); + background: none; + color: var(--color-text-body); + font-size: var(--text-caption); + text-align: left; + cursor: pointer; +} + +.b05-drainage__basin:hover, +.b05-drainage__basin.is-selected { + border-color: var(--color-border); + background: var(--color-surface); +} + +/* 지도 위 서클 번호와 같은 파스텔 색을 써서 목록 항목과 유역을 눈으로 잇는다. */ +.b05-drainage__basin-index { + display: inline-flex; + width: 20px; + height: 20px; + flex: 0 0 auto; + align-items: center; + justify-content: center; + border: 1px solid var(--color-border); + border-radius: 50%; + color: #1f2937; + font-size: 11px; + font-weight: 600; +} + +.b05-drainage__basin-metrics { + min-width: 0; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} diff --git a/main.py b/main.py index e32c62ea..50680a2a 100644 --- a/main.py +++ b/main.py @@ -32,6 +32,7 @@ from B04_wf1_Surface.B04_wf1_Surface_Router_Contour import router as b04_surface from B04_wf1_Surface.B04_wf1_Surface_Router_GIS import router as b04_surface_gis_router from B04_wf1_Surface.B04_wf1_Surface_Router_GIS import tiles_router from B05_wf2_Route.B05_wf2_Route_Router import router as b05_route_router +from B05_wf2_Route.B05_wf2_Route_Router_Drainage import router as b05_drainage_router from B06_wf3_ProfileCross.B06_wf3_ProfileCross_Router import router as b06_section_router from B07_wf4_DesignDetail.B07_wf4_DesignDetail_Router import router as b07_design_router from common_util.common_util_auth import require_company, verify_session @@ -274,6 +275,7 @@ app.include_router(b04_surface_contour_router, dependencies=protected_with_compa app.include_router(b04_surface_gis_router, dependencies=protected_with_company) app.include_router(tiles_router, dependencies=protected_with_company) app.include_router(b05_route_router, dependencies=protected_with_company) +app.include_router(b05_drainage_router, dependencies=protected_with_company) app.include_router(b06_section_router, dependencies=protected_with_company) app.include_router(b07_design_router, dependencies=protected_with_company)