321 lines
12 KiB
Python
321 lines
12 KiB
Python
"""배수유역 산정 엔진.
|
|
|
|
관 매설 구조물 측점 후보를 제안하고, 각 측점이 받는 배수유역 경계를 산정한다.
|
|
지형 판단은 **도엽 등고선·세류선(하천중심선)·표고점**만 사용한다 — 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
|
|
# 같은 세류 교차로 볼 최소 이격(m). 이보다 가까운 교차점은 하나로 묶는다.
|
|
MIN_STRUCTURE_SPACING_M = 5.0
|
|
# 유역 경계 탐색 반경(m). 측점에서 이 거리를 넘는 지형은 해당 유역으로 보지 않는다.
|
|
MAX_BASIN_RADIUS_M = 1000.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를 넘으면 보충 측점을 만든다.
|
|
|
|
종단도상 상대적으로 물이 모일 것으로 예상되는 지점(절토부 내 종단 저점)을
|
|
우선 배치한다(2026-07-29 사용자 지시). 저점이 없으면 목표 인근 절토부로 대체한다.
|
|
"""
|
|
added: list[StructureCandidate] = []
|
|
cursor = start_m
|
|
while end_m - cursor > MAX_STRUCTURE_SPACING_M:
|
|
target = cursor + MAX_STRUCTURE_SPACING_M
|
|
placed = _gather_low_point(vertices, cursor, target, end_m)
|
|
if placed is None:
|
|
placed = _nearest_uphill(vertices, target, end_m)
|
|
if placed is None:
|
|
# 도로 연장 기준 300m 규칙 — 저점·절토부가 없어도 관 배치는 보장한다
|
|
# (2026-07-29 사용자 지시: 도로 340m면 최소 1개).
|
|
placed = min(target, (cursor + end_m) / 2.0)
|
|
x, y, _ = _interpolate_vertex(vertices, placed)
|
|
added.append(StructureCandidate(chainage_m=placed, x=x, y=y, reason="spacing"))
|
|
cursor = placed
|
|
return added
|
|
|
|
|
|
def _gather_low_point(
|
|
vertices: list[RouteVertex],
|
|
cursor_m: float,
|
|
target_m: float,
|
|
limit_m: float,
|
|
step_m: float = 10.0,
|
|
) -> float | None:
|
|
"""탐색창 [cursor+150, target] 안 절토부의 종단 국소 저점(사그) 중 가장 낮은 지점.
|
|
|
|
창 하한을 간격의 절반으로 두어 보충 측점이 과밀하게 몰리지 않게 하고,
|
|
국소 저점만 인정해 일정 오르막에서는 None(300m 규칙 폴백)을 돌려준다.
|
|
"""
|
|
window_start = cursor_m + MAX_STRUCTURE_SPACING_M / 2.0
|
|
probes: list[float] = []
|
|
probe = window_start - step_m
|
|
while probe <= target_m + step_m:
|
|
probes.append(probe)
|
|
probe += step_m
|
|
heights = [_interpolate_vertex(vertices, position)[2] for position in probes]
|
|
best: tuple[float, float] | None = None # (계획고 z, 누가거리)
|
|
for i in range(1, len(probes) - 1):
|
|
position = probes[i]
|
|
if position >= limit_m or position > target_m or position < window_start:
|
|
continue
|
|
# 국소 저점(양쪽이 같거나 높음) = 물이 모여 더 못 흐르는 지점. 앞쪽이 오르막인
|
|
# 조건을 내포하므로 별도의 절토부(is_uphill_at) 판정은 두지 않는다.
|
|
if heights[i] > heights[i - 1] or heights[i] > heights[i + 1]:
|
|
continue
|
|
if best is None or heights[i] < best[0]:
|
|
best = (heights[i], position)
|
|
return best[1] if best else None
|
|
|
|
|
|
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
|