Aislo/resources/tester/test_route_shapefile_input.py

"""계획노선 shapefile 입력 + PRJ 2개 분리 검증 (2026-08-31).

원청 실물 자료(일월.용화.2.2 — UTM-K 노선 shapefile, 동부원점 지형 PRJ/TFW)를
그대로 써서 판독·좌표계 선택·필수 세트 판정을 확인한다.
"""

from __future__ import annotations

import sys
from pathlib import Path

import pytest

PROJECT_ROOT = Path(__file__).resolve().parents[2]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from B03_FileInput.B03_FileInput_Engine import reconcile_shapefile_members  # noqa: E402
from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata  # noqa: E402
from B03_FileInput.B03_FileInput_Engine_Shapefile import (  # noqa: E402
    read_shapefile_attributes,
    read_shapefile_parts,
    shapefile_crs_input,
    shapefile_encoding,
    shapefile_epsg_label,
)
from B03_FileInput.B03_FileInput_Repository_Temp import (  # noqa: E402
    is_batch_required_complete,
)
from B03_FileInput.B03_FileInput_Router import _missing_required_file_types  # noqa: E402
from common_util.common_util_crs import find_project_prj  # noqa: E402
from common_util.common_util_route_geometry import (  # noqa: E402
    find_planned_route_file,
    read_planned_route,
)

# 업로드 원본을 그대로 둔 시험용 세트 (basename만 route/terrain으로 맞췄다).
FIXTURE_DIR = Path(
    r"C:\Users\umsan\AppData\Local\Temp\claude"
    r"\d--02-Software-Prog---------------------"
    r"\72ca32d1-3b09-48f8-90f8-b4c851bf490e\scratchpad"
)
ROUTE_SHP = FIXTURE_DIR / "shpset" / "route.shp"
TERRAIN_PRJ = FIXTURE_DIR / "terrain.prj"

pytestmark = pytest.mark.skipif(
    not ROUTE_SHP.exists(), reason="원청 실물 shapefile 세트가 없는 환경"
)


def test_shapefile_geometry_matches_source() -> None:
    """실물 노선: PolyLine 1파트·121정점, 2D(z=0)."""
    parts = read_shapefile_parts(ROUTE_SHP)
    assert len(parts) == 1
    assert len(parts[0]) == 121
    assert all(point[2] == 0.0 for point in parts[0])


def test_cpg_949_decodes_korean_attributes() -> None:
    """`.cpg` 내용이 `949` 한 줄이어도 CP949로 읽어 한글 속성이 살아야 한다."""
    assert shapefile_encoding(ROUTE_SHP) == "cp949"
    attributes = read_shapefile_attributes(ROUTE_SHP)
    assert attributes["대상지"] == "일월.용화.2.2"
    assert attributes["길이"] == "2.200"


def test_crs_input_is_wkt_not_epsg_code() -> None:
    """노선 PRJ는 `to_epsg()`가 None이다 — 변환 입력은 원문 WKT로 나와야 한다."""
    crs_input = shapefile_crs_input(ROUTE_SHP)
    assert crs_input is not None
    assert crs_input.startswith("PROJCS[")
    assert "PCS_ITRF2000_UTM_K" in crs_input
    # 라벨은 별도로 붙는다(표시·로그용).
    assert shapefile_epsg_label(ROUTE_SHP) == 5179


def test_route_reprojects_into_terrain_raster_extent() -> None:
    """노선(UTM-K)을 지형 계로 옮기면 TFW 원점 안쪽에 앉아야 한다."""
    from pyproj import Transformer

    from common_util.common_util_crs import crs_input_from_prj

    planned = read_planned_route(ROUTE_SHP)
    assert planned is not None
    target = crs_input_from_prj(TERRAIN_PRJ.read_text(encoding="utf-8"))
    transformer = Transformer.from_crs(planned.crs_input, target, always_xy=True)
    moved = [transformer.transform(vertex.x, vertex.y) for vertex in planned.vertices]

    xs = [point[0] for point in moved]
    ys = [point[1] for point in moved]
    # TFW: 화소 0.2m, 좌상단 원점 (208288.8376, 368092.2820), y는 아래로 진행.
    origin_x, origin_y = 208288.8376409571, 368092.2819918637
    assert min(xs) > origin_x
    assert max(ys) < origin_y
    # 노선 연장은 도형 기준 약 2.1km (속성 길이 2.2km와 같은 자릿수).
    assert 2000.0 < planned.vertices[-1].chainage_m < 2300.0


def test_planned_route_reader_prefers_shapefile(tmp_path: Path) -> None:
    """노선 파일이 CSV·shapefile 둘 다 있으면 shapefile을 고른다."""
    input_dir = tmp_path / "input"
    (input_dir / "shp").mkdir(parents=True)
    (input_dir / "csv").mkdir(parents=True)
    for extension in (".shp", ".shx", ".dbf", ".cpg", ".prj"):
        target = input_dir / "shp" / f"route{extension}"
        target.write_bytes(ROUTE_SHP.with_suffix(extension).read_bytes())
    (input_dir / "csv" / "old.csv").write_text(
        "x,y,z,crs_epsg\n1,2,3,5186\n4,5,6,5186\n", encoding="utf-8"
    )

    found = find_planned_route_file(input_dir)
    assert found is not None and found.suffix == ".shp"
    planned = read_planned_route(found)
    assert planned is not None
    assert len(planned.vertices) == 121
    assert planned.name == "일월.용화.2.2"


