Aislo/resources/tester/test_530_1_plan_terrain.py

"""D 예정노선 지형 받기(PLAN 530-1 · 561-2) — 사업 전체 한 벌 · 범위 보기 = 내려받기 없음 ·
옛 도엽 유지 · 새 도엽만 · 범위는 줄지 않음 · 빠짐 없음.

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

import json
from pathlib import Path

import pytest
from pyproj import Transformer

from B04_PreProcess import B04_PreProcess_Engine_SheetStore as store_mod
from B04_PreProcess.B04_PreProcess_Engine_MapSheet import sheets_for_points
from common_util.common_util_crs import resolve_project_crs
from config.config_system_terrain import SHEET_SURFACE_MARGIN_M
from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain

EPSG = 5186
A, B, C = (230000.0, 450000.0), (231500.0, 451200.0), (238000.0, 455000.0)


@pytest.fixture
def fake(tmp_path, monkeypatch):
    seen = {"download": [], "merge": [], "satellite": [], "dem": []}

    def download_sheet(store_dir, sheet_no, version=2):
        seen["download"].append(sheet_no)
        store_dir = Path(store_dir)
        store_dir.mkdir(parents=True, exist_ok=True)
        (store_dir / f"{sheet_no}.zip").write_bytes(b"PK")
        index = store_mod._load_index(store_dir)
        index["sheets"][sheet_no] = {"file": f"{sheet_no}.zip"}
        store_mod._save_index(store_dir, index)

    def merge(store_dir, sheets, output_dir, layers=None):
        seen["merge"].append(list(sheets))
        (output_dir / "도엽_등고선.geojson").write_text("{}", encoding="utf-8")
        return {}

    def satellite(prj, bounds, output_dir, layer_name="Satellite", ext="jpeg"):
        seen["satellite"].append(bounds)
        meta = {"x_min": bounds["x"][0], "x_max": bounds["x"][1]}
        meta.update({"y_min": bounds["y"][0], "y_max": bounds["y"][1]})
        (output_dir / "vworld_satellite_meta.json").write_text(json.dumps(meta), "utf-8")
        return output_dir / "vworld_satellite.png"

    def dem(root, processed, models_dir, route_xy, crs, methods=None, grid_m=None):
        seen["dem"].append((models_dir.name, route_xy.tolist(), crs))
        seen["grid_m"] = grid_m
        return [{"model_file_path": f"B04_PreProcess/{models_dir.name}/dtm.npz"}]

    monkeypatch.setattr(store_mod, "download_sheet", download_sheet)
    monkeypatch.setattr(terrain, "parse_and_merge_sheets", merge)
    monkeypatch.setattr(terrain, "download_vworld_satellite_map", satellite)
    monkeypatch.setattr(terrain, "build_sheet_surface_model", dem)
    monkeypatch.setattr(terrain.layers, "ensure_cadastral", lambda *a: {"count": 0})  # 569-1
    return tmp_path, seen


def _zips(root):
    return sorted(p.stem for p in store_mod.get_project_map_sheets_dir(root).glob("*.zip"))


def _dem_box(seen) -> list[float]:
    """가짜 DEM 이 받은 꼭짓점 상자 + B 절취 여유 = 실제 DEM 상자."""
    (x0, y0), (x1, y1) = seen["dem"][-1][1]
    m = SHEET_SURFACE_MARGIN_M
    return [x0 - m, y0 - m, x1 + m, y1 + m]


def test_one_set_and_extend_keeps_old(fake) -> None:
    root, seen = fake
    first = terrain.ensure_plan_terrain(root, [A, B], EPSG)
    old = _zips(root)
    assert sorted(seen["download"]) == old == sorted(first["sheets"]["downloaded"])
    assert first["satellite"] == "done" and len(first["models"]) == 1
    assert seen["dem"][-1][0] == "models_plan"  # 연도 없는 한 벌
    m = terrain.DEFAULT_MARGIN_M
    box = [min(A[0], B[0]) - m, min(A[1], B[1]) - m, max(A[0], B[0]) + m, max(A[1], B[1]) + m]
    assert _dem_box(seen) == pytest.approx(box) == first["box_m"]
    assert terrain.received_box(root) == pytest.approx(tuple(box))
    # 작업 좌표계 .prj — B 좌표계 창구가 읽음
    assert resolve_project_crs(root) == f"EPSG:{EPSG}"

    # 그은 점이 범위 밖 — 더 받기 = 옛 범위 ∪ 새 범위 · 새 도엽만
    seen["download"].clear()
    second = terrain.ensure_plan_terrain(root, [A, B, C], EPSG)
    now = _zips(root)
    assert set(old) <= set(now)  # 옛 도엽 그대로(prune 없음)
    assert sorted(seen["download"]) == sorted(set(now) - set(old))
    assert seen["merge"][-1] == now  # 병합 = 옛 ∪ 새
    grown = second["box_m"]
    assert grown[0] <= box[0] and grown[1] <= box[1] and grown[2] >= C[0] + m
    first_sat, union = seen["satellite"][0], seen["satellite"][-1]
    assert union["x"][0] <= first_sat["x"][0] and union["x"][1] >= first_sat["x"][1]
    assert second["sheets"]["failed"] == {}

    # 같은 범위 다시 — 받기 0 · 위성 건너뜀 · 병합 열쇠 같음 · 작은 점이어도 범위 줄지 않음
    seen["download"].clear()
    merges = len(seen["merge"])
    third = terrain.ensure_plan_terrain(root, [A, B], EPSG)
    assert seen["download"] == [] and third["satellite"] == "skip"
    assert len(seen["merge"]) == merges and third["box_m"] == grown


def test_extent_no_download_and_covered(fake) -> None:
    root, seen = fake
    before = terrain.plan_extent(root, [A, B], EPSG)
    assert seen["download"] == [] and before["have"] is None and before["covered"] is False
    assert before["margin_m"] == terrain.DEFAULT_MARGIN_M and len(before["box"]) == 4
    x0, y0, x1, y1 = before["box_m"]
    assert before["cells"] == int((x1 - x0) * (y1 - y0) / 4)  # D 지형 격자 2 m
    assert all(not s["stored"] and len(s["ring"]) == 4 for s in before["sheets"])
    terrain.ensure_plan_terrain(root, [A, B], EPSG)
    assert seen["grid_m"] == 2.0  # B 공용 1 m 가 아니라 D 전용 값
    after = terrain.plan_extent(root, [A, B], EPSG)
    assert after["covered"] and after["have_m"] == after["box_m"]
    assert all(s["stored"] for s in after["sheets"])
    assert {s["no"] for s in after["sheets"]} == {s["no"] for s in before["sheets"]}
    assert not terrain.plan_extent(root, [A, C], EPSG)["covered"]  # 그은 점이 밖 = 더 받기
    # 여유 — 바닥 B 절취 여유 · 빈 값 = 처음 값
    assert terrain.clamp_margin(100.0) == SHEET_SURFACE_MARGIN_M
    assert terrain.clamp_margin(None) == terrain.DEFAULT_MARGIN_M
    assert terrain.clamp_margin(1e9) == terrain.MAX_MARGIN_M


def test_box_sheets_leave_no_gap(fake) -> None:
    root, _ = fake
    far = (A[0] + 9000.0, A[1] + 7000.0)  # 11 km
    got = terrain.ensure_plan_terrain(root, [A, far], EPSG)
    x0, y0, x1, y1 = got["box_m"]
    to_wgs = Transformer.from_crs(f"EPSG:{EPSG}", "EPSG:4326", always_xy=True)
    inside = []
    for i in range(41):
        for j in range(41):
            lon, lat = to_wgs.transform(x0 + (x1 - x0) * i / 40, y0 + (y1 - y0) * j / 40)
            inside.append((lat, lon))
    assert set(sheets_for_points(inside)) <= set(got["sheets"]["merged"])


def test_empty_surface_raises_and_keeps_no_box(fake, monkeypatch) -> None:
    root, _ = fake
    monkeypatch.setattr(terrain, "build_sheet_surface_model", lambda *a, **k: [])
    with pytest.raises(ValueError, match="사업 지형면을 만들지 못했습니다"):
        terrain.ensure_plan_terrain(root, [A, B], EPSG)
    assert terrain.received_box(root) is None