Aislo/resources/tester/test_route_initial.py

"""초기 계획노선 갈래(PLAN 20장) — `B05_Profile_Engine_RouteInitial` 합성 지형 시험.

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

공통 제약(종단기울기 상한 · 최소 곡선반지름 · 계류 이격)은 넘지 못함 — 지형상 불가능할 때만
`metrics.violations` 에 측점 · 값 · 사유로 뜨고 `over_limit.total` 은 그 수와 같다.
"""

import numpy as np
import pytest

from B05_Profile.B05_Profile_Engine_RouteInitial import (
    MODES,
    PENDING,
    STREAM_OFFSET_M,
    Terrain,
    generate_initial_route,
    read_initial_choice,
    resolve_weights,
    stream_lines,
    write_initial_choice,
)
from B05_Profile.B05_Profile_Engine_RouteInitial_Metrics import grade_line, station_label

GRADE = 0.14
RADIUS = 12.0


def _terrain(z_of) -> Terrain:
    x = np.arange(0.0, 802.0, 2.0)
    y = np.arange(0.0, 402.0, 2.0)
    xx, yy = np.meshgrid(x, y)
    return Terrain(x, y, z_of(xx, yy), np.ones(xx.shape, dtype=bool))


def _line(y: float) -> list[tuple[float, float]]:
    return [(float(x), y) for x in np.arange(50.0, 752.0, 4.0)]


def _run(terrain: Terrain, mode: str, streams=(), weights=None) -> dict:
    return generate_initial_route(
        _line(200.0),
        terrain,
        list(streams),
        mode=mode,
        weights=weights,
        max_grade=GRADE,
        min_radius_m=RADIUS,
    )


def _kept(result: dict) -> None:
    """제약을 지켰거나, 못 지킨 곳이 목록에 있고 수가 같음."""
    metrics = result["metrics"]
    assert metrics["over_limit"]["total"] == len(metrics["violations"])
    if not metrics["violations"]:
        assert metrics["max_grade_pct"] <= GRADE * 100.0 + 1e-6
        assert result["relaxed"] is False
    for curve in result["curves"]:
        assert curve["radius_m"] >= RADIUS - 1e-6 or curve["violations"]


@pytest.fixture(scope="module")
def hill() -> Terrain:
    """가운데 솟은 언덕(높이 40 m) — 곧은 예상노선이 꼭대기(측면 27 %)를 넘는다."""
    return _terrain(
        lambda xx, yy: 40.0 * np.exp(-((xx - 400.0) ** 2 + (yy - 200.0) ** 2) / 90.0**2)
    )


@pytest.mark.parametrize("mode", MODES)
def test_every_mode_keeps_grade_limit_on_hill(hill: Terrain, mode: str) -> None:
    result = _run(hill, mode)
    _kept(result)
    metrics = result["metrics"]
    assert metrics["over_limit"]["total"] == 0
    assert metrics["max_grade_pct"] <= GRADE * 100.0 + 1e-6
    # 기점 · 종점은 옮기지 않고 복도(100 m) 밖으로 안 나감
    assert result["planned"][0] == [50.0, 200.0]
    assert result["planned"][-1] == [750.0, 200.0]
    assert metrics["offset_max_m"] <= result["criteria"]["corridor_m"] + 4.0


def test_follow_stays_on_expected_where_allowed() -> None:
    """고른 경사(2 %) — 추종은 예상노선을 그대로(격자 칸 반 안쪽)."""
    ramp = _terrain(lambda xx, yy: 0.02 * xx)
    result = _run(ramp, "follow")
    _kept(result)
    assert result["metrics"]["offset_max_m"] <= 3.0


def test_weighted_expected_share_pulls_route_back(hill: Terrain) -> None:
    near = _run(hill, "weighted", weights={"expected": 90, "earthwork": 5, "grade": 5})
    far = _run(hill, "weighted", weights={"expected": 10, "earthwork": 45, "grade": 45})
    assert near["metrics"]["offset_avg_m"] <= far["metrics"]["offset_avg_m"]


def test_stream_offset_is_hard() -> None:
    """평지 · 예상노선 20 m 옆을 나란히 흐르는 소하천 — 모든 갈래가 버퍼를 비켜 감."""
    flat = _terrain(lambda xx, yy: np.zeros_like(xx))
    stream = [[(0.0, 220.0), (800.0, 220.0)]]
    for mode in MODES:
        result = _run(flat, mode, streams=stream)
        _kept(result)
        assert result["metrics"]["over_limit"]["stream"] == 0
        assert result["metrics"]["offset_avg_m"] > STREAM_OFFSET_M - 20.0 - 30.0


def test_stream_crossing_is_allowed() -> None:
    flat = _terrain(lambda xx, yy: np.zeros_like(xx))
    crossing = [[(400.0, 0.0), (400.0, 400.0)]]
    result = _run(flat, "follow", streams=crossing)
    _kept(result)
    assert result["metrics"]["over_limit"]["total"] == 0


def test_terrain_impossible_is_listed() -> None:
    """복도 전체를 가로지르는 20 m 벽 — 어떤 길도 상한을 넘음 → 목록 · 사유 · 막은 기울기."""
    wall = _terrain(lambda xx, yy: np.where(xx > 400.0, 20.0, 0.0))
    result = _run(wall, "weighted")
    _kept(result)
    assert result["relaxed"] is True
    assert result["terrain_bound_pct"] > GRADE * 100.0
    grades = [v for v in result["metrics"]["violations"] if v["kind"] == "grade"]
    assert grades and all("지형상 불가능" in v["reason"] for v in grades)
    assert all(v["station"].startswith("No.") for v in grades)


