diff --git a/E05_StabReport/E05_StabReport_Area.py b/E05_StabReport/E05_StabReport_Area.py new file mode 100644 index 000000000..f62bb196e --- /dev/null +++ b/E05_StabReport/E05_StabReport_Area.py @@ -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"])) diff --git a/E05_StabReport/E05_StabReport_Build.py b/E05_StabReport/E05_StabReport_Build.py index d924938cf..9717c6813 100644 --- a/E05_StabReport/E05_StabReport_Build.py +++ b/E05_StabReport/E05_StabReport_Build.py @@ -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 diff --git a/E07_StabArea/E07_StabArea_Engine.py b/E07_StabArea/E07_StabArea_Engine.py new file mode 100644 index 000000000..add8f867a --- /dev/null +++ b/E07_StabArea/E07_StabArea_Engine.py @@ -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) + } diff --git a/E07_StabArea/E07_StabArea_Router.py b/E07_StabArea/E07_StabArea_Router.py new file mode 100644 index 000000000..dfae623ff --- /dev/null +++ b/E07_StabArea/E07_StabArea_Router.py @@ -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), " 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} diff --git a/common_util/common_util_project_kind.py b/common_util/common_util_project_kind.py index d4d4e997d..5a82d734c 100644 --- a/common_util/common_util_project_kind.py +++ b/common_util/common_util_project_kind.py @@ -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", + ), } diff --git a/config/config_system.py b/config/config_system.py index 2fa29bb33..efe11fac1 100644 --- a/config/config_system.py +++ b/config/config_system.py @@ -220,6 +220,7 @@ PROJECT_STORAGE_STAGE_DIRS = frozenset( "E03_StabSection", "E04_Stab3D", "E05_StabReport", + "E07_StabArea", } ) diff --git a/main.py b/main.py index f9d6e072d..c3b89b238 100644 --- a/main.py +++ b/main.py @@ -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) diff --git a/resources/master_template/report/stab_report.json b/resources/master_template/report/stab_report.json index 9ded03dad..42a350b7e 100644 --- a/resources/master_template/report/stab_report.json +++ b/resources/master_template/report/stab_report.json @@ -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" diff --git a/resources/master_template/report/stab_report_slots.md b/resources/master_template/report/stab_report_slots.md index 3fb11f4c7..b9bafb828 100644 --- a/resources/master_template/report/stab_report_slots.md +++ b/resources/master_template/report/stab_report_slots.md @@ -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 이상)이 있는 측점. diff --git a/resources/tester/test_534_3_e05_report.py b/resources/tester/test_534_3_e05_report.py index e4dfc66f6..3b0c93ede 100644 --- a/resources/tester/test_534_3_e05_report.py +++ b/resources/tester/test_534_3_e05_report.py @@ -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 diff --git a/resources/tester/test_534_3_stab_report_form.py b/resources/tester/test_534_3_stab_report_form.py index 89ac63e32..8eee5c646 100644 --- a/resources/tester/test_534_3_stab_report_form.py +++ b/resources/tester/test_534_3_stab_report_form.py @@ -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(" 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"], " 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