perf(계측): 격자 확장·하강 방향장 단계 마크 추가

PLAN 0-10. 1차 영역이 2.6초로 떨어진 뒤 남은 병목이 `expand_by_red_boundary` 102초라
그 안을 가르기 위한 계측. 로직 불변.

- 확장 회차별 시간·격자·셀 수·방향장 재사용 여부를 한 줄씩
- 격자 해석 1회: 하강 방향장(재사용/새로) / 지형 조립 / 도로 굽기 / 세류 새김 / 도달 판정
- 하강 방향장: 등고선 굽기 / 밴드 표고(EDT 1회) / 밴드별 하강거리(EDT N회) /
  위치에너지 조립 / 흐름 경로

실측 로그상 방향장이 회차마다 새로 만들어지고 밴드가 100단이라, 격자 전체 EDT 가
회차당 100회씩 도는 구조로 보임 — 그 가설을 수치로 가르려는 것.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

B04_PreProcess/B04_PreProcess_Engine_Watershed_Descent.py

@@ -23,6 +23,7 @@
 from __future__ import annotations
 
 import logging
+import time
 import math
 from dataclasses import dataclass
 from typing import Any
@@ -127,8 +128,12 @@ def build_contour_descent(
     elevation_floor_m: float | None = None,
 ) -> ContourDescent:
     """등고선만으로 셀별 흐름 방향을 세운다. 보간면을 만들지 않는다."""
+    from B03_FileInput.B03_FileInput_Service_Chain import _log_steps
+
+    marks = [("시작", time.perf_counter())]
     rows, cols = spec.n_rows, spec.n_cols
     burned, levels = rasterize_contours(spec, contour_features, elevation_floor_m)
+    marks.append(("등고선 굽기(rasterize_contours)", time.perf_counter()))
     empty = ContourDescent(
         spec=spec,
         band_elevation=np.full((rows, cols), np.nan, dtype=np.float32),
@@ -146,6 +151,7 @@ def build_contour_descent(
     # ② 셀마다 가장 가까운 등고 라인의 표고 = 그 셀의 밴드.
     _, (near_row, near_col) = distance_transform_edt(~on_contour, return_indices=True)
     band_elevation = burned[near_row, near_col].astype(np.float32)
+    marks.append(("밴드 표고(EDT 1회)", time.perf_counter()))
     inside = domain if domain is not None else np.ones((rows, cols), dtype=bool)
     band_elevation = np.where(inside, band_elevation, np.nan)
 
@@ -169,6 +175,7 @@ def build_contour_descent(
 
     # 최하단 밴드는 더 내려갈 등고 라인이 없다. 무효로 빼지 않고 **정지 셀**로 남긴다 —
     # 그래야 그 자리가 도로·세류선이면 적색으로 잡히고, 아니면 물이 고이는 지점으로 보인다.
+    marks.append((f"밴드별 하강거리(EDT {len(levels) - 1}회)", time.perf_counter()))
     lowest = inside & (band_elevation == levels[-1]) & (band_rank < 0)
     if lowest.any():
         distance[lowest] = 0.0
@@ -185,9 +192,12 @@ def build_contour_descent(
     distance[valid & ~np.isfinite(distance)] = 0.0
     potential = np.where(valid, band_rank.astype(np.float64) * span + distance, np.inf)
 
+    marks.append(("위치에너지 조립", time.perf_counter()))
     receiver, step_length, azimuth = _route_by_potential(
         spec, potential, valid, target_row, target_col
     )
+    marks.append(("흐름 경로(_route_by_potential)", time.perf_counter()))
+    _log_steps(f"하강 방향장({rows}×{cols})", marks)
     logger.info(
         "배수유역: 등고선 하강 방향 %d셀 (밴드 %d단), 최하단 정지 %d셀",
         int(valid.sum()),

B04_PreProcess/B04_PreProcess_Engine_Watershed_Expand.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import logging
+import time
 from dataclasses import dataclass
 from typing import Any
 
@@ -80,8 +81,13 @@ def analyze_domain(
     회차마다 다시 계산하지 않고, 격자가 커졌을 때만 새로 만들어 넘겨받는다(`descent`).
     해석 영역은 마지막에 마스크로만 씌운다.
     """
-    if descent is None or descent.spec != spec:
+    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:
         descent = build_contour_descent(spec, contour_features, None, elevation_floor_m)
+    marks.append(("하강 방향장" + ("(재사용)" if reused else "(새로 만듦)"), time.perf_counter()))
     valid = descent.valid & domain
     if not valid.any():
         return None
@@ -92,9 +98,14 @@ def analyze_domain(
         receiver=descent.receiver,
         step_length=descent.step_length,
     )
+    marks.append(("지형 격자 조립", time.perf_counter()))
     road = rasterize_road(spec, route_line, DRAINAGE_ROAD_WIDTH_M)
+    marks.append(("도로 굽기(rasterize_road)", time.perf_counter()))
     terrain, burned = burn_stream_flow(terrain, road, upstream_streams)
+    marks.append(("세류 새김(burn_stream_flow)", time.perf_counter()))
     flow = classify_flow(terrain, road, burned, azimuth=descent.azimuth)
+    marks.append(("도로 도달 판정(classify_flow)", time.perf_counter()))
+    _log_steps(f"격자 해석 1회({spec.n_rows}×{spec.n_cols})", marks)
     return GridAnalysis(
         spec=spec, domain=domain, descent=descent, terrain=terrain, road=road, flow=flow
     )
@@ -111,6 +122,7 @@ def expand_by_red_boundary(
     max_rounds: int = DRAINAGE_RED_EXPAND_MAX_ROUNDS,
 ) -> RedExpansion | None:
     """최외곽 적색 셀 주변으로 넓히며, 새로 추가한 셀에 적색이 없을 때까지 반복한다."""
+    _expand_started = time.perf_counter()
     band_cells = max(1, int(round(band_m / spec.cell_m)))
     # 1차 영역의 bbox는 영역에 딱 붙어 있어 첫 회차부터 격자를 넓혀야 한다. 미리 여유를
     # 두면 방향장을 다시 만들지 않고 해석 영역만 넓히며 몇 회차를 돌 수 있다.
@@ -122,9 +134,17 @@ def expand_by_red_boundary(
     if analysis is None:
         return None
 
+    logger.info(
+        "[계측] 확장 0회차(첫 해석) %.1fs · 격자 %d×%d · 셀 %d",
+        time.perf_counter() - _expand_started,
+        analysis.spec.n_rows,
+        analysis.spec.n_cols,
+        started_cells,
+    )
     rounds = 0
     closed = False
     for attempt in range(max_rounds):
+        _round_started = time.perf_counter()
         current = analysis.spec
         reaches = analysis.flow.reaches_road.reshape(current.n_rows, current.n_cols)
         rim_red = outermost_cells(analysis.domain) & reaches
@@ -162,6 +182,16 @@ def expand_by_red_boundary(
             break
         analysis = widened_analysis
         rounds += 1
+        logger.info(
+            "[계측] 확장 %d회차 %.1fs · 격자 %d×%d · 셀 %d (+%d) · 방향장 %s",
+            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(
             grown_spec.n_rows, grown_spec.n_cols