feat(536-2): E07 광역 자연사면 위험 엔진 · 길 · E05 광역요약 시트

- 엔진 E07_StabArea_Engine — DTM 고른 격자 · Horn 경사 · D8 집수면적(B04 웅덩이 채움 · 수신 셀 재사용) · 무한사면 Fs(포화비 m) · SINMAP 안정 지수(c · tanφ 고른 분포 닫힌 꼴) · 6 등급 · 노선 띠 · 측점표
- 길 E07_StabArea_Router — POST 해석(뒤에서 · 진행률 · 지문 같으면 cached) · GET 기본값(E02 붕적토 · 범위 위 +5 kPa · +3°) · 파생 저장본 _derived/area.json(격자 base64)
- E05 보고서 s7 광역요약 — 조건 · 등급별 면적 · 위험 측점 줄 · 양식 · 자리표 덧붙임 · 해석 지문에 E07 저장본
- 시험 test_536_2 새로 · test_534_3 두 파일은 시트 7 · PDF 쪽 +1 로 고침(광역요약 시트 더함)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-25 21:25:33 +09:00
co-authored by Claude Opus 5.5
parent 17810f00ab
commit 0fd6437628
12 changed files with 1452 additions and 13 deletions
+78
View File
@@ -0,0 +1,78 @@
"""E05 보고서 s7 광역요약 — E07 광역 자연사면 위험 저장본 → 시트 (PLAN 536-2).
칸 자리 = `resources/master_template/report/stab_report_slots.md` s7.
E07 저장본이 없으면 「광역 해석 전」.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
from E07_StabArea.E07_StabArea_Engine import CLASS_KEYS, RISK_CLASS
from E07_StabArea.E07_StabArea_Router import RESULT, load
CLASS_NAMES = ("", "안정", "보통 안정", "준안정", "위험 하한", "위험 상한", "불안정")
def stamp(project_root: Path) -> list[int] | None:
"""해석 지문용 — 저장본 크기 · 고친 때."""
path = Path(project_root) / RESULT
return [path.stat().st_size, path.stat().st_mtime_ns] if path.is_file() else None
def read(project_root: Path) -> dict[str, Any] | None:
"""저장본에서 보고서에 쓰는 것만(격자 배열 뺌)."""
doc = load(project_root)
if not doc:
return None
return {k: doc[k] for k in ("settings", "computed_at", "stations", "summary")}
def _set(sheet: dict[str, Any], a1: str, value: Any, style: int | None = None) -> None:
cell = dict(sheet["칸"].get(a1) or {})
cell["값"] = round(value, 6) if isinstance(value, float) else value
if style is not None:
cell["서식"] = style
sheet["칸"][a1] = cell
def fill_area(sheet: dict[str, Any], area: dict[str, Any] | None, labels: dict[float, str]) -> None:
"""s7 — 조건 넷 · 등급별 면적 여섯 · 위험 측점(22 행부터 한 줄씩)."""
if not area:
_set(sheet, "B5", "광역 해석 전 — E07 광역에서 [해석] 뒤 [보고서 갱신]")
return
s, summary = area["settings"], area["summary"]
_set(sheet, "B5", f"중심선 ± {s['band_m']:g} m · 격자 {s['cell_m']:g} m")
_set(
sheet,
"B6",
f"γt {s['gamma_t']:g} · γsat {s['gamma_sat']:g} kN/m³ · c {s['c']:g}~{s['c_hi']:g} kPa · "
f"φ {s['phi']:g}~{s['phi_hi']:g}° · 뿌리 점착 {s['c_root']:g} kPa · "
f"토층 {s['soil_depth_m']:g} m",
)
rain = (
"지표 포화(m = 1)"
if s["scenario"] == "saturated"
else f"강우 {s['rain_mm_h']:g} mm/h · 투수계수 {s['k_m_s']:g} m/s(정상류 포화비)"
)
_set(sheet, "B7", rain)
_set(sheet, "B8", str(area["computed_at"]).replace("T", " "))
for k in range(1, 7):
_set(sheet, f"C{11 + k}", float(summary["areas_m2"][CLASS_KEYS[k]]))
rows = [r for r in area["stations"] if r.get("worst_class", 0) >= RISK_CLASS]
template = {col: dict(sheet["칸"].get(f"{col}22") or {}) for col in "ABCDEFG"}
if not rows:
_set(sheet, "A22", "위험 측점 없음")
for i, row in enumerate(rows):
r = 22 + i
for col, cell in template.items():
sheet["칸"][f"{col}{r}"] = dict(cell)
ch = float(row["chainage_m"])
_set(sheet, f"A{r}", row.get("label") or labels.get(round(ch, 3)) or f"{ch:.2f}")
_set(sheet, f"B{r}", ch)
_set(sheet, f"C{r}", float(row["fs_min"]))
_set(sheet, f"D{r}", float(row["si_min"]))
_set(sheet, f"E{r}", CLASS_NAMES[int(row["worst_class"])])
_set(sheet, f"F{r}", float(row["risk_area_m2"]))
_set(sheet, f"G{r}", float(row["slope_max_deg"]))
+8 -3
View File
@@ -1,12 +1,13 @@
"""E05 안정성평가 보고서 빌더(요구 347-5 · 351 · 요구서 4-3 · PLAN 534-3).
입력 = E02 지반 조건(`ground.json`) · E03 2D 결과 요약(`_derived/result2d.json`) · 측점별 저장본 ·
대표 단면(`representative.json`) · E04 3D 저장본 요약(있으면) · E01 복사본(측점 · 횡단 · 노선).
대표 단면(`representative.json`) · E04 3D 저장본 요약(있으면) · E07 광역 저장본(있으면) ·
E01 복사본(측점 · 횡단 · 노선).
양식 = `resources/master_template/report/stab_report.json`
칸 자리 = 같은 폴더 `stab_report_slots.md`.
몫: 사용자(표지 s1 · 개요 s2) = 처음 만들 때만 채움 ·
프로그램(입력조건 s3 · 측점표 s4 · 대표단면 s5 · 3D요약 s6)
프로그램(입력조건 s3 · 측점표 s4 · 대표단면 s5 · 3D요약 s6 · 광역요약 s7)
= [보고서 갱신] 때 양식에서 통째로 다시 씀.
판정 글 칸 = 엔진 판정 값 + N.G 서식 10 · O.K 서식 11 을 직접 붙임
(인쇄 HTML · xlsx 는 조건부 서식을 안 먹음).
@@ -30,6 +31,7 @@ from E01_StabRegister import E01_StabRegister_Source as source
from E02_StabGround import E02_StabGround_Model as ground_model
from E03_StabSection.E03_StabSection_Engine_Model import final_surface
from E04_Stab3D import E04_Stab3D_Run as run3d
from E05_StabReport import E05_StabReport_Area as area
from E05_StabReport import E05_StabReport_Figure as figure
ROOT = Path(__file__).resolve().parents[1]
@@ -37,7 +39,7 @@ FORM_PATH = ROOT / "resources" / "master_template" / "report" / "stab_report.jso
STAGE = "E05_StabReport"
REPORT, META = "report.json", "report_meta.json"
USER_SHEETS = ("s1", "s2")
PROGRAM_SHEETS = ("s3", "s4", "s5", "s6")
PROGRAM_SHEETS = ("s3", "s4", "s5", "s6", "s7")
BLOCK = 28 # 대표단면 한 쪽 = 28 행
E03 = Path("E03_StabSection")
RESULT2D = E03 / "_derived" / "result2d.json"
@@ -128,6 +130,7 @@ def analysis_fingerprint(project_root: Path) -> str:
]
if folder.is_dir()
else [],
"area": area.stamp(project_root),
"form": hashlib.sha1(FORM_PATH.read_bytes()).hexdigest(),
}
text = json.dumps(body, ensure_ascii=False, sort_keys=True, default=str)
@@ -181,6 +184,7 @@ def gather(project_root: Path) -> dict[str, Any]:
"representatives": reps,
"ground": ground,
"sections3d": sorted(sections3d, key=lambda s: s["start_m"]),
"area": area.read(project_root),
"fingerprint": analysis_fingerprint(project_root),
}
@@ -571,6 +575,7 @@ def build_program(book: dict[str, Any], inputs: dict[str, Any]) -> list[tuple[st
fill_stations(by["s4"], inputs)
jobs = fill_sections(by["s5"], inputs)
jobs.append(fill_3d(by["s6"], inputs))
area.fill_area(by["s7"], inputs["area"], _labels(inputs["stations"]))
return jobs
+283
View File
@@ -0,0 +1,283 @@
"""E07 광역 자연사면 위험 — 무한사면 안전율 · SINMAP 안정 지수 (요구 345 · 351 · PLAN 536-2).
순수 계산(storage 모름) — D 예정노선 재해 안전성도 같은 엔진을 씀(한 벌).
근거 = 지식DB `original/예정노선/02` §1.1 · §2 ① · `original/안정성평가/04`(붕적토 범위).
지형 DEM → 고른 격자(cell_m) · 경사 β(Horn 3×3) · D8 흐름(웅덩이 채움 · 수신 셀 = B04 배수유역
함수 재사용) · 단위 집수면적 a = 상류 셀 수 × cell²/cell(등고선 폭 = cell)
포화비 m = 1(지표 포화 · 기본) | min(R·a / (T·sinβ), 1) · T = K·z(정상류 · SINMAP · SHALSTAB)
안전율 Fs = (c + c_r + [γt z(1−m) + (γsat−γw) m z] cos²β tanφ)
/ ([γt z(1−m) + γsat m z] sinβ cosβ)
안정 지수 SI(SINMAP · Pack 1998) — c ∈ [c, c_hi] · tanφ ∈ [tanφ, tanφ_hi] 고른 분포
Fs_min ≥ 1 → SI = Fs_min · Fs_max < 1 → SI = 0
· 그 사이 → SI = P(Fs > 1)(두 고른 분포 합 · 닫힌 꼴)
등급 1 안정(SI ≥ 1.5) · 2 보통 안정(≥ 1.25) · 3 준안정(≥ 1.0) · 4 위험 하한(≥ 0.5)
· 5 위험 상한(> 0) · 6 불안정(SI = 0) · 0 = 자료 없음
격자 꼴 = 행 0 이 남쪽(y0) · 셀 (iy, ix) 중심 = (x0 + ix·dx, y0 + iy·dx) — E04 래스터와 같음.
"""
from __future__ import annotations
import math
from dataclasses import dataclass
from types import SimpleNamespace
from typing import Any
import numpy as np
from scipy.interpolate import RegularGridInterpolator
from scipy.spatial import cKDTree
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import (
compute_receivers,
condition_surface,
)
GAMMA_W = 9.81 # kN/m³
SI_EDGES = (1.5, 1.25, 1.0, 0.5) # 등급 1~4 아래 경계 · 5 = (0, 0.5) · 6 = 0
CLASS_KEYS = ("", "stable", "moderate", "quasi", "lower", "upper", "defended")
RISK_CLASS = 4 # 이 등급부터 위험(SI < 1)
FLAT_SLOPE_RAD = 1e-4 # 이보다 평평하면 활동 없음(Fs = ∞)
@dataclass(frozen=True)
class AreaParams:
"""한 번 해석 조건 — 단위 kN/m³ · kPa · ° · m · mm/h · m/s."""
gamma_t: float
gamma_sat: float
c: float # 점착력 아래(결정 Fs · SI 범위 아래)
c_hi: float # SI 범위 위
phi: float
phi_hi: float
c_root: float = 0.0 # 뿌리 점착(나지 = 0)
soil_depth_m: float = 1.5
scenario: str = "saturated" # saturated = 지표 포화 m = 1 · rain = 정상류 강우
rain_mm_h: float = 30.0
k_m_s: float = 1e-4 # 포화 투수계수
# ---------------------------------------------------------------------------
# 지형
# ---------------------------------------------------------------------------
def resample_dem(dem: dict[str, Any], cell_m: float) -> tuple[np.ndarray, float, float]:
"""DTM npz(x · y 오름 축 · z (ny, nx) · valid_mask) → 고른 격자 (z, x0, y0) · 밖 = NaN."""
x = np.asarray(dem["x"], float)
y = np.asarray(dem["y"], float)
z = np.where(np.asarray(dem["valid_mask"], bool), np.asarray(dem["z"], float), np.nan)
nx = int(math.floor((x[-1] - x[0]) / cell_m)) + 1
ny = int(math.floor((y[-1] - y[0]) / cell_m)) + 1
gx = x[0] + np.arange(nx) * cell_m
gy = y[0] + np.arange(ny) * cell_m
yy, xx = np.meshgrid(gy, gx, indexing="ij")
f = RegularGridInterpolator((y, x), z, bounds_error=False, fill_value=np.nan)
return f(np.column_stack([yy.ravel(), xx.ravel()])).reshape(ny, nx), float(x[0]), float(y[0])
def slope_rad(z: np.ndarray, cell_m: float) -> np.ndarray:
"""Horn 3×3 경사(rad) — 이웃에 NaN 이 있거나 격자 끝이면 NaN."""
p = np.pad(z, 1, constant_values=np.nan)
def at(dy: int, dx: int) -> np.ndarray:
return p[1 + dy : 1 + dy + z.shape[0], 1 + dx : 1 + dx + z.shape[1]]
gx = ((at(-1, 1) + 2 * at(0, 1) + at(1, 1)) - (at(-1, -1) + 2 * at(0, -1) + at(1, -1))) / (
8 * cell_m
)
gy = ((at(1, -1) + 2 * at(1, 0) + at(1, 1)) - (at(-1, -1) + 2 * at(-1, 0) + at(-1, 1))) / (
8 * cell_m
)
return np.arctan(np.hypot(gx, gy))
def catchment_area(z: np.ndarray, cell_m: float) -> np.ndarray:
"""D8 단위 집수면적 a(m) = 자기 포함 상류 셀 수 × cell. 밖 = NaN.
ponytail: 지형 격자 끝 밖 상류는 모름(복사 DTM = 노선 띠) — 띠 끝 셀은 a 가 작게 나옴 ·
넓은 DEM(D 예정노선 등)이면 그대로 맞음.
"""
filled, valid = condition_surface(z.astype(np.float32))
spec = SimpleNamespace(n_rows=z.shape[0], n_cols=z.shape[1], cell_m=cell_m)
receiver, _ = compute_receivers(spec, filled, valid)
flat = filled.ravel()
order = np.flatnonzero(valid.ravel())
order = order[np.argsort(-flat[order], kind="stable")].tolist() # 높은 데서 낮은 데로
acc = valid.ravel().astype(float).tolist()
recv = receiver.tolist()
for i in order:
r = recv[i]
if r != i:
acc[r] += acc[i]
out = np.asarray(acc).reshape(z.shape) * cell_m
return np.where(valid, out, np.nan)
# ---------------------------------------------------------------------------
# 무한사면 · SINMAP
# ---------------------------------------------------------------------------
def wetness(beta: np.ndarray, a: np.ndarray, p: AreaParams) -> np.ndarray:
"""포화비 m(0~1) — 지표 포화 = 1 · 강우 = min(R·a/(T sinβ), 1)."""
if p.scenario == "saturated":
return np.ones_like(beta)
recharge = p.rain_mm_h / 1000.0 / 3600.0 # m/s
transmissivity = p.k_m_s * p.soil_depth_m # m²/s
with np.errstate(divide="ignore", invalid="ignore"):
m = recharge * a / (transmissivity * np.sin(beta))
return np.clip(np.nan_to_num(m, nan=1.0, posinf=1.0), 0.0, 1.0)
def _terms(beta: np.ndarray, m: np.ndarray, p: AreaParams) -> tuple[np.ndarray, np.ndarray]:
"""Fs = (c + c_r)·A + tanφ·B 의 A · B(평평한 셀 = ∞)."""
z = p.soil_depth_m
cos, sin = np.cos(beta), np.sin(beta)
normal = (p.gamma_t * z * (1 - m) + (p.gamma_sat - GAMMA_W) * m * z) * cos**2
shear = (p.gamma_t * z * (1 - m) + p.gamma_sat * m * z) * sin * cos
flat = beta < FLAT_SLOPE_RAD
with np.errstate(divide="ignore", invalid="ignore"):
a = np.where(flat, np.inf, 1.0 / shear)
b = np.where(flat, np.inf, normal / shear)
return a, b
def infinite_slope_fs(beta: np.ndarray, m: np.ndarray, p: AreaParams) -> np.ndarray:
"""결정 안전율(c · φ = 범위 아래 = 보수) — 지식DB 02 §1.1 식."""
a, b = _terms(beta, m, p)
with np.errstate(invalid="ignore"):
return np.where(
np.isinf(a), np.inf, (p.c + p.c_root) * a + math.tan(math.radians(p.phi)) * b
)
def _cdf_sum_uniform(
s: np.ndarray, a1: np.ndarray, a2: np.ndarray, b1: np.ndarray, b2: np.ndarray
) -> np.ndarray:
"""P(X + Y ≤ s) · X ~ U[a1, a2] · Y ~ U[b1, b2] 독립 — 폭 0 이면 그 변수는 상수."""
wa, wb = a2 - a1, b2 - b1
eps = 1e-12
def g(u: np.ndarray) -> np.ndarray:
return np.maximum(u, 0.0) ** 2 / 2
with np.errstate(divide="ignore", invalid="ignore"):
both = (g(s - a1 - b1) - g(s - a2 - b1) - g(s - a1 - b2) + g(s - a2 - b2)) / (wa * wb)
only_a = (s - a1 - b1) / wa
only_b = (s - a1 - b1) / wb
step = (a1 + b1 <= s).astype(float)
out = np.where(
(wa > eps) & (wb > eps), both, np.where(wa > eps, only_a, np.where(wb > eps, only_b, step))
)
return np.clip(np.nan_to_num(out, nan=0.0), 0.0, 1.0)
def stability_index(beta: np.ndarray, m: np.ndarray, p: AreaParams) -> np.ndarray:
"""SINMAP 안정 지수 — Fs_min ≥ 1 → Fs_min · Fs_max < 1 → 0 · 그 사이 → P(Fs > 1)."""
a, b = _terms(beta, m, p)
t1, t2 = math.tan(math.radians(p.phi)), math.tan(math.radians(max(p.phi_hi, p.phi)))
c1, c2 = p.c + p.c_root, max(p.c_hi, p.c) + p.c_root
flat = np.isinf(a)
a = np.where(flat, 0.0, a)
b = np.where(flat, 0.0, b)
lo, hi = c1 * a + t1 * b, c2 * a + t2 * b
prob = 1.0 - _cdf_sum_uniform(np.ones_like(a), c1 * a, c2 * a, t1 * b, t2 * b)
si = np.where(lo >= 1.0, lo, np.where(hi < 1.0, 0.0, prob))
return np.where(flat, np.inf, si)
def classify(si: np.ndarray) -> np.ndarray:
"""SI → 등급 1~6(0 = 자료 없음 · NaN)."""
cls = np.full(si.shape, 6, dtype=np.uint8)
cls[si > 0] = 5
for k, edge in enumerate(reversed(SI_EDGES)): # 0.5 → 4 · 1.0 → 3 · 1.25 → 2 · 1.5 → 1
cls[si >= edge] = 4 - k
cls[np.isnan(si)] = 0
return cls
def analyze_grid(z: np.ndarray, cell_m: float, p: AreaParams) -> dict[str, np.ndarray]:
"""고른 격자 지형 → {slope(rad) · a · m · fs · si · cls} · 밖 = NaN / 0."""
beta = slope_rad(z, cell_m)
a = catchment_area(z, cell_m) if p.scenario == "rain" else np.full(z.shape, np.nan)
m = wetness(beta, a, p)
fs = infinite_slope_fs(beta, m, p)
si = stability_index(beta, m, p)
nodata = np.isnan(beta)
fs[nodata] = np.nan
si[nodata] = np.nan
m = np.where(nodata, np.nan, m)
return {"slope": beta, "a": a, "m": m, "fs": fs, "si": si, "cls": classify(si)}
# ---------------------------------------------------------------------------
# 노선 띠 · 측점
# ---------------------------------------------------------------------------
def dense_line(line: np.ndarray, step_m: float = 0.5) -> np.ndarray:
"""중심선 [x, y, 누가] 를 step 간격으로 촘촘히(꺾인 점 사이 셀도 바른 누가)."""
out = [line[:1]]
for a, b in zip(line[:-1], line[1:]):
n = max(1, int(math.ceil(math.hypot(*(b[:2] - a[:2])) / step_m)))
out.append(a + (b - a) * np.linspace(0, 1, n + 1)[1:, None])
return np.vstack(out)
def band_cells(
shape: tuple[int, int], x0: float, y0: float, cell_m: float, line: np.ndarray, half_m: float
) -> tuple[np.ndarray, np.ndarray]:
"""노선 띠(중심선 ± half_m) 셀 → (띠 안, 셀의 가장 가까운 중심선 누가)."""
dense = dense_line(np.asarray(line, float))
yy, xx = np.meshgrid(
y0 + np.arange(shape[0]) * cell_m, x0 + np.arange(shape[1]) * cell_m, indexing="ij"
)
dist, at = cKDTree(dense[:, :2]).query(np.column_stack([xx.ravel(), yy.ravel()]))
return (dist <= half_m).reshape(shape), dense[at, 2].reshape(shape)
def station_table(
chainages: list[float],
cell_chain: np.ndarray,
inside: np.ndarray,
out: dict[str, np.ndarray],
cell_m: float,
) -> list[dict[str, Any]]:
"""측점마다 맡은 띠 셀(이웃 측점과 중간 누가로 나눔) —
최소 Fs · SI · 가장 나쁜 등급 · 위험 면적."""
chs = np.asarray(sorted(chainages), float)
if not len(chs):
return []
mids = (chs[:-1] + chs[1:]) / 2
owner = np.searchsorted(mids, cell_chain)
have = inside & (out["cls"] > 0)
rows = []
for k, ch in enumerate(chs):
sel = have & (owner == k)
n = int(sel.sum())
if not n:
rows.append({"chainage_m": float(ch), "cells": 0})
continue
cls = out["cls"][sel]
fs = out["fs"][sel]
si = out["si"][sel]
rows.append(
{
"chainage_m": float(ch),
"cells": n,
"fs_min": round(float(np.min(fs)), 3),
"si_min": round(float(np.min(si)), 3),
"worst_class": int(cls.max()),
"risk_area_m2": round(float((cls >= RISK_CLASS).sum()) * cell_m**2, 1),
"slope_max_deg": round(float(np.degrees(np.nanmax(out["slope"][sel]))), 1),
}
)
return rows
def class_areas(cls: np.ndarray, inside: np.ndarray, cell_m: float) -> dict[str, float]:
"""띠 안 등급별 면적 m²."""
return {
CLASS_KEYS[k]: round(float(((cls == k) & inside).sum()) * cell_m**2, 1) for k in range(1, 7)
}
+337
View File
@@ -0,0 +1,337 @@
"""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-1" # 엔진 판 — 바뀌면 지문이 달라져 다시 셈
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)
c_root: float = Field(0.0, ge=0.0, le=50.0)
soil_depth_m: float = Field(1.5, ge=0.1, le=10.0)
scenario: Literal["saturated", "rain"] = "saturated"
rain_mm_h: float = Field(30.0, ge=0.0, le=300.0)
k_m_s: float = Field(1e-4, ge=1e-8, le=1e-1)
force: bool = False
@model_validator(mode="after")
def _ranges(self) -> AreaRequest:
if self.c_hi < self.c or self.phi_hi < self.phi:
raise ValueError("범위 위가 아래보다 작습니다(c · φ).")
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",
"rain_mm_h",
"k_m_s",
)
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)
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,
"risk_stations": [
r["chainage_m"] for r in rows if r.get("worst_class", 0) >= engine.RISK_CLASS
],
"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}
+7 -1
View File
@@ -20,7 +20,13 @@ from config.config_db import get_db_pool
KIND_STAGE_DIRS = {
"PLAN": ("D02_PlanBase", "D03_PlanAnalysis", "D04_PlanConfirm"),
"STABILITY": ("E02_StabGround", "E03_StabSection", "E04_Stab3D", "E05_StabReport"),
"STABILITY": (
"E02_StabGround",
"E03_StabSection",
"E04_Stab3D",
"E05_StabReport",
"E07_StabArea",
),
}
+1
View File
@@ -220,6 +220,7 @@ PROJECT_STORAGE_STAGE_DIRS = frozenset(
"E03_StabSection",
"E04_Stab3D",
"E05_StabReport",
"E07_StabArea",
}
)
+2
View File
@@ -83,6 +83,7 @@ from E01_StabRegister.E01_StabRegister_Router import router as e01_stab_register
from E02_StabGround.E02_StabGround_Router import router as e02_stab_ground_router
from E04_Stab3D.E04_Stab3D_Router import router as e04_stab_3d_router
from E05_StabReport.E05_StabReport_Router import router as e05_stab_report_router
from E07_StabArea.E07_StabArea_Router import router as e07_stab_area_router
from E03_StabSection.E03_StabSection_Router import router as e03_stab_section_router
from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router
from M02_MasterTemplete.M02_Labor_Router import router as m02_labor_router
@@ -493,6 +494,7 @@ app.include_router(d04_plan_link_router, dependencies=protected_with_company)
app.include_router(e02_stab_ground_router, dependencies=protected_with_company)
app.include_router(e04_stab_3d_router, dependencies=protected_with_company)
app.include_router(e05_stab_report_router, dependencies=protected_with_company)
app.include_router(e07_stab_area_router, dependencies=protected_with_company)
app.include_router(e03_stab_section_router, dependencies=protected_with_company)
app.include_router(b05_corridor_router, dependencies=protected_with_company)
app.include_router(b05_route_replan_router, dependencies=protected_with_company)
@@ -2600,6 +2600,484 @@
}
}
]
},
{
"id": "s7",
"이름": "광역요약",
"칸": {
"A1": {
"값": "광역 자연사면 위험 요약",
"서식": 1
},
"B1": {
"서식": 1
},
"C1": {
"서식": 1
},
"D1": {
"서식": 1
},
"E1": {
"서식": 1
},
"F1": {
"서식": 1
},
"G1": {
"서식": 1
},
"H1": {
"서식": 1
},
"A2": {
"값": "노선 띠 무한사면 · SINMAP 안정 지수 — 노선 위 · 아래 자연사면 표층 붕괴 위험 · 문헌 추정치",
"서식": 2
},
"B2": {
"서식": 2
},
"C2": {
"서식": 2
},
"D2": {
"서식": 2
},
"E2": {
"서식": 2
},
"F2": {
"서식": 2
},
"G2": {
"서식": 2
},
"H2": {
"서식": 2
},
"A4": {
"값": "1. 해석 조건",
"서식": 3
},
"B4": {
"서식": 3
},
"C4": {
"서식": 3
},
"D4": {
"서식": 3
},
"E4": {
"서식": 3
},
"F4": {
"서식": 3
},
"G4": {
"서식": 3
},
"H4": {
"서식": 3
},
"A5": {
"값": "노선 띠",
"서식": 4
},
"B5": {
"서식": 5
},
"C5": {
"서식": 5
},
"D5": {
"서식": 5
},
"E5": {
"서식": 5
},
"F5": {
"서식": 5
},
"G5": {
"서식": 5
},
"H5": {
"서식": 5
},
"A6": {
"값": "토층 물성",
"서식": 4
},
"B6": {
"서식": 5
},
"C6": {
"서식": 5
},
"D6": {
"서식": 5
},
"E6": {
"서식": 5
},
"F6": {
"서식": 5
},
"G6": {
"서식": 5
},
"H6": {
"서식": 5
},
"A7": {
"값": "강우 · 포화",
"서식": 4
},
"B7": {
"서식": 5
},
"C7": {
"서식": 5
},
"D7": {
"서식": 5
},
"E7": {
"서식": 5
},
"F7": {
"서식": 5
},
"G7": {
"서식": 5
},
"H7": {
"서식": 5
},
"A8": {
"값": "해석 때",
"서식": 4
},
"B8": {
"서식": 5
},
"C8": {
"서식": 5
},
"D8": {
"서식": 5
},
"E8": {
"서식": 5
},
"F8": {
"서식": 5
},
"G8": {
"서식": 5
},
"H8": {
"서식": 5
},
"A10": {
"값": "2. 등급별 면적",
"서식": 3
},
"B10": {
"서식": 3
},
"C10": {
"서식": 3
},
"D10": {
"서식": 3
},
"E10": {
"서식": 3
},
"F10": {
"서식": 3
},
"G10": {
"서식": 3
},
"H10": {
"서식": 3
},
"A11": {
"값": "등급",
"서식": 6
},
"B11": {
"값": "안정 지수 SI",
"서식": 6
},
"C11": {
"값": "면적 (m²)",
"서식": 6
},
"D11": {
"값": "비율 (%)",
"서식": 6
},
"A12": {
"값": "안정",
"서식": 7
},
"B12": {
"값": "SI ≥ 1.5",
"서식": 7
},
"C12": {
"서식": 14
},
"D12": {
"식": "IF(SUM(C$12:C$17)>0,C12/SUM(C$12:C$17)*100,\"\")",
"서식": 16
},
"A13": {
"값": "보통 안정",
"서식": 7
},
"B13": {
"값": "1.25 ≤ SI < 1.5",
"서식": 7
},
"C13": {
"서식": 14
},
"D13": {
"식": "IF(SUM(C$12:C$17)>0,C13/SUM(C$12:C$17)*100,\"\")",
"서식": 16
},
"A14": {
"값": "준안정",
"서식": 7
},
"B14": {
"값": "1.0 ≤ SI < 1.25",
"서식": 7
},
"C14": {
"서식": 14
},
"D14": {
"식": "IF(SUM(C$12:C$17)>0,C14/SUM(C$12:C$17)*100,\"\")",
"서식": 16
},
"A15": {
"값": "위험 하한",
"서식": 7
},
"B15": {
"값": "0.5 ≤ SI < 1.0",
"서식": 7
},
"C15": {
"서식": 14
},
"D15": {
"식": "IF(SUM(C$12:C$17)>0,C15/SUM(C$12:C$17)*100,\"\")",
"서식": 16
},
"A16": {
"값": "위험 상한",
"서식": 7
},
"B16": {
"값": "0 < SI < 0.5",
"서식": 7
},
"C16": {
"서식": 14
},
"D16": {
"식": "IF(SUM(C$12:C$17)>0,C16/SUM(C$12:C$17)*100,\"\")",
"서식": 16
},
"A17": {
"값": "불안정",
"서식": 7
},
"B17": {
"값": "SI = 0",
"서식": 7
},
"C17": {
"서식": 14
},
"D17": {
"식": "IF(SUM(C$12:C$17)>0,C17/SUM(C$12:C$17)*100,\"\")",
"서식": 16
},
"A18": {
"값": "합계",
"서식": 4
},
"B18": {
"서식": 4
},
"C18": {
"식": "SUM(C12:C17)",
"서식": 14
},
"D18": {
"서식": 16
},
"A20": {
"값": "3. 위험 측점 (SI < 1 셀이 있는 측점)",
"서식": 3
},
"B20": {
"서식": 3
},
"C20": {
"서식": 3
},
"D20": {
"서식": 3
},
"E20": {
"서식": 3
},
"F20": {
"서식": 3
},
"G20": {
"서식": 3
},
"H20": {
"서식": 3
},
"A21": {
"값": "측점",
"서식": 6
},
"B21": {
"값": "누가거리 (m)",
"서식": 6
},
"C21": {
"값": "최소 Fs",
"서식": 6
},
"D21": {
"값": "최소 SI",
"서식": 6
},
"E21": {
"값": "가장 나쁜 등급",
"서식": 6
},
"F21": {
"값": "위험 면적 (m²)",
"서식": 6
},
"G21": {
"값": "최대 경사 (°)",
"서식": 6
},
"A22": {
"서식": 7
},
"B22": {
"서식": 17
},
"C22": {
"서식": 8
},
"D22": {
"서식": 8
},
"E22": {
"서식": 7
},
"F22": {
"서식": 14
},
"G22": {
"서식": 16
}
},
"병합": [
"A1:H1",
"A2:H2",
"A4:H4",
"B5:H5",
"B6:H6",
"B7:H7",
"B8:H8",
"A10:H10",
"A20:H20"
],
"열": {
"A": {
"폭": 14
},
"B": {
"폭": 15
},
"C": {
"폭": 11
},
"D": {
"폭": 11
},
"E": {
"폭": 13
},
"F": {
"폭": 13
},
"G": {
"폭": 12
},
"H": {
"폭": 6
}
},
"행": {
"1": {
"높이": 36
},
"2": {
"높이": 20
},
"4": {
"높이": 22
},
"10": {
"높이": 22
},
"20": {
"높이": 22
},
"21": {
"높이": 32
}
},
"보기": {
"눈금선": false
},
"인쇄": {
"용지": "A4",
"방향": "portrait",
"맞춤": {
"폭": 1,
"높이": 0
},
"여백": {
"위": 15,
"아래": 15,
"왼": 10,
"오른": 10
},
"머리행": [
21,
21
]
}
}
],
"활성": "s1"
@@ -79,3 +79,13 @@
| 20 행부터 구간 하나 = 한 줄 | A 구간 · B 시작 측점 · C 끝 측점(서식 7) · D 3D 최소 Fs · E 2D 최소 Fs(서식 8) · **F 3D/2D 비 = 수식 `IF(E{r}>0,D{r}/E{r},"")`**(서식 8) · G 임계 체적 m³(서식 14) · H 판정(서식 7 → N.G 10 · O.K 11) |
- 20 행 서식을 줄마다 복사.
## s7 광역요약 (프로그램 몫 · 머리 줄 21 행 반복 · E07 광역 저장본 · 없으면 `B5` 에 「광역 해석 전」)
| 칸 | 내용 |
| --- | --- |
| `B5` `B6` `B7` `B8` (각 B:H 병합 · 서식 5) | 노선 띠 · 토층 물성 · 강우 · 포화 · 해석 때 글 |
| `C12`~`C17` | 등급 1~6(안정 · 보통 안정 · 준안정 · 위험 하한 · 위험 상한 · 불안정) 면적 m²(서식 14) · `D` 비율 = 양식 수식 · `C18` 합계 = 양식 수식 |
| 22 행부터 위험 측점 하나 = 한 줄 | A 측점(서식 7) · B 누가거리(17) · C 최소 Fs · D 최소 SI(8) · E 가장 나쁜 등급 글(7) · F 위험 면적 m²(14) · G 최대 경사 °(16) · 없으면 `A22` 「위험 측점 없음」 |
- 22 행 서식을 줄마다 복사. 위험 측점 = 맡은 띠 셀에 SI < 1(등급 4 이상)이 있는 측점.
+77 -4
View File
@@ -203,6 +203,79 @@ def test_처음_만듦_측점수_NG_그림(project):
assert saved["fingerprint"] == build.analysis_fingerprint(project)
def test_광역요약_E07_저장본(project):
"""E07 저장본 없음 = 「광역 해석 전」 · 생김 = 지문 바뀜 → 갱신 뒤 등급 면적 · 위험 측점 줄."""
_create(project)
book, meta = store.load(project)
assert _sheets(book)["광역요약"]["칸"]["B5"]["값"].startswith("광역 해석 전")
keys = ("stable", "moderate", "quasi", "lower", "upper", "defended")
area = {
"settings": {
"band_m": 100.0,
"cell_m": 2.0,
"gamma_t": 17.5,
"gamma_sat": 18.5,
"c": 5.0,
"c_hi": 10.0,
"phi": 27.0,
"phi_hi": 30.0,
"c_root": 0.0,
"soil_depth_m": 1.5,
"scenario": "saturated",
"rain_mm_h": 30.0,
"k_m_s": 1e-4,
},
"computed_at": "2026-10-25T21:00:00",
"stations": [
{
"chainage_m": 0.0,
"cells": 9,
"fs_min": 1.6,
"si_min": 1.6,
"worst_class": 1,
"risk_area_m2": 0.0,
"slope_max_deg": 12.0,
"label": "No.0",
},
{
"chainage_m": 20.0,
"cells": 9,
"fs_min": 0.8,
"si_min": 0.0,
"worst_class": 6,
"risk_area_m2": 36.0,
"slope_max_deg": 41.5,
"label": "No.1",
},
{
"chainage_m": 40.0,
"cells": 9,
"fs_min": 0.95,
"si_min": 0.7,
"worst_class": 4,
"risk_area_m2": 8.0,
"slope_max_deg": 33.0,
"label": "No.2",
},
],
"summary": {"areas_m2": dict(zip(keys, (100.0, 20.0, 12.0, 8.0, 0.0, 36.0)))},
}
path = project / "E07_StabArea" / "_derived" / "area.json"
path.parent.mkdir(parents=True)
path.write_text(json.dumps(area, ensure_ascii=False), encoding="utf-8")
assert build.analysis_fingerprint(project) != meta["fingerprint"]
store.refresh(project)
book, _ = store.load(project)
s7 = _sheets(book)["광역요약"]["칸"]
assert [s7[f"C{r}"]["값"] for r in range(12, 18)] == [100.0, 20.0, 12.0, 8.0, 0.0, 36.0]
assert [s7[f"A{r}"]["값"] for r in (22, 23)] == ["No.1", "No.2"] and "A24" not in s7
assert (
s7["E22"]["값"] == "불안정" and s7["F23"]["값"] == 8.0 and s7["B7"]["값"].startswith("지표")
)
vals = book["계산값"]["s7"]
assert vals["C18"] == {"수": "176"} and vals["D17"]["수"].startswith("20.45")
def test_갱신_사용자몫_불변_덮임(project):
_create(project)
book, _ = store.load(project)
@@ -262,7 +335,7 @@ def test_xlsx_되읽기_pdf_쪽수(project):
figs = store.figures_of(project, book)
assert len(figs) == len(REPS) + 1
wb = load_workbook(io.BytesIO(workbook_to_xlsx(book, figs)))
assert wb.sheetnames == ["표지", "개요", "입력조건", "측점표", "대표단면", "3D요약"]
assert wb.sheetnames == ["표지", "개요", "입력조건", "측점표", "대표단면", "3D요약", "광역요약"]
assert len(wb["대표단면"]._images) == len(REPS) and len(wb["3D요약"]._images) == 1
assert [b.id for b in wb["대표단면"].row_breaks.brk] == [28, 56, 84]
assert wb["측점표"]["J9"].value == "N.G" and wb["측점표"]["J9"].fill.fgColor.rgb.endswith(
@@ -278,8 +351,8 @@ def test_xlsx_되읽기_pdf_쪽수(project):
texts = [p.extract_text() for p in PdfReader(io.BytesIO(pdf)).pages]
stab = [i for i, t in enumerate(texts) if "산정 안전율" in t]
assert stab and all("산정 안전율" in texts[i] for i in stab)
# 표지 1 + 개요 1 + 입력조건 1 + 측점표 n + 대표단면 3 + 3D 1
assert len(texts) == 3 + len(stab) + len(REPS) + 1
# 표지 1 + 개요 1 + 입력조건 1 + 측점표 n + 대표단면 3 + 3D 1 + 광역 1
assert len(texts) == 3 + len(stab) + len(REPS) + 1 + 1
def test_단면_그림_SVG():
@@ -317,7 +390,7 @@ def test_길_저장_409_갱신_초기화_내보내기(project, monkeypatch):
got = client.get(url).json()
except PrintError as error:
pytest.skip(f"브라우저 없음 — {error}")
assert got["editable"] and got["stale"] is False and len(got["book"]["시트"]) == 6
assert got["editable"] and got["stale"] is False and len(got["book"]["시트"]) == 7
book = got["book"]
_sheets(book)["표지"]["칸"]["C10"] = {"값": "산림청", "서식": 5}
assert client.put(url, json={"rev": "옛판", "book": book}).status_code == 409
@@ -36,7 +36,7 @@ STYLE_KEYS = {
} # fmt: skip
PRINT_KEYS = {"용지", "방향", "배율", "맞춤", "여백", "머리행", "쪽나눔"}
A1 = re.compile(r"^([A-Z]{1,3})(\d+)$")
NAMES = ["표지", "개요", "입력조건", "측점표", "대표단면", "3D요약"]
NAMES = ["표지", "개요", "입력조건", "측점표", "대표단면", "3D요약", "광역요약"]
needs_node = pytest.mark.skipif(shutil.which("node") is None, reason="node 가 없음")
@@ -148,7 +148,7 @@ def test_양식_계약():
book = _form()
_contract(book)
assert [s["이름"] for s in book["시트"]] == NAMES
assert [s["id"] for s in book["시트"]] == ["s1", "s2", "s3", "s4", "s5", "s6"]
assert [s["id"] for s in book["시트"]] == ["s1", "s2", "s3", "s4", "s5", "s6", "s7"]
def test_양식_인쇄_쪽나눔_머리행_그림():
@@ -159,7 +159,15 @@ def test_양식_인쇄_쪽나눔_머리행_그림():
assert by["측점표"]["인쇄"]["방향"] == "landscape" and by["측점표"]["인쇄"]["머리행"] == [4, 5]
assert by["대표단면"]["인쇄"]["쪽나눔"] == [BLOCK]
figs = {n: len(s.get("그림", [])) for n, s in by.items()}
assert figs == {"표지": 0, "개요": 0, "입력조건": 0, "측점표": 0, "대표단면": 1, "3D요약": 1}
assert figs == {
"표지": 0,
"개요": 0,
"입력조건": 0,
"측점표": 0,
"대표단면": 1,
"3D요약": 1,
"광역요약": 0,
}
# N.G 빨강 조건부 서식 — 측점표 판정 열 · 서식 표에 N.G 칸 서식(빌더가 인쇄 · xlsx 용으로 직접 씀)
assert by["측점표"]["조건부서식"][0]["범위"][0]["c0"] == 9
assert any(st.get("글자색") == "#C00000" and st.get("채움") for st in _form()["서식"])
@@ -222,8 +230,8 @@ def test_견본_pdf_쪽수_머리행(png):
)
sections = [i for i, t in enumerate(texts) if "대표 단면" in t and "안전율 Fs" in t]
assert len(sections) == 2 # 단면마다 한 쪽
# 표지 1 + 개요 1 + 입력조건 1 + 측점표 n + 대표단면 2 + 3D 1
assert len(pages) == 1 + 1 + 1 + len(stab) + 2 + 1
# 표지 1 + 개요 1 + 입력조건 1 + 측점표 n + 대표단면 2 + 3D 1 + 광역 1
assert len(pages) == 1 + 1 + 1 + len(stab) + 2 + 1 + 1
assert all(float(pages[i].mediabox.width) > float(pages[i].mediabox.height) for i in stab)
html = spreadsheet.print_htmls([{"book": book, "figures": {k: png for k in figs}}])[0]
assert html.count("<img ") == 3 # 단면 2 + 3D 띠 1
+158
View File
@@ -0,0 +1,158 @@
"""536-2 — E07 광역 자연사면 위험 `E07_StabArea_Engine` · `_Router`.
무한사면 해석해(건조 · 포화 · 점착) · SINMAP 안정 지수(닫힌 꼴 = 표본 · 등급 경계) ·
D8 집수면적 견본(기운 평면 · V 골짜기) · 강우 포화비 · 가짜 복사본으로 띠 · 측점표 · 저장본 · 지문.
"""
import base64
import json
import math
import numpy as np
import pytest
from E07_StabArea import E07_StabArea_Engine as engine
from E07_StabArea import E07_StabArea_Router as router_mod
SOIL = dict(gamma_t=18.0, gamma_sat=20.0, c=0.0, c_hi=0.0, phi=35.0, phi_hi=35.0)
def _fs(beta_deg: float, m: float, **kw) -> float:
p = engine.AreaParams(**{**SOIL, **kw})
return float(engine.infinite_slope_fs(np.array([math.radians(beta_deg)]), np.array([m]), p)[0])
def test_무한사면_해석해_건조_포화_점착():
t = math.tan(math.radians(35)) / math.tan(math.radians(30))
assert _fs(30, 0.0) == pytest.approx(t, rel=1e-12) # 건조 c = 0 → tanφ / tanβ
assert _fs(30, 1.0) == pytest.approx((20 - 9.81) / 20 * t, rel=1e-12) # 포화 → γ' / γsat 배
# 손계산 — c 5 · 뿌리 2 · z 1.5 · β 35° · m 0.5
b, z = math.radians(35), 1.5
w = 18 * z * 0.5 + 20 * z * 0.5
n = (18 * z * 0.5 + (20 - 9.81) * z * 0.5) * math.cos(b) ** 2
hand = (7 + n * math.tan(math.radians(35))) / (w * math.sin(b) * math.cos(b))
assert _fs(35, 0.5, c=5.0, c_root=2.0) == pytest.approx(hand, rel=1e-12)
assert math.isinf(_fs(0, 1.0)) # 평평 = 활동 없음
def test_SINMAP_안정지수_닫힌꼴_표본():
p = engine.AreaParams(gamma_t=17.5, gamma_sat=18.5, c=0.0, c_hi=8.0, phi=25.0, phi_hi=38.0)
beta = np.radians(np.array([10.0, 30.0, 35.0, 40.0, 60.0]))
m = np.ones_like(beta)
si = engine.stability_index(beta, m, p)
a, b = engine._terms(beta, m, p)
rng = np.random.default_rng(1)
c = rng.uniform(0, 8, 400_000)
t = rng.uniform(math.tan(math.radians(25)), math.tan(math.radians(38)), 400_000)
for k in range(len(beta)):
lo = 0 * a[k] + math.tan(math.radians(25)) * b[k]
fs = c * a[k] + t * b[k]
if lo >= 1:
assert si[k] == pytest.approx(lo) # Fs_min ≥ 1 → SI = Fs_min
elif fs.max() < 1:
assert si[k] == 0.0
else:
assert si[k] == pytest.approx((fs > 1).mean(), abs=3e-3) # P(Fs > 1)
assert si[0] >= 1 and 0 < si[1] < 1 and si[4] == 0 # 세 갈래 모두 지남
# 폭 0(범위 없음) — 결정 Fs 와 같거나 0
q = engine.AreaParams(**SOIL)
fs = engine.infinite_slope_fs(beta, m, q)
si0 = engine.stability_index(beta, m, q)
assert np.allclose(si0, np.where(fs >= 1, fs, 0.0))
def test_SINMAP_등급_경계():
si = np.array([np.inf, 1.5, 1.4999, 1.25, 1.2499, 1.0, 0.9999, 0.5, 0.4999, 1e-9, 0.0, np.nan])
assert engine.classify(si).tolist() == [1, 1, 2, 2, 3, 3, 4, 4, 5, 5, 6, 0]
def test_D8_집수면적_기운평면_V골짜기():
cell = 2.0
ny, nx = 5, 7
xx = np.arange(nx)[None, :] * cell
plane = np.repeat(100.0 - 0.5 * xx, ny, axis=0) # 동쪽으로 기움 → 행마다 동쪽으로 흐름
a = engine.catchment_area(plane, cell)
assert np.allclose(a, (np.arange(nx)[None, :] + 1) * cell)
yy = np.arange(9)[:, None] * cell
valley = 100.0 + np.abs(yy - 8.0) * 1.0 - 0.1 * np.arange(nx)[None, :] * cell
a = engine.catchment_area(valley, cell)
assert a[4, -1] == pytest.approx(9 * nx * cell) # 전부가 골 출구로
assert np.all(a[4, :-1] < a[4, 1:]) # 골 따라 늘어남
def test_강우_포화비():
p = engine.AreaParams(**SOIL, scenario="rain", rain_mm_h=36.0, k_m_s=1e-4, soil_depth_m=2.0)
beta = np.radians(np.array([30.0, 30.0]))
a = np.array([10.0, 1e6])
m = engine.wetness(beta, a, p)
assert m[0] == pytest.approx(1e-5 * 10 / (2e-4 * 0.5)) # R a / (T sinβ)
assert m[1] == 1.0
assert engine.wetness(beta, a, engine.AreaParams(**SOIL)).tolist() == [1.0, 1.0]
def _decode(text: str, dtype: str, grid: dict) -> np.ndarray:
return np.frombuffer(base64.b64decode(text), dtype=dtype).reshape(grid["ny"], grid["nx"])
@pytest.fixture()
def project(tmp_path, monkeypatch):
"""가짜 E 복사본 — 서쪽이 낮은 30° 비탈(동서 200 m · 남북 100 m) · 노선 = 남북 중심선."""
x = np.arange(0.0, 200.5, 0.5)
y = np.arange(0.0, 100.5, 0.5)
z = np.repeat((x * math.tan(math.radians(30)))[None, :], len(y), axis=0)
valid = np.ones_like(z, bool)
valid[:, x > 180] = False # 동쪽 끝 = 지형 없음
dem = tmp_path / "dtm.npz"
np.savez(
dem,
x=x.astype(np.float32),
y=y.astype(np.float32),
z=z.astype(np.float32),
valid_mask=valid,
)
stations = [
{"chainage_m": float(c), "label": f"No.{i}", "center_x": 100.0, "center_y": float(c)}
for i, c in enumerate(range(0, 101, 20))
]
route = [
{"sequence_num": i, "model_x": 100.0, "model_y": float(c), "chainage_m": float(c)}
for i, c in enumerate((0, 100))
]
src = router_mod.source
monkeypatch.setattr(src, "terrain_grid_path", lambda _r: dem)
monkeypatch.setattr(src, "stations", lambda _r: stations)
monkeypatch.setattr(src, "db_snapshot", lambda _r: {"route_points": route})
return tmp_path
def test_길_띠_측점표_저장본(project):
settings = {**router_mod.defaults(project), "band_m": 30.0, "cell_m": 2.0}
assert settings["material_id"] == "colluvium" and settings["c_hi"] == settings["c"] + 5
steps = []
summary = router_mod.run(project, project, settings, "fp", lambda p, m: steps.append(p))
doc = router_mod.load(project)
assert steps == sorted(steps) and doc["fingerprint"] == "fp"
g = doc["grid"]
cls = _decode(doc["cls"], "u1", g)
fs = _decode(doc["fs"], "<u2", g)
# 붕적토 포화 30° — 손계산 Fs
b = math.radians(30)
n = (18.5 - 9.81) * 1.5 * math.cos(b) ** 2
hand = (5 + n * math.tan(math.radians(27))) / (18.5 * 1.5 * math.sin(b) * math.cos(b))
inner = fs[cls > 0]
assert np.median(inner) / 1000 == pytest.approx(hand, abs=2e-3)
assert summary["fs_min"] == pytest.approx(hand, abs=5e-3)
# 띠 = 중심선 ± 30 m(격자 상자도 띠만큼) · 등급 면적 합 = 덮인 면적
assert g["nx"] * g["dx"] <= 62 + 1e-9 and g["ny"] * g["dx"] <= 102 + 1e-9
assert sum(summary["areas_m2"].values()) == pytest.approx(summary["covered_area_m2"])
assert summary["covered_area_m2"] == pytest.approx((cls > 0).sum() * 4.0)
rows = doc["stations"]
assert [r["chainage_m"] for r in rows] == [0, 20, 40, 60, 80, 100]
assert all(r["label"] == f"No.{i}" and r["worst_class"] >= 4 for i, r in enumerate(rows))
assert summary["risk_stations"] == [0, 20, 40, 60, 80, 100]
assert sum(r["cells"] for r in rows) == (cls > 0).sum()
# 지문 = 설정에 따라 바뀜
assert router_mod.fingerprint("s", settings) != router_mod.fingerprint(
"s", {**settings, "c": 6}
)
json.dumps(doc) # 그대로 JSON