auto: 2026-07-29 18:23 (EOMSANGDON-HOME)

B05_wf2_Route/B05_wf2_Route_Engine_Drainage.py

@@ -241,7 +241,9 @@ def _fill_spacing(
         if placed is None:
             placed = _nearest_uphill(vertices, target, end_m)
         if placed is None:
-            break
+            # 도로 연장 기준 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

B05_wf2_Route/B05_wf2_Route_Engine_Drainage_Watershed.py

@@ -19,7 +19,7 @@ from dataclasses import dataclass, field
 from typing import Any
 
 from shapely.geometry import LineString, Point, Polygon
-from shapely.ops import substring, unary_union
+from shapely.ops import split, substring, unary_union
 
 from B05_wf2_Route.B05_wf2_Route_Engine_Drainage import (
     StructureCandidate,
@@ -33,6 +33,7 @@ from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Trace import (
     ContourIndex,
     DividerStep,
     _explode_lines,
+    side_sign,
     trace_divider,
     trace_upstream_network,
 )
@@ -42,6 +43,8 @@ logger = logging.getLogger(__name__)
 # 유효 유역 최소 면적(m²)과 상류망 커버 보정 버퍼(m).
 MIN_BASIN_AREA_M2 = 100.0
 STREAM_COVER_BUFFER_M = 20.0
+# 도로 절단용 노선 양끝 연장 길이(m). 노선 끝 너머로 새는 하류측 조각까지 잘라낸다.
+ROAD_EXTEND_M = 500.0
 # 산측 판정: 도로 접선의 좌우 수직 오프셋(m)과 표고 조회 반경(m).
 UPHILL_PROBE_OFFSET_M = 40.0
 UPHILL_PROBE_RADIUS_M = 60.0
@@ -211,6 +214,50 @@ def _assemble_polygon(
     return polygon
 
 
+def _extended_road_line(vertices: list[Any]) -> LineString:
+    """양끝 접선 방향으로 연장한 도로선 — 폴리곤 절단이 끝에서 끊기지 않게 한다."""
+    coords = [(vertex.x, vertex.y) for vertex in vertices]
+    hx, hy = coords[0]
+    dx, dy = coords[1][0] - hx, coords[1][1] - hy
+    norm = math.hypot(dx, dy) or 1.0
+    head = (hx - dx / norm * ROAD_EXTEND_M, hy - dy / norm * ROAD_EXTEND_M)
+    tx, ty = coords[-1]
+    dx, dy = tx - coords[-2][0], ty - coords[-2][1]
+    norm = math.hypot(dx, dy) or 1.0
+    tail = (tx + dx / norm * ROAD_EXTEND_M, ty + dy / norm * ROAD_EXTEND_M)
+    return LineString([head, *coords, tail])
+
+
+def _clip_to_uphill(
+    polygon: Polygon,
+    road_line: LineString,
+    extended_road: LineString,
+    uphill_sign: int,
+) -> Polygon | None:
+    """폴리곤을 도로선으로 절단해 산측 조각만 남긴다 — 도로가 유역의 한쪽 경계.
+
+    도로 하류측(성토부 아래)은 유역에 포함하지 않는다(2026-07-29 사용자 지시).
+    """
+    try:
+        pieces = split(polygon, extended_road)
+    except Exception:  # noqa: BLE001 - 절단 실패 시 원본 유지
+        return polygon
+    kept = [
+        piece
+        for piece in getattr(pieces, "geoms", [pieces])
+        if piece.geom_type == "Polygon"
+        and side_sign(road_line, piece.representative_point()) == uphill_sign
+    ]
+    if not kept:
+        return None
+    merged = unary_union(kept)
+    if merged.geom_type == "MultiPolygon":
+        merged = max(merged.geoms, key=lambda part: part.area)
+    if merged.is_empty or merged.geom_type != "Polygon":
+        return None
+    return merged
+
+
 def build_watershed_basins(
     vertices: list[Any],
     candidates: list[StructureCandidate],
@@ -233,6 +280,7 @@ def build_watershed_basins(
         return []
     spot_index = ContourIndex(spot_features, elevation_keys)
     road_line = LineString([(vertex.x, vertex.y) for vertex in vertices])
+    extended_road = _extended_road_line(vertices)
 
     ordered = sorted(candidates, key=lambda item: item.chainage_m)
     divides = _divide_chainages(vertices, ordered)
@@ -298,7 +346,9 @@ def build_watershed_basins(
                 covered = max(covered.geoms, key=lambda part: part.area)
             if covered.geom_type == "Polygon" and not covered.is_empty:
                 polygon = covered
-        if polygon.area < MIN_BASIN_AREA_M2:
+        # 마지막에 도로선으로 절단해 하류측을 제거한다 — 도로가 유역의 한쪽 경계.
+        polygon = _clip_to_uphill(polygon, road_line, extended_road, signs[position])
+        if polygon is None or polygon.area < MIN_BASIN_AREA_M2:
             continue
 
         outlet_z = contour_index.nearest_elevation(outlet, UPHILL_PROBE_RADIUS_M)