157 lines
5.9 KiB
Python
157 lines
5.9 KiB
Python
"""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,
|
|
resolve_cross_file,
|
|
)
|
|
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 = [row for row in db_snapshot(root)["cross_sections"] if row["chainage_m"] is not None]
|
|
return sorted(rows, key=lambda row: (float(row["chainage_m"]), row.get("sequence_num") or 0))
|
|
|
|
|
|
def _row_data(row: dict[str, Any]) -> dict[str, Any]:
|
|
data = row.get("data")
|
|
return (json.loads(data) if isinstance(data, str) else data) or {}
|
|
|
|
|
|
def stations(root: Path) -> list[dict[str, Any]]:
|
|
"""측점 목록(누가 차례) — chainage_m · label · 중심 좌표 · 방위각 · 오르막 쪽 · 지반 유형.
|
|
|
|
횡단 파일을 못 찾은 측점(`missing_stations`)은 「횡단 없음」으로 빠짐 — 전체는 안 깨짐."""
|
|
out = []
|
|
for row in _cross_rows(root):
|
|
path = resolve_cross_file(root, row)
|
|
if path is None:
|
|
continue
|
|
cross = _json(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": (_row_data(row).get("design") or {}).get("ground_type"),
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def missing_stations(root: Path) -> list[float]:
|
|
"""횡단 없음으로 빠진 측점 누가거리 — 화면 알림용(개수 = len)."""
|
|
return [
|
|
float(row["chainage_m"])
|
|
for row in _cross_rows(root)
|
|
if resolve_cross_file(root, row) is None
|
|
]
|
|
|
|
|
|
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(암 경계 = 지반선을 그만큼 내린 선 · 암 측점만) …).
|
|
측점이 없거나 횡단 없음이면 KeyError."""
|
|
for row in _cross_rows(root):
|
|
if abs(float(row["chainage_m"]) - float(chainage_m)) < 1e-6:
|
|
path = resolve_cross_file(root, row)
|
|
if path is None:
|
|
raise KeyError(f"횡단 없음: {chainage_m}")
|
|
cross = _json(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": _row_data(row).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
|