def test_terrain_prj_wins_over_route_prj(tmp_path: Path) -> None:
    """PRJ가 둘이면 지형(input/prj/)을 프로젝트 좌표계로 고른다."""
    stage_input = tmp_path / "B03_FileInput" / "input"
    (stage_input / "shp").mkdir(parents=True)
    (stage_input / "prj").mkdir(parents=True)
    # 노선 PRJ 파일명이 정렬상 앞서도(a < z) 지형이 이겨야 한다.
    (stage_input / "shp" / "a_route.prj").write_text(
        ROUTE_SHP.with_suffix(".prj").read_text(encoding="utf-8"), encoding="utf-8"
    )
    (stage_input / "prj" / "z_terrain.prj").write_text(
        TERRAIN_PRJ.read_text(encoding="utf-8"), encoding="utf-8"
    )

    chosen = find_project_prj(tmp_path)
    assert chosen is not None
    assert chosen.name == "z_terrain.prj"
    assert "Modified East Belt" in chosen.read_text(encoding="utf-8")


def test_terrain_prj_pairs_with_terrain_data(tmp_path: Path) -> None:
    """`input/prj/`에 옛 PRJ가 남아 있어도 지금 지형 자료와 짝인 것을 고른다.

    실측 사고(2026-08-31): 옛 `result.prj`(East Belt 2010, FN 600,000)가 이름 정렬에서
    새 `용화.prj`(Modified East Belt, FN 500,000)보다 앞서 뽑혀 노선이 Y로 100,000m
    어긋났고 B05 경로 계산이 400으로 실패했다.
    """
    stage_input = tmp_path / "B03_FileInput" / "input"
    for name in ("prj", "las", "tfw"):
        (stage_input / name).mkdir(parents=True)
    # 옛 잔재 — 이름이 앞서고, 짝이 되는 지형 자료는 이제 없다.
    (stage_input / "prj" / "result.prj").write_text("stale", encoding="utf-8")
    # 이번 지형 — LAS/TFW와 basename이 같다.
    (stage_input / "prj" / "용화.prj").write_text(
        TERRAIN_PRJ.read_text(encoding="utf-8"), encoding="utf-8"
    )
    (stage_input / "las" / "용화.las").write_bytes(b"LASF")
    (stage_input / "tfw" / "용화.tfw").write_text("0.2 0 0 -0.2 1 2", encoding="utf-8")

    chosen = find_project_prj(tmp_path)
    assert chosen is not None
    assert chosen.name == "용화.prj"
    assert "Modified East Belt" in chosen.read_text(encoding="utf-8")


def test_reconcile_pulls_prj_that_arrived_first(tmp_path: Path) -> None:
    """PRJ가 SHP보다 먼저 도착해 input/prj/에 앉아도 세트 폴더로 끌어온다."""
    stage_input = tmp_path / "input"
    (stage_input / "shp").mkdir(parents=True)
    (stage_input / "prj").mkdir(parents=True)
    (stage_input / "shp" / "route.shp").write_bytes(ROUTE_SHP.read_bytes())
    (stage_input / "prj" / "route.prj").write_text("dummy", encoding="utf-8")
    (stage_input / "prj" / "terrain.prj").write_text("keep", encoding="utf-8")

    moved = reconcile_shapefile_members(tmp_path)

    assert [path.name for path in moved] == ["route.prj"]
    assert (stage_input / "shp" / "route.prj").exists()
    assert not (stage_input / "prj" / "route.prj").exists()
    assert (stage_input / "prj" / "terrain.prj").exists()  # 지형 PRJ는 그대로


def test_analyze_dispatches_shapefile() -> None:
    """업로드 분석기가 .shp를 계획노선으로 인식한다."""
    metadata = analyze_input_metadata(ROUTE_SHP)
    assert metadata["purpose"] == "planned_route"
    assert metadata["shape_type"] == "PolyLine"
    assert metadata["point_count"] == 121
    assert metadata["epsg"] == 5179
    assert metadata["missing_members"] == []


def test_required_set_accepts_shapefile_instead_of_csv() -> None:
    """필수 세트: 노선은 csv|shp 택일, shp면 형제 파일까지 본다."""
    complete = {"shp", "shx", "dbf", "route_prj", "prj", "tfw", "las"}
    assert _missing_required_file_types(complete) == []

    # 형제 파일이 빠지면 막는다 — 노선 PRJ(route_prj)도 형제다.
    assert _missing_required_file_types({"shp", "prj", "tfw", "las"}) == [
        "dbf",
        "route_prj",
        "shx",
    ]

    # 노선 PRJ만 있고 지형 PRJ가 없으면 프로젝트 좌표계를 못 정한다.
    assert "prj" in _missing_required_file_types({"shp", "shx", "dbf", "route_prj", "tfw", "las"})

    # 보관함 묶음은 확장자만 알아 노선/지형 PRJ를 못 가린다 — shx·dbf까지만 본다.
    assert is_batch_required_complete({"shp", "shx", "dbf", "prj", "tfw", "las"})
    assert not is_batch_required_complete({"shp", "prj", "tfw", "las"})

    # 노선이 아예 없으면 막는다.
    assert "csv/shp" in _missing_required_file_types({"prj", "tfw", "las"})

    # 기존 CSV 흐름은 그대로 통과한다(회귀).
    assert _missing_required_file_types({"csv", "prj", "tfw", "laz"}) == []
    assert is_batch_required_complete({"csv", "prj", "tfw", "laz"})