Aislo/E03_StabSection/E03_StabSection_Router_Prob.py

"""E03 확률 해석 · 민감도 API (요구 351 · PLAN 537-1) — 대표 단면만.

- `GET /api/projects/{id}/stab/prob/setup/{station_key}?side=&condition=` — 그 단면 지층 물성(E02 값 = 평균) ·
  기본값 후보(`prob_defaults.json` · 근거 · 사용자 확정 전) · 지하수 갈래 · 기준안전율 · 마지막 결과
- `POST …/stab/prob/{station_key}` — 분포 · 표본 수 · 갈래 → 뒤에서(진행 `GET …/stab/prob/progress`) ·
  같은 지문 = 저장본(fresh)
- `GET …/stab/prob` — 지금 2D 결과에 붙은 기록 목록(측점 · 비탈 · 조건마다 가장 최근 · E05 「확률해석」 시트)

평균 = E02 지반 조건(작업본) 값 · 바꾸려면 E02 에서. 임계면 = 지금 2D 결과의 그 방법 최소 면.
결과 = 파생 저장본 `_derived/stab2d_prob/<지문>` · 지문 = 원본 측점 지문 + 요청 + 코드.
엔진 = `Engine_Prob` — 임계면 고정 = 서버 스레드(빠름) · 매회 탐색 = 일꾼 프로세스(풀 · 느림).
계산 자리 = ③ 서버 단독.
"""

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import os
import subprocess
import sys
import tempfile
import time
from functools import lru_cache
from pathlib import Path
from typing import Any, Literal
from uuid import UUID

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field

from common_util.common_util_derived_store import (
    derived_dir,
    derived_key,
    read_derived,
    write_derived,
)
from E02_StabGround import E02_StabGround_Router as e02
from E03_StabSection import E03_StabSection_Engine_Prob as prob
from E03_StabSection import E03_StabSection_Router as e03
from E03_StabSection import E03_StabSection_Router_WhatIf as whatif
from E03_StabSection import E03_StabSection_View as view
from E03_StabSection.E03_StabSection_Engine_LEM import WALL_ID
from E03_StabSection.E03_StabSection_Engine_Model import section_model_from_b06
from E03_StabSection.E03_StabSection_Types import groundwater as norm_groundwater

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/projects", tags=["E03 Stab Probabilistic"])
KIND = "stab2d_prob"
KEEP = 200
DEFAULTS_PATH = (
    Path(__file__).resolve().parents[1] / "resources/master_template/stability/prob_defaults.json"
)
#: 매회 탐색 표본 수 상한(측점 하나 탐색 수 초 × N)
SEARCH_MAX_N = 2000
#: 일꾼 프로세스 수 — None = 코어 − 2 · 1 = 서버 안에서 차례로(시험)
WORKERS: int | None = None

_JOBS: dict[str, dict[str, Any]] = {}
_TASKS: set[asyncio.Task] = set()


@lru_cache(maxsize=1)
def code_hash() -> str:
    digest = hashlib.sha1(e03.engine_hash().encode())
    for name in ("E03_StabSection_Engine_Prob.py", Path(__file__).name):
        digest.update((Path(__file__).parent / name).read_bytes())
    return digest.hexdigest()[:12]


def load_defaults() -> dict[str, Any]:
    return json.loads(DEFAULTS_PATH.read_text(encoding="utf-8"))


class VarIn(BaseModel):
    """확률 변수 하나 — c · φ · γ = 변동계수 · 지하수위 이동(m) · r_u = 표준편차 · 절단 = 평균 ± k·σ."""

    material: str | None = None
    prop: Literal["c", "phi", "gamma", "water", "ru"]
    dist: Literal["normal", "lognormal"] = "normal"
    cov: float | None = Field(None, ge=0.0, le=2.0)
    std: float | None = Field(None, ge=0.0, le=20.0)
    trunc_sigma: float | None = Field(3.0, gt=0.0, le=10.0)


class ProbRequest(BaseModel):
    side: Literal["left", "right"]
    condition: Literal["dry", "wet", "seismic"]
    method: Literal["ordinary", "bishop", "janbu", "spencer", "morgenstern_price"] = "bishop"
    mode: Literal["fixed", "search"] = "fixed"
    n: int = Field(1000, ge=50, le=20000)
    sampling: Literal["lhs", "mc"] = "lhs"
    seed: int = Field(0, ge=0, le=2**31 - 1)
    variables: list[VarIn] = Field(default_factory=list, max_length=40)
    rho_c_phi: float = Field(0.0, ge=-0.95, le=0.95)


