Aislo/resources/tester/test_b05_extra_stations_supplier.py

"""구조물(비정규) 측점의 단일 공급자 — 2026-08-28 [초기화] 후 측점 실종 건.

배경: 측점 정본은 파일(longitudinal.json stations + cross_*.json)인데, 그 파일을 쓰는
`run_section_generation`에는 공급자가 없었다. 유일한 공급자가 프론트 [임시저장]이라
초기화·반폭 재생성이 파일을 다시 쓸 때마다 구조물 측점이 통째로 사라졌다(실측: 초기화
직후 cross 23 → 19, irregular 4 → 0).
"""

import json

import numpy as np
import pytest

from B05_Profile.B05_Profile_Engine_Sections import (
    _load_pipe_points,
    cross_filename,
    resolve_extra_stations,
    run_section_generation,
)
from B05_Profile.B05_Profile_Engine_Sections_Core import (
    SectionGenerationOptions,
    generate_sections,
)
from common_util.common_util_drainage_pipes import parse_pipe_points, route_signature
from common_util.common_util_route_geometry import RouteVertex

# 실측 정본(프로젝트 c1bb453f) 그대로 — 라벨 문자열이 종단 정본과 글자까지 같아야 한다.
LIVE_POINTS = [
    {
        "chainage_m": 84.3,
        "source": "spacing",
        "options": {"pipe_kind": "파형강관", "pipe_diameter_mm": 1000},
    },
    {"chainage_m": 149.73, "source": "stream", "facility": "ford_bridge"},
    {"chainage_m": 200.92, "source": "spacing", "facility": "box_culvert"},
    {"chainage_m": 275.71, "source": "spacing", "options": {"pipe_diameter_mm": 800}},
]
LIVE_LABELS = (
    (84.3, "파형강관 D1000"),
    (149.73, "세월교"),
    (200.92, "BOX암거"),
    (275.71, "파형강관 D800"),
)

STRAIGHT_LINE = [[0.0, 0.0, 100.0], [350.0, 0.0, 100.0]]


class _FlatSampler:
    """표고 고정 스텁 — 측점 구성만 보는 테스트라 지형은 상수로 둔다."""

    def sample_xy(self, xy):
        count = len(np.asarray(xy, dtype=np.float64))
        return np.full(count, 100.0), np.ones(count, dtype=bool)


