Merge remote-tracking branch 'origin/sub_laptop_3' into main_laptop_1

B04_PreProcess/B04_PreProcess_Engine_Watershed_Analyze.py

@@ -5,7 +5,7 @@
 일반 사용자가 쓰는 B05는 그 결과를 읽어 쓰기만 한다(2026-07-31 사용자 지시).
 
   ① 도로 교차 세류망 상류측 추출 → 반경 버퍼 = 1차 배수유역
-  ② 도로 시작점 기준 격자 생성 (1차 영역에 걸치는 셀만)
+  ② 좌표 정수 m 기준 격자 생성 (1차 영역에 걸치는 셀만)
   ③ 등고선 하강 방향 — 높은 등고 라인에서 낮은 등고 라인으로. 보간면을 쓰지 않으므로
      가짜 웅덩이·평탄면이 원리적으로 생기지 않는다
   ④ 세류망 흐름 새김 → 최외곽부터 사슬 추적 → 도로 도달 여부(적/청) 판정
@@ -24,6 +24,7 @@ import logging
 import math
 import time
 from dataclasses import dataclass, field
+from pathlib import Path
 from typing import Any
 
 import numpy as np
@@ -152,6 +153,7 @@ def preview_stages(
     vertices: list[RouteVertex],
     contour_features: list[dict[str, Any]],
     stream_features: list[dict[str, Any]],
+    descent_store: tuple[Path, str] | None = None,
 ) -> StagePreview | None:
     """지금까지 구현·검증된 단계를 순서대로 돌려 결과를 모은다.
 
@@ -184,6 +186,7 @@ def preview_stages(
         route_line,
         region.split.upstream,
         floor,
+        descent_store=descent_store,
     )
     if expansion is None:
         logger.warning("배수유역: 등고선 하강 방향을 세우지 못해 흐름 판정을 건너뜁니다.")

B04_PreProcess/B04_PreProcess_Engine_Watershed_Descent.py

@@ -22,10 +22,14 @@
 
 from __future__ import annotations
 
+import hashlib
+import json
 import logging
-import time
 import math
+import time
+from collections.abc import Iterator
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any
 
 import numpy as np
@@ -349,6 +353,99 @@ def build_contour_descent(
     )
 
 
+# ── 흐름 방향 저장본(347-1) ─────────────────────────────────────────────
+#
+# 방향장은 등고선 · 남긴 등고 단 · 격자(원점 · 셀 · 범위)만으로 정해진다. 원점이 좌표 정수 m 라
+# 노선이 저장 범위 안이면 읽어서 잘라 쓴다. 파일 이름 = descent_<입력 지문>_<범위 지문>.npz ·
+# 종류마다 최근 2벌만 남긴다. 계산 규칙이 바뀌면 `DESCENT_FILE_VERSION` 을 올린다.
+DESCENT_FILE_VERSION = 1
+_DESCENT_KEEP = 2
+
+
+def descent_input_key(
+    contour_key: str,
+    contour_features: list[dict[str, Any]],
+    elevation_floor_m: float | None,
+    cell_m: float,
+) -> str:
+    """방향장 입력 지문 — 도엽 등고선 지문 + 남긴 등고 단 + 셀 + 계산 규칙 값.
+
+    남긴 단은 속성 표고만 본다 — 같은 도엽 · 같은 단 집합이면 남는 선도 같다(길이 거름은 결정적).
+    """
+    levels = sorted(
+        {
+            elevation
+            for feature in contour_features
+            if feature.get("geometry")
+            and (elevation := _feature_elevation(feature.get("properties") or {})) is not None
+            and (elevation_floor_m is None or elevation >= elevation_floor_m)
+        }
+    )
+    text = json.dumps(
+        [
+            DESCENT_FILE_VERSION,
+            contour_key,
+            levels,
+            cell_m,
+            DRAINAGE_CONTOUR_MIN_LENGTH_M,
+            DESCENT_WINDOW_MARGIN_CELLS,
+            DESCENT_TILE_CELLS,
+        ]
+    )
+    return hashlib.sha1(text.encode(), usedforsecurity=False).hexdigest()[:16]
+
+
+def iter_saved_descents(directory: Path, input_key: str) -> Iterator[ContourDescent]:
+    """같은 입력 지문으로 저장한 방향장 — 최근 것부터. 못 읽는 파일은 건너뜀."""
+    paths = sorted(
+        directory.glob(f"descent_{input_key}_*.npz"),
+        key=lambda path: path.stat().st_mtime,
+        reverse=True,
+    )
+    for path in paths:
+        try:
+            with np.load(path, allow_pickle=False) as data:
+                x_min, y_max, cell_m, n_rows, n_cols = data["spec"].tolist()
+                yield ContourDescent(
+                    spec=GridSpec(
+                        float(x_min), float(y_max), float(cell_m), int(n_rows), int(n_cols)
+                    ),
+                    band_elevation=data["band_elevation"],
+                    valid=data["valid"],
+                    receiver=data["receiver"],
+                    step_length=data["step_length"],
+                    azimuth=data["azimuth"],
+                    levels=data["levels"].tolist(),
+                )
+        except (OSError, KeyError, ValueError):
+            logger.warning("배수유역: 흐름 방향 저장본을 읽지 못해 건너뜁니다 (%s).", path)
+
+
+def save_descent(directory: Path, input_key: str, descent: ContourDescent) -> None:
+    """방향장을 저장하고 최근 2벌만 남긴다. 실패는 로그만(계산은 이어감)."""
+    spec = descent.spec
+    extent = f"{spec.x_min:.3f},{spec.y_max:.3f},{spec.n_rows},{spec.n_cols}"
+    extent_key = hashlib.sha1(extent.encode(), usedforsecurity=False).hexdigest()[:8]
+    path = directory / f"descent_{input_key}_{extent_key}.npz"
+    try:
+        directory.mkdir(parents=True, exist_ok=True)
+        np.savez(
+            path,
+            spec=np.array([spec.x_min, spec.y_max, spec.cell_m, spec.n_rows, spec.n_cols]),
+            band_elevation=descent.band_elevation,
+            valid=descent.valid,
+            receiver=descent.receiver,
+            step_length=descent.step_length,
+            azimuth=descent.azimuth,
+            levels=np.array(descent.levels, dtype=np.float64),
+        )
+        olds = sorted(directory.glob("descent_*.npz"), key=lambda item: item.stat().st_mtime)
+        for old in olds[:-_DESCENT_KEEP]:
+            old.unlink(missing_ok=True)
+    except OSError:
+        logger.warning("배수유역: 흐름 방향 저장본을 남기지 못했습니다 (%s).", path)
+
+
 def _route_by_potential(
     spec: GridSpec,
     potential: np.ndarray,

B04_PreProcess/B04_PreProcess_Engine_Watershed_Expand.py

@@ -9,10 +9,12 @@
   ③ 넓힌 영역으로 흐름 방향·색을 다시 분석한다
   ④ **새로 추가된 셀에 적색이 하나도 없으면 종료** (2026-07-31 사용자 지시)
 
-격자 bbox에 닿으면 격자 자체도 셀 정수배로 넓힌다 — 도로 시작점 기준 격자점은 유지된다.
+격자 bbox에 닿으면 격자 자체도 셀 정수배로 넓힌다 — 좌표 정수 m 기준 격자점은 유지된다.
 
 하강 방향장은 처음에 **넉넉한 큰 격자**로 한 번 짓고, 회차마다 그 격자에서 잘라 쓴다
 (`DESCENT_SPARE_M` · 342-3). 격자가 커질 때마다 다시 짓던 것(8회 · 30 s)을 한 번으로 줄인다.
+저장소(`descent_store` = 폴더 · 도엽 등고선 지문)를 주면 지은 것을 남기고, 덮는 저장본이 있으면
+읽기만 한다(347-1).
 """
 
 from __future__ import annotations
