Aislo/resources/tester/test_616_zero_order_fast.py

"""PLAN 616 — 0차곡 첫 계산 빠르게: 창고 격자에서 셈(616-1) · 도엽 선 미리 뽑아 이어 붙임(616-2) ·
무거운 셈 = 서버 밖 일꾼(616-3) · 급경사 값 그림 바이너리(616-4).

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

from __future__ import annotations

import base64
import os
import sys
from pathlib import Path

import numpy as np
import pytest
import shapely

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))

from common_util import common_util_terrain_store as store  # noqa: E402
from common_util.common_util_atomic import atomic_write_npz  # noqa: E402
from common_util.common_util_json import atomic_write_json  # noqa: E402
from D03_PlanAnalysis import D03_PlanAnalysis_Layers as layers  # noqa: E402
from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zo  # noqa: E402
from tools import precompute_terrain as tool  # noqa: E402

EPSG = 5187
WEST, EAST = "36901074", "36901075"
CELL = 10.0  # 시험은 굵은 칸(여유 200 m = 20 칸 · 기본 임계 = 10 ~ 100 칸)


def _height(x, y):
    """기운 면 + 골 · 마루 — 웅덩이 없음 · 작은 집수."""
    return 0.08 * x + 6.0 * np.sin(y / 38.0) + 2.0 * np.sin(x / 55.0)


def _put_sheet(no: str) -> None:
    ix0, iy0, own = store._owned(no, EPSG, CELL)
    rows, cols = own.shape
    gx, gy = np.meshgrid((np.arange(cols) + ix0 + 0.5) * CELL, (np.arange(rows) + iy0 + 0.5) * CELL)
    z = _height(gx - 300000.0, gy).astype(np.float32)
    z[~own] = np.nan
    npz, key_path, _ = store._names(no, EPSG, CELL)
    atomic_write_npz(npz, z=z, ix0=np.int64(ix0), iy0=np.int64(iy0), cell_m=np.float64(CELL))
    atomic_write_json(key_path, {**store._key(EPSG, CELL, {}), "source": "gpkg", "seconds": 1})


@pytest.fixture
def shared(tmp_path, monkeypatch):
    """임시 창고 · 칸 10 m · d8 부름 셈."""
    monkeypatch.setattr(store, "TERRAIN_ROOT", tmp_path / "terrain")
    monkeypatch.setattr(zo, "ZERO_ORDER_CELL_M", CELL)
    calls = []
    real = zo.d8_accumulation
    monkeypatch.setattr(zo, "d8_accumulation", lambda *a: calls.append(1) or real(*a))
    return calls


def _models(tmp_path: Path, shift: float = 0.0) -> Path:
    """두 도엽 가운데를 덮는 사업 DEM(칸 = 시험 칸 · 값은 안 씀 · 범위만)."""
    grid = store.load_terrain(EPSG, sheets=[WEST, EAST], cell_m=CELL)
    x, y = grid["x"][30:-30] + shift, grid["y"][30:-30]
    models = tmp_path / f"models{shift:g}"
    models.mkdir()
    np.savez(
        models / f"dtm_sheet_{zo.SHEET_SURFACE_DEFAULT_METHOD}.npz",
        x=x,
        y=y,
        z=np.zeros((len(y), len(x))),
        valid_mask=np.ones((len(y), len(x)), bool),
        resolution=np.array([CELL], np.float32),
    )
    return models


def test_sheet_lines_join_equals_box(shared, tmp_path) -> None:
    """도엽 선(여유 200 m 에서 뽑아 끝점 도엽만) 이어 붙임 = 두 도엽 통째 상자에서 뽑은 선."""
    _put_sheet(WEST)
    _put_sheet(EAST)
    paths = [zo.sheet_lines(no, EPSG, CELL) for no in (WEST, EAST)]
    joined = [g for p in paths for g in zo._read_lines(p)]
    grid = store.load_terrain(EPSG, sheets=[WEST, EAST], cell_m=CELL)
    half = CELL / 2
    box = (grid["x"][0] - half, grid["y"][0] - half, grid["x"][-1] + half, grid["y"][-1] + half)
    accs = [zo._acc_names(no, EPSG, CELL)[0] for no in (WEST, EAST)]
    whole = zo._box_lines(accs, EPSG, box, CELL, zo.ZERO_ORDER_MIN_M2, zo.ZERO_ORDER_MAX_M2)
    assert len(whole) >= 20
    # 여유 200 m 보다 긴 선(시험 지형 = 동서로 긴 골)만 창 끝에서 잘려 다를 수 있음 — 1 % 안
    assert abs(len(joined) - len(whole)) <= 0.01 * len(whole)
    total = sum(g.length for g in whole)
    assert abs(sum(g.length for g in joined) - total) <= 0.01 * total
    keys = {tuple(map(tuple, shapely.get_coordinates(g).round(3))) for g in joined}
    assert len(keys) == len(joined)  # 겹쳐 뽑힌 선은 한 도엽에만


def test_layer_default_reads_lines_light(shared, tmp_path) -> None:
    """기본 임계 = 창고 도엽 선 이어 붙여 자름 — 창고에 없으면 compute=False 가 None(일꾼 몫) ·
    있으면 d8 · 선 뽑기 없이 바로."""
    calls = shared
    _put_sheet(WEST)
    _put_sheet(EAST)
    models = _models(tmp_path)
    assert zo.zero_order_layer(tmp_path, models, EPSG, compute=False) is None
    assert calls == []
    first = zo.zero_order_layer(tmp_path, models, EPSG)
    assert first["count"] >= 10 and first["cell_m"] == CELL
    assert len(calls) == 2  # 도엽 둘 누적 한 번씩
    light = zo.zero_order_layer(tmp_path, models, EPSG, compute=False)
    assert (light["count"], light["length_m"]) == (first["count"], first["length_m"])
    assert len(calls) == 2
    assert light["collection"]["features"][0]["geometry"]["type"] == "LineString"
    # 받은 범위 밖 선은 잘림 — 모든 점이 범위 안
    with np.load(models / f"dtm_sheet_{zo.SHEET_SURFACE_DEFAULT_METHOD}.npz") as dem:
        x0, x1 = dem["x"][0] - CELL / 2, dem["x"][-1] + CELL / 2
    from pyproj import Transformer

    to_m = Transformer.from_crs("EPSG:4326", f"EPSG:{EPSG}", always_xy=True)
    pts = np.array(
        [p for f in light["collection"]["features"] for p in f["geometry"]["coordinates"]]
    )
    mx, _ = to_m.transform(pts[:, 0], pts[:, 1])
    assert mx.min() >= x0 - 0.01 and mx.max() <= x1 + 0.01


def test_layer_custom_threshold_saved_and_grid_free(shared, tmp_path) -> None:
    """다른 임계 = 창고 누적에서 셈(사업 DEM 격자 안 씀 · 반 칸 어긋나도 창고) · 둘째 = 저장본."""
    calls = shared
    _put_sheet(WEST)
    _put_sheet(EAST)
    models = _models(tmp_path)
    cells = CELL * CELL
    assert zo.zero_order_layer(tmp_path, models, EPSG, 5 * cells, 40 * cells, False) is None
    first = zo.zero_order_layer(tmp_path, models, EPSG, 5 * cells, 40 * cells)
    assert first["count"] >= 1 and len(calls) == 2
    again = zo.zero_order_layer(tmp_path, models, EPSG, 5 * cells, 40 * cells, False)
    assert again == first  # 저장본
    shifted = zo.zero_order_layer(
        tmp_path, _models(tmp_path, CELL / 2), EPSG, 5 * cells, 40 * cells
    )
    assert shifted["count"] >= 1 and len(calls) == 2  # 옛 격자 갈래 없음 = 누적 다시 안 셈


def test_lines_recomputed_when_neighbor_acc_appears(shared) -> None:
    _put_sheet(WEST)
    _put_sheet(EAST)
    path = zo.sheet_lines(WEST, EPSG, CELL)  # 동쪽 누적 없음 = 빼고 셈
    stamp = path.stat().st_mtime_ns
    assert zo.sheet_lines(WEST, EPSG, CELL, compute=False) == path
    zo.sheet_accumulation(EAST, EPSG, CELL)
    with pytest.raises(zo.NeedsCompute):
        zo.sheet_lines(WEST, EPSG, CELL, compute=False)
    assert zo.sheet_lines(WEST, EPSG, CELL).stat().st_mtime_ns != stamp


def test_acc_key_version_unchanged() -> None:
    """누적 열쇠 판을 올리면 전국 미리 계산 누적을 통째 다시 셈 — 616 은 안 올림."""
    assert zo.ACC_VERSION == 2


def test_precompute_zero_lines_step(shared, monkeypatch) -> None:
    _put_sheet(WEST)
    _put_sheet(EAST)
    monkeypatch.setattr(tool, "CELL_M", CELL)
    assert "zero_lines" in tool.STEPS and tool.STEPS.index("zero_lines") == 2
    _, _, info = tool.run_step("zero_lines", WEST)
    assert info["ok"] and info["fresh"] and not info["empty"]
    _, _, again = tool.run_step("zero_lines", WEST)
    assert again["ok"] and not again["fresh"]


def test_raster_bytes_roundtrip() -> None:
    png = b"\x89PNG\r\n\x1a\nfake"
    value = {"image": "data:image/png;base64," + base64.b64encode(png).decode(), "counts": [1, 2]}
    got, meta = layers.raster_bytes(value)
    assert got == png and meta == {"counts": [1, 2]}


def test_heavy_pool_runs_outside() -> None:
    """일꾼 = 다른 프로세스 · 띄울 때 __main__ 을 되돌려 놓음."""
    main = sys.modules["__main__"]
    pool = zo.heavy_pool()
    try:
        assert sys.modules["__main__"] is main
        assert pool.submit(os.getpid).result(timeout=120) != os.getpid()
        assert zo.heavy_pool() is pool
    finally:
        pool.shutdown()
        zo._POOL = None