Aislo/resources/tester/test_609_1_descent_shared.py

"""609-1 도엽별 공용 하강장 — 도엽 넷 이어 붙임 = 통째 짓기 · 두 번째는 짓지 않음 ·
원천 바뀌면 다시 · 번호 없는 좌표계 · 배수 확장이 공용 길 = 프로젝트 길 · 못 지으면 프로젝트 길.

./venv/Scripts/python.exe -m pytest resources/tester/test_609_1_descent_shared.py -q
"""

import numpy as np
import pytest
from pyproj import Transformer
from shapely.geometry import LineString, box, mapping

from B04_PreProcess import B04_PreProcess_Engine_Watershed_DescentShared as shared
from B04_PreProcess import B04_PreProcess_Engine_Watershed_Expand as expand
from B04_PreProcess.B04_PreProcess_Engine_MapSheet import sheet5k_to_bounds
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Descent import build_contour_descent
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import GridSpec
from common_util import common_util_terrain_store as store

EPSG = 5187
CELL = 10.0  # 시험만 10 m(실제 1 m) — 도엽 넷을 빨리 지음


def _corner():
    """도엽 36901074 남동 모서리(도엽 넷이 만나는 점) · 사업 좌표 10 m 눈금."""
    _, lat0, lon1, _ = sheet5k_to_bounds("36901074")
    x, y = Transformer.from_crs("EPSG:4326", f"EPSG:{EPSG}", always_xy=True).transform(lon1, lat0)
    return round(x / CELL) * CELL, round(y / CELL) * CELL


def _features():
    """모서리 곁 봉우리 — 30 m 마다 5 m 내려가는 고리(중심을 비켜 도엽 경계를 비스듬히 지남)."""
    cx, cy = _corner()
    out = []
    for i, radius in enumerate(range(30, 4500, 30)):
        t = np.linspace(0.0, 2 * np.pi, max(64, radius // 5) + 1)
        line = LineString(
            np.column_stack([cx + 137 + radius * np.cos(t), cy - 91 + radius * np.sin(t)])
        )
        out.append({"geometry": mapping(line), "properties": {"elevation": 1000.0 - 5.0 * i}})
    return out


@pytest.fixture
def world(tmp_path, monkeypatch):
    monkeypatch.setattr(store, "TERRAIN_ROOT", tmp_path)
    monkeypatch.setattr(shared, "TERRAIN_ROOT", tmp_path)
    monkeypatch.setattr(shared, "_gpkg_stamp", lambda: "gpkg-1")
    features = _features()

    def from_gpkg(epsg, region):
        area = box(*region)
        return [f for f in features if LineString(f["geometry"]["coordinates"]).intersects(area)]

    monkeypatch.setattr(shared, "_gpkg_features", from_gpkg)
    built = []
    real = shared.build_contour_descent
    monkeypatch.setattr(shared, "build_contour_descent", lambda *a: built.append(1) or real(*a))
    return features, built


def _spec(half=1500.0):
    cx, cy = _corner()
    n = int(2 * half / CELL)
    return GridSpec(cx - half, cy + half, CELL, n, n)


def test_stitched_equals_whole(world):
    features, built = world
    spec = _spec()
    stitched = shared.shared_descent(EPSG, spec)
    assert len(built) == 4  # 도엽 넷
    pad = 20
    big = GridSpec(
        spec.x_min - pad * CELL,
        spec.y_max + pad * CELL,
        CELL,
        spec.n_rows + 2 * pad,
        spec.n_cols + 2 * pad,
    )
    whole = expand.crop_descent(build_contour_descent(big, features), spec)
    assert stitched.valid.sum() > 0.99 * spec.size
    assert np.array_equal(stitched.valid, whole.valid)
    assert np.array_equal(stitched.receiver, whole.receiver)
    assert np.array_equal(stitched.azimuth, whole.azimuth)
    assert np.array_equal(stitched.step_length, whole.step_length)
    assert np.array_equal(stitched.band_elevation, whole.band_elevation, equal_nan=True)


def test_second_read_builds_nothing_and_source_change_rebuilds(world, monkeypatch):
    _, built = world
    shared.shared_descent(EPSG, _spec(500.0))
    first = len(built)
    shared.shared_descent(EPSG, _spec(800.0))  # 다른 프로젝트 · 같은 도엽
    assert first == 4 and len(built) == first
    monkeypatch.setattr(shared, "_gpkg_stamp", lambda: "gpkg-2")  # 전국 파일 바뀜
    shared.shared_descent(EPSG, _spec(500.0))
    assert len(built) == first + 4


def _expand(store_tuple):
    cx, cy = _corner()
    spec = GridSpec(cx - 400, cy + 400, CELL, 80, 80)
    domain = np.zeros((80, 80), bool)
    domain[20:60, 20:60] = True
    route = LineString([(cx - 300, cy - 250), (cx + 300, cy - 250)])
    return expand.expand_by_red_boundary(
        spec, domain, _features(), route, [], None, descent_store=store_tuple
    )


def test_expand_shared_equals_project_path(world, tmp_path):
    _, built = world
    project = _expand((tmp_path / "p1", "도엽-1"))
    via_shared = _expand((tmp_path / "p2", "도엽-1", f"EPSG:{EPSG}"))
    assert built  # 공용 길로 지음
    assert not list((tmp_path / "p2").glob("descent_*.npz"))  # 프로젝트엔 안 남김
    assert via_shared.analysis.spec == project.analysis.spec
    assert np.array_equal(via_shared.analysis.flow.reaches_road, project.analysis.flow.reaches_road)
    assert np.array_equal(via_shared.analysis.descent.receiver, project.analysis.descent.receiver)


def test_shared_failure_falls_back_to_project(world, tmp_path, monkeypatch):
    def fail(*_):
        raise RuntimeError("받기 실패")

    monkeypatch.setattr(shared, "ensure_sheet_descent", fail)
    result = _expand((tmp_path / "p3", "도엽-1", f"EPSG:{EPSG}"))
    assert result is not None and list((tmp_path / "p3").glob("descent_*.npz"))


def test_crs_without_epsg_number(world, tmp_path):
    """PRJ WKT(번호 없음 · 용화 등)도 공용 길 — 좌표계 지문으로 따로 저장."""
    text = (
        "+proj=tmerc +lat_0=38 +lon_0=129 +k=1 +x_0=200000 +y_0=600000 +ellps=GRS80"
        " +towgs84=0,0,0,0,0,0,0 +units=m +no_defs +type=crs"
    )  # 5187 과 같은 격자 · 번호 없음
    crs = shared.descent_crs(text)
    assert crs == text and shared.descent_crs("EPSG:5187") == EPSG
    spec = _spec(500.0)
    by_text = shared.shared_descent(crs, spec)
    by_number = shared.shared_descent(EPSG, spec)
    assert np.array_equal(by_text.receiver, by_number.receiver)
    assert {p.name.split("_")[1][:3] for p in tmp_path.glob("*/descent_*.npz")} == {"wkt", "518"}