@@ -20,6 +22,7 @@ from __future__ import annotations
 import logging
 import time
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any
 
 import numpy as np
@@ -28,6 +31,9 @@ from shapely.geometry import LineString
 from B04_PreProcess.B04_PreProcess_Engine_Watershed_Descent import (
     ContourDescent,
     build_contour_descent,
+    descent_input_key,
+    iter_saved_descents,
+    save_descent,
 )
 from B04_PreProcess.B04_PreProcess_Engine_Watershed_Flow import (
     FlowClassification,
@@ -135,6 +141,7 @@ def expand_by_red_boundary(
     elevation_floor_m: float | None = None,
     band_m: float = DRAINAGE_RED_EXPAND_BAND_M,
     max_rounds: int = DRAINAGE_RED_EXPAND_MAX_ROUNDS,
+    descent_store: tuple[Path, str] | None = None,
 ) -> RedExpansion | None:
     """최외곽 적색 셀 주변으로 넓히며, 새로 추가한 셀에 적색이 없을 때까지 반복한다."""
     _expand_started = time.perf_counter()
@@ -144,9 +151,33 @@ 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
-    )
+    store = None
+    if descent_store is not None:
+        folder, contour_key = descent_store
+        store = (
+            folder,
+            descent_input_key(contour_key, contour_features, elevation_floor_m, spec.cell_m),
+        )
+
+    def source_for(grid: GridSpec, mask: np.ndarray) -> ContourDescent:
+        """`grid` 를 덮는 큰 방향장 — 저장본이 덮으면 읽기만 · 아니면 둘레 여유째 짓고 남김."""
+        if store is not None:
+            for saved in iter_saved_descents(*store):
+                if crop_descent(saved, grid) is not None:
+                    logger.info(
+                        "배수유역: 흐름 방향 저장본을 읽어 씁니다(%d×%d).",
+                        saved.spec.n_rows,
+                        saved.spec.n_cols,
+                    )
+                    return saved
+        built = build_contour_descent(
+            _pad_spec(grid, mask, spare_cells)[0], contour_features, None, elevation_floor_m
+        )
+        if store is not None:
+            save_descent(*store, built)
+        return built
+
+    source = source_for(spec, domain)
     analysis = analyze_domain(
         spec, domain, contour_features, route_line, upstream_streams, elevation_floor_m, source
     )
