perf(배수유역): 342-3 빠른 길 — 흐름 방향 큰 격자 한 번 · 도로 닿음 판정 한꺼번에

- 하강 방향장 = 해석 격자 둘레 400 m 큰 격자로 한 번 짓고 회차마다 잘라 씀(crop_descent) · 넘으면 그때 다시 지음
- classify_flow 한 칸씩 걷기 → 포인터 더블링(_reach_by_doubling) · 기록용 출발 셀 수는 최외곽 · 나머지 셀 수로
- 거울 시험 test_342_3_watershed_fast(옛 걷기와 칸마다 같음 · 잘라 씀 안쪽 같음) · 실측 bench_342_watershed
- 검증_sub3 실측: 적색 차이 0칸 · 도로 귀속 · 강도 · 관 자리 같음 · 108 → 27 s(붐빈 랩탑 · 340 기준 약 15 s)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XWh5wNNycMkbwMpACz1rte

B04_PreProcess/B04_PreProcess_Engine_Watershed_Expand.py

@@ -10,6 +10,9 @@
   ④ **새로 추가된 셀에 적색이 하나도 없으면 종료** (2026-07-31 사용자 지시)
 
 격자 bbox에 닿으면 격자 자체도 셀 정수배로 넓힌다 — 도로 시작점 기준 격자점은 유지된다.
+
+하강 방향장은 처음에 **넉넉한 큰 격자**로 한 번 짓고, 회차마다 그 격자에서 잘라 쓴다
+(`DESCENT_SPARE_M` · 342-3). 격자가 커질 때마다 다시 짓던 것(8회 · 30 s)을 한 번으로 줄인다.
 """
 
 from __future__ import annotations
@@ -43,6 +46,11 @@ from config.config_system import (
 
 logger = logging.getLogger(__name__)
 
+# 하강 방향장을 지을 때 해석 격자 둘레에 더 두는 여유(m). 확장이 이 안에서 끝나면 방향장을
+# 한 번만 짓는다 — 검증_sub3 실측 확장은 첫 격자에서 남 350 m · 서 200 m 까지 넓어졌다.
+# 넘으면 그때 넓힌 격자 둘레로 다시 짓는다(결과는 같고 시간만 듦).
+DESCENT_SPARE_M = 400.0
+
 
 @dataclass
 class GridAnalysis:
@@ -78,16 +86,23 @@ def analyze_domain(
     """주어진 격자·해석 영역에 대해 등고선 하강 방향과 도로 도달 색을 한 번 계산한다.
 
     **하강 방향장은 해석 영역과 무관하다** — 등고선 기하만으로 정해진다. 그래서 확장
