Aislo/B08_Quantity/B08_Quantity_Hydro_Site.py

"""B08 4탭 홍수량 계산서 — 공사지 지역 · 확률강우량 · 저장 파일 (PLAN 69-5 · 69-6).

읽기만: 전처리 강우량표(`B04_PreProcess/drainage/rainfall_table.json` · 관측소 확률강우량)
· 행정구역(`B04_PreProcess/processed/행정구역_*_bounds.geojson` · WGS84) · 프로젝트 등록 지역(DB).
배수유역 계산 코드는 부르지 않음 — 이미 저장된 파일만 읽음.

쓰기: B08 자기 파일 `B08_Quantity/quantities/hydro_inputs.json` 하나([저장] 때만).
"""

from __future__ import annotations

import json
import os
from pathlib import Path
from typing import Any

from shapely.geometry import Point, shape

RAINFALL_FILE = Path("B04_PreProcess") / "drainage" / "rainfall_table.json"
PROCESSED_DIR = Path("B04_PreProcess") / "processed"
EMD_FILE = "행정구역_읍면동_bounds.geojson"
SGG_FILE = "행정구역_시군구_bounds.geojson"
SAVED_FILE = Path("B08_Quantity") / "quantities" / "hydro_inputs.json"

#: 강우량표가 없을 때 재현기간 드롭다운 — 확률강우량 자료 표의 재현기간
FALLBACK_PERIODS = (2, 3, 5, 10, 20, 30, 50, 70, 80, 100, 200, 300, 500)
DAY_MINUTES = 1440


def _read_json(path: Path) -> dict[str, Any] | None:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    return data if isinstance(data, dict) else None


def _features(path: Path) -> list[dict[str, Any]]:
    data = _read_json(path) or {}
    return [f for f in data.get("features") or [] if isinstance(f, dict)]


def _feature_at(path: Path, lon: float, lat: float) -> dict[str, Any] | None:
    """점이 드는 행정구역 속성 — 경계 파일은 WGS84."""
    point = Point(lon, lat)
    for feature in _features(path):
        try:
            if shape(feature["geometry"]).covers(point):
                return feature.get("properties") or {}
        except (KeyError, TypeError, ValueError):
            continue
    return None


def _region(root: Path, rain: dict[str, Any] | None, db_region: str | None) -> dict[str, str]:
    """공사지 지역 — 전처리 행정구역(노선 대표점이 드는 곳) > 시군구 하나뿐 > 프로젝트 등록 지역."""
    lat, lon = (rain or {}).get("lat"), (rain or {}).get("lon")
    if isinstance(lat, (int, float)) and isinstance(lon, (int, float)):
        emd = _feature_at(root / PROCESSED_DIR / EMD_FILE, lon, lat)
        sgg = _feature_at(root / PROCESSED_DIR / SGG_FILE, lon, lat)
        name = (sgg or {}).get("sig_kor_nm") or ""
        full = (emd or {}).get("full_nm") or (sgg or {}).get("full_nm") or ""
        if name or full:
            return {"name": name or full.split()[-1], "full": full, "source": "전처리 행정구역"}
    sggs = _features(root / PROCESSED_DIR / SGG_FILE)
    if len(sggs) == 1:
        props = sggs[0].get("properties") or {}
        return {
            "name": props.get("sig_kor_nm") or "",
            "full": props.get("full_nm") or "",
            "source": "전처리 행정구역",
        }
    if db_region:
        return {"name": db_region, "full": db_region, "source": "프로젝트 등록"}
    return {"name": "", "full": "", "source": ""}


def _r24_table(rain: dict[str, Any] | None) -> dict[int, float]:
    """재현기간(년) → 24시간 확률강우량(mm)."""
    table: dict[int, float] = {}
    for row in (rain or {}).get("values") or []:
        try:
            if int(row["duration_min"]) == DAY_MINUTES:
                table[int(row["return_period_yr"])] = float(row["depth_mm"])
        except (KeyError, TypeError, ValueError):
            continue
    return table


def site_info(root: Path, db_region: str | None) -> dict[str, Any]:
    """화면 머리 — 지역 · 관측소 · 재현기간 목록 · 재현기간별 R24(없으면 빈 표)."""
    rain = _read_json(root / RAINFALL_FILE)
    r24 = _r24_table(rain)
    station = (rain or {}).get("station") or {}
    return {
        "region": _region(root, rain, db_region),
        "station": station.get("name") or "",
        "periods": sorted(r24) or list(FALLBACK_PERIODS),
        "r24": {str(k): v for k, v in sorted(r24.items())},
    }


def site_r24(root: Path, return_period: float) -> float | None:
    """계산 때 서버가 정본으로 쓰는 R24 — 자료가 있으면 그 값 · 없으면 None(화면 입력)."""
    return _r24_table(_read_json(root / RAINFALL_FILE)).get(int(return_period))


def read_saved(root: Path) -> dict[str, dict[str, Any]]:
    data = _read_json(root / SAVED_FILE) or {}
    stations = data.get("stations")
    return stations if isinstance(stations, dict) else {}


def write_saved(root: Path, stations: dict[str, dict[str, Any]]) -> None:
    """원자적 쓰기 — 임시 파일에 쓴 뒤 바꿔 끼움."""
    path = root / SAVED_FILE
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(".part")
    payload = {"version": 1, "stations": stations}
    tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=1), encoding="utf-8")
    os.replace(tmp, path)