Aislo/E01_StabRegister/E01_StabRegister_Source.py

"""E 안정성평가 복사본 읽기 — storage 를 아는 유일한 자리 (요구 345 · 532-1).

복사본 = E 프로젝트 저장소 `E01_StabRegister/source/`(`common_util_project_copy.copy_stability`
이 생성 때 한 번 만듦 · 그 뒤 불변). E 엔진 · 화면은 이 함수들로만 받음 — 원본 임도 프로젝트를
읽지 않음.
"""

from __future__ import annotations

import json
from functools import lru_cache
from pathlib import Path
from typing import Any

import aiomysql

from common_util.common_util_project_copy import (
    MANIFEST_NAME,
    STABILITY_DB_NAME,
    STABILITY_SOURCE_DIR,
)
from common_util.common_util_storage import resolve_stored_project_path
from config.config_db import get_db_pool


def source_dir(project_root: Path | str) -> Path:
    """E 프로젝트 루트 → 복사본 폴더."""
    return Path(project_root) / STABILITY_SOURCE_DIR


async def project_source_dir(project_id: str) -> Path:
    """E project_id → 복사본 폴더(없으면 LookupError)."""
    async with (
        get_db_pool().acquire() as connection,
        connection.cursor(aiomysql.DictCursor) as cursor,
    ):
        await cursor.execute(
            "SELECT storage_path FROM projects"
            " WHERE id = %s AND project_kind = 'STABILITY' AND deleted_at IS NULL",
            (project_id,),
        )
        row = await cursor.fetchone()
    if not row:
        raise LookupError(f"안정성평가 프로젝트가 아님: {project_id}")
    root = source_dir(resolve_stored_project_path(row["storage_path"]))
    if not (root / MANIFEST_NAME).is_file():
        raise LookupError(f"복사본이 없음: {project_id}")
    return root


@lru_cache(maxsize=8)
def _read_json(path: str, mtime_ns: int) -> Any:
    return json.loads(Path(path).read_text(encoding="utf-8"))


def _json(path: Path) -> Any:
    return _read_json(str(path), path.stat().st_mtime_ns)


def manifest(root: Path) -> dict[str, Any]:
    """복사 목록 — 원본 id · 복사 시각 · 지문 · 파일."""
    return _json(root / MANIFEST_NAME)


def db_snapshot(root: Path) -> dict[str, Any]:
    """동결 DB 행 — project · project_workflow_stages · surface_params · route ·
    route_points · route_statistics · longitudinal_sections · cross_sections."""
    return _json(root / STABILITY_DB_NAME)


def _cross_rows(root: Path) -> list[dict[str, Any]]:
    rows = db_snapshot(root)["cross_sections"]
    return sorted(rows, key=lambda row: (float(row["chainage_m"] or 0.0), row["sequence_num"] or 0))


def stations(root: Path) -> list[dict[str, Any]]:
    """측점 목록(누가 차례) — chainage_m · label · 중심 좌표 · 방위각 · 오르막 쪽 · 지반 유형."""
    out = []
    for row in _cross_rows(root):
        data = json.loads(row["data"]) if isinstance(row["data"], str) else row["data"]
        cross = _json(root / row["cross_section_file_path"])
        out.append(
            {
                "chainage_m": float(row["chainage_m"]),
                "label": cross.get("label"),
                "kind": cross.get("kind"),
                "center_x": cross.get("center_x"),
                "center_y": cross.get("center_y"),
                "center_z": cross.get("center_z"),
                "azimuth_deg": cross.get("azimuth_deg"),
                "uphill_side": cross.get("uphill_side"),
                "ground_type": (data.get("design") or {}).get("ground_type"),
            }
        )
    return out


def cross_section(root: Path, chainage_m: float) -> dict[str, Any]:
    """한 측점 횡단 — 지반선(samples: offset_m · x · y · z · valid) · 좌표틀 · B06 설계값
    (`design`: design_line · road_edges · section_mode · 절성토 경사 · 소단 · ground_type ·
    rock_boundary_offset_m(암 경계 = 지반선을 그만큼 내린 선 · 암 측점만) …)."""
    for row in _cross_rows(root):
        if abs(float(row["chainage_m"]) - float(chainage_m)) < 1e-6:
            data = json.loads(row["data"]) if isinstance(row["data"], str) else row["data"]
            cross = _json(root / row["cross_section_file_path"])
            return {
                "chainage_m": float(row["chainage_m"]),
                "label": cross.get("label"),
                "frame": cross.get("frame"),
                "uphill_side": cross.get("uphill_side"),
                "samples": cross.get("samples") or [],
                "design": data.get("design") or {},
            }
    raise KeyError(f"측점 없음: {chainage_m}")


def terrain_grid_path(root: Path) -> Path:
    """복사한 지형 격자(DTM npz — 원본 설계가 쓴 그 하나)."""
    found = sorted((root / "B04_PreProcess" / "models").glob("dtm_*.npz"))
    if len(found) != 1:
        raise FileNotFoundError(f"복사한 지형 격자가 하나가 아님: {[p.name for p in found]}")
    return found[0]


def corridor_path(root: Path) -> Path | None:
    """최신 노선 3D 노선면(없으면 None)."""
    found = sorted((root / "B05_Profile" / "corridor").glob("corridor_*.json"))
    return found[0] if found else None


def structures_path(root: Path) -> Path | None:
    """노선 구조물 정본(벽 등 · 없으면 None)."""
    path = root / "B05_Profile" / "route" / "structures.json"
    return path if path.is_file() else None