diff --git a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Analyze.py b/B04_PreProcess/B04_PreProcess_Engine_Watershed_Analyze.py index 2fadf6ddb..298df74b3 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Analyze.py +++ b/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("배수유역: 등고선 하강 방향을 세우지 못해 흐름 판정을 건너뜁니다.") diff --git a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Descent.py b/B04_PreProcess/B04_PreProcess_Engine_Watershed_Descent.py index e854be436..5afc21cdb 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Descent.py +++ b/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, diff --git a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Expand.py b/B04_PreProcess/B04_PreProcess_Engine_Watershed_Expand.py index 324b8ab02..d1e09fa39 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Expand.py +++ b/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 diff --git a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Grid.py b/B04_PreProcess/B04_PreProcess_Engine_Watershed_Grid.py index 1bad541d4..57d9f62ff 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Grid.py +++ b/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 diff --git a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Stream.py b/B04_PreProcess/B04_PreProcess_Engine_Watershed_Stream.py index 6a6e8cf4f..f89224ea9 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Stream.py +++ b/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, diff --git a/B04_PreProcess/B04_PreProcess_Router_Watershed.py b/B04_PreProcess/B04_PreProcess_Router_Watershed.py index 1011a16b8..c1d55c4a4 100644 --- a/B04_PreProcess/B04_PreProcess_Router_Watershed.py +++ b/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( diff --git a/resources/tester/bench_342_watershed.py b/resources/tester/bench_342_watershed.py index 497903663..ef0653783 100644 --- a/resources/tester/bench_342_watershed.py +++ b/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 diff --git a/resources/tester/test_347_1_descent_store.py b/resources/tester/test_347_1_descent_store.py new file mode 100644 index 000000000..932f5923d --- /dev/null +++ b/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