def _model(ctx: dict[str, Any], condition: str) -> dict[str, Any]:
    """그 조건 모형 — `analyze_station` 과 같은 입력(지층 · 지반선 · 벽 · 하중 · E02 지하수 갈래)."""
    kw = ctx["kw"]
    model = section_model_from_b06(
        ctx["cross"],
        ctx["design"],
        materials=kw["materials"],
        layers=kw["layers"],
        ground=kw["ground"],
        wet=condition == "wet",
        seismic_kh=kw["seismic_kh"] if condition == "seismic" else 0.0,
        road_surcharge_kpa=kw["road_surcharge_kpa"],
        walls=kw.get("walls"),
    )
    model["groundwater"] = norm_groundwater(kw["groundwater_for"](condition, model))
    return model


def _slot(root: Path, station_key: str, side: str, condition: str, method: str) -> dict:
    """측점 · 비탈 · 조건 — 대표 단면만 · 모형 · 임계면 · 기준안전율."""
    ctx = whatif._context(root, station_key)
    rows = [dict(r) for r in e03._latest_summary(root, ctx["plan"]["fp"])[0]["stations"]]
    view.mark_representative(rows, e03._overrides(root))
    mark = next((r.get("representative") or {} for r in rows if r["key"] == station_key), {})
    if not mark.get("on"):
        raise HTTPException(
            status_code=422, detail="대표 단면만 확률 해석합니다 — 먼저 대표 단면으로."
        )
    if condition not in ctx["kw"]["conditions"]:
        raise HTTPException(status_code=422, detail="그 조건은 해석하지 않았습니다.")
    case = next(
        (
            c
            for c in ctx["value"]["out"]["cases"]
            if c["condition"] == condition and c["side"] == side
        ),
        None,
    )
    if case is None:
        raise HTTPException(status_code=422, detail="그 쪽 비탈이 없습니다.")
    methods = case["result"]["methods"]
    surface = (methods.get(method) or {}).get("min") or None
    model = _model(ctx, condition)
    sd = next(s for s in model["sides"] if s["side"] == side)
    return {
        "ctx": ctx,
        "model": model,
        "side": sd,
        "role": case["role"],
        "surface": surface,
        "methods": methods,
        "required": view.required_factor(ctx["plan"]["factors"], condition, case["role"]),
    }


def _materials(model: dict[str, Any]) -> list[dict[str, Any]]:
    """단면에 쓰인 물성(벽 뺌 · 지층 차례)."""
    used = list(dict.fromkeys(r["material_id"] for r in model["regions"]))
    by = {m["id"]: m for m in model["materials"]}
    return [by[i] for i in used if i != WALL_ID and i in by]


def _variables(req: ProbRequest, model: dict[str, Any]) -> tuple[list[dict], list]:
    """요청 → 엔진 변수(평균 = 모형 값 · σ 0 은 뺌) · c–φ 상관 짝."""
    mats = {m["id"]: m for m in _materials(model)}
    gw = model.get("groundwater") or {}
    out = []
    for v in req.variables:
        if v.prop in ("water", "ru"):
            need = "line" if v.prop == "water" else "ru"
            if gw.get("mode") != need:
                raise HTTPException(
                    status_code=422,
                    detail=f"{'지하수위' if v.prop == 'water' else 'r_u'} 변수는 그 조건 지하수 갈래가 "
                    f"{'수위선' if need == 'line' else 'r_u'} 일 때만.",
                )
            mean, std = (0.0, v.std) if v.prop == "water" else (float(gw["ru"]), v.std)
            name = "지하수위" if v.prop == "water" else "r_u"
        else:
            m = mats.get(v.material or "")
            if m is None:
                raise HTTPException(status_code=422, detail=f"단면에 없는 물성: {v.material}")
            mean = float(m["gamma_t"] if v.prop == "gamma" else m[v.prop])
            std = mean * float(v.cov) if v.cov is not None else v.std
            name = f"{m.get('name') or m['id']} {dict(c='c', phi='φ', gamma='γ')[v.prop]}"
        if not std:
            continue
        if v.dist == "lognormal" and (mean <= 0 or v.prop in ("water",)):
            raise HTTPException(status_code=422, detail=f"대수정규는 평균 > 0 만: {name}")
        k = v.trunc_sigma
        out.append(
            {
                "name": name,
                "material": v.material if v.prop not in ("water", "ru") else None,
                "prop": v.prop,
                "dist": v.dist,
                "mean": mean,
                "std": float(std),
                "lo": None if k is None else k * float(std),
                "hi": None if k is None else k * float(std),
            }
        )
    pairs = []
    if req.rho_c_phi:
        for i, a in enumerate(out):
            for j, b in enumerate(out):
                if a["prop"] == "c" and b["prop"] == "phi" and a["material"] == b["material"]:
                    pairs.append((i, j, req.rho_c_phi))
    return out, pairs


