Aislo/resources/tester/test_common_util_crs.py

# -*- coding: utf-8 -*-
"""PRJ 좌표계 판별 검증 — common_util_crs + 수리된 get_epsg_from_prj.

실행: ./venv/Scripts/python.exe -m pytest resources/tester/test_common_util_crs.py -q
"""

from __future__ import annotations

import sys
from pathlib import Path

import pytest
from pyproj import CRS, Transformer

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))

from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_prj_metadata  # noqa: E402
from B04_PreProcess.B04_PreProcess_Engine_VWorld import get_epsg_from_prj  # noqa: E402
from common_util.common_util_crs import (  # noqa: E402
    crs_input_from_prj,
    identify_epsg,
)

UPLOADS = Path(r"C:/Users/umsan/.claude/uploads/ab448e1d-3fa5-4f88-8e02-b26428c8dfc7")
LAS_PRJ = UPLOADS / "a450e281-__.prj"          # Korean 1985 Modified East Belt → 5176
ROUTE_PRJ = UPLOADS / "105c7734-______.__.2.2.prj"  # UTM-K, AUTHORITY 없음 → 5179
EXISTING_COMPD_PRJ = (
    ROOT / "storage/1/3/2f940d8a-2065-4cf6-8bf8-dc3f0af84e57/B03_FileInput/input/prj/result.prj"
)

needs_uploads = pytest.mark.skipif(not LAS_PRJ.exists(), reason="업로드 PRJ 없음")
needs_storage = pytest.mark.skipif(not EXISTING_COMPD_PRJ.exists(), reason="저장소 PRJ 없음")


# ── 사다리 라벨 판별 ──────────────────────────────────────────────────────────
@needs_uploads
def test_uploaded_las_prj_labels_5176():
    text = LAS_PRJ.read_text(encoding="utf-8")
    assert identify_epsg(CRS.from_wkt(text), text) == 5176


@needs_uploads
def test_uploaded_route_prj_labels_5179():
    text = ROUTE_PRJ.read_text(encoding="utf-8")
    # AUTHORITY 태그가 아예 없다 — 파라미터 지문(③)으로만 잡힌다.
    assert 'AUTHORITY["EPSG"' not in text
    assert identify_epsg(CRS.from_wkt(text), text) == 5179


def test_standard_codes_round_trip():
    for code in (5185, 5186, 5187, 5188, 5179, 5174, 5176, 32652, 4326):
        wkt = CRS.from_epsg(code).to_wkt("WKT1_GDAL")
        assert identify_epsg(CRS.from_wkt(wkt), wkt) == code, code


@needs_storage
def test_compound_kngeoid_prj_labels_5187():
    text = EXISTING_COMPD_PRJ.read_text(encoding="utf-8")
    # KNGeoid24 결합(COMPD_CS) — 수평 성분으로 판별한다. 사설 수직 코드는
    # B03 _prepare_prj_wkt가 떼므로 여기서는 pyproj가 그대로 읽는지만 대비한다.
    from B03_FileInput.B03_FileInput_Engine_Analyze import _prepare_prj_wkt

    parse_text, _ = _prepare_prj_wkt(text)
    assert identify_epsg(CRS.from_wkt(parse_text), parse_text) == 5187


# ── 회귀: 죽은 문자열 분기 ────────────────────────────────────────────────────
def test_central_belt_no_longer_misread_as_east():
    """예전 코드는 false_easting의 EAST 때문에 모든 PRJ를 5187로 판정했다."""
    central = CRS.from_epsg(5186).to_wkt("WKT1_GDAL")
    assert get_epsg_from_prj(central) == "EPSG:5186"
    west = CRS.from_epsg(5185).to_wkt("WKT1_GDAL")
    assert get_epsg_from_prj(west) == "EPSG:5185"
    utm = CRS.from_epsg(32652).to_wkt("WKT1_GDAL")
    assert get_epsg_from_prj(utm) == "EPSG:32652"


def test_empty_prj_keeps_legacy_default():
    assert get_epsg_from_prj("") == "EPSG:5186"
    assert get_epsg_from_prj("이것은 WKT가 아니다") == "EPSG:5186"


# ── 변환 정본 = WKT (TOWGS84 보존) ───────────────────────────────────────────
@needs_uploads
def test_las_prj_returns_wkt_and_transforms_into_korea():
    text = LAS_PRJ.read_text(encoding="utf-8")
    crs_input = get_epsg_from_prj(text)
    # 비표준 TOWGS84라 pyproj DB 확정이 안 된다 → 원문 WKT 그대로 반환해야 한다.
    assert crs_input.lstrip().startswith("PROJCS"), crs_input[:40]
    lon, lat = Transformer.from_crs(crs_input, "EPSG:4326", always_xy=True).transform(
        209014.2, 367805.3
    )
    assert 128.5 < lon < 129.7 and 36.3 < lat < 37.3, (lon, lat)  # 울진 일대


