Aislo/resources/tester/test_420_1_cross_earth.py

"""420 추천 횡단 토공 · 저장본 판 · [재계산] 서버(요구 254).

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

횡단 토공 = 공식 횡단 단면적 평균단면법(저장 정본 ±6 %) · 차 = 바뀐 측점만 · 결과 표 절 · 성 =
횡단 · [재계산] = 잠금 안 지우고 새로 셈(셈 수 1 → 2) · 다른 지문은 남기고 30일 지난 것만 지움 ·
표준 횡단면 = 저장 패널값(키 · 엔진에 감) · `initial_choice` with_profile 되살림.
"""

import asyncio
import json
import os
import time
from pathlib import Path
from uuid import uuid4

import numpy as np
import pytest
from test_224_3_recommend_store import EXPECTED, STREAMS, _ask, _key, _terrain, wired  # noqa: F401

from B05_Profile import B05_Profile_Router_RouteInitial as router
from B05_Profile.B05_Profile_Engine_RouteInitial import (
    Terrain,
    read_initial_choice,
    write_initial_choice,
)
from B05_Profile.B05_Profile_Engine_RouteInitial_CrossEarth import (
    attach_cross_earth,
    cross_earthwork,
    cross_earthwork_delta,
)
from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import _station_areas, route_sections
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
    ALGO_VERSION,
    DEFAULT_MODES,
    source_fingerprint,
    store_dir,
)

ROOT = Path(__file__).resolve().parents[2]
SUB2 = ROOT / "storage/1/3/cb15892a-5788-4a26-9316-20a15421f68b"


def _slope() -> Terrain:
    """옆경사 30 % 사면(노선 = x 방향) — 절 · 성 둘 다 생김."""
    x = np.arange(0.0, 200.0, 2.0)
    y = np.arange(-80.0, 80.0, 2.0)
    xx, yy = np.meshgrid(x, y)
    return Terrain(x, y, 0.05 * xx + 0.3 * yy, np.ones(xx.shape, dtype=bool))


LINE = [(20.0, 0.0), (100.0, 0.0), (180.0, 0.0)]


def test_횡단_토공_평균단면법() -> None:
    terrain = _slope()
    earth = cross_earthwork(LINE, None, None, terrain)
    rows = earth["per_station"]
    assert [r["chainage_m"] for r in rows] == [float(s) for s in range(0, 161, 20)]
    # 직접 셈한 단면적 · Σ (A_i + A_i+1) / 2 · L
    at, ground, sections, modes, widen = route_sections(LINE, terrain)
    areas = [_station_areas(sections[i], modes[i], ground[i], widen[i]) for i in range(len(at))]
    cut = sum((areas[i][0] + areas[i + 1][0]) / 2 * 20.0 for i in range(len(at) - 1))
    fill = sum((areas[i][2] + areas[i + 1][2]) / 2 * 20.0 for i in range(len(at) - 1))
    assert earth["cut"] == pytest.approx(cut, abs=0.1) and earth["cut"] > 0
    assert earth["fill"] == pytest.approx(fill, abs=0.1) and earth["fill"] > 0
    # 계획고를 1 m 올리면 성토가 늚 · 측점 수와 다른 계획고는 거절
    up = cross_earthwork(LINE, None, np.asarray(ground) + 1.0, terrain)
    assert up["fill"] > earth["fill"] and up["cut"] < earth["cut"]
    with pytest.raises(ValueError):
        cross_earthwork(LINE, None, [1.0, 2.0], terrain)


def test_표준_횡단면이_단면적을_바꿈() -> None:
    from config.config_system_design import STANDARD_CROSS_SECTION

    terrain = _slope()
    wide = json.loads(json.dumps(STANDARD_CROSS_SECTION))
    for group in wide.values():
        group["road_width_m"] += 2.0
    base = cross_earthwork(LINE, None, None, terrain)
    assert cross_earthwork(LINE, None, None, terrain, wide)["cut"] > base["cut"]


def test_토공_차_바뀐_측점만() -> None:
    terrain = _slope()
    moved = [(20.0, 0.0), (100.0, 6.0), (180.0, 0.0)]
    changed = [60.0, 80.0, 100.0]
    same = cross_earthwork_delta(LINE, LINE, changed, terrain)
    assert same["cut"] == 0 and same["fill"] == 0 and same["stations"] == 3
    delta = cross_earthwork_delta(LINE, moved, changed, terrain)
    old = cross_earthwork(LINE, changed, None, terrain)["per_station"]
    new = cross_earthwork(moved, changed, None, terrain)["per_station"]
    expect = 20.0 * sum(n["fill_m2"] - o["fill_m2"] for n, o in zip(new, old))
    assert delta["fill"] == pytest.approx(expect, abs=0.2)
    # 옛 결과를 넘기면 옛 단면적을 다시 안 셈 · 전체 어림 = 옛 합 + 차
    whole = cross_earthwork(LINE, None, None, terrain)
    reuse = cross_earthwork_delta(whole, moved, changed, terrain)
    assert reuse["fill"] == pytest.approx(delta["fill"], abs=0.2)
    assert reuse["fill_total"] == pytest.approx(whole["fill"] + reuse["fill"], abs=0.2)
    assert cross_earthwork_delta(LINE, moved, [], terrain)["earth"] == 0


