"""계획 노선 기하 공용 유틸 — 정점·누가거리·세류 교차점. 배수유역 분석(B04)과 관 편집·세부유역(B05)이 같은 노선 표현을 써야 하므로 여기 한 곳에만 정의한다. 어느 한쪽 페이지 폴더에 두면 반대 방향 import가 생긴다. 노선 원천은 두 가지다. · B03에 업로드된 **계획 노선 파일**(CSV) — 배수유역 분석의 입력 · DB `route_points` — B05에서 탐색·확정한 노선 둘 다 같은 `RouteVertex` 목록으로 바꿔 아래 함수들이 그대로 받는다. """ from __future__ import annotations import csv import logging import math from dataclasses import dataclass from pathlib import Path from typing import Any from shapely.geometry import LineString, Point, shape logger = logging.getLogger(__name__) # 계획 노선 CSV 열 이름 후보. B03이 여러 형식을 받게 되므로 흔한 표기를 모두 받아 준다. _X_KEYS = ("x", "X", "동", "easting", "EASTING") _Y_KEYS = ("y", "Y", "북", "northing", "NORTHING") _Z_KEYS = ("z", "Z", "표고", "elevation", "ELEV") _ORDER_KEYS = ("sequence", "order", "seq", "no", "번호") _EPSG_KEYS = ("crs_epsg", "epsg", "EPSG") @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"=최대 간격 규칙 보충, "confirmed"=사용자 확정 reason: str stream_name: str | None = None @dataclass class PlannedRoute: """계획 노선 파일에서 읽은 노선.""" vertices: list[RouteVertex] epsg: int | None name: str | None source: Path @property def line(self) -> LineString: return LineString([(vertex.x, vertex.y) for vertex in self.vertices]) def read_planned_route_csv(path: Path) -> PlannedRoute | None: """계획 노선 CSV를 읽어 정점 목록으로 바꾼다. 열 이름은 대소문자·한글 표기를 함께 받아 준다(B03이 여러 형식을 수용할 예정). `sequence`가 있으면 그 순서로 정렬하고, 없으면 파일에 적힌 순서를 그대로 쓴다. """ try: with path.open("r", encoding="utf-8-sig", newline="") as file: rows = list(csv.DictReader(file)) except (OSError, csv.Error, UnicodeDecodeError): logger.warning("계획 노선 CSV를 읽지 못했습니다: %s", path) return None if not rows: return None epsg = _first_int(rows[0], _EPSG_KEYS) name = _first_text(rows[0], ("route_name", "name", "노선명")) parsed: list[tuple[float, float, float, float]] = [] # (정렬키, x, y, z) for index, row in enumerate(rows): x = _first_float(row, _X_KEYS) y = _first_float(row, _Y_KEYS) if x is None or y is None: continue order = _first_float(row, _ORDER_KEYS) parsed.append( (float(index) if order is None else order, x, y, _first_float(row, _Z_KEYS) or 0.0) ) if len(parsed) < 2: logger.warning("계획 노선 CSV에 좌표가 2점 미만입니다: %s", path) return None parsed.sort(key=lambda item: item[0]) vertices: list[RouteVertex] = [] cumulative = 0.0 previous: tuple[float, float] | None = None for _, x, y, z in parsed: if previous is not None: cumulative += math.dist(previous, (x, y)) vertices.append(RouteVertex(x=x, y=y, z=z, chainage_m=cumulative)) previous = (x, y) logger.info( "계획 노선 %s: 정점 %d개, 연장 %.0fm, EPSG %s", path.name, len(vertices), cumulative, epsg ) return PlannedRoute(vertices=vertices, epsg=epsg, name=name, source=path) def find_planned_route_file(input_dir: Path) -> Path | None: """B03 입력 폴더에서 계획 노선 파일을 찾는다. 여러 개면 가장 최근 것.""" if not input_dir.exists(): return None candidates = sorted( input_dir.rglob("*.csv"), key=lambda item: item.stat().st_mtime, reverse=True ) return candidates[0] if candidates else 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: """해당 위치가 오르막(절토부)인지 종단 계획선의 국소 기울기 부호로 판정한다.""" _, _, 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 _first_text(row: dict[str, Any], keys: tuple[str, ...]) -> str | None: for key in keys: value = row.get(key) if value not in (None, ""): return str(value).strip() return None def _first_float(row: dict[str, Any], keys: tuple[str, ...]) -> float | None: text = _first_text(row, keys) if text is None: return None try: return float(text) except ValueError: return None def _first_int(row: dict[str, Any], keys: tuple[str, ...]) -> int | None: value = _first_float(row, keys) return None if value is None else int(value)