def test_grade_line_stays_within_limit() -> None:
    ground = np.array([0.0, 0.0, 5.0, 10.0, 10.0, 2.0, 0.0])
    line = grade_line(ground, 2.0, GRADE)
    assert np.all(np.abs(np.diff(line)) <= GRADE * 2.0 + 1e-9)
    assert station_label(125.0) == "No.6+5.0"


def test_weights_and_stream_filter() -> None:
    assert resolve_weights("weighted", None) == {"expected": 50.0, "earthwork": 25.0, "grade": 25.0}
    assert resolve_weights("gentle", {"expected": 99})["grade"] == 100.0
    assert set(MODES) == {"follow", "earthwork", "gentle", "weighted"}
    line = {"type": "LineString", "coordinates": [[0, 0], [1, 1]]}
    features = [
        {"properties": {"구분": "세류"}, "geometry": line},
        {"properties": {"구분": "소하천"}, "geometry": line},
    ]
    assert len(stream_lines(features)) == 1


def test_unknown_mode_rejected(hill: Terrain) -> None:
    with pytest.raises(ValueError):
        _run(hill, "shortest")


def test_criteria_confirmed(hill: Terrain) -> None:
    """기준값 확정(2026-09-29 사용자 결정) — `pending` 은 이제 늘 빈 목록."""
    assert _run(hill, "follow")["criteria"]["pending"] == [] == list(PENDING)


def test_criteria_overrides_change_result(hill: Terrain) -> None:
    """화면 제약 칸(계류 이격 · 지반 고름)을 겹치면 그 값으로 셈하고, 안 겹치면 기본값."""
    default = generate_initial_route(
        _line(200.0),
        hill,
        [],
        mode="follow",
        weights=None,
        max_grade=GRADE,
        min_radius_m=RADIUS,
    )
    assert default["criteria"]["stream_offset_m"] == STREAM_OFFSET_M
    overridden = generate_initial_route(
        _line(200.0),
        hill,
        [],
        mode="follow",
        weights=None,
        max_grade=GRADE,
        min_radius_m=RADIUS,
        stream_offset_m=30.0,
        grade_smooth_m=2.0,
    )
    assert overridden["criteria"]["stream_offset_m"] == 30.0
    assert overridden["criteria"]["grade_smooth_m"] == 2.0
    assert overridden["criteria"]["max_grade_pct"] == default["criteria"]["max_grade_pct"]
    _kept(overridden)


def test_initial_choice_roundtrip(tmp_path) -> None:
    """[확인] 때 적은 현재 초기값(갈래 · 비중 · 제약 겹침)을 다시 읽음 · 없거나 깨지면 추종."""
    assert read_initial_choice(tmp_path) == {
        "logic": "follow",
        "weights": {"expected": 100.0, "earthwork": 0.0, "grade": 0.0},
        "criteria": None,
    }
    write_initial_choice(tmp_path, "weighted", {"expected": 60, "earthwork": 20, "grade": 20})
    assert read_initial_choice(tmp_path) == {
        "logic": "weighted",
        "weights": {"expected": 60.0, "earthwork": 20.0, "grade": 20.0},
        "criteria": None,
    }
    # 사용자가 화면 칸에 겹쳐 넣은 값만(빈 칸은 안 담김) — 남으면 다시 열 때 그 칸만 채운다.
    write_initial_choice(
        tmp_path, "earthwork", None, {"stream_offset_m": 30.0, "grade_smooth_m": 2.0}
    )
    assert read_initial_choice(tmp_path) == {
        "logic": "earthwork",
        "weights": {"expected": 0.0, "earthwork": 100.0, "grade": 0.0},
        "criteria": {"stream_offset_m": 30.0, "grade_smooth_m": 2.0},
    }
    (tmp_path / "B05_Profile" / "route" / "initial_choice.json").write_text("{", encoding="utf-8")
    assert read_initial_choice(tmp_path)["logic"] == "follow"
    assert read_initial_choice(tmp_path)["criteria"] is None


def test_request_accepts_logic_name() -> None:
    from B05_Profile.B05_Profile_Router_Replan import RouteInitialRequest, RouteReplanRequest

    assert RouteInitialRequest.model_validate({"logic": "gentle"}).mode == "gentle"
    assert RouteInitialRequest.model_validate({"mode": "earthwork"}).mode == "earthwork"
    replan = RouteReplanRequest.model_validate({"vertices": [], "initial": {"logic": "weighted"}})
    assert replan.initial is not None and replan.initial.mode == "weighted"


def test_request_accepts_criteria_overrides() -> None:
    """화면 제약 칸 — 비운 칸은 None(서버 기본값), 채운 칸만 값이 실린다(계획서 20)."""
    from B05_Profile.B05_Profile_Router_Replan import RouteInitialRequest, RouteReplanRequest

    request = RouteInitialRequest.model_validate(
        {"mode": "earthwork", "criteria": {"stream_offset_m": 30, "grade_smooth_m": 2}}
    )
    assert request.criteria is not None
    assert request.criteria.stream_offset_m == 30
    assert request.criteria.grade_smooth_m == 2
    assert request.criteria.max_grade_pct is None
    assert request.criteria.min_radius_m is None
    replan = RouteReplanRequest.model_validate(
        {"vertices": [], "initial": {"mode": "weighted", "criteria": {"max_grade_pct": 12}}}
    )
    assert replan.initial is not None and replan.initial.criteria is not None
    assert replan.initial.criteria.max_grade_pct == 12