Aislo/resources/tester/test_624_1_layers_any_project.py

"""D03 지도 겹 길 — 임도 프로젝트도 받음 · 지형 원천 인자 · 받은 겹 자리(요구 530 · PLAN 624-1).

- 예정노선(D) · 임도(B) 둘 다 급경사 · 음영 · 지적 받음 · 원천 머리(`source`)
- 원천별 파생 저장본 열쇠 다름 · 없는 원천 404 · 틀린 원천 400
- 받은 겹 = `D03_PlanAnalysis/layers/` — B04 `processed/` 파일 안 덮음 · 새 파일 0
- D 옛 자리 파일은 한 번 복사(다시 안 받음) · D 급경사 값 = 옛 길 값
- 라이다 0차곡 = 격자 통째 · 저장본

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

import json
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager

import numpy as np
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from D03_PlanAnalysis import D03_PlanAnalysis_Layers as layers
from D03_PlanAnalysis import D03_PlanAnalysis_Router as router_mod
from D03_PlanAnalysis import D03_PlanAnalysis_TerrainSource as source
from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zero_order

PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
EPSG = 5186
X0, Y0 = 200000.0, 450000.0
BOX = (X0, Y0, X0 + 400.0, Y0 + 400.0)
B04_CADASTRAL = b'{"type": "FeatureCollection", "features": [], "b04": true}'


def _dem(path, tilt: float, n: int = 60, cell: float = 2.0) -> None:
    x = X0 + np.arange(n) * cell
    y = Y0 + np.arange(n) * cell
    z = np.add.outer(np.zeros(n), np.arange(n) * cell * tilt)
    path.parent.mkdir(parents=True, exist_ok=True)
    np.savez(path, x=x, y=y, z=z, valid_mask=np.ones(z.shape, bool), resolution=[cell])


class _Pool:
    @asynccontextmanager
    async def acquire(self):
        yield None


class FakeWfs:
    def __init__(self) -> None:
        self.calls = 0

    def __call__(self, box, epsg, typename):
        self.calls += 1
        lon, lat = 127.0, 37.0
        ring = [[lon, lat], [lon + 1e-4, lat], [lon + 1e-4, lat + 1e-4], [lon, lat]]
        geom = {"type": "Polygon", "coordinates": [ring]}
        return [{"type": "Feature", "geometry": geom, "properties": {"pnu": str(box)}}]


@pytest.fixture
def world(tmp_path, monkeypatch):
    """D 프로젝트(수치지형도) · B 프로젝트(라이다 ptd 확정 + 수치지형도 서피스 · B04 지적도)."""
    import common_util.common_util_crs as crs
    import common_util.common_util_surface_confirmation as confirm
    import config.config_db as config_db

    d_root, b_root = tmp_path / "D", tmp_path / "B"
    _dem(d_root / "B04_PreProcess" / "models_plan" / "dtm_sheet_laplace.npz", 0.3)
    b_models = b_root / "B04_PreProcess" / "models"
    _dem(b_models / "dtm_ptd.npz", 0.6)
    _dem(b_models / "dtm_ptd_smooth.npz", 0.5)
    _dem(b_models / "dtm_sheet_laplace_smooth.npz", 0.2)
    processed = b_root / "B04_PreProcess" / "processed"
    processed.mkdir(parents=True)
    (processed / layers.CADASTRAL_FILE).write_bytes(B04_CADASTRAL)
    meta = {"x_min": BOX[0], "y_min": BOX[1], "x_max": BOX[2], "y_max": BOX[3]}
    (processed / "vworld_white_meta.json").write_text(json.dumps(meta), "utf-8")
    kinds = {"D": "PLAN", "B": "ROAD"}
    state = {"pid": "D"}
    params = {"source_filter": "ptd", "method": "dtm", "smooth": True}

    async def project_row(project_id):
        return {"project_kind": kinds[state["pid"]], "storage_path": state["pid"]}

    async def plan_ends(project_id, center=None, reset=False):
        return d_root, EPSG, [], None

    async def surface_params(connection, project_id):
        return dict(params)

    monkeypatch.setattr(router_mod.store, "project_row", project_row)
    monkeypatch.setattr(router_mod, "_plan_ends", plan_ends)
    monkeypatch.setattr(router_mod.terrain, "received_box", lambda root, epsg: BOX)
    monkeypatch.setattr(source, "resolve_stored_project_path", lambda p: str(tmp_path / p))
    monkeypatch.setattr(confirm, "get_surface_confirmation_params", surface_params)
    monkeypatch.setattr(config_db, "get_db_pool", lambda: _Pool())
    monkeypatch.setattr(crs, "resolve_project_crs", lambda root: f"EPSG:{EPSG}")
    wfs = FakeWfs()
    monkeypatch.setattr(layers, "_wfs", wfs)
    monkeypatch.setattr(zero_order, "heavy_pool", lambda: ThreadPoolExecutor(1))
    app = FastAPI()
    app.include_router(router_mod.router)
    client = TestClient(app)

    def get(pid: str, path: str, **query):
        state["pid"] = pid
        return client.get(f"/api/projects/{PID}/plan/layers/{path}", params=query)

    return {"get": get, "D": d_root, "B": b_root, "wfs": wfs, "params": params}


def _raster(res) -> dict:
    assert res.status_code == 200, res.text
    return json.loads(res.headers["X-Plan-Raster"])


def test_both_kinds_get_slope_with_source(world) -> None:
    get = world["get"]
    assert _raster(get("D", "raster", kind="slope"))["source"] == "sheet"
    assert _raster(get("B", "raster", kind="slope"))["source"] == "lidar"  # 설계 지형 = 라이다
    assert _raster(get("B", "raster", kind="slope", terrain="sheet"))["source"] == "sheet"
    shade = get("B", "raster", kind="hillshade", terrain="lidar").json()
    assert shade["source"] == "lidar" and len(shade["box"]) == 4
    assert get("D", "raster", kind="slope", terrain="lidar").status_code == 404  # 624-4 전
    assert get("B", "raster", kind="slope", terrain="las").status_code == 400


def test_source_keys_differ(world) -> None:
    get = world["get"]
    for terrain in ("lidar", "sheet", "design"):
        get("B", "raster", kind="slope", terrain=terrain)
    saved = sorted((world["B"] / "_derived" / layers.SLOPE_KIND).glob("*.json"))
    assert len(saved) == 2  # design = lidar 같은 격자 → 같은 열쇠
    lidar = _raster(get("B", "raster", kind="slope", terrain="lidar"))
    sheet = _raster(get("B", "raster", kind="slope", terrain="sheet"))
    assert lidar["counts"] != sheet["counts"]


def test_road_lidar_missing_when_sheet_only(world) -> None:
    world["params"].update(source_filter="sheet_laplace")  # 라이다 없는 임도
    assert get_status(world, "lidar") == 404
    assert get_status(world, "design") == 200


def get_status(world, terrain: str) -> int:
    return world["get"]("B", "raster", kind="slope", terrain=terrain).status_code


def test_road_layers_never_touch_b04_files(world) -> None:
    processed = world["B"] / "B04_PreProcess" / "processed"
    before = {p.name: p.read_bytes() for p in processed.iterdir()}
    res = world["get"]("B", "cadastral")
    assert res.status_code == 200 and res.json()["features"]
    after = {p.name: p.read_bytes() for p in processed.iterdir()}
    assert after == before  # 같은 이름 덮음 0 · 새 파일 0
    mine = world["B"] / source.LAYER_DIR / layers.CADASTRAL_FILE
    assert mine.is_file() and mine.read_bytes() != B04_CADASTRAL


def test_plan_old_files_copied_once(world) -> None:
    """D 옛 자리(B04 processed) 지적도 + 열쇠 = 새 자리로 복사 · 다시 받지 않음 · 값 같음."""
    old = world["D"] / "B04_PreProcess" / "processed"
    old.mkdir(parents=True)
    body = {"type": "FeatureCollection", "features": [{"type": "Feature", "properties": {}}]}
    (old / layers.CADASTRAL_FILE).write_text(json.dumps(body), "utf-8")
    key = {"box": list(BOX), "epsg": EPSG, "count": 1, "failed": 0, "failed_tiles": []}
    (old / layers.CADASTRAL_KEY).write_text(json.dumps(key), "utf-8")
    res = world["get"]("D", "cadastral")
    assert res.status_code == 200 and res.json() == body and world["wfs"].calls == 0
    assert (world["D"] / source.LAYER_DIR / layers.CADASTRAL_KEY).is_file()


def test_plan_slope_same_as_before(world, tmp_path) -> None:
    """D 급경사 = 옛 길(`slope_layer(root, models_plan, epsg)`) 값 그대로."""
    res = world["get"]("D", "raster", kind="slope")
    meta = _raster(res)
    old = layers.slope_layer(world["D"], world["D"] / "B04_PreProcess" / "models_plan", EPSG)
    png, old_meta = layers.raster_bytes(old)
    assert res.content == png and meta == {"kind": "slope", "source": "sheet", **old_meta}


def test_lidar_zero_order_whole_grid(world) -> None:
    """라이다 0차곡 = 격자 통째(1 m → 2 m 솎음) · 저장본 · 수치지형도 = 창고 길."""
    n, cell = 200, 1.0
    x, y = X0 + np.arange(n) * cell, Y0 + np.arange(n) * cell
    valley = np.abs(x - x.mean())[None, :] * 0.3 + (y - Y0)[:, None] * 0.2
    npz = world["B"] / "B04_PreProcess" / "models" / "dtm_ptd_smooth.npz"
    np.savez(npz, x=x, y=y, z=valley, valid_mask=np.ones(valley.shape, bool), resolution=[cell])
    res = world["get"]("B", "zero-order", terrain="lidar")
    got = res.json()
    assert res.status_code == 200 and got["source"] == "lidar" and got["cell_m"] == 2.0
    assert got["count"] >= 1 and got["length_m"] > 0
    again = zero_order.dem_zero_order_layer(world["B"], npz, EPSG, compute=False)
    assert again is not None and again["count"] == got["count"]
    frame = source.plan_frame(world["D"], EPSG, BOX, None)
    work, light, _ = source.zero_order_calls(frame, 1.0, 2.0)
    assert work is zero_order.zero_order_layer and light[-1] == frame.dem