def _run_search(job: dict[str, Any], emit: Any) -> dict[str, Any]:
    """매회 탐색 일꾼(`python -m …Engine_Prob job.json`) — 한 줄마다 emit · 실패면 RuntimeError."""
    with tempfile.TemporaryDirectory(prefix="aislo_e03p_") as folder:
        path, out = Path(folder) / "job.json", Path(folder) / "out.json"
        path.write_text(json.dumps({**job, "out": str(out)}, ensure_ascii=False), encoding="utf-8")
        log = Path(folder) / "stderr.txt"
        with log.open("w", encoding="utf-8") as errors:
            process = subprocess.Popen(
                [sys.executable, "-m", "E03_StabSection.E03_StabSection_Engine_Prob", str(path)],
                cwd=e03.PROJECT_ROOT,
                stdout=subprocess.PIPE,
                stderr=errors,
                text=True,
                encoding="utf-8",
            )
            for line in process.stdout:
                if line.startswith("{"):
                    emit(json.loads(line).get("done"))
            code = process.wait()
        if code != 0:
            tail = log.read_text(encoding="utf-8", errors="replace").strip().splitlines()[-1:]
            raise RuntimeError(f"확률 해석 일꾼 실패: {tail}")
        return json.loads(out.read_text(encoding="utf-8"))