@pytest.mark.skipif(not (SUB2 / "B05_Profile").is_dir(), reason="검증_sub2 storage 없음")
def test_검증_sub2_저장_정본_6퍼센트() -> None:
    """저장된 노선 측점 · 계획고 = 지반 — 공식 [저장] 단면적 합과 ±6 %(2 m 추천 지형 · 1 m 표본)."""
    from B05_Profile.B05_Profile_Engine_RouteInitial import load_terrain_and_streams
    from B05_Profile.B05_Profile_Engine_Solver import _MODELS_SUBDIR
    from common_util.common_util_route_geometry import read_planned_route_csv
    from common_util.common_util_surface_sampler import build_surface_sampler

    route = read_planned_route_csv(SUB2 / "B05_Profile/route/planned_route_initial.csv")
    curves = json.loads(
        (SUB2 / "B05_Profile/route/planned_route_initial_curves.json").read_text("utf-8")
    )
    curves = curves.get("curves") if isinstance(curves, dict) else curves
    line = [(v.x, v.y) for v in route.vertices]
    files = sorted((SUB2 / "B06_Section/cross_sections").glob("cross_*.json"))
    at = [json.loads(p.read_text("utf-8"))["chainage_m"] for p in files]
    # 공식 [저장] 단면적(DB cross_sections.design · 2026-10-18 뜸 · 계획고 = 지반) 평균단면법 합
    stored_cut, stored_fill = 5166.0, 23079.3
    selection = {"source_filter": "classification", "method": "dtm", "smooth": True}
    terrain, _ = load_terrain_and_streams(SUB2, selection)
    surface = build_surface_sampler(SUB2 / _MODELS_SUBDIR, "classification", "dtm", True)
    for ground in (terrain, surface):
        earth = cross_earthwork(line, at, None, ground, design_curves=curves)
        assert earth["cut"] == pytest.approx(stored_cut, rel=0.06)
        assert earth["fill"] == pytest.approx(stored_fill, rel=0.06)


def test_결과_절성_횡단_값() -> None:
    terrain = _slope()
    plain = {"planned": [list(p) for p in LINE], "curves": [], "metrics": {"cut_m3": 1.0}}
    profiled = {
        "planned": [list(p) for p in LINE],
        "metrics": {"cut_m3": 1.0, "fill_m3": 2.0, "profile": {"cut_m3": 30.0, "fill_m3": 20.0}},
    }
    result = {**plain, "unsnapped": profiled}
    attach_cross_earth(result, terrain)
    earth = cross_earthwork(LINE, None, None, terrain)
    assert result["metrics"]["cut_m3"] == earth["cut"] and result["metrics"]["band_cut_m3"] == 1.0
    assert result["metrics"]["surplus_m3"] == pytest.approx(earth["cut"] - earth["fill"], abs=0.2)
    shown = result["unsnapped"]["metrics"]
    assert (shown["cut_m3"], shown["fill_m3"], shown["band_fill_m3"]) == (30.0, 20.0, 2.0)
    attach_cross_earth(result, terrain)  # 두 번 붙여도 띠 가늠을 안 덮음
    assert result["metrics"]["band_cut_m3"] == 1.0


def test_표준_횡단면_키_엔진에_감(wired) -> None:  # noqa: F811
    assert _key(standard={"a": 1}) != _key() and _key(standard={"a": 1}) == _key(standard={"a": 1})
    wired.standard = {"road": {"width_m": 5.0}}
    _ask(mode="earthwork")
    assert wired.inputs[-1]["standard"] == wired.standard


def test_재계산_잠금_안_지우고_새로(wired) -> None:  # noqa: F811
    first = _ask(mode="earthwork")
    _ask(mode="gentle")
    _ask(mode="earthwork", criteria={"min_radius_m": 15.0})  # 다른 묶음
    assert wired.count == 3 and first["stored"] is False
    files = {p.name: p.read_text("utf-8") for p in store_dir(wired.root).glob("*.json")}
    again = _ask(mode="earthwork", recompute=True)
    assert again["stored"] is False and wired.count == 4  # 셈 1 → 2(earthwork)
    now = {p.name: json.loads(p.read_text("utf-8")) for p in store_dir(wired.root).glob("*.json")}
    # 같은 묶음 gentle 은 지워짐 · 다른 묶음은 그대로 · earthwork 파일은 새로 씀
    assert len(now) == 2
    other = [n for n, d in now.items() if d["request"]["criteria"]]
    assert other and now[other[0]] == json.loads(files[other[0]])
    fresh = next(d for n, d in now.items() if not d["request"]["criteria"])
    assert "recompute" not in fresh["request"] and fresh["version"] == ALGO_VERSION
    assert fresh["result"]["metrics"]["cut_m3"] == 10.0  # 횡단(균형 종단) 값
    assert _ask(mode="earthwork")["stored"] and wired.count == 4


