eomsangdonandClaude Opus 5.5
cfd1e3eb2c
- 엔진 = 미리보기(지형 경사 가짜 래스터) · 533-1 · 2 결과 꼴이 오면 run_engine 만 갈아 끼움 - 권한 = protected_with_company(지정받은 회사 GET 만) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
276 lines
9.9 KiB
Python
276 lines
9.9 KiB
Python
"""E04 3D 해석 한 번 — 입력 모으기 · 입력 지문 · 엔진 부르기 · 파생 저장본 (PLAN 533-3).
|
|
|
|
계산 자리 = ③ 서버 단독. 엔진(탐색 · 기둥 솔버 · 합성 지형) 주인 = 533-1 · 2 — 여기는 부르기만.
|
|
저장본 = E 프로젝트 `E04_Stab3D/results/<구간 열쇠>.json` · 입력 지문(복사본 지문 · E02 지반 ·
|
|
요청 · 엔진 판)이 같으면 다시 셈하지 않음.
|
|
|
|
래스터 꼴(화면 `logic/E04_Stab3D_Raster.ts` 와 같은 열쇠):
|
|
grid = {x0, y0, dx, nx, ny} — 셀 (iy, ix) 중심 = (x0 + ix·dx, y0 + iy·dx) · 줄 우선 평탄 배열
|
|
z = 합성 지표(원지형 + 노선면) · f = 셀별 최소 안전율 · v = 그 임계 체적 m³ · m = 방법(0 Bishop
|
|
· 1 Janbu 보정) · 닿지 않은 셀 = null
|
|
best = 전체 최소 활동면 {x, y, z(중심), r, ratio, azimuth_deg, f, volume, method, ix, iy}
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import time
|
|
from collections.abc import Callable
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
|
|
from common_util.common_util_surface_sampler import DtmGridSampler
|
|
from E01_StabRegister import E01_StabRegister_Source as source
|
|
|
|
ENGINE_VERSION = "e04-preview-1" # 엔진 판 — 바뀌면 지문이 달라져 다시 셈
|
|
STAGE_DIR = "E04_Stab3D"
|
|
GROUND_FILE = Path("E02_StabGround") / "ground.json"
|
|
SECTION_LENGTH_M = 100.0 # 자동 구간 길이
|
|
MAX_SECTION_M = 500.0 # 한 번에 셀 구간 상한(시간 · 메모리)
|
|
METHOD_NAMES = ("bishop", "janbu_corrected")
|
|
|
|
Progress = Callable[[int, str], None]
|
|
StopFlag = Callable[[], bool]
|
|
|
|
|
|
def results_dir(project_root: Path) -> Path:
|
|
return Path(project_root) / STAGE_DIR / "results"
|
|
|
|
|
|
def section_key(start_m: float, end_m: float) -> str:
|
|
return f"{start_m:.2f}_{end_m:.2f}"
|
|
|
|
|
|
def auto_sections(stations: list[dict[str, Any]]) -> list[dict[str, float]]:
|
|
"""측점을 앞에서부터 SECTION_LENGTH_M 이상 묶음 — 경계 = 측점 · 꼬리가 짧으면 앞 구간에 붙임."""
|
|
chainages = sorted({float(s["chainage_m"]) for s in stations})
|
|
out: list[dict[str, float]] = []
|
|
start = chainages[0] if chainages else 0.0
|
|
for ch in chainages[1:]:
|
|
if ch - start >= SECTION_LENGTH_M - 1e-6:
|
|
out.append({"start_m": start, "end_m": ch})
|
|
start = ch
|
|
if chainages and start < chainages[-1]:
|
|
if out and chainages[-1] - start < SECTION_LENGTH_M / 2:
|
|
out[-1]["end_m"] = chainages[-1]
|
|
else:
|
|
out.append({"start_m": start, "end_m": chainages[-1]})
|
|
return out
|
|
|
|
|
|
def read_ground(project_root: Path) -> dict[str, Any] | None:
|
|
"""E02 작업본(없으면 None = 기본값)."""
|
|
path = Path(project_root) / GROUND_FILE
|
|
return json.loads(path.read_text(encoding="utf-8")) if path.is_file() else None
|
|
|
|
|
|
def fingerprint(
|
|
source_fingerprint: str, ground: dict[str, Any] | None, settings: dict[str, Any]
|
|
) -> str:
|
|
body = json.dumps(
|
|
{
|
|
"source": source_fingerprint,
|
|
"ground": ground,
|
|
"settings": settings,
|
|
"engine": ENGINE_VERSION,
|
|
},
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
)
|
|
return hashlib.sha1(body.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def section_stations(stations: list[dict[str, Any]], start_m: float, end_m: float) -> list[dict]:
|
|
return [s for s in stations if start_m - 1e-6 <= float(s["chainage_m"]) <= end_m + 1e-6]
|
|
|
|
|
|
def bbox_of(
|
|
stations: list[dict[str, Any]], half_width_m: float
|
|
) -> tuple[float, float, float, float]:
|
|
xs = [float(s["center_x"]) for s in stations]
|
|
ys = [float(s["center_y"]) for s in stations]
|
|
return (
|
|
min(xs) - half_width_m,
|
|
max(xs) + half_width_m,
|
|
min(ys) - half_width_m,
|
|
max(ys) + half_width_m,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 엔진 — 533-1 · 2 결과 꼴이 오기 전 = 미리보기(지형 경사로 만든 가짜 래스터 · 틀 · 배치 확인용)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _preview_engine(
|
|
root: Path,
|
|
stations: list[dict[str, Any]],
|
|
settings: dict[str, Any],
|
|
on_progress: Progress,
|
|
should_stop: StopFlag,
|
|
) -> dict[str, Any]:
|
|
dem = DtmGridSampler.open_cached(source.terrain_grid_path(root))
|
|
x0, x1, y0, y1 = bbox_of(stations, float(settings["box_half_width_m"]))
|
|
dx = float(settings["cell_m"])
|
|
xs = np.arange(x0, x1 + dx / 2, dx)
|
|
ys = np.arange(y0, y1 + dx / 2, dx)
|
|
xx, yy = np.meshgrid(xs, ys)
|
|
z, valid = dem.sample_xy(np.column_stack([xx.ravel(), yy.ravel()]))
|
|
z = np.where(valid, z, np.nan).reshape(xx.shape)
|
|
gy, gx = np.gradient(z, dx)
|
|
slope = np.hypot(gx, gy)
|
|
f = np.full(z.shape, np.nan)
|
|
rows = z.shape[0]
|
|
for iy in range(rows):
|
|
if should_stop():
|
|
raise InterruptedError
|
|
f[iy] = 0.8 + 2.4 / (1.0 + 2.5 * slope[iy] ** 2)
|
|
if iy % max(1, rows // 20) == 0:
|
|
on_progress(int(100 * iy / rows), f"미리보기 {iy}/{rows} 줄")
|
|
volume = np.where(np.isfinite(f), 50.0 + 400.0 * slope**2, np.nan)
|
|
iy, ix = np.unravel_index(int(np.nanargmin(f)), f.shape)
|
|
r = 10.0
|
|
best = {
|
|
"x": float(xs[ix]),
|
|
"y": float(ys[iy]),
|
|
"z": float(z[iy, ix] + 0.6 * r),
|
|
"r": r,
|
|
"ratio": 1.0,
|
|
"azimuth_deg": 0.0,
|
|
"f": float(f[iy, ix]),
|
|
"volume": float(volume[iy, ix]),
|
|
"method": 0,
|
|
}
|
|
return {
|
|
"grid": {"x0": float(xs[0]), "y0": float(ys[0]), "dx": dx, "nx": len(xs), "ny": len(ys)},
|
|
"z": z,
|
|
"f": f,
|
|
"v": volume,
|
|
"m": np.where(np.isfinite(f), 0.0, np.nan),
|
|
"best": best,
|
|
"trials": int(np.isfinite(f).sum()),
|
|
"preview": True,
|
|
}
|
|
|
|
|
|
def run_engine(
|
|
root: Path,
|
|
stations: list[dict[str, Any]],
|
|
ground: dict[str, Any] | None,
|
|
settings: dict[str, Any],
|
|
on_progress: Progress,
|
|
should_stop: StopFlag,
|
|
) -> dict[str, Any]:
|
|
"""엔진 한 번 → {grid, z, f, v, m(넘파이 (ny, nx)), best, trials}."""
|
|
return _preview_engine(root, stations, settings, on_progress, should_stop)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 저장본
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _flat(values: np.ndarray, digits: int) -> list[float | None]:
|
|
return [None if not math.isfinite(v) else round(float(v), digits) for v in values.ravel()]
|
|
|
|
|
|
def _nearest_chainage(stations: list[dict[str, Any]], x: float, y: float) -> float:
|
|
return min(
|
|
stations, key=lambda s: (float(s["center_x"]) - x) ** 2 + (float(s["center_y"]) - y) ** 2
|
|
)["chainage_m"]
|
|
|
|
|
|
def pack(
|
|
raw: dict[str, Any],
|
|
stations: list[dict[str, Any]],
|
|
settings: dict[str, Any],
|
|
finger: str,
|
|
elapsed_s: float,
|
|
) -> dict[str, Any]:
|
|
"""엔진 결과 → 저장본(JSON) — 요약은 best 에서."""
|
|
grid = raw["grid"]
|
|
best = dict(raw["best"])
|
|
best["ix"] = int(round((best["x"] - grid["x0"]) / grid["dx"]))
|
|
best["iy"] = int(round((best["y"] - grid["y0"]) / grid["dx"]))
|
|
f = np.asarray(raw["f"], dtype=float)
|
|
return {
|
|
"fingerprint": finger,
|
|
"engine": ENGINE_VERSION,
|
|
"preview": bool(raw.get("preview")),
|
|
"settings": settings,
|
|
"computed_at": datetime.now().isoformat(timespec="seconds"),
|
|
"elapsed_s": round(elapsed_s, 2),
|
|
"grid": grid,
|
|
"z": _flat(np.asarray(raw["z"], dtype=float), 2),
|
|
"f": _flat(f, 3),
|
|
"v": _flat(np.asarray(raw["v"], dtype=float), 1),
|
|
"m": _flat(np.asarray(raw["m"], dtype=float), 0),
|
|
"best": best,
|
|
"summary": {
|
|
"start_m": settings["start_m"],
|
|
"end_m": settings["end_m"],
|
|
"f_min": round(float(best["f"]), 3),
|
|
"volume": round(float(best["volume"]), 1),
|
|
"method": METHOD_NAMES[int(best.get("method") or 0)],
|
|
"chainage_m": float(_nearest_chainage(stations, best["x"], best["y"])),
|
|
"cells": int(np.isfinite(f).sum()),
|
|
"trials": int(raw.get("trials") or 0),
|
|
},
|
|
}
|
|
|
|
|
|
def save(project_root: Path, key: str, result: dict[str, Any]) -> None:
|
|
folder = results_dir(project_root)
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
tmp = folder / f"{key}.json.tmp"
|
|
tmp.write_text(json.dumps(result, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
|
|
tmp.replace(folder / f"{key}.json")
|
|
|
|
|
|
def load(project_root: Path, key: str) -> dict[str, Any] | None:
|
|
path = results_dir(project_root) / f"{key}.json"
|
|
return json.loads(path.read_text(encoding="utf-8")) if path.is_file() else None
|
|
|
|
|
|
def saved_summaries(project_root: Path) -> list[dict[str, Any]]:
|
|
"""저장본 요약(래스터 빼고) — 구간 목록 색 · 판정용."""
|
|
out = []
|
|
for path in sorted(results_dir(project_root).glob("*.json")):
|
|
doc = json.loads(path.read_text(encoding="utf-8"))
|
|
out.append(
|
|
{
|
|
"key": path.stem,
|
|
"fingerprint": doc["fingerprint"],
|
|
"settings": doc["settings"],
|
|
"preview": doc.get("preview", False),
|
|
**doc["summary"],
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def analyze(
|
|
project_root: Path,
|
|
root: Path,
|
|
settings: dict[str, Any],
|
|
finger: str,
|
|
on_progress: Progress,
|
|
should_stop: StopFlag,
|
|
) -> dict[str, Any]:
|
|
"""한 구간 3D 해석 → 저장본 쓰고 요약 돌려줌."""
|
|
started = time.perf_counter()
|
|
stations = source.stations(root)
|
|
picked = section_stations(stations, settings["start_m"], settings["end_m"])
|
|
if len(picked) < 2:
|
|
raise ValueError("구간 안 측점이 둘 미만입니다.")
|
|
ground = read_ground(project_root)
|
|
raw = run_engine(root, picked, ground, settings, on_progress, should_stop)
|
|
result = pack(raw, picked, settings, finger, time.perf_counter() - started)
|
|
save(project_root, section_key(settings["start_m"], settings["end_m"]), result)
|
|
return result["summary"]
|