eomsangdonandClaude Fable 5
c1f139527d
등고선 아크 추적 + 능선 행진 방식이 능선/계곡을 안정적으로 분리하지 못해 D8 물 방향 + 도로 기준 상류 추적 방식으로 교체한다. - 능선을 따로 탐지하지 않는다. 물길을 따라가 도로에 닿는 셀만 유역이고, 그 경계가 곧 능선이다. 유역 내부 봉우리는 자동으로 포함된다. - 등고선 TIN 보간 후 웅덩이 채움(형태학적 재구성) + 평탄면 미세경사로 가짜 웅덩이/평탄 삼각형에서 흐름이 끊기는 문제를 없앤다. - 상류 추적은 포인터 더블링으로 전 셀을 한 번에 푼다. 셀의 흐름 종착 도로 셀(root)이 유역 판정·흐름 강도·세부유역 라벨의 공통 근거가 되어, 관을 옮겨도 격자 해석 없이 측구 라우팅만 다시 돌면 된다(.npz 캐시). - 활성 셀이 격자 최외곽에 닿은 방향으로만 확장하고, 경계 링이 전부 비활성이 되면(띠 폐합) 멈춘다. 변경 사항 - 신규 엔진 3종: Engine_Watershed_Grid / _Flow / _Basin - 폐기 엔진 4종은 _legacy_watershed/ 로 원본 보관(ruff 제외) - config_system.py §5-3-1 에 DRAINAGE_* 파라미터 18개 (격자 1m, 반경 300m) - 표고점 데이터 사용 중단(유효 데이터 부족), 프론트 능선 토글 제거 (전체 유역 외곽선과 같은 선이므로 중복) - 응답에 main_polygon_lonlat / strength_profile 추가, 계획선 위 흐름 강도 표기 합성 지형 검증: 유역 179,919㎡ vs 이론 180,000㎡ (오차 0.04%), 능선 자동 검출, 확장 3회 후 자동 정지, 캐시 재사용 1.2s -> 0.1s Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
323 lines
12 KiB
Python
323 lines
12 KiB
Python
"""배수유역 산정 엔진.
|
|
|
|
관 매설 구조물 측점 후보를 제안하고, 각 측점이 받는 배수유역 경계를 산정한다.
|
|
지형 판단은 **도엽 등고선·세류선(하천중심선)**만 사용한다 — 3D 포인트클라우드나 지형
|
|
메시는 쓰지 않고(2026-07-28 사용자 지시), 표고점도 유효 데이터가 적어 뺐다(2026-07-31).
|
|
유역 경계 산정 자체는 격자 흐름 해석(`..._Engine_Watershed_Basin`)이 맡고, 이 모듈은
|
|
측점 후보 제안과 노선 정점·누가거리 보간만 담당한다.
|
|
|
|
유역을 나누는 최종 목적은 각 지점의 파이프 관경 결정이다. 유역 경사면에 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
|