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_plants_its_station(tmp_path):
    """물넘이포장도 기준 측점 하나를 심는다 — 그 측점 횡단도에 파인 노면(286-1)."""
    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] == [40.0, 60.0]
    assert extras[0][1] == "물넘이포장"


# ── 구조물 정본 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):
    """E·G군은 측점을 만들지 않는다 — 켠 것은 A군과 C·D군 구간형뿐이다(178-1)."""
    _write_structures(
        tmp_path,
        [
            {
                "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_slope_group_interval_plants_start_anchor_end(tmp_path):
    """C군 구간형(옹벽·흙막이·소단)도 시작·기준·종료 셋을 심는다 — 격자 사이 벽도 횡단에 얹힘(178-1)."""
    _write_structures(
        tmp_path,
        [
            {
                "type_id": "soil_guard",
                "chainage_m": 45.0,
                "start_m": 40.0,
                "end_m": 50.0,
                "placement": "interval",
            },
            {
                "type_id": "retaining_wall",
                "chainage_m": 125.0,
                "start_m": 120.0,
                "end_m": 130.0,
                "placement": "interval",
            },
            {
                "type_id": "berm",
                "chainage_m": 165.0,
                "start_m": 160.0,
                "end_m": 170.0,
                "placement": "interval",
            },
        ],
    )
    extras = resolve_extra_stations(tmp_path, parse_pipe_points(LIVE_POINTS))
    assert [c for c, label in extras if label == "흙막이"] == [40.0, 45.0, 50.0]
    assert [c for c, label in extras if label == "옹벽"] == [120.0, 125.0, 130.0]
    assert [c for c, label in extras if label == "소단"] == [160.0, 165.0, 170.0]


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,
    ]


# ── 소수 측점 횡단 — 관 · 구조물 자리마다 따로(이름 소수 2자리 · PLAN 20장) ──────────────


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_in_grid_meter_keeps_both_with_own_names():
    stations = _stations(((100.4, "X"),))
    chainages = [station["chainage_m"] for station in stations]
    assert 100.4 in chainages and 100.0 in chainages
    labelled = [station for station in stations if station.get("structure") == "X"]
    assert [(station["chainage_m"], station["kind"]) for station in labelled] == [
        (100.4, "irregular")
    ]
    assert cross_filename(100.4) == "cross_00100.40m.json"
    assert len({cross_filename(c) for c in chainages}) == len(chainages)


def test_extra_station_on_grid_within_5mm_labels_grid_station():
    stations = _stations(((100.003, "X"), (0.3, "S")))
    labelled = {station["chainage_m"]: station for station in stations if station.get("structure")}
    assert sorted(labelled) == [0.3, 100.0]
    # 같은 m 면 측점 하나 · 구조물 측점이 위(272) — 격자 자리 표시(`grid`)로 떠나면 되돌림
    assert labelled[100.0]["kind"] == "irregular" and labelled[100.0].get("grid") is True
    assert labelled[0.3]["kind"] == "irregular" and "grid" not in labelled[0.3]


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


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_00084.30m", "cross_00149.73m", "cross_00200.92m", "cross_00275.71m"):
        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))