-    회차마다 다시 계산하지 않고, 격자가 커졌을 때만 새로 만들어 넘겨받는다(`descent`).
+    회차마다 다시 계산하지 않고, 이 격자를 덮는 큰 방향장(`descent`)을 받으면 잘라 쓴다.
     해석 영역은 마지막에 마스크로만 씌운다.
     """
     from B03_FileInput.B03_FileInput_Service_Chain import _log_steps
 
     marks = [("시작", time.perf_counter())]
-    reused = descent is not None and descent.spec == spec
-    if not reused:
+    fitted = crop_descent(descent, spec) if descent is not None else None
+    if fitted is None:
         descent = build_contour_descent(spec, contour_features, None, elevation_floor_m)
-    marks.append(("하강 방향장" + ("(재사용)" if reused else "(새로 만듦)"), time.perf_counter()))
+    else:
+        descent = fitted
+    marks.append(
+        (
+            "하강 방향장" + ("(잘라 씀)" if fitted is not None else "(새로 만듦)"),
+            time.perf_counter(),
+        )
+    )
     valid = descent.valid & domain
     if not valid.any():
         return None
@@ -128,8 +143,12 @@ def expand_by_red_boundary(
     # 두면 방향장을 다시 만들지 않고 해석 영역만 넓히며 몇 회차를 돌 수 있다.
     spec, domain = _pad_spec(spec, domain, band_cells * 2)
     started_cells = int(domain.sum())
+    spare_cells = max(0, int(round(DESCENT_SPARE_M / spec.cell_m)))
+    source = build_contour_descent(
+        _pad_spec(spec, domain, spare_cells)[0], contour_features, None, elevation_floor_m
+    )
     analysis = analyze_domain(
-        spec, domain, contour_features, route_line, upstream_streams, elevation_floor_m
+        spec, domain, contour_features, route_line, upstream_streams, elevation_floor_m, source
     )
     if analysis is None:
         return None
@@ -160,6 +179,14 @@ def expand_by_red_boundary(
         if not added_mask.any():
             closed = True
             break
+        if crop_descent(source, grown_spec) is None:
+            # 넉넉히 둔 방향장 밖으로 넓어짐 — 넓힌 격자 둘레로 다시 짓는다.
+            source = build_contour_descent(
+                _pad_spec(grown_spec, grown_domain, spare_cells)[0],
+                contour_features,
+                None,
+                elevation_floor_m,
+            )
 
         logger.info(
             "배수유역: %d회차 확장 — 최외곽 적색 %d셀 주변 %.0fm, 셀 %d개 추가",
@@ -175,22 +202,20 @@ def expand_by_red_boundary(
             route_line,
             upstream_streams,
             elevation_floor_m,
-            # 격자가 그대로면 방향장을 재사용한다 — 등고선 기하가 안 바뀌었으므로 결과는 같다.
-            descent=analysis.descent if grown_spec == current else None,
+            descent=source,
         )
         if widened_analysis is None:
             break
         analysis = widened_analysis
         rounds += 1
         logger.info(
-            "[계측] 확장 %d회차 %.1fs · 격자 %d×%d · 셀 %d (+%d) · 방향장 %s",
+            "[계측] 확장 %d회차 %.1fs · 격자 %d×%d · 셀 %d (+%d)",
             attempt + 1,
             time.perf_counter() - _round_started,
             grown_spec.n_rows,
             grown_spec.n_cols,
             int(widened.sum()),
             int(added_mask.sum()),
-            "재사용" if grown_spec == current else "새로",
         )
 
         added_reaches = widened_analysis.flow.reaches_road.reshape(
@@ -230,6 +255,59 @@ def _dilate_by(mask: np.ndarray, steps: int) -> np.ndarray:
     return result
 
 
+def crop_descent(source: ContourDescent, spec: GridSpec) -> ContourDescent | None:
+    """큰 방향장에서 `spec` 격자 몫을 잘라 낸다. 덮지 못하거나 셀이 안 맞으면 None.
+
+    잘린 가장자리 밖을 가리키던 셀은 제자리(싱크)로 둔다 — 해석 영역 밖으로 빠지는 물과
+    같은 판정(미도달)이다. 화살표(방위)는 원래 방향 그대로 둔다.
+    """
+    big = source.spec
+    if big == spec:
+        return source
+    if big.cell_m != spec.cell_m:
+        return None
+    row_shift = (big.y_max - spec.y_max) / spec.cell_m
+    col_shift = (spec.x_min - big.x_min) / spec.cell_m
+    row0, col0 = int(round(row_shift)), int(round(col_shift))
+    if (
+        abs(row_shift - row0) > 1e-6
+        or abs(col_shift - col0) > 1e-6
+        or row0 < 0
+        or col0 < 0
+        or row0 + spec.n_rows > big.n_rows
+        or col0 + spec.n_cols > big.n_cols
+    ):
+        return None
+    rows = slice(row0, row0 + spec.n_rows)
+    cols = slice(col0, col0 + spec.n_cols)
+
+    def window(flat: np.ndarray) -> np.ndarray:
+        return flat.reshape(big.n_rows, big.n_cols)[rows, cols]
+
+    target = window(source.receiver).astype(np.int64)
+    target_row = target // big.n_cols - row0
+    target_col = target % big.n_cols - col0
+    inside = (
+        (target_row >= 0)
+        & (target_row < spec.n_rows)
+        & (target_col >= 0)
+        & (target_col < spec.n_cols)
+    )
+    local = np.arange(spec.size, dtype=np.int64).reshape(spec.n_rows, spec.n_cols)
+    receiver = np.where(inside, target_row * spec.n_cols + target_col, local)
+    return ContourDescent(
+        spec=spec,
+        band_elevation=source.band_elevation[rows, cols].copy(),
+        valid=source.valid[rows, cols].copy(),
+        receiver=receiver.reshape(-1).astype(np.int32),
+        step_length=np.where(inside, window(source.step_length), 0.0)
+        .reshape(-1)
+        .astype(np.float32),
+        azimuth=window(source.azimuth).reshape(-1).copy(),
+        levels=source.levels,
+    )
+
+
 def _pad_spec(spec: GridSpec, domain: np.ndarray, cells: int) -> tuple[GridSpec, np.ndarray]:
     """격자에 사방 여유를 두고 해석 영역 마스크를 그 안으로 옮겨 담는다."""
     if cells <= 0: