432 lines
19 KiB
Python
432 lines
19 KiB
Python
"""E03 보강 검토 API — 비탈 완화 · 소단 가정 재해석 · 벽 외적 안정 (요구 345 · 351 · PLAN 536-3).
|
|
|
|
- `POST /api/projects/{id}/stab/whatif/{station_key}` — 가정(비탈 · 1:n · 소단 폭 · 높이 ·
|
|
최소 완화 찾기 · 네일 · 앵커 `supports` · 최소 길이 찾기) → 형상만 바꾼 모형으로
|
|
`analyze_station` 다시 · 전후 안전율 · 판정 · 바뀐 단면(화면 꼴) ·
|
|
네일마다 동원력 · mode(엔진 `cases[i].result.support_forces`)
|
|
- `POST …/stab/wallcheck/{station_key}` — 그 측점 벽마다 활동 · 전도 · 지지력(평상시 · 지진 시)
|
|
- `GET …/stab/whatif` — 지금 2D 결과에 붙은 가정 · 벽 검토 목록(단면 꼴 뺌 · E05 「보강검토」 시트)
|
|
|
|
가정 결과 = 파생 저장본 `_derived/stab2d_whatif/<지문>` + 단면 꼴 `stab2d_whatif_view/` (원본 결과 그대로) ·
|
|
지문 = 원본 측점 지문 + 가정. 엔진 = 부르기만(`analyze_station` · `section_model_from_b06`).
|
|
계산 자리 = ③ 서버 단독.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import inspect
|
|
import math
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
from functools import lru_cache, partial
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel, Field, model_validator
|
|
from shapely.geometry import Point, Polygon
|
|
|
|
from common_util.common_util_derived_store import (
|
|
derived_dir,
|
|
derived_key,
|
|
read_derived,
|
|
write_derived,
|
|
)
|
|
from E01_StabRegister import E01_StabRegister_Source as src
|
|
from E02_StabGround import E02_StabGround_Model as ground_model
|
|
from E02_StabGround import E02_StabGround_Router as e02
|
|
from E03_StabSection import E03_StabSection_Router as e03
|
|
from E03_StabSection import E03_StabSection_View as view
|
|
from E03_StabSection import E03_StabSection_Walls as walls_mod
|
|
from E03_StabSection import E03_StabSection_WallCheck as wallcheck
|
|
from E03_StabSection import E03_StabSection_WhatIf as whatif
|
|
from E03_StabSection.E03_StabSection_Engine_Model import WALL_MATERIAL, section_model_from_b06
|
|
from E03_StabSection.E03_StabSection_Engine_Station import analyze_station
|
|
|
|
router = APIRouter(prefix="/api/projects", tags=["E03 Stab What-if"])
|
|
KIND = "stab2d_whatif"
|
|
KIND_VIEW = "stab2d_whatif_view" # 바뀐 단면 화면 꼴(무거움 · 목록은 안 읽음)
|
|
KEEP = 300
|
|
#: 최소 길이 찾기 — 가장 긴 길이 · 찾는 단위(m)
|
|
SUPPORT_MAX_LEN = 25.0
|
|
SUPPORT_LEN_STEP = 0.5
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def code_hash() -> str:
|
|
"""가정 · 벽 검토 코드 지문 — 바뀌면 저장본을 다시 셈(엔진 지문은 원본 측점 지문에 있음)."""
|
|
digest = hashlib.sha1()
|
|
for name in ("E03_StabSection_WhatIf.py", "E03_StabSection_WallCheck.py", Path(__file__).name):
|
|
digest.update((Path(__file__).parent / name).read_bytes())
|
|
return digest.hexdigest()[:12]
|
|
|
|
|
|
class SupportIn(BaseModel):
|
|
"""네일 · 앵커 하나 — 브레인 정함 꼴(sub4 엔진 같은 이름) · side 없으면 요청 비탈."""
|
|
|
|
type: Literal["nail", "anchor"]
|
|
side: Literal["left", "right"] | None = None
|
|
x: float
|
|
z: float
|
|
angle_deg: float = Field(ge=-10.0, le=80.0)
|
|
length_m: float = Field(gt=0.0, le=50.0)
|
|
spacing_m: float = Field(gt=0.1, le=10.0)
|
|
tensile_kN: float = Field(gt=0.0)
|
|
bond_kPa: float = Field(gt=0.0)
|
|
bond_length_m: float = Field(0.0, ge=0.0, le=50.0)
|
|
hole_dia_mm: float = Field(gt=0.0, le=500.0)
|
|
prestress_kN: float = Field(0.0, ge=0.0)
|
|
|
|
@model_validator(mode="after")
|
|
def _anchor_bond(self) -> "SupportIn":
|
|
if self.type == "anchor" and not 0 < self.bond_length_m < self.length_m:
|
|
raise ValueError("앵커 정착장은 0 초과 · 길이 미만")
|
|
return self
|
|
|
|
|
|
class WhatIfRequest(BaseModel):
|
|
side: Literal["left", "right"]
|
|
ratio: float | None = Field(None, gt=0.05, le=whatif.MAX_RATIO)
|
|
bench_w: float = Field(0.0, ge=0.0, le=10.0)
|
|
bench_h: float = Field(0.0, ge=0.0, le=30.0)
|
|
search: bool = False
|
|
#: 최소 완화 찾기 목표 — 조건 하나(None = 모든 조건) · 목표 안전율(None = 그 조건 기준안전율)
|
|
condition: Literal["dry", "wet", "seismic"] | None = None
|
|
target_fs: float | None = Field(None, gt=0.5, le=5.0)
|
|
#: 네일 · 앵커 — 있으면 search = 최소 길이 찾기(모두 같은 길이 · 앵커 정착장 그대로)
|
|
supports: list[SupportIn] = Field(default_factory=list, max_length=40)
|
|
|
|
|
|
def engine_takes_supports() -> bool:
|
|
"""sub4 엔진에 supports 인자가 들어왔는지 — 없으면 네일 · 앵커는 그림만(결과 = 보강 없음)."""
|
|
return "supports" in inspect.signature(analyze_station).parameters
|
|
|
|
|
|
def _sized(supports: list[dict], length: float | None) -> list[dict]:
|
|
if length is None:
|
|
return supports
|
|
return [
|
|
{**s, "length_m": max(length, s["bond_length_m"] + 0.5 * (s["type"] == "anchor"))}
|
|
for s in supports
|
|
]
|
|
|
|
|
|
def _forces(out: dict, side: str) -> dict[str, list]:
|
|
"""조건마다 그 비탈 임계면 기준 네일 힘(엔진 `cases[i].result.support_forces`)."""
|
|
return {
|
|
c["condition"]: c["result"]["support_forces"]
|
|
for c in out.get("cases") or []
|
|
if c.get("side") == side and (c.get("result") or {}).get("support_forces") is not None
|
|
}
|
|
|
|
|
|
class WallRequest(BaseModel):
|
|
method: Literal["coulomb", "rankine"] = "coulomb"
|
|
water_ratio: float = Field(0.0, ge=0.0, le=1.0)
|
|
kh: float | None = Field(None, ge=0.0, le=0.5)
|
|
|
|
|
|
def _context(root: Path, station_key: str) -> dict[str, Any]:
|
|
"""측점 하나 — 지금 입력과 같은 2D 결과 · 원본 저장본 · 엔진 입력(일꾼과 같은 값)."""
|
|
ground = e03._ground(root)
|
|
plan = e03.plan_run(root, ground)
|
|
summary, fresh = e03._latest_summary(root, plan["fp"])
|
|
if not summary or not fresh:
|
|
raise HTTPException(status_code=409, detail="2D 결과가 지금 입력과 다릅니다 — 먼저 [해석].")
|
|
row = next((r for r in summary["stations"] if r["key"] == station_key), None)
|
|
saved = read_derived(root, e03.KIND_STATION, row["fp"]) if row else None
|
|
if not saved or not saved["value"]["out"]:
|
|
raise HTTPException(status_code=404, detail="그 측점 해석 결과가 없습니다.")
|
|
value = saved["value"]
|
|
cross = src.cross_section(plan["source"], row["chainage_m"])
|
|
entry = next((it["walls"] for it in plan["items"] if it["key"] == station_key), None)
|
|
shared = plan["shared"]
|
|
kw = {
|
|
"materials": shared["materials"],
|
|
"conditions": list(shared["conditions"]),
|
|
"seismic_kh": shared["seismic_kh"],
|
|
"road_surcharge_kpa": shared["road_surcharge_kpa"],
|
|
"layers": value["inputs"]["layers"],
|
|
"ground": [tuple(p) for p in value["inputs"]["ground"]],
|
|
}
|
|
walls = walls_mod.walls_of(entry)
|
|
if walls:
|
|
kw["walls"] = walls
|
|
kw["groundwater_for"] = partial(ground_model.station_groundwater, ground)
|
|
return {
|
|
"ground": ground,
|
|
"plan": plan,
|
|
"row": row,
|
|
"value": value,
|
|
"cross": cross,
|
|
"design": walls_mod.design_with_walls(cross["design"], entry),
|
|
"walls": walls or [],
|
|
"kw": kw,
|
|
}
|
|
|
|
|
|
def _wall_side(wall: dict[str, Any]) -> str:
|
|
return "left" if Polygon(wall["polygon"]).centroid.x > 0 else "right"
|
|
|
|
|
|
def _section(ctx: dict[str, Any], design: dict[str, Any], row: dict, out: dict) -> dict:
|
|
"""바뀐 단면 화면 꼴 — `GET …/section` 과 같은 꼴(지하수선 = E02 고른 갈래)."""
|
|
kw = {k: ctx["kw"][k] for k in ("materials", "layers", "ground", "road_surcharge_kpa")}
|
|
kw["walls"] = ctx["kw"].get("walls")
|
|
dry = section_model_from_b06(ctx["cross"], design, **kw)
|
|
water = {}
|
|
for cond in ctx["kw"]["conditions"]:
|
|
model = section_model_from_b06(ctx["cross"], design, wet=cond == "wet", **kw)
|
|
gw = ground_model.station_groundwater(ctx["ground"], cond, model)
|
|
water[cond] = gw.get("line") if gw.get("mode") == "line" else None
|
|
return view.section_view(row, out, dry, water)
|
|
|
|
|
|
def _side_cases(row: dict[str, Any], side: str) -> dict[str, Any]:
|
|
return {cond: by[side] for cond, by in (row.get("cases") or {}).items() if side in by}
|
|
|
|
|
|
def run_whatif(root: Path, station_key: str, req: WhatIfRequest) -> dict[str, Any]:
|
|
ctx = _context(root, station_key)
|
|
row, design, ground = ctx["row"], ctx["design"], ctx["kw"]["ground"]
|
|
role = {s["side"]: s["role"] for s in row["sides"]}.get(req.side)
|
|
if role is None:
|
|
raise HTTPException(status_code=422, detail="그 쪽 비탈이 없습니다.")
|
|
if any(_wall_side(w) == req.side for w in ctx["walls"]):
|
|
raise HTTPException(status_code=422, detail="벽이 선 비탈 — 벽 외적 안정으로 검토하세요.")
|
|
before_ratio = whatif.current_ratio(design, req.side, role)
|
|
assume = {
|
|
"side": req.side,
|
|
"role": role,
|
|
"ratio": float(req.ratio or before_ratio),
|
|
"bench_w": req.bench_w if req.bench_h > 0 else 0.0,
|
|
"bench_h": req.bench_h if req.bench_w > 0 else 0.0,
|
|
}
|
|
goal = {"condition": req.condition, "target_fs": req.target_fs} if req.search else None
|
|
supports = [{**s.model_dump(), "side": s.side or req.side} for s in req.supports]
|
|
engine = engine_takes_supports()
|
|
key = derived_key(code_hash(), row["fp"], assume, goal, supports, engine)
|
|
cached, shown = read_derived(root, KIND, key), read_derived(root, KIND_VIEW, key)
|
|
if cached and shown:
|
|
return {**cached["value"], "view": shown["value"]}
|
|
t0 = time.perf_counter()
|
|
meta, factors = ctx["value"]["meta"], ctx["plan"]["factors"]
|
|
|
|
memo: dict[tuple, tuple[dict, dict] | None] = {}
|
|
same = not assume["bench_w"] # 지금 기울기 그대로 = 원본 설계선(다시 그리지 않음)
|
|
|
|
def run(n: float, length: float | None = None) -> tuple[dict, dict] | None:
|
|
n = round(n, 4)
|
|
if (n, length) not in memo:
|
|
if same and abs(n - before_ratio) < 1e-9:
|
|
variant = design
|
|
else: # None = 해석 폭 안에서 원지반을 못 만남(자연 비탈보다 완만)
|
|
variant = whatif.design_variant(design, ground, {**assume, "ratio": n})
|
|
kw = dict(ctx["kw"])
|
|
if supports and engine:
|
|
kw["supports"] = _sized(supports, length)
|
|
memo[(n, length)] = variant and (analyze_station(ctx["cross"], variant, **kw), variant)
|
|
return memo[(n, length)]
|
|
|
|
reached: list[float] = [] # 원지반을 만난 기울기
|
|
|
|
def meets(done: tuple[dict, dict] | None) -> bool:
|
|
if done is None:
|
|
return False
|
|
cases = _side_cases(view.station_row(meta, done[0], factors), req.side)
|
|
if req.condition:
|
|
cases = {k: v for k, v in cases.items() if k == req.condition}
|
|
need = (lambda c: c["required"]) if req.target_fs is None else (lambda c: req.target_fs)
|
|
return bool(cases) and all(
|
|
c["fs"] is not None and need(c) is not None and c["fs"] >= need(c)
|
|
for c in cases.values()
|
|
)
|
|
|
|
def passes(n: float) -> bool:
|
|
done = run(n) # 원본과 같은 방법 · 조건(방법을 줄이면 탐색 씨앗이 달라 값이 바뀜)
|
|
if done is not None:
|
|
reached.append(n)
|
|
return meets(done)
|
|
|
|
search = None
|
|
length = None
|
|
if req.search and supports:
|
|
start = max([2.0] + [s["bond_length_m"] + 0.5 for s in supports if s["type"] == "anchor"])
|
|
if engine:
|
|
found = whatif.search_ratio(
|
|
lambda L: meets(run(assume["ratio"], L)),
|
|
start,
|
|
max_ratio=SUPPORT_MAX_LEN,
|
|
step=SUPPORT_LEN_STEP,
|
|
)
|
|
else:
|
|
found = {"found": None, "tried": []}
|
|
search = {**found, "by": "length", "longest": SUPPORT_MAX_LEN}
|
|
length = found["found"] or SUPPORT_MAX_LEN
|
|
elif req.search:
|
|
top = whatif.gentlest_ratio(design, ground, assume, before_ratio) or before_ratio
|
|
search = {**whatif.search_ratio(passes, before_ratio, max_ratio=top), "gentlest": top}
|
|
# 못 찾으면 원지반을 만난 가장 완만한 기울기로 보여 줌
|
|
assume["ratio"] = search["found"] or max(reached, default=before_ratio)
|
|
assume["ratio"] = round(assume["ratio"], 4)
|
|
done = run(assume["ratio"], length)
|
|
if done is None:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"가정 비탈(1:{assume['ratio']:g} · 소단 포함)이 해석 폭 안에서 원지반을 못 만납니다"
|
|
" — 자연 비탈보다 완만.",
|
|
)
|
|
out, variant = done
|
|
after = view.station_row(meta, out, factors)
|
|
if supports:
|
|
assume["supports"] = _sized(supports, length)
|
|
record = {
|
|
"type": "support" if supports else "slope",
|
|
"station_key": station_key,
|
|
"chainage_m": row["chainage_m"],
|
|
"label": row.get("label"),
|
|
"base_fp": row["fp"],
|
|
"assume": {**assume, "ratio_before": before_ratio},
|
|
"search": search and {**search, **goal},
|
|
"before": _side_cases(row, req.side),
|
|
"after": _side_cases(after, req.side),
|
|
# 엔진 고리 전 = 네일 · 앵커 힘 없음(결과 = 보강 없는 재해석 · 화면은 그림만)
|
|
"support_forces": _forces(out, req.side) if supports else None,
|
|
"supports_pending": bool(supports) and not engine,
|
|
"computed_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
|
|
"elapsed_s": round(time.perf_counter() - t0, 2),
|
|
}
|
|
shown = _section(ctx, variant, after, out)
|
|
write_derived(root, KIND_VIEW, key, shown, keep=KEEP)
|
|
write_derived(root, KIND, key, record, keep=KEEP)
|
|
return {**record, "view": shown}
|
|
|
|
|
|
def _material_at(model: dict[str, Any], x: float, z: float, skip: str = "wall") -> str | None:
|
|
pt = Point(x, z)
|
|
for region in model["regions"]:
|
|
if region["material_id"] != skip and Polygon(region["polygon"]).buffer(1e-6).contains(pt):
|
|
return region["material_id"]
|
|
return None
|
|
|
|
|
|
def _surface_z(surface: list, x: float) -> float:
|
|
pts = sorted(surface)
|
|
for (xa, za), (xb, zb) in zip(pts, pts[1:]):
|
|
if xa <= x <= xb:
|
|
return za + (zb - za) * ((x - xa) / (xb - xa) if xb > xa else 0.0)
|
|
return pts[0][1] if x < pts[0][0] else pts[-1][1]
|
|
|
|
|
|
def run_wallcheck(root: Path, station_key: str, req: WallRequest) -> dict[str, Any]:
|
|
ctx = _context(root, station_key)
|
|
if not ctx["walls"]:
|
|
raise HTTPException(status_code=404, detail="그 측점에 벽이 없습니다.")
|
|
row = ctx["row"]
|
|
seismic = ctx["ground"].get("seismic") or {}
|
|
kh = (
|
|
req.kh
|
|
if req.kh is not None
|
|
else float(seismic.get("kh") or 0.0) * bool(seismic.get("enabled"))
|
|
)
|
|
params = {"method": req.method, "water_ratio": req.water_ratio, "kh": kh}
|
|
key = derived_key(code_hash(), row["fp"], "wall", params)
|
|
cached = read_derived(root, KIND, key)
|
|
if cached:
|
|
return cached["value"]
|
|
kw = {k: ctx["kw"][k] for k in ("materials", "layers", "ground", "road_surcharge_kpa", "walls")}
|
|
model = section_model_from_b06(ctx["cross"], ctx["design"], **kw)
|
|
mats = {m["id"]: m for m in model["materials"]}
|
|
roles = {s["side"]: s["role"] for s in row["sides"]}
|
|
walls = []
|
|
for wall in ctx["walls"]:
|
|
poly = Polygon(wall["polygon"])
|
|
side = _wall_side(wall)
|
|
role = roles.get(side, "fill")
|
|
outward = 1 if side == "left" else -1
|
|
back = -outward if role == "fill" else outward # 쌓기 = 노면 쪽 흙 · 깎기 = 산 쪽 흙
|
|
x0, z0, x1, z1 = poly.bounds
|
|
H = z1 - z0
|
|
heel = x1 if back > 0 else x0
|
|
toe = x0 if back > 0 else x1
|
|
fill_id = _material_at(model, heel + back * 0.3, z0 + 0.5 * H) or "fill"
|
|
found_id = _material_at(model, 0.5 * (x0 + x1), z0 - 0.2) or fill_id
|
|
backfill, foundation = mats.get(fill_id, mats.get("fill")), mats.get(found_id)
|
|
surface = model["surface"]
|
|
dz = _surface_z(surface, heel + back * max(H, 1.0)) - _surface_z(surface, heel)
|
|
beta = math.degrees(math.atan2(dz, max(H, 1.0)))
|
|
beta = min(max(beta, 0.0), float(backfill["phi"]) - 1.0)
|
|
common = {
|
|
"backfill": backfill,
|
|
"foundation": foundation,
|
|
"method": req.method,
|
|
"beta_deg": beta,
|
|
"surcharge_kpa": ctx["kw"]["road_surcharge_kpa"] if role == "fill" else 0.0,
|
|
"water_ratio": req.water_ratio,
|
|
"wall_gamma": float(WALL_MATERIAL["gamma_t"]),
|
|
"embed_m": max(_surface_z(surface, toe - back * 0.05) - z0, 0.0),
|
|
}
|
|
cases = [wallcheck.check_wall(wall["polygon"], back, **common)]
|
|
if kh > 0:
|
|
cases.append(wallcheck.check_wall(wall["polygon"], back, kh=kh, **common))
|
|
walls.append(
|
|
{
|
|
"name": wall.get("name"),
|
|
"kind": wall.get("kind"),
|
|
"side": side,
|
|
"role": role,
|
|
"backfill": fill_id,
|
|
"foundation": found_id,
|
|
"cases": cases,
|
|
}
|
|
)
|
|
record = {
|
|
"type": "wall",
|
|
"station_key": station_key,
|
|
"chainage_m": row["chainage_m"],
|
|
"label": row.get("label"),
|
|
"base_fp": row["fp"],
|
|
"params": params,
|
|
"walls": walls,
|
|
"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"):
|
|
saved = read_derived(root, KIND, path.stem)
|
|
value = (saved or {}).get("value")
|
|
if not value or value.get("base_fp") not in fps:
|
|
continue
|
|
slot = (value["station_key"], value["type"], (value.get("assume") or {}).get("side"))
|
|
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["type"]))
|
|
|
|
|
|
@router.post("/{project_id}/stab/whatif/{station_key}", response_model=None)
|
|
async def post_whatif(project_id: UUID, station_key: str, payload: WhatIfRequest) -> dict:
|
|
root, _ = await e02._project(project_id)
|
|
return await asyncio.to_thread(run_whatif, root, station_key, payload)
|
|
|
|
|
|
@router.post("/{project_id}/stab/wallcheck/{station_key}", response_model=None)
|
|
async def post_wallcheck(project_id: UUID, station_key: str, payload: WallRequest) -> dict:
|
|
root, _ = await e02._project(project_id)
|
|
return await asyncio.to_thread(run_wallcheck, root, station_key, payload)
|
|
|
|
|
|
@router.get("/{project_id}/stab/whatif", response_model=None)
|
|
async def get_whatif(project_id: UUID) -> dict:
|
|
root, _ = await e02._project(project_id)
|
|
return {"records": await asyncio.to_thread(records, root)}
|