eomsangdonandClaude Opus 5
fe72ea041b
원청 정식 계획노선이 shapefile(UTM-K)로, 지형이 별도 PRJ(동부원점 Bessel)로 들어오는데 입력 경로가 shapefile 확장자를 막고 PRJ를 프로젝트당 1개로 전제했다. - 업로드 허용에 .shp/.shx/.dbf/.cpg 추가, 한 번에 보낼 파일 수 5 -> 10 - B03_FileInput_Engine_Shapefile: ESRI 규격 직접 파싱(GDAL 미사용). 형제 파일이 아직 안 왔어도 .shp 하나로 기하를 읽는다. .cpg 내용이 949뿐인 실물을 CP949로 정규화해 한글 속성을 살린다. - 노선 판독을 read_planned_route로 일원화(CSV/shapefile), PlannedRoute에 crs_input 추가 - 변환 입력은 EPSG 코드가 아니라 crs_input_from_prj가 주는 값(EPSG:n 또는 원문 WKT)이다. 실물 PRJ 2종 모두 to_epsg가 None이다. - shapefile 세트를 input/shp/ 한 폴더에 모은다(GDAL 요건). 노선 PRJ가 그 안에 남으므로 지형 PRJ(input/prj/)와 파일명 정렬 운에 기대지 않고 갈린다. find_project_prj가 지형 PRJ를 프로젝트 좌표계로 고른다. - 필수 세트를 노선 1종(csv 또는 shp) + prj + tfw로 완화, shp면 shx/dbf 동반 필수. - UI: 확장자 단독 슬롯 매칭을 basename 그룹핑으로 바꿔 노선 PRJ와 지형 PRJ가 같은 슬롯을 다투지 않게 하고, 노선 슬롯이 파일 한 벌을 담아 함께 전송한다. 자체검증: tmp/tests/test_route_shapefile_input.py 9개 통과, tsc --noEmit 통과, ruff check/format 통과. 전체 스위트 잔여 실패 11건은 HEAD 사본(git archive)에서 동일하게 재현되는 기존 실패다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
344 lines
12 KiB
Python
344 lines
12 KiB
Python
"""계획 노선 기하 공용 유틸 — 정점·누가거리·세류 교차점.
|
|
|
|
배수유역 분석(B04)과 관 편집·세부유역(B05)이 같은 노선 표현을 써야 하므로 여기 한 곳에만
|
|
정의한다. 어느 한쪽 페이지 폴더에 두면 반대 방향 import가 생긴다.
|
|
|
|
노선 원천은 두 가지다.
|
|
· B03에 업로드된 **계획 노선 파일**(CSV·shapefile) — 배수유역 분석의 입력
|
|
· DB `route_points` — B05에서 탐색·확정한 노선
|
|
둘 다 같은 `RouteVertex` 목록으로 바꿔 아래 함수들이 그대로 받는다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import logging
|
|
import math
|
|
import struct
|
|
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
|
|
# `Transformer.from_crs` 입력 문자열("EPSG:n" 또는 **원문 WKT**). 좌표계를 EPSG 코드로
|
|
# 가리면 라벨이 안 붙는 PRJ가 통째로 막히므로, 변환은 언제나 이 값으로 한다.
|
|
# `epsg`는 표시·로그용 라벨일 뿐이다 (2026-08-31 사용자 확정).
|
|
crs_input: str | None = None
|
|
|
|
@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,
|
|
crs_input=f"EPSG:{epsg}" if epsg else None,
|
|
)
|
|
|
|
|
|
def read_planned_route_shapefile(path: Path) -> PlannedRoute | None:
|
|
"""계획 노선 shapefile을 읽어 정점 목록으로 바꾼다.
|
|
|
|
원청 자료는 **2차원 폴리라인**이라 z가 없다. 지반고는 확정 서피스에서 뽑으므로
|
|
(`build_surface_sampler`) 여기서는 0으로 채운다 — CSV의 z와 같은 취급이다.
|
|
파트가 여럿이면 가장 긴 것을 노선으로 본다.
|
|
"""
|
|
from B03_FileInput.B03_FileInput_Engine_Shapefile import (
|
|
read_shapefile_attributes,
|
|
read_shapefile_parts,
|
|
shapefile_crs_input,
|
|
shapefile_epsg_label,
|
|
)
|
|
|
|
try:
|
|
parts = read_shapefile_parts(path)
|
|
except (OSError, ValueError, struct.error) as error:
|
|
logger.warning("계획 노선 shapefile을 읽지 못했습니다: %s (%s)", path, error)
|
|
return None
|
|
if not parts:
|
|
logger.warning("계획 노선 shapefile에 폴리라인이 없습니다: %s", path)
|
|
return None
|
|
|
|
points = max(parts, key=len)
|
|
if len(points) < 2:
|
|
logger.warning("계획 노선 shapefile에 좌표가 2점 미만입니다: %s", path)
|
|
return None
|
|
|
|
vertices: list[RouteVertex] = []
|
|
cumulative = 0.0
|
|
previous: tuple[float, float] | None = None
|
|
for x, y, z in points:
|
|
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)
|
|
|
|
attributes = read_shapefile_attributes(path)
|
|
name = next(
|
|
(
|
|
attributes[key]
|
|
for key in ("대상지", "노선명", "route_name", "name")
|
|
if attributes.get(key)
|
|
),
|
|
path.stem,
|
|
)
|
|
epsg = shapefile_epsg_label(path)
|
|
logger.info(
|
|
"계획 노선 %s: 정점 %d개, 연장 %.0fm, EPSG %s(라벨)",
|
|
path.name,
|
|
len(vertices),
|
|
cumulative,
|
|
epsg,
|
|
)
|
|
return PlannedRoute(
|
|
vertices=vertices,
|
|
epsg=epsg,
|
|
name=name,
|
|
source=path,
|
|
crs_input=shapefile_crs_input(path),
|
|
)
|
|
|
|
|
|
def read_planned_route(path: Path) -> PlannedRoute | None:
|
|
"""계획 노선 파일을 확장자에 맞는 판독기로 읽는다(CSV·shapefile)."""
|
|
if path.suffix.lower() == ".shp":
|
|
return read_planned_route_shapefile(path)
|
|
return read_planned_route_csv(path)
|
|
|
|
|
|
def find_planned_route_file(input_dir: Path) -> Path | None:
|
|
"""B03 입력 폴더에서 계획 노선 파일을 찾는다.
|
|
|
|
shapefile을 CSV보다 우선한다 — 원청 정식 노선이 shapefile로 오고, CSV는 예전
|
|
자료거나 좌표만 뽑아 둔 보조본인 경우가 많다. 같은 종류가 여럿이면 가장 최근 것.
|
|
"""
|
|
if not input_dir.exists():
|
|
return None
|
|
for pattern in ("*.shp", "*.csv"):
|
|
candidates = sorted(
|
|
input_dir.rglob(pattern), key=lambda item: item.stat().st_mtime, reverse=True
|
|
)
|
|
if candidates:
|
|
return candidates[0]
|
|
return 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)
|