351 lines
14 KiB
Python
351 lines
14 KiB
Python
"""E07 광역 자연사면 위험 지도 API (요구 345 · 351 · 요구서 4-2 2차 · PLAN 536-2).
|
||
|
||
- `POST /api/projects/{id}/stab/area/analyze` — 해석(뒤에서) · 입력 지문이 저장본과 같으면
|
||
`cached` · force = 다시
|
||
- `GET …/stab/area/progress` — 진행률
|
||
- `GET …/stab/area` — 기본값(E02 붕적토) · 물성 목록 · 노선 · 측점 · 저장본(격자 · 측점표 · 요약)
|
||
|
||
계산 자리 = ③ 서버 단독(엔진 `E07_StabArea_Engine` · 순수). 지형 = 복사본 원지형 DTM
|
||
(노선면 합성 안 함 — 노선 위 · 아래 자연사면).
|
||
파생 저장본 = `E07_StabArea/_derived/area.json`(지문이 같으면 안 셈).
|
||
권한 = main.py `protected_with_company`(지정받은 회사 = GET 만).
|
||
|
||
격자 꼴(화면 `logic/E07_StabArea_Raster.ts` 와 같은 열쇠) — grid = {x0, y0, dx, nx, ny}
|
||
(행 0 = 남쪽) · fs · si = uint16 ×1000(65535 = 자료 없음 · 65.534 위로 자름) ·
|
||
cls = uint8 등급(0 없음 · 1~6) · slope = uint16 ×10 ° · m = uint8 ×100(255 없음)
|
||
— 모두 줄 우선 · 작은 끝 base64 · 띠 밖 셀 = 자료 없음.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import base64
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
import time
|
||
from collections.abc import Callable
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any, Literal
|
||
from uuid import UUID
|
||
|
||
import aiomysql
|
||
import numpy as np
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from pydantic import BaseModel, Field, model_validator
|
||
|
||
from common_util.common_util_auth import require_company
|
||
from common_util.common_util_json import atomic_write_json
|
||
from common_util.common_util_storage import resolve_stored_project_path
|
||
from config.config_db import get_db_pool
|
||
from E01_StabRegister import E01_StabRegister_Source as source
|
||
from E02_StabGround import E02_StabGround_Model as ground_model
|
||
from E07_StabArea import E07_StabArea_Engine as engine
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter(prefix="/api/projects", tags=["E07 Stab Area"])
|
||
|
||
ENGINE_VERSION = "e07-3" # 엔진 판 — 바뀌면 지문이 달라져 다시 셈
|
||
RESULT = Path("E07_StabArea") / "_derived" / "area.json"
|
||
DEFAULT_MATERIAL = "colluvium" # 표층 붕괴에 가장 취약(지식DB 안정성평가/04 붕적토)
|
||
RANGE_C_KPA, RANGE_PHI_DEG = (
|
||
5.0,
|
||
3.0,
|
||
) # SI 범위 위 = 아래 + 이 폭(04 붕적토 범위 c 0~10 · φ 25~30 폭의 반)
|
||
|
||
_JOBS: dict[str, dict[str, Any]] = {}
|
||
_TASKS: set[asyncio.Task] = set()
|
||
_IDLE = {"running": False, "percent": 0, "message": None, "error": None}
|
||
|
||
|
||
class AreaRequest(BaseModel):
|
||
band_m: float = Field(100.0, ge=10.0, le=500.0) # 노선 띠 = 중심선 ± 이 폭
|
||
cell_m: float = Field(2.0, ge=1.0, le=10.0)
|
||
material_id: str | None = None # 이름표만(값은 아래 칸)
|
||
gamma_t: float = Field(..., gt=0.0, le=30.0)
|
||
gamma_sat: float = Field(..., gt=engine.GAMMA_W, le=30.0)
|
||
c: float = Field(..., ge=0.0, le=500.0)
|
||
c_hi: float = Field(..., ge=0.0, le=500.0)
|
||
phi: float = Field(..., ge=0.0, le=60.0)
|
||
phi_hi: float = Field(..., ge=0.0, le=60.0)
|
||
# 기본값 = 근거 후보(바꿀 수 있음 · 확정은 사용자) — 출처 = 지식DB 예정노선/02 §1.1 ·
|
||
# SINMAP 2 설명서 기본 보정 · 국내 적용 연구(김동엽 2026 봉화 TRIGRS Cr 0~5 kPa · 우면산 1 kPa)
|
||
c_root: float = Field(1.0, ge=0.0, le=50.0) # 산림 뿌리 점착 · 나지 · 절성토면 = 0
|
||
soil_depth_m: float = Field(1.5, ge=0.1, le=10.0)
|
||
scenario: Literal["sinmap", "saturated"] = "sinmap" # 표준(습윤지수) · 최악(완전 포화)
|
||
tr_lo: float = Field(2000.0, ge=1.0, le=1e5) # T/R(m) — SINMAP 기본 보정 2000~3000
|
||
tr_hi: float = Field(3000.0, ge=1.0, le=1e5)
|
||
# 측점 등급 문턱(%) — 영향 구역 안 SI < 1 셀 비율 · 브레인 정함(사용자 확정 전)
|
||
level_lo_pct: float = Field(engine.LEVEL_LO_PCT, gt=0.0, le=100.0)
|
||
level_hi_pct: float = Field(engine.LEVEL_HI_PCT, gt=0.0, le=100.0)
|
||
force: bool = False
|
||
|
||
@model_validator(mode="after")
|
||
def _ranges(self) -> AreaRequest:
|
||
if self.level_hi_pct < self.level_lo_pct:
|
||
raise ValueError("측점 등급 문턱: 높음이 보통보다 작습니다.")
|
||
if self.c_hi < self.c or self.phi_hi < self.phi or self.tr_hi < self.tr_lo:
|
||
raise ValueError("범위 위가 아래보다 작습니다(c · φ · T/R).")
|
||
return self
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 입력 · 지문 · 저장본
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _ground(project_root: Path) -> dict[str, Any]:
|
||
for path in (ground_model.ground_path(project_root), ground_model.initial_path(project_root)):
|
||
if path.is_file():
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
return ground_model.pick(ground_model.load_defaults())
|
||
|
||
|
||
def defaults(project_root: Path) -> dict[str, Any]:
|
||
"""기본 조건 — E02 붕적토(없으면 첫 물성) · 범위 위 = + 5 kPa · + 3°."""
|
||
materials = _ground(project_root).get("materials") or []
|
||
mat = next((m for m in materials if m.get("id") == DEFAULT_MATERIAL), None) or materials[0]
|
||
return {
|
||
**{
|
||
k: f.default
|
||
for k, f in AreaRequest.model_fields.items()
|
||
if not f.is_required() and k != "force"
|
||
},
|
||
"material_id": mat.get("id"),
|
||
"gamma_t": float(mat["gamma_t"]),
|
||
"gamma_sat": float(mat["gamma_sat"]),
|
||
"c": float(mat["c"]),
|
||
"c_hi": float(mat["c"]) + RANGE_C_KPA,
|
||
"phi": float(mat["phi"]),
|
||
"phi_hi": float(mat["phi"]) + RANGE_PHI_DEG,
|
||
}
|
||
|
||
|
||
def fingerprint(source_fingerprint: str, settings: dict[str, Any]) -> str:
|
||
body = {"source": source_fingerprint, "settings": settings, "engine": ENGINE_VERSION}
|
||
text = json.dumps(body, ensure_ascii=False, sort_keys=True)
|
||
return hashlib.sha1(text.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def load(project_root: Path) -> dict[str, Any] | None:
|
||
path = Path(project_root) / RESULT
|
||
return json.loads(path.read_text(encoding="utf-8")) if path.is_file() else None
|
||
|
||
|
||
def _route_line(root: Path) -> np.ndarray:
|
||
points = sorted(
|
||
source.db_snapshot(root).get("route_points") or [], key=lambda p: p["sequence_num"]
|
||
)
|
||
return np.array([[p["model_x"], p["model_y"], p["chainage_m"]] for p in points], float)
|
||
|
||
|
||
def _b64(values: np.ndarray, dtype: str) -> str:
|
||
return base64.b64encode(np.ascontiguousarray(values, dtype=dtype).tobytes()).decode("ascii")
|
||
|
||
|
||
def _u16(values: np.ndarray, scale: float, keep: np.ndarray) -> str:
|
||
v = np.round(np.clip(np.nan_to_num(values * scale, nan=-1.0, posinf=65534.0), -1, 65534))
|
||
return _b64(np.where(keep & (v >= 0), v, 65535), "<u2")
|
||
|
||
|
||
def run(
|
||
project_root: Path,
|
||
root: Path,
|
||
settings: dict[str, Any],
|
||
finger: str,
|
||
report: Callable[[int, str], None],
|
||
) -> dict[str, Any]:
|
||
"""해석 한 번 → 저장본 쓰고 요약 돌려줌."""
|
||
started = time.perf_counter()
|
||
cell = float(settings["cell_m"])
|
||
report(5, "지형 격자")
|
||
with np.load(source.terrain_grid_path(root)) as npz:
|
||
dem = {k: npz[k] for k in ("x", "y", "z", "valid_mask")}
|
||
z, x0, y0 = engine.resample_dem(dem, cell)
|
||
report(25, "경사 · 집수면적 · 안전율")
|
||
keys = (
|
||
"gamma_t",
|
||
"gamma_sat",
|
||
"c",
|
||
"c_hi",
|
||
"phi",
|
||
"phi_hi",
|
||
"c_root",
|
||
"soil_depth_m",
|
||
"scenario",
|
||
"tr_lo",
|
||
"tr_hi",
|
||
)
|
||
out = engine.analyze_grid(z, cell, engine.AreaParams(**{k: settings[k] for k in keys}))
|
||
report(75, "노선 띠 · 측점")
|
||
stations = sorted(source.stations(root), key=lambda s: float(s["chainage_m"]))
|
||
inside, chain = engine.band_cells(
|
||
z.shape, x0, y0, cell, _route_line(root), float(settings["band_m"])
|
||
)
|
||
rows = engine.station_table(
|
||
[s["chainage_m"] for s in stations],
|
||
chain,
|
||
inside,
|
||
out,
|
||
cell,
|
||
float(settings["level_lo_pct"]),
|
||
float(settings["level_hi_pct"]),
|
||
)
|
||
labels = {round(float(s["chainage_m"]), 3): s.get("label") for s in stations}
|
||
for row in rows:
|
||
row["label"] = labels.get(round(row["chainage_m"], 3))
|
||
# 띠 상자로 잘라 저장
|
||
iy, ix = np.nonzero(inside)
|
||
if not len(iy):
|
||
raise ValueError("노선 띠가 지형 격자와 겹치지 않습니다.")
|
||
win = (slice(iy.min(), iy.max() + 1), slice(ix.min(), ix.max() + 1))
|
||
keep = inside[win] & (out["cls"][win] > 0)
|
||
areas = engine.class_areas(out["cls"], inside, cell)
|
||
have = inside & (out["cls"] > 0)
|
||
summary = {
|
||
"areas_m2": areas,
|
||
"band_area_m2": round(float(inside.sum()) * cell**2, 1),
|
||
"covered_area_m2": round(float(have.sum()) * cell**2, 1),
|
||
"risk_area_m2": round(
|
||
sum(v for k, v in areas.items() if engine.CLASS_KEYS.index(k) >= engine.RISK_CLASS), 1
|
||
),
|
||
"fs_min": round(float(np.min(out["fs"][have])), 3) if have.any() else None,
|
||
"station_levels": {k: sum(r.get("level") == k for r in rows) for k in engine.LEVEL_KEYS},
|
||
"stations": len(rows),
|
||
}
|
||
doc = {
|
||
"fingerprint": finger,
|
||
"engine": ENGINE_VERSION,
|
||
"settings": settings,
|
||
"computed_at": datetime.now().isoformat(timespec="seconds"),
|
||
"elapsed_s": round(time.perf_counter() - started, 2),
|
||
"grid": {
|
||
"x0": x0 + int(ix.min()) * cell,
|
||
"y0": y0 + int(iy.min()) * cell,
|
||
"dx": cell,
|
||
"nx": int(keep.shape[1]),
|
||
"ny": int(keep.shape[0]),
|
||
},
|
||
"fs": _u16(out["fs"][win], 1000, keep),
|
||
"si": _u16(out["si"][win], 1000, keep),
|
||
"cls": _b64(np.where(keep, out["cls"][win], 0), "u1"),
|
||
"slope": _u16(np.degrees(out["slope"][win]), 10, keep),
|
||
"m": _b64(np.where(keep, np.round(np.nan_to_num(out["m"][win], nan=0) * 100), 255), "u1"),
|
||
"stations": rows,
|
||
"summary": summary,
|
||
}
|
||
report(95, "저장")
|
||
atomic_write_json(Path(project_root) / RESULT, doc)
|
||
return summary
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 길
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def _project_root(project_id: UUID) -> Path:
|
||
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",
|
||
(str(project_id),),
|
||
)
|
||
row = await cursor.fetchone()
|
||
if not row or not row["storage_path"]:
|
||
raise HTTPException(status_code=404, detail="안정성평가 프로젝트가 아닙니다.")
|
||
return Path(resolve_stored_project_path(row["storage_path"]))
|
||
|
||
|
||
def _source_root(project_root: Path) -> Path:
|
||
root = source.source_dir(project_root)
|
||
if not (root / source.MANIFEST_NAME).is_file():
|
||
raise HTTPException(status_code=409, detail="원본 복사본이 없습니다.")
|
||
return root
|
||
|
||
|
||
@router.post("/{project_id}/stab/area/analyze", response_model=None)
|
||
async def analyze(project_id: UUID, payload: AreaRequest) -> dict[str, Any]:
|
||
"""[해석] — 지문이 같으면 `cached`(안 셈) · 다르거나 force 면 뒤에서 셈."""
|
||
pid = str(project_id)
|
||
project_root = await _project_root(project_id)
|
||
root = _source_root(project_root)
|
||
settings = payload.model_dump(exclude={"force"})
|
||
finger = fingerprint(source.manifest(root)["source_fingerprint"], settings)
|
||
saved = await asyncio.to_thread(load, project_root)
|
||
if not payload.force and saved and saved["fingerprint"] == finger:
|
||
return {"status": "cached", "summary": saved["summary"]}
|
||
if (_JOBS.get(pid) or {}).get("running"):
|
||
raise HTTPException(status_code=409, detail="이미 광역 해석이 돌고 있습니다.")
|
||
job = {**_IDLE, "running": True, "message": "시작"}
|
||
_JOBS[pid] = job
|
||
|
||
def report(percent: int, message: str) -> None:
|
||
job.update(percent=max(0, min(99, int(percent))), message=message)
|
||
|
||
async def work() -> None:
|
||
try:
|
||
await asyncio.to_thread(run, project_root, root, settings, finger, report)
|
||
job.update(percent=100, message="끝")
|
||
except Exception as error: # noqa: BLE001 — 진행률에 까닭을 그대로 올림
|
||
logger.exception("E07 광역 해석 실패: project_id=%s", pid)
|
||
job["error"] = str(error)
|
||
finally:
|
||
job["running"] = False
|
||
|
||
task = asyncio.create_task(work(), name=f"E07 area {pid}")
|
||
_TASKS.add(task)
|
||
task.add_done_callback(_TASKS.discard)
|
||
return {"status": "accepted"}
|
||
|
||
|
||
@router.get("/{project_id}/stab/area/progress", response_model=None)
|
||
async def progress(project_id: UUID) -> dict[str, Any]:
|
||
return {"status": "success", **dict(_JOBS.get(str(project_id)) or _IDLE)}
|
||
|
||
|
||
def _overview(project_root: Path, root: Path) -> dict[str, Any]:
|
||
saved = load(project_root)
|
||
if saved:
|
||
current = fingerprint(source.manifest(root)["source_fingerprint"], saved["settings"])
|
||
saved["stale"] = saved["fingerprint"] != current
|
||
return {
|
||
"defaults": defaults(project_root),
|
||
"materials": _ground(project_root).get("materials") or [],
|
||
"line": _route_line(root).round(3).tolist(),
|
||
"stations": [
|
||
{
|
||
"chainage_m": s["chainage_m"],
|
||
"label": s["label"],
|
||
"x": s["center_x"],
|
||
"y": s["center_y"],
|
||
}
|
||
for s in sorted(source.stations(root), key=lambda s: float(s["chainage_m"]))
|
||
],
|
||
"result": saved,
|
||
}
|
||
|
||
|
||
async def _editable(project_id: UUID, session: dict[str, Any]) -> bool:
|
||
"""[해석] 가능 = 시스템 관리자 · 같은 회사(지정받은 회사는 보기만)."""
|
||
if session.get("role") == "SYSTEM_ADMIN":
|
||
return True
|
||
async with get_db_pool().acquire() as connection, connection.cursor() as cursor:
|
||
await cursor.execute("SELECT company_id FROM projects WHERE id = %s", (str(project_id),))
|
||
row = await cursor.fetchone()
|
||
return bool(row) and row[0] == session.get("company_id")
|
||
|
||
|
||
@router.get("/{project_id}/stab/area", response_model=None)
|
||
async def area(
|
||
project_id: UUID, session: dict[str, Any] = Depends(require_company)
|
||
) -> dict[str, Any]:
|
||
project_root = await _project_root(project_id)
|
||
root = _source_root(project_root)
|
||
overview = await asyncio.to_thread(_overview, project_root, root)
|
||
return {"status": "success", "editable": await _editable(project_id, session), **overview}
|