@needs_uploads
def test_route_prj_transform_matches_utmk():
    text = ROUTE_PRJ.read_text(encoding="utf-8")
    crs_input = get_epsg_from_prj(text)
    lon, lat = Transformer.from_crs(crs_input, "EPSG:4326", always_xy=True).transform(
        1142863.4, 1869357.4
    )
    # 정식 EPSG:5179와 십cm 수준에서 같아야 한다 (ITRF2000≈GRS80 무보정)
    lon2, lat2 = Transformer.from_crs("EPSG:5179", "EPSG:4326", always_xy=True).transform(
        1142863.4, 1869357.4
    )
    assert abs(lon - lon2) < 1e-5 and abs(lat - lat2) < 1e-5


def test_wkt_string_accepted_by_transformer():
    """반환값이 EPSG든 WKT든 Transformer.from_crs 입력으로 동작해야 한다."""
    wkt = CRS.from_epsg(5187).to_wkt("WKT1_GDAL")
    x, y = Transformer.from_crs(wkt, "EPSG:4326", always_xy=True).transform(208457.1, 467857.4)
    assert 128.0 < x < 130.0 and 36.0 < y < 38.0


# ── B03 메타데이터 라벨 보강 ─────────────────────────────────────────────────
@needs_uploads
def test_analyze_prj_metadata_fills_epsg_labels():
    las_meta = analyze_prj_metadata(LAS_PRJ)
    assert las_meta["is_valid"] and las_meta["epsg"] == 5176, las_meta["epsg"]
    route_meta = analyze_prj_metadata(ROUTE_PRJ)
    assert route_meta["is_valid"] and route_meta["epsg"] == 5179, route_meta["epsg"]


@needs_storage
def test_analyze_existing_compound_prj_still_5187():
    meta = analyze_prj_metadata(EXISTING_COMPD_PRJ)
    assert meta["epsg"] == 5187 and meta["crs_status"] in ("identified", "custom_vertical_crs")


# ── 광역 스윕: 한국 전체 + 해외, WKT 3형식, 익명화 악조건 ────────────────────
KOREAN_CODES = (5173, 5174, 5175, 5176, 5177, 5178, 5179, 5185, 5186, 5187, 5188, 32651, 32652, 4326)
FOREIGN_CODES = (3857, 32610, 25832, 2154, 27700, 26910, 2193, 6669, 6677, 3095, 28355, 31370)
WKT_FORMATS = ("WKT1_GDAL", "WKT2_2019", "WKT1_ESRI")


@pytest.mark.parametrize("fmt", WKT_FORMATS)
def test_sweep_korean_and_foreign_codes(fmt):
    """한국 14종 + 해외 12종을 세 가지 WKT 방언으로 직렬화해 전부 재판별한다."""
    for code in KOREAN_CODES + FOREIGN_CODES:
        wkt = CRS.from_epsg(code).to_wkt(fmt)
        assert identify_epsg(CRS.from_wkt(wkt), wkt) == code, (code, fmt)
        assert crs_input_from_prj(wkt) == f"EPSG:{code}", (code, fmt)


def _anonymize(wkt: str) -> str:
    """AUTHORITY 태그 제거 + 좌표계 이름 익명화 — 원청 노선 PRJ와 같은 악조건."""
    import re

    wkt = re.sub(r",?AUTHORITY\[[^\]]*\]", "", wkt)
    return re.sub(r'PROJCS\["[^"]+"', 'PROJCS["unknown"', wkt)


def test_sweep_anonymized_wkt_still_identified():
    """AUTHORITY 없고 이름도 지운 ESRI WKT — 파라미터만으로 판별돼야 한다."""
    for code in (5185, 5186, 5187, 5188, 5179, 5174, 5176, 32652, 32610, 2154, 26910):
        wkt = _anonymize(CRS.from_epsg(code).to_wkt("WKT1_ESRI"))
        got = identify_epsg(CRS.from_wkt(wkt), wkt)
        assert got == code, (code, got)


def test_anonymized_never_mislabels_as_other_code():
    """익명화 WKT가 엉뚱한 코드로 오판되면 좌표가 통째로 어긋난다 — 정답 아니면 None만 허용."""
    for code in FOREIGN_CODES:
        wkt = _anonymize(CRS.from_epsg(code).to_wkt("WKT1_ESRI"))
        got = identify_epsg(CRS.from_wkt(wkt), wkt)
        assert got in (code, None), (code, got)


@needs_storage
def test_existing_compound_prj_keeps_legacy_epsg_string():
    """기존 프로젝트(수평 5187 + KNGeoid24 수직)는 예전처럼 "EPSG:5187"이 나와야 한다.

    수직 성분의 bound(지오이드)는 2D 변환에 무관하므로 WKT 폴백 사유가 아니다.
    """
    text = EXISTING_COMPD_PRJ.read_text(encoding="utf-8")
    assert get_epsg_from_prj(text) == "EPSG:5187"