def _write_pipe_points(root, points, polyline=STRAIGHT_LINE, signature=None):
    edits = root / "B04_PreProcess" / "drainage" / "edits"
    edits.mkdir(parents=True, exist_ok=True)
    vertices = [RouteVertex(x=p[0], y=p[1], z=p[2], chainage_m=0.0) for p in polyline]
    (edits / "pipe_points.json").write_text(
        json.dumps(
            {"route_signature": signature or route_signature(vertices), "points": points},
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )


def _write_structures(root, structures):
    route_dir = root / "B05_Profile" / "route"
    route_dir.mkdir(parents=True, exist_ok=True)
    (route_dir / "structures.json").write_text(
        json.dumps({"revision": 1, "structures": structures}, ensure_ascii=False),
        encoding="utf-8",
    )


# ── 라벨 파생 ──────────────────────────────────────────────────────────────


def test_pipe_points_become_labelled_stations(tmp_path):
    assert resolve_extra_stations(tmp_path, parse_pipe_points(LIVE_POINTS)) == LIVE_LABELS


def test_pipe_without_options_falls_back_to_config_defaults(tmp_path):
    extras = resolve_extra_stations(tmp_path, parse_pipe_points([{"chainage_m": 30.0}]))
    assert extras == ((30.0, "파형강관 D800"),)


def test_ford_pavement_is_not_an_irregular_station(tmp_path):
    """물넘이포장은 비정규 측점이 아니다 — 종단 그래프 구조물 표시 몫(2026-08-28 사용자 확정)."""
    points = [{"chainage_m": 40.0, "facility": "ford_pavement"}, {"chainage_m": 60.0}]
    extras = resolve_extra_stations(tmp_path, parse_pipe_points(points))
    assert [chainage for chainage, _label in extras] == [60.0]


# ── 구조물 정본 A군 합류 ───────────────────────────────────────────────────


def test_group_a_point_and_revetment_facility_join(tmp_path):
    """A군 점형은 기준 한 곳, 기슭막이(관 시설)는 시작·기준·종료 세 곳(2026-08-28).

    기슭막이는 구조물 정본에서 관 지점 정본으로 옮겨갔다(`managed_by: pipe_points`) —
    구간형 구조물이 아니라 관 시설 목록에서 온다.
    """
    _write_structures(
        tmp_path,
        [{"type_id": "cross_drain_exposed", "chainage_m": 60.0, "placement": "point"}],
    )
    points = [
        *LIVE_POINTS,
        {
            "chainage_m": 100.0,
            "source": "user",
            "facility": "revetment",
            "start_m": 95.0,
            "end_m": 115.0,
        },
    ]
    extras = resolve_extra_stations(tmp_path, parse_pipe_points(points))
    assert (60.0, "노출형 횡단수로") in extras
    assert [chainage for chainage, label in extras if label == "기슭막이"] == [95.0, 100.0, 115.0]
    assert [chainage for chainage, _label in extras] == [
        60.0,
        84.3,
        95.0,
        100.0,
        115.0,
        149.73,
        200.92,
        275.71,
    ]


def test_other_groups_do_not_plant_stations(tmp_path):
    """C·E·G군은 측점을 만들지 않는다 — 지금 켠 것은 A군과 D군 구간형뿐이다."""
    _write_structures(
        tmp_path,
        [
            {
                "type_id": "retaining_wall",
                "chainage_m": 40.0,
                "start_m": 40.0,
                "end_m": 60.0,
                "placement": "interval",
            },
            {
                "type_id": "pavement_concrete",
                "chainage_m": 300.0,
                "start_m": 295.0,
                "end_m": 305.0,
                "placement": "interval",
            },
        ],
    )
    extras = resolve_extra_stations(tmp_path, parse_pipe_points(LIVE_POINTS))
    assert [chainage for chainage, _label in extras] == [84.3, 149.73, 200.92, 275.71]


def test_broken_structures_file_keeps_pipe_derived_stations(tmp_path):
    route_dir = tmp_path / "B05_Profile" / "route"
    route_dir.mkdir(parents=True, exist_ok=True)
    (route_dir / "structures.json").write_text("{not json", encoding="utf-8")
    assert resolve_extra_stations(tmp_path, parse_pipe_points(LIVE_POINTS)) == LIVE_LABELS


# ── 관 정본 로드·지문 게이트 ───────────────────────────────────────────────


def test_missing_pipe_points_file_is_empty_not_error(tmp_path):
    assert _load_pipe_points(tmp_path, STRAIGHT_LINE) == []


def test_route_signature_mismatch_drops_points_with_warning(tmp_path, caplog):
    """지문이 다르고 좌표도 없는 구 저장분만 버린다 — 침묵하지 않고 경고를 남긴다.

    좌표가 있으면 그 선에 투영해 이월한다(2026-08-30). LIVE_POINTS에는 좌표가 없다.
    """
    _write_pipe_points(tmp_path, LIVE_POINTS, signature="옛노선-0000")
    with caplog.at_level("WARNING"):
        assert _load_pipe_points(tmp_path, STRAIGHT_LINE) == []
    assert any("좌표 없는 구 저장분" in record.getMessage() for record in caplog.records)


def test_matching_signature_loads_points(tmp_path):
    _write_pipe_points(tmp_path, LIVE_POINTS)
    assert [pipe.chainage_m for pipe in _load_pipe_points(tmp_path, STRAIGHT_LINE)] == [
        84.3,
        149.73,
        200.92,
        275.71,
    ]


# ── 횡단 파일명 충돌 스냅(격자 우선 불변식) ────────────────────────────────


def _stations(extra_stations):
    result = generate_sections(
        STRAIGHT_LINE,
        _FlatSampler(),
        SectionGenerationOptions(station_interval_m=20.0, extra_stations=extra_stations),
    )
    return result["longitudinal"]["stations"]


def test_extra_station_within_one_meter_snaps_to_grid():
    stations = _stations(((100.4, "X"),))
    assert 100.4 not in [station["chainage_m"] for station in stations]
    labelled = [station for station in stations if station.get("structure") == "X"]
    assert [station["chainage_m"] for station in labelled] == [100.0]
    assert labelled[0]["kind"] == "regular"  # 격자 측점 우선 — 기존 불변식


def test_two_extras_in_same_meter_bucket_keep_one():
    stations = _stations(((80.2, "A"), (80.4, "B")))
    assert [station["chainage_m"] for station in stations if station.get("structure")] == [80.0]


def test_live_extras_stay_irregular_and_filenames_stay_unique():
    stations = _stations(LIVE_LABELS)
    irregular = [station for station in stations if station["kind"] == "irregular"]
    assert [station["chainage_m"] for station in irregular] == [84.3, 149.73, 200.92, 275.71]
    assert [station["structure"] for station in irregular] == [
        label for _chainage, label in LIVE_LABELS
    ]
    assert len({cross_filename(station["chainage_m"]) for station in stations}) == len(stations)


# ── 단일 공급자 계약 (run_section_generation) ──────────────────────────────


@pytest.fixture
def project(tmp_path, monkeypatch):
    route_dir = tmp_path / "B05_Profile" / "route"
    route_dir.mkdir(parents=True, exist_ok=True)
    (route_dir / "route_main.geojson").write_text(
        json.dumps({"geometry": {"type": "LineString", "coordinates": STRAIGHT_LINE}}),
        encoding="utf-8",
    )
    monkeypatch.setattr(
        "B05_Profile.B05_Profile_Engine_Sections.build_surface_sampler",
        lambda *args, **kwargs: _FlatSampler(),
    )
    return tmp_path


def _generate(project_root, options=None):
    return run_section_generation(
        project_root,
        "B05_Profile/route/route_main.geojson",
        "csf",
        "dtm",
        True,
        options=options or SectionGenerationOptions(station_interval_m=20.0),
    )


def _saved_stations(project_root):
    path = project_root / "B06_Section" / "longitudinal" / "longitudinal.json"
    return json.loads(path.read_text(encoding="utf-8"))["stations"]


def test_run_section_generation_derives_stations_from_truth_files(project):
    _write_pipe_points(project, LIVE_POINTS)
    result = _generate(project)

    stations = _saved_stations(project)
    irregular = [station for station in stations if station["kind"] == "irregular"]
    assert [station["chainage_m"] for station in irregular] == [84.3, 149.73, 200.92, 275.71]
    assert [station["structure"] for station in irregular] == [
        label for _chainage, label in LIVE_LABELS
    ]

    cross_dir = project / "B06_Section" / "cross_sections"
    for name in ("cross_00084m", "cross_00150m", "cross_00201m", "cross_00276m"):
        assert (cross_dir / f"{name}.json").is_file()
    # 파일 = DB 행 — 종전에는 파일 23 vs DB 19로 갈렸다.
    assert len(result["cross_sections"]) == len(stations)


def test_caller_supplied_extra_stations_are_ignored(project):
    """호출자 값을 되먹이면 B04에서 지운 관이 스냅샷을 타고 부활한다 — 정본 파일만 쓴다."""
    _write_pipe_points(project, LIVE_POINTS)
    _generate(
        project,
        SectionGenerationOptions(station_interval_m=20.0, extra_stations=((10.0, "가짜"),)),
    )
    stations = _saved_stations(project)
    assert all(station.get("structure") != "가짜" for station in stations)
    assert [station["chainage_m"] for station in stations if station["kind"] == "irregular"] == [
        84.3,
        149.73,
        200.92,
        275.71,
    ]


def test_without_pipe_points_only_grid_stations(project):
    _generate(project)
    assert all(station["kind"] != "irregular" for station in _saved_stations(project))