def run_prob(root: Path, station_key: str, req: ProbRequest, report: Any) -> dict[str, Any]:
    slot = _slot(root, station_key, req.side, req.condition, req.method)
    if slot["surface"] is None:
        raise HTTPException(status_code=422, detail="그 방법 임계면이 없습니다(해 없음).")
    if req.mode == "search" and req.n > SEARCH_MAX_N:
        raise HTTPException(status_code=422, detail=f"매회 탐색은 표본 {SEARCH_MAX_N} 회까지.")
    variables, pairs = _variables(req, slot["model"])
    if not variables:
        raise HTTPException(status_code=422, detail="확률 변수가 없습니다(변동계수 · 표준편차 0).")
    row = slot["ctx"]["row"]
    key = derived_key(code_hash(), row["fp"], req.model_dump())
    cached = read_derived(root, KIND, key)
    if cached:
        return cached["value"]
    job = {
        "model": slot["model"],
        "surface": slot["surface"],
        "direction": slot["side"]["direction"],
        "limits": list(slot["side"]["limits"]),
        "variables": variables,
        "correlations": pairs,
        "n": req.n,
        "sampling": req.sampling,
        "seed": req.seed,
        "mode": req.mode,
        "method": req.method,
        "required": slot["required"],
    }

    def emit(done: int | None) -> None:
        if done:
            report(2 + 93 * int(done) // req.n, f"표본 {done} / {req.n}")

    report(1, "시작")
    workers = WORKERS or max(1, (os.cpu_count() or 2) - 2)
    if req.mode == "search" and workers > 1:
        result = _run_search({**job, "workers": workers}, emit)
    else:
        result = prob.run(**job, emit=emit)
    record = {
        "station_key": station_key,
        "chainage_m": row["chainage_m"],
        "label": row.get("label"),
        "base_fp": row["fp"],
        "side": req.side,
        "role": slot["role"],
        "condition": req.condition,
        "required": slot["required"],
        "fs_deterministic": slot["surface"]["F"],
        "surface_kind": slot["surface"]["kind"],
        "request": req.model_dump(),
        "result": result,
        "computed_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
    }
    write_derived(root, KIND, key, record, keep=KEEP)
    return record


def records(root: Path) -> list[dict[str, Any]]:
    """가장 최근 2D 요약 측점에 붙은 기록 — (측점 · 비탈 · 조건)마다 가장 최근 하나 · 누가거리 순."""
    summary, _ = e03._latest_summary(root, "")
    fps = {r["fp"] for r in (summary or {}).get("stations") or []}
    latest: dict[tuple, dict[str, Any]] = {}
    for path in derived_dir(root, KIND).glob("*.json"):
        value = (read_derived(root, KIND, path.stem) or {}).get("value")
        if not value or value.get("base_fp") not in fps:
            continue
        slot = (value["station_key"], value["side"], value["condition"])
        if slot not in latest or value["computed_at"] > latest[slot]["computed_at"]:
            latest[slot] = value
    return sorted(
        latest.values(), key=lambda v: (float(v["chainage_m"]), v["side"], v["condition"])
    )


def setup(root: Path, station_key: str, side: str, condition: str) -> dict[str, Any]:
    slot = _slot(root, station_key, side, condition, "bishop")
    gw = slot["model"].get("groundwater") or {}
    last = next(
        (
            r
            for r in records(root)
            if (r["station_key"], r["side"], r["condition"]) == (station_key, side, condition)
        ),
        None,
    )
    return {
        "station_key": station_key,
        "side": side,
        "condition": condition,
        "role": slot["role"],
        "required": slot["required"],
        "fs_deterministic": {m: (v.get("min") or {}).get("F") for m, v in slot["methods"].items()},
        "materials": [
            {k: m.get(k) for k in ("id", "name", "c", "phi", "gamma_t")}
            for m in _materials(slot["model"])
        ],
        "groundwater": {"mode": gw.get("mode", "none"), "ru": gw.get("ru")},
        "defaults": load_defaults(),
        "search_max_n": SEARCH_MAX_N,
        "last": last,
    }


@router.get("/{project_id}/stab/prob/setup/{station_key}", response_model=None)
async def get_setup(project_id: UUID, station_key: str, side: str, condition: str) -> dict:
    root, _ = await e02._project(project_id)
    return await asyncio.to_thread(setup, root, station_key, side, condition)


@router.post("/{project_id}/stab/prob/{station_key}", response_model=None)
async def post_prob(project_id: UUID, station_key: str, payload: ProbRequest) -> dict:
    """뒤에서 — 같은 지문이면 바로 저장본(fresh) · 진행 = `GET …/stab/prob/progress`."""
    root, _ = await e02._project(project_id)
    key = str(project_id)
    if (_JOBS.get(key) or {}).get("running"):
        raise HTTPException(status_code=409, detail="이미 확률 해석 중입니다.")
    # 입력 검사 · 저장본 확인은 바로(422 를 그 자리에서 돌려줌)
    slot = await asyncio.to_thread(
        _slot, root, station_key, payload.side, payload.condition, payload.method
    )
    _variables(payload, slot["model"])
    fp = derived_key(code_hash(), slot["ctx"]["row"]["fp"], payload.model_dump())
    saved = read_derived(root, KIND, fp)
    if saved:
        return {"status": "fresh", "record": saved["value"]}
    job = {
        "running": True,
        "percent": 0,
        "message": "시작",
        "error": None,
        "station_key": station_key,
    }
    _JOBS[key] = job

    def report(percent: int, message: str) -> None:
        job.update(percent=int(percent), message=message)

    async def work() -> None:
        try:
            await asyncio.to_thread(run_prob, root, station_key, payload, report)
            job.update(percent=100, message="끝")
        except Exception as error:  # noqa: BLE001 — 진행률에 까닭을 그대로 올림
            logger.exception("E03 확률 해석 실패: project_id=%s", project_id)
            job["error"] = str(getattr(error, "detail", None) or error)
        finally:
            job["running"] = False

    task = asyncio.create_task(work(), name=f"E03 prob {project_id}")
    _TASKS.add(task)
    task.add_done_callback(_TASKS.discard)
    return {"status": "accepted", **job}


@router.get("/{project_id}/stab/prob/progress", response_model=None)
async def get_progress(project_id: UUID) -> dict:
    await e02._project(project_id)
    job = _JOBS.get(str(project_id))
    return dict(job) if job else {"running": False, "percent": 0, "message": "", "error": None}


@router.get("/{project_id}/stab/prob", response_model=None)
async def get_records(project_id: UUID) -> dict:
    root, _ = await e02._project(project_id)
    return {"records": await asyncio.to_thread(records, root)}