@@ -181,12 +212,7 @@ def expand_by_red_boundary(
             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,
-            )
+            source = source_for(grown_spec, grown_domain)
 
         logger.info(
             "배수유역: %d회차 확장 — 최외곽 적색 %d셀 주변 %.0fm, 셀 %d개 추가",
@@ -329,7 +355,7 @@ def _grow_for_rim(
 ) -> tuple[GridSpec, np.ndarray, np.ndarray]:
     """적색 최외곽이 격자 bbox에 닿았으면 그 방향으로 격자를 넓히고 마스크를 옮겨 담는다.
 
-    격자는 셀 정수배로만 넓히므로 도로 시작점 기준 격자점이 그대로 유지된다.
+    격자는 셀 정수배로만 넓히므로 좌표 정수 m 기준 격자점이 그대로 유지된다.
     """
     north = band_cells if rim[0, :].any() else 0
     south = band_cells if rim[-1, :].any() else 0

B04_PreProcess/B04_PreProcess_Engine_Watershed_Grid.py

@@ -216,6 +216,10 @@ def _outside_bounds(
     return bounds[2] < clip[0] or bounds[0] > clip[2] or bounds[3] < clip[1] or bounds[1] > clip[3]
 
 
+# 배수유역 격자 원점 — 셀 모서리가 좌표 정수 m 에 놓인다(347-1 · 프로젝트 공통).
+GRID_ORIGIN_XY = (0.0, 0.0)
+
+
 def grid_spec_from_bounds(
     x_min: float,
     y_min: float,
@@ -226,9 +230,10 @@ def grid_spec_from_bounds(
 ) -> GridSpec:
     """범위를 덮는 격자를 만든다. `anchor_xy`를 주면 그 점에 셀 모서리를 맞춘다.
 
-    격자 원점은 **도로 시작점**에 고정한다(2026-07-31 사용자 지시). bbox 좌상단에 맞추면
-    1차 영역이 조금만 달라져도 격자가 통째로 밀려 이전 결과와 셀이 대응되지 않는다.
-    도로 시작점에 맞추면 반경·영역을 바꿔도 같은 자리의 셀은 같은 자리에 남는다.
+    격자 원점은 **좌표 정수 m(`GRID_ORIGIN_XY`)**에 고정한다(347-1 사용자 확정 · 옛 「도로
+    시작점」 지시를 바꿈). bbox 좌상단에 맞추면 1차 영역이 조금만 달라져도 격자가 통째로 밀려
+    이전 결과와 셀이 대응되지 않는다. 좌표 정수 m 에 맞추면 노선 · 반경이 바뀌어도 같은 자리의
+    셀은 같은 자리에 남아 흐름 방향 저장본(`drainage/descent_*.npz`)을 그대로 잘라 쓴다.
 
     격자 크기는 절대 자동으로 바꾸지 않는다 — config 값이 그대로 쓰인다(사용자 지시).
     셀 수가 많으면 경고만 남기고 그대로 진행한다.
@@ -258,7 +263,7 @@ def grid_spec_from_bounds(
 def expand_grid_spec(spec: GridSpec, sides: dict[str, bool], step_m: float) -> GridSpec:
     """활성 셀이 닿은 방향으로만 격자를 넓힌다.
 
-    셀 정수배로만 넓혀 격자 격자점(도로 시작점 기준)이 그대로 유지되게 한다.
+    셀 정수배로만 넓혀 격자점(좌표 정수 m 기준)이 그대로 유지되게 한다.
     """
     steps = max(1, int(math.ceil(step_m / spec.cell_m)))
     west = steps if sides.get("west") else 0

B04_PreProcess/B04_PreProcess_Engine_Watershed_Stream.py

@@ -21,6 +21,7 @@ from shapely.geometry import LineString, MultiPolygon, Polygon, shape
 from shapely.ops import substring, unary_union
 
 from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import (
+    GRID_ORIGIN_XY,
     ContourCloud,
     GridSpec,
     build_cell_mask,
@@ -344,7 +345,7 @@ def build_primary_region(
     노선 버퍼는 세류 교차가 없는 구간의 도로도 격자 안에 들어오게 한다 — 그래야 그 구간
     사면이 유역으로 잡힌다. 상류 세류망 선정이 정확해진 뒤 다시 포함했다(2026-07-31 사용자 지시).
 
-    격자는 bbox를 통째로 채우지 않는다. **도로 시작점에 셀 모서리를 맞춘 뒤, 1차 영역에
+    격자는 bbox를 통째로 채우지 않는다. **좌표 정수 m(원점 0,0)에 셀 모서리를 맞춘 뒤, 1차 영역에
     조금이라도 걸치는 셀만** 생성한다(2026-07-31 사용자 지시). bbox 전체를 쓰면 영역 밖
     빈 셀이 대부분이라 의미가 없고, 원점을 bbox 좌상단에 두면 영역이 조금만 변해도 격자가
     통째로 밀려 이전 결과와 셀이 대응되지 않는다.
@@ -359,16 +360,13 @@ def build_primary_region(
         logger.warning("배수유역: 상류 세류망이 없어 노선 버퍼만으로 1차 영역을 잡습니다.")
     area = unary_union(geometries)
     x_min, y_min, x_max, y_max = area.bounds
-    road_start = route_line.coords[0]
-    spec = grid_spec_from_bounds(
-        x_min, y_min, x_max, y_max, cell_m, anchor_xy=(float(road_start[0]), float(road_start[1]))
-    )
+    spec = grid_spec_from_bounds(x_min, y_min, x_max, y_max, cell_m, anchor_xy=GRID_ORIGIN_XY)
     cell_mask = build_cell_mask(spec, area)
     outside = route_line.difference(area)
     road_outside_m = float(outside.length) if not outside.is_empty else 0.0
     active = int(cell_mask.sum())
     logger.info(
-        "배수유역: 1차 영역 %.0f㎡ → 격자 %d×%d (%.2fm, 도로 시점 기준) 중 %d셀 생성 "
+        "배수유역: 1차 영역 %.0f㎡ → 격자 %d×%d (%.2fm, 좌표 정수 m 기준) 중 %d셀 생성 "
         "(bbox %d셀의 %.0f%%), 노선 이탈 %.0fm/%.0fm",
         area.area,
         spec.n_rows,

B04_PreProcess/B04_PreProcess_Router_Watershed.py

@@ -8,9 +8,11 @@
 """
 
 import asyncio
+import hashlib
 import json
 import logging
 import math
+import pickle
 import time
 from pathlib import Path
 from typing import Any
@@ -149,6 +151,42 @@ def _map_coordinates(coordinates: Any, transformer: Transformer) -> Any:
     return [item for item in mapped if item is not None]
 
 
+# 도엽을 사업지 좌표로 바꾼 저장본(347-1) — 원본 파일 자국 · 좌표계가 같으면 읽기만.
+SHEET_METRIC_FILENAME = "00_sheet_metric.pkl"
+
+
+def _metric_sheets(
+    directory: Path, cache_dir: Path, source_crs: str, transformer: Transformer
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]], str]:
+    """(하천, 등고선, 도엽 등고선 지문). 피처 4,200개 재투영(3.8 s)을 한 번만 한다."""
+    marks = []
+    for name in (STREAM_FILE, CONTOUR_FILE):
+        try:
+            info = (directory / name).stat()
+            marks.append(f"{name}:{info.st_mtime_ns}:{info.st_size}")
+        except OSError:
+            marks.append(f"{name}:-")
+    text = "|".join([*marks, str(source_crs)])
+    key = hashlib.sha1(text.encode(), usedforsecurity=False).hexdigest()[:16]
+    path = cache_dir / SHEET_METRIC_FILENAME
+    try:
+        with path.open("rb") as file:
+            saved = pickle.load(file)
+        if saved.get("key") == key:
+            return saved["streams"], saved["contours"], key
+    except (OSError, pickle.PickleError, EOFError, AttributeError, KeyError, TypeError):
+        pass
+    streams = _reproject_features(_load_features(directory, STREAM_FILE), transformer)
+    contours = _reproject_features(_load_features(directory, CONTOUR_FILE), transformer)
+    try:
+        cache_dir.mkdir(parents=True, exist_ok=True)
+        with path.open("wb") as file:
+            pickle.dump({"key": key, "streams": streams, "contours": contours}, file)
+    except OSError:
+        logger.warning("배수유역: 도엽 좌표 저장본을 남기지 못했습니다 (%s).", path)
+    return streams, contours, key
+
+
 def _candidate_payload(
     candidate: StructureCandidate,
     to_lonlat: Any,
@@ -204,10 +242,8 @@ async def _prepare(project_id: UUID) -> dict[str, Any] | JSONResponse:
     to_lonlat_transformer = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True)
     to_metric_transformer = Transformer.from_crs("EPSG:4326", source_crs, always_xy=True)
 
-    directory = _sheet_dir(stored_path)
-    streams = _reproject_features(_load_features(directory, STREAM_FILE), to_metric_transformer)
-    contour_features = _reproject_features(
-        _load_features(directory, CONTOUR_FILE), to_metric_transformer
+    streams, contour_features, contour_key = _metric_sheets(
+        _sheet_dir(stored_path), drainage_dir(stored_path), source_crs, to_metric_transformer
     )
     return {
         "route_source": route_file.name,
@@ -215,6 +251,7 @@ async def _prepare(project_id: UUID) -> dict[str, Any] | JSONResponse:
         "route_line": planned.line,
         "streams": streams,
         "contours": contour_features,
+        "contour_key": contour_key,
         "stored_path": stored_path,
         "to_lonlat": lambda x, y: to_lonlat_transformer.transform(x, y),
     }
@@ -403,6 +440,8 @@ async def get_primary_region(
         prepared["vertices"],
         prepared["contours"],
         prepared["streams"],
+        # 흐름 방향 저장본(347-1) — 노선이 저장 범위 안이면 읽기만
+        (drainage_dir(prepared["stored_path"]), prepared["contour_key"]),
     )
     if preview is None:
         return JSONResponse(

resources/tester/bench_342_watershed.py

@@ -3,9 +3,10 @@
 
   ./venv/Scripts/python.exe resources/tester/bench_342_watershed.py base          # 기준 저장
   ./venv/Scripts/python.exe resources/tester/bench_342_watershed.py new base      # 새로 돌려 대조
+  ./venv/Scripts/python.exe resources/tester/bench_342_watershed.py s1 base store # 흐름 방향 저장본
 
-pytest 대상 아님(DB · 프로젝트 파일이 있어야 돎). 저장 안 함 — 프로젝트 파일은 읽기만
-(흐름 방향 저장본 drainage/descent_*.npz 는 347-1 부터 씀).
+pytest 대상 아님(DB · 프로젝트 파일이 있어야 돎). 분석 결과는 저장 안 함 — 도엽 좌표 저장본
+(drainage/00_sheet_metric.pkl)만 프로젝트에 남고 · 흐름 방향 저장본은 tmp/342/descent 에 둠.
 """
 
 import asyncio
@@ -40,7 +41,8 @@ async def _prepare():
 
 def main() -> None:
     name = sys.argv[1]
-    base = sys.argv[2] if len(sys.argv) > 2 else None
+    base = sys.argv[2] if len(sys.argv) > 2 and sys.argv[2] != "-" else None
+    store = len(sys.argv) > 3 and sys.argv[3] == "store"
     OUT_DIR.mkdir(parents=True, exist_ok=True)
     logging.basicConfig(
         level=logging.INFO,
@@ -51,7 +53,12 @@ def main() -> None:
     from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import preview_stages
 
     started = time.perf_counter()
-    pv = preview_stages(prepared["vertices"], prepared["contours"], prepared["streams"])
+    pv = preview_stages(
+        prepared["vertices"],
+        prepared["contours"],
+        prepared["streams"],
+        (OUT_DIR / "descent", prepared["contour_key"]) if store else None,
+    )
     preview_s = time.perf_counter() - started
     spec = pv.spec
     red = pv.flow.reaches_road & pv.flow.analyzed

resources/tester/test_347_1_descent_store.py

@@ -0,0 +1,128 @@
+"""347-1 흐름 방향 · 도엽 좌표 저장 — 저장본을 읽어도 결과 같음 · 두 번째는 짓지 않음.
+
+- 격자 원점 = 좌표 정수 m(셀 모서리 0,0) — 노선이 바뀌어도 같은 자리 셀은 같은 자리.
+- 방향장 저장본: 입력 지문(도엽 지문 · 남긴 등고 단 · 셀) 같고 범위가 덮으면 읽기만 · 최근 2벌.
+- 도엽 좌표 저장본: 원본 자국 · 좌표계가 같으면 다시 안 바꿈.
+실측(검증_sub3 · 차이 0칸 · 108 → 10.8 s 붐빔 · 340 기준 약 6 s)은 bench_342_watershed.py.
+"""
+
+import math
+
+import numpy as np
+from shapely.geometry import LineString
+
+from B04_PreProcess import B04_PreProcess_Engine_Watershed_Expand as expand
+from B04_PreProcess import B04_PreProcess_Router_Watershed as router
+from B04_PreProcess.B04_PreProcess_Engine_Watershed_Descent import (
+    descent_input_key,
+    iter_saved_descents,
+)
+from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import (
+    GRID_ORIGIN_XY,
+    GridSpec,
+    grid_spec_from_bounds,
+)
+
+
+def _contours():
+    features = []
+    for radius in range(10, 200, 10):
+        coords = [
+            [150.0 + radius * math.cos(t), -150.0 + radius * math.sin(t)]
+            for t in np.linspace(0, 2 * math.pi, 241)
+        ]
+        features.append(
+            {
+                "geometry": {"type": "LineString", "coordinates": coords},
+                "properties": {"elevation": 300.0 - radius},
+            }
+        )
+    return features
+
+
+def test_grid_origin_is_integer_metre():
+    assert GRID_ORIGIN_XY == (0.0, 0.0)
+    spec = grid_spec_from_bounds(10.37, -80.6, 90.2, -3.4, 1.0, anchor_xy=GRID_ORIGIN_XY)
+    assert spec.x_min == 10.0 and spec.y_max == -3.0
+
+
+def test_input_key_follows_contours_and_floor():
+    contours = _contours()
+    base = descent_input_key("도엽-1", contours, None, 1.0)
+    assert base == descent_input_key("도엽-1", contours, None, 1.0)
+    assert base != descent_input_key("도엽-2", contours, None, 1.0)  # 도엽 · 좌표계 바뀜
+    assert base != descent_input_key("도엽-1", contours, 200.0, 1.0)  # 남긴 등고 단 바뀜
+    assert base == descent_input_key("도엽-1", contours, 100.5, 1.0)  # 단 집합 그대로(전부 남음)
+
+
+def _run(store, monkeypatch, counter):
+    real = expand.build_contour_descent
+
+    def counting(*args, **kwargs):
+        counter.append(1)
+        return real(*args, **kwargs)
+
+    monkeypatch.setattr(expand, "build_contour_descent", counting)
+    spec = GridSpec(x_min=100.0, y_max=-100.0, cell_m=1.0, n_rows=100, n_cols=100)
+    domain = np.zeros((100, 100), bool)
+    domain[20:80, 20:80] = True
+    route = LineString([(110.0, -150.0), (190.0, -150.0)])
+    result = expand.expand_by_red_boundary(
+        spec, domain, _contours(), route, [], None, descent_store=store
+    )
+    return result.analysis
+
+
+def test_store_gives_same_result_and_reads_second_time(tmp_path, monkeypatch):
+    monkeypatch.setattr(expand, "DESCENT_SPARE_M", 40.0)
+    built: list[int] = []
+    plain = _run(None, monkeypatch, built)
+    store = (tmp_path, "도엽-1")
+    first = _run(store, monkeypatch, built)
+    count_after_first = len(built)
+    second = _run(store, monkeypatch, built)
+    assert len(built) == count_after_first  # 두 번째는 짓지 않음
+    for analysis in (first, second):
+        assert analysis.spec == plain.spec
+        assert np.array_equal(analysis.flow.reaches_road, plain.flow.reaches_road)
+        assert np.array_equal(analysis.descent.receiver, plain.descent.receiver)
+    saved = list(tmp_path.glob("descent_*.npz"))
+    assert 1 <= len(saved) <= 2
+    key = descent_input_key("도엽-1", _contours(), None, 1.0)
+    assert next(iter_saved_descents(tmp_path, key)).spec.n_rows >= plain.spec.n_rows
+
+
+def test_store_keeps_two_newest(tmp_path, monkeypatch):
+    from B04_PreProcess.B04_PreProcess_Engine_Watershed_Descent import save_descent
+
+    monkeypatch.setattr(expand, "DESCENT_SPARE_M", 0.0)
+    analysis = _run(None, monkeypatch, [])
+    for index in range(3):
+        moved = GridSpec(100.0 + index, -100.0, 1.0, analysis.spec.n_rows, analysis.spec.n_cols)
+        descent = analysis.descent
+        save_descent(tmp_path, "key", type(descent)(**{**descent.__dict__, "spec": moved}))
+    assert len(list(tmp_path.glob("descent_*.npz"))) == 2
+
+
+def test_metric_sheets_cached_until_source_changes(tmp_path, monkeypatch):
+    sheets = tmp_path / "sheets"
+    sheets.mkdir()
+    feature = {
+        "type": "Feature",
+        "properties": {"elevation": 10},
+        "geometry": {"type": "LineString", "coordinates": [[0, 0], [1, 1]]},
+    }
+    import json
+
+    for name in (router.STREAM_FILE, router.CONTOUR_FILE):
+        (sheets / name).write_text(json.dumps({"features": [feature]}), "utf-8")
+    calls = []
+    real = router._reproject_features
+    monkeypatch.setattr(
+        router, "_reproject_features", lambda f, t: calls.append(1) or real(f, None)
+    )
+    first = router._metric_sheets(sheets, tmp_path / "drainage", "EPSG:5186", None)
+    again = router._metric_sheets(sheets, tmp_path / "drainage", "EPSG:5186", None)
+    assert len(calls) == 2 and first == again  # 두 번째는 저장본
+    router._metric_sheets(sheets, tmp_path / "drainage", "EPSG:5179", None)  # 좌표계 바뀜
+    assert len(calls) == 4