def test_재계산_메모리_묶음도_버림(wired) -> None:  # noqa: F811
    from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS
    from B05_Profile.B05_Profile_Engine_RouteInitial_Store import request_fingerprint

    _ask(mode="earthwork")
    pool = request_fingerprint(
        EXPECTED,
        _terrain(),
        STREAMS,
        mode="earthwork",
        weights=None,
        max_grade=0.14,
        min_radius_m=12.0,
        stream_offset_m=None,
        grade_smooth_m=None,
        target_grade=None,
        with_profile=True,
        design_speed=20,
    )[1]
    _POOLS[bytes.fromhex(pool)] = "옛 묶음"
    _ask(mode="earthwork")
    assert bytes.fromhex(pool) in _POOLS
    _ask(mode="earthwork", recompute=True)
    assert bytes.fromhex(pool) not in _POOLS


def _foreign(folder: Path, name: str, days: float, **fields) -> None:
    """다른 판 · 지문 저장본(다른 PC · 워크트리 코드) — `days` 전 수정."""
    data = {"version": ALGO_VERSION, "source": "0" * 16, "request": {}, **fields}
    path = folder / f"{name}.json"
    path.write_text(json.dumps(data), encoding="utf-8")
    stamp = time.time() - days * 86400.0
    os.utime(path, (stamp, stamp))


def test_다른_지문은_남기고_30일_지난_것만_지움(wired) -> None:  # noqa: F811
    folder = store_dir(wired.root)
    folder.mkdir(parents=True)
    _foreign(folder, "old_new", 1, version=ALGO_VERSION - 1, source=source_fingerprint())
    _foreign(folder, "other_new", 29)
    _foreign(folder, "old_stale", 31, version=ALGO_VERSION - 1)
    _foreign(folder, "other_stale", 40)
    (folder / "broken.json").write_text("{", encoding="utf-8")
    asyncio.run(router.precompute_defaults(uuid4()))
    names = {p.stem for p in folder.glob("*.json")}
    assert {"old_new", "other_new", "broken"} <= names
    assert not names & {"old_stale", "other_stale"}
    assert len(names) == 3 + len(DEFAULT_MODES) and wired.count == len(DEFAULT_MODES)


def test_다른_지문은_재계산_정리에서도_안_건드림(wired) -> None:  # noqa: F811
    _ask(mode="earthwork")
    saved = next(store_dir(wired.root).glob("*.json"))
    pool = json.loads(saved.read_text("utf-8"))["pool"]
    # 같은 묶음 · 같은 종류지만 다른 지문 — 최근 두 벌 정리 · [재계산] 둘 다 남김
    folder = store_dir(wired.root)
    for index in range(3):
        _foreign(folder, f"other{index}", 0, pool=pool, request={"mode": "earthwork"})
    _ask(mode="earthwork", criteria={"min_radius_m": 15.0})
    _ask(mode="earthwork", recompute=True)
    names = {p.stem for p in folder.glob("*.json")}
    assert {"other0", "other1", "other2"} <= names


def test_소스_지문이_바뀌면_다른_키(monkeypatch: pytest.MonkeyPatch) -> None:
    import B05_Profile.B05_Profile_Engine_RouteInitial_Store as store

    before = _key()
    monkeypatch.setattr(store, "source_fingerprint", lambda: "f" * 16)
    assert _key() != before


def test_with_profile_되살림(tmp_path: Path) -> None:
    assert read_initial_choice(tmp_path)["with_profile"] is True  # 없으면 켬
    write_initial_choice(tmp_path, "gentle", None, None, False)
    assert read_initial_choice(tmp_path)["with_profile"] is False
    write_initial_choice(tmp_path, "gentle", None, None)
    assert read_initial_choice(tmp_path)["with_profile"] is True
    # 옛 파일(칸 없음) = 켬
    path = tmp_path / "B05_Profile/route/initial_choice.json"
    path.write_text(json.dumps({"logic": "gentle"}), encoding="utf-8")
    assert read_initial_choice(tmp_path)["with_profile"] is True


def test_적용이_종단_반영을_적음_소스() -> None:
    source = (ROOT / "B05_Profile/B05_Profile_Router_Replan.py").read_text(encoding="utf-8")
    call = source[source.index("write_initial_choice,\n") :][:600]
    assert "request.balanced_profile" in call
    initial = (ROOT / "B05_Profile/B05_Profile_Router_RouteInitial.py").read_text(encoding="utf-8")
    # 지우기는 잠금 안
    lock = initial.index("async with pool_lock(pool_key):")
    assert lock < initial.index("drop_pool_results, project_root, pool_key") < lock + 400
    assert 'model_dump(exclude={"recompute"})' in initial
    # [적용] 균형 종단도 저장 패널값
    assert "standard=standard" in initial[initial.index("async def balanced_profile_of") :]