From cfd1e3eb2c7fffe51aaf5145898dea3d916158e1 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sun, 25 Oct 2026 20:14:29 +0900 Subject: [PATCH] =?UTF-8?q?feat(E04):=203D=20=ED=95=B4=EC=84=9D=20?= =?UTF-8?q?=EA=B8=B8=20=E2=80=94=20analyze3d=20=C2=B7=20=EB=81=8A=EA=B8=B0?= =?UTF-8?q?=20=C2=B7=20progress3d=20=C2=B7=20result3d=20=C2=B7=20=EC=9E=85?= =?UTF-8?q?=EB=A0=A5=20=EC=A7=80=EB=AC=B8=20=ED=8C=8C=EC=83=9D=20=EC=A0=80?= =?UTF-8?q?=EC=9E=A5=EB=B3=B8=20=C2=B7=20=EC=8B=9C=ED=97=98=20(533-3=20?= =?UTF-8?q?=EA=B8=B8=20=EB=B6=80=EB=B6=84)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 엔진 = 미리보기(지형 경사 가짜 래스터) · 533-1 · 2 결과 꼴이 오면 run_engine 만 갈아 끼움 - 권한 = protected_with_company(지정받은 회사 GET 만) Co-Authored-By: Claude Opus 5.5 --- E04_Stab3D/E04_Stab3D_Router.py | 221 +++++++++++++++++ E04_Stab3D/E04_Stab3D_Run.py | 275 ++++++++++++++++++++++ main.py | 2 + resources/tester/test_533_3_e04_router.py | 264 +++++++++++++++++++++ 4 files changed, 762 insertions(+) create mode 100644 E04_Stab3D/E04_Stab3D_Router.py create mode 100644 E04_Stab3D/E04_Stab3D_Run.py create mode 100644 resources/tester/test_533_3_e04_router.py diff --git a/E04_Stab3D/E04_Stab3D_Router.py b/E04_Stab3D/E04_Stab3D_Router.py new file mode 100644 index 000000000..30a1b8e82 --- /dev/null +++ b/E04_Stab3D/E04_Stab3D_Router.py @@ -0,0 +1,221 @@ +"""E04 안정성평가 3D 해석 API (요구 347-1 · 351 · 요구서 4-2 · 4-3 · PLAN 533-3). + +- `POST /api/projects/{id}/stab/analyze3d` — 한 구간 3D 해석(뒤에서) · 입력 지문이 저장본과 같으면 + 셈하지 않고 `cached` · `force` 면 다시 셈 +- `POST …/stab/analyze3d/cancel` — 도는 해석 끊기(다시 시작 = analyze3d 다시) +- `GET …/stab/progress3d` — 진행률 +- `GET …/stab/result3d` — 자동 구간 · 노선 · 측점 · 저장본 요약(+ 지문이 지금과 다른지) +- `GET …/stab/result3d?key=` — 한 구간 저장본(래스터 · 임계 활동면) + +계산 자리 = ③ 서버 단독. 권한 = main.py `protected_with_company`(지정받은 회사 = GET 만 · 그 밖 403). +""" + +from __future__ import annotations + +import asyncio +import logging +from pathlib import Path +from typing import Any, Literal +from uuid import UUID + +import aiomysql +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, Field + +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 E04_Stab3D import E04_Stab3D_Run as run + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/projects", tags=["E04 Stab 3D"]) + +# 진행 — 프로젝트 → {running, percent, message, error, cancelled, key, summary} · 프로세스 메모리 +_JOBS: dict[str, dict[str, Any]] = {} +_STOPS: dict[str, bool] = {} +_TASKS: set[asyncio.Task] = set() +_IDLE = { + "running": False, + "percent": 0, + "message": None, + "error": None, + "cancelled": False, + "key": None, + "summary": None, +} + + +class SearchSettings(BaseModel): + """탐색 조건 — 길이 m · 부피 m³ (Scoops3D 방식 · 요구서 4-2).""" + + shape: Literal["sphere", "ellipsoid", "both"] = "both" + ellipsoid_ratio: float = Field(2.0, ge=1.0, le=5.0) # 타원체 = 회전축 방향 반축 배수 + box_half_width_m: float = Field(30.0, ge=5.0, le=100.0) # 탐색 상자 = 중심선 ± 이 폭 + cell_m: float = Field(1.0, ge=0.5, le=5.0) # 기둥(셀) 크기 + spacing_m: float = Field(2.0, ge=0.5, le=10.0) # 구 중심 수평 간격(가장 고운) + h_min_m: float = Field(2.0, ge=0.0, le=100.0) # 구 중심 높이(지표 위) 범위 · 간격 + h_max_m: float = Field(30.0, ge=1.0, le=200.0) + h_step_m: float = Field(2.0, ge=0.5, le=20.0) + dr_m: float = Field(1.0, ge=0.2, le=5.0) # 반지름 증분 + max_depth_m: float = Field(15.0, ge=1.0, le=60.0) + v_min_m3: float = Field(10.0, ge=0.0) + v_max_m3: float = Field(10_000.0, gt=0.0) + + +class Analyze3dRequest(BaseModel): + start_m: float = Field(..., ge=0.0) + end_m: float = Field(..., ge=0.0) + search: SearchSettings = SearchSettings() + method: Literal["bishop", "hovland"] = "bishop" # bishop = Hungr 3D Bishop(+Janbu 보정) + force: bool = False + + +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 + + +def _settings(payload: Analyze3dRequest) -> dict[str, Any]: + """지문 · 저장본에 남는 요청 — 소수 둘째 자리로 맞춤(같은 구간 = 같은 열쇠).""" + if payload.end_m <= payload.start_m: + raise HTTPException(status_code=400, detail="구간 끝이 시작보다 앞입니다.") + if payload.end_m - payload.start_m > run.MAX_SECTION_M: + raise HTTPException( + status_code=400, detail=f"구간은 {run.MAX_SECTION_M:.0f} m 이하로 고르세요." + ) + search = payload.search + if search.h_max_m < search.h_min_m or search.v_max_m3 <= search.v_min_m3: + raise HTTPException(status_code=400, detail="탐색 범위의 최소가 최대보다 큽니다.") + return { + "start_m": round(payload.start_m, 2), + "end_m": round(payload.end_m, 2), + "method": payload.method, + **search.model_dump(), + } + + +def _current_fingerprint(project_root: Path, root: Path, settings: dict[str, Any]) -> str: + return run.fingerprint( + source.manifest(root)["source_fingerprint"], run.read_ground(project_root), settings + ) + + +@router.post("/{project_id}/stab/analyze3d", response_model=None) +async def analyze3d(project_id: UUID, payload: Analyze3dRequest) -> dict[str, Any]: + """[해석] — 지문이 같으면 `cached`(안 셈) · 다르거나 force 면 뒤에서 셈.""" + pid = str(project_id) + project_root = await _project_root(project_id) + root = _source_root(project_root) + settings = _settings(payload) + key = run.section_key(settings["start_m"], settings["end_m"]) + finger = await asyncio.to_thread(_current_fingerprint, project_root, root, settings) + saved = await asyncio.to_thread(run.load, project_root, key) + if not payload.force and saved and saved["fingerprint"] == finger: + return {"status": "cached", "key": key, "summary": saved["summary"]} + if (_JOBS.get(pid) or {}).get("running"): + raise HTTPException(status_code=409, detail="이미 3D 해석이 돌고 있습니다.") + + job = {**_IDLE, "running": True, "message": "시작", "key": key} + _JOBS[pid] = job + _STOPS[pid] = False + + def report(percent: int, message: str) -> None: + job.update(percent=max(0, min(99, int(percent))), message=message) + + async def work() -> None: + try: + job["summary"] = await asyncio.to_thread( + run.analyze, project_root, root, settings, finger, report, lambda: _STOPS[pid] + ) + job.update(percent=100, message="끝") + except InterruptedError: + job.update(cancelled=True, message="끊음") + except Exception as error: # noqa: BLE001 — 진행률에 까닭을 그대로 올림 + logger.exception("E04 3D 해석 실패: project_id=%s", pid) + job["error"] = str(getattr(error, "detail", None) or error) + finally: + job["running"] = False + + task = asyncio.create_task(work(), name=f"E04 3D {pid}") + _TASKS.add(task) + task.add_done_callback(_TASKS.discard) + return {"status": "accepted", "key": key} + + +@router.post("/{project_id}/stab/analyze3d/cancel", response_model=None) +async def cancel3d(project_id: UUID) -> dict[str, Any]: + """[끊기] — 엔진이 다음 확인 때 멈춤 · 저장본은 그대로.""" + pid = str(project_id) + running = bool((_JOBS.get(pid) or {}).get("running")) + if running: + _STOPS[pid] = True + return {"status": "success", "stopping": running} + + +@router.get("/{project_id}/stab/progress3d", response_model=None) +async def progress3d(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]: + stations = source.stations(root) + finger_source = source.manifest(root)["source_fingerprint"] + ground = run.read_ground(project_root) + saved = run.saved_summaries(project_root) + for item in saved: + current = run.fingerprint(finger_source, ground, item.pop("settings")) + item["stale"] = item.pop("fingerprint") != current + points = source.db_snapshot(root).get("route_points") or [] + return { + "sections": run.auto_sections(stations), + "stations": [ + { + "chainage_m": s["chainage_m"], + "label": s["label"], + "x": s["center_x"], + "y": s["center_y"], + } + for s in stations + ], + "line": [ + [float(p["model_x"]), float(p["model_y"]), float(p["chainage_m"])] + for p in sorted(points, key=lambda p: p["sequence_num"]) + ], + "saved": saved, + "defaults": SearchSettings().model_dump(), + "max_section_m": run.MAX_SECTION_M, + } + + +@router.get("/{project_id}/stab/result3d", response_model=None) +async def result3d(project_id: UUID, key: str | None = None) -> dict[str, Any]: + """key 없음 = 구간 · 노선 · 저장본 요약 / key = 그 구간 래스터 · 임계 활동면.""" + project_root = await _project_root(project_id) + root = _source_root(project_root) + if key is None: + return {"status": "success", **await asyncio.to_thread(_overview, project_root, root)} + if not all(ch.isdigit() or ch in "._" for ch in key): + raise HTTPException(status_code=400, detail="구간 열쇠가 틀렸습니다.") + doc = await asyncio.to_thread(run.load, project_root, key) + if doc is None: + raise HTTPException(status_code=404, detail="이 구간 3D 해석 결과가 없습니다.") + current = await asyncio.to_thread(_current_fingerprint, project_root, root, doc["settings"]) + return {"status": "success", "key": key, "stale": doc["fingerprint"] != current, **doc} diff --git a/E04_Stab3D/E04_Stab3D_Run.py b/E04_Stab3D/E04_Stab3D_Run.py new file mode 100644 index 000000000..2a0e1a003 --- /dev/null +++ b/E04_Stab3D/E04_Stab3D_Run.py @@ -0,0 +1,275 @@ +"""E04 3D 해석 한 번 — 입력 모으기 · 입력 지문 · 엔진 부르기 · 파생 저장본 (PLAN 533-3). + +계산 자리 = ③ 서버 단독. 엔진(탐색 · 기둥 솔버 · 합성 지형) 주인 = 533-1 · 2 — 여기는 부르기만. +저장본 = E 프로젝트 `E04_Stab3D/results/<구간 열쇠>.json` · 입력 지문(복사본 지문 · E02 지반 · +요청 · 엔진 판)이 같으면 다시 셈하지 않음. + +래스터 꼴(화면 `logic/E04_Stab3D_Raster.ts` 와 같은 열쇠): + grid = {x0, y0, dx, nx, ny} — 셀 (iy, ix) 중심 = (x0 + ix·dx, y0 + iy·dx) · 줄 우선 평탄 배열 + z = 합성 지표(원지형 + 노선면) · f = 셀별 최소 안전율 · v = 그 임계 체적 m³ · m = 방법(0 Bishop + · 1 Janbu 보정) · 닿지 않은 셀 = null + best = 전체 최소 활동면 {x, y, z(중심), r, ratio, azimuth_deg, f, volume, method, ix, iy} +""" + +from __future__ import annotations + +import hashlib +import json +import math +import time +from collections.abc import Callable +from datetime import datetime +from pathlib import Path +from typing import Any + +import numpy as np + +from common_util.common_util_surface_sampler import DtmGridSampler +from E01_StabRegister import E01_StabRegister_Source as source + +ENGINE_VERSION = "e04-preview-1" # 엔진 판 — 바뀌면 지문이 달라져 다시 셈 +STAGE_DIR = "E04_Stab3D" +GROUND_FILE = Path("E02_StabGround") / "ground.json" +SECTION_LENGTH_M = 100.0 # 자동 구간 길이 +MAX_SECTION_M = 500.0 # 한 번에 셀 구간 상한(시간 · 메모리) +METHOD_NAMES = ("bishop", "janbu_corrected") + +Progress = Callable[[int, str], None] +StopFlag = Callable[[], bool] + + +def results_dir(project_root: Path) -> Path: + return Path(project_root) / STAGE_DIR / "results" + + +def section_key(start_m: float, end_m: float) -> str: + return f"{start_m:.2f}_{end_m:.2f}" + + +def auto_sections(stations: list[dict[str, Any]]) -> list[dict[str, float]]: + """측점을 앞에서부터 SECTION_LENGTH_M 이상 묶음 — 경계 = 측점 · 꼬리가 짧으면 앞 구간에 붙임.""" + chainages = sorted({float(s["chainage_m"]) for s in stations}) + out: list[dict[str, float]] = [] + start = chainages[0] if chainages else 0.0 + for ch in chainages[1:]: + if ch - start >= SECTION_LENGTH_M - 1e-6: + out.append({"start_m": start, "end_m": ch}) + start = ch + if chainages and start < chainages[-1]: + if out and chainages[-1] - start < SECTION_LENGTH_M / 2: + out[-1]["end_m"] = chainages[-1] + else: + out.append({"start_m": start, "end_m": chainages[-1]}) + return out + + +def read_ground(project_root: Path) -> dict[str, Any] | None: + """E02 작업본(없으면 None = 기본값).""" + path = Path(project_root) / GROUND_FILE + return json.loads(path.read_text(encoding="utf-8")) if path.is_file() else None + + +def fingerprint( + source_fingerprint: str, ground: dict[str, Any] | None, settings: dict[str, Any] +) -> str: + body = json.dumps( + { + "source": source_fingerprint, + "ground": ground, + "settings": settings, + "engine": ENGINE_VERSION, + }, + ensure_ascii=False, + sort_keys=True, + ) + return hashlib.sha1(body.encode("utf-8")).hexdigest() + + +def section_stations(stations: list[dict[str, Any]], start_m: float, end_m: float) -> list[dict]: + return [s for s in stations if start_m - 1e-6 <= float(s["chainage_m"]) <= end_m + 1e-6] + + +def bbox_of( + stations: list[dict[str, Any]], half_width_m: float +) -> tuple[float, float, float, float]: + xs = [float(s["center_x"]) for s in stations] + ys = [float(s["center_y"]) for s in stations] + return ( + min(xs) - half_width_m, + max(xs) + half_width_m, + min(ys) - half_width_m, + max(ys) + half_width_m, + ) + + +# --------------------------------------------------------------------------- +# 엔진 — 533-1 · 2 결과 꼴이 오기 전 = 미리보기(지형 경사로 만든 가짜 래스터 · 틀 · 배치 확인용) +# --------------------------------------------------------------------------- + + +def _preview_engine( + root: Path, + stations: list[dict[str, Any]], + settings: dict[str, Any], + on_progress: Progress, + should_stop: StopFlag, +) -> dict[str, Any]: + dem = DtmGridSampler.open_cached(source.terrain_grid_path(root)) + x0, x1, y0, y1 = bbox_of(stations, float(settings["box_half_width_m"])) + dx = float(settings["cell_m"]) + xs = np.arange(x0, x1 + dx / 2, dx) + ys = np.arange(y0, y1 + dx / 2, dx) + xx, yy = np.meshgrid(xs, ys) + z, valid = dem.sample_xy(np.column_stack([xx.ravel(), yy.ravel()])) + z = np.where(valid, z, np.nan).reshape(xx.shape) + gy, gx = np.gradient(z, dx) + slope = np.hypot(gx, gy) + f = np.full(z.shape, np.nan) + rows = z.shape[0] + for iy in range(rows): + if should_stop(): + raise InterruptedError + f[iy] = 0.8 + 2.4 / (1.0 + 2.5 * slope[iy] ** 2) + if iy % max(1, rows // 20) == 0: + on_progress(int(100 * iy / rows), f"미리보기 {iy}/{rows} 줄") + volume = np.where(np.isfinite(f), 50.0 + 400.0 * slope**2, np.nan) + iy, ix = np.unravel_index(int(np.nanargmin(f)), f.shape) + r = 10.0 + best = { + "x": float(xs[ix]), + "y": float(ys[iy]), + "z": float(z[iy, ix] + 0.6 * r), + "r": r, + "ratio": 1.0, + "azimuth_deg": 0.0, + "f": float(f[iy, ix]), + "volume": float(volume[iy, ix]), + "method": 0, + } + return { + "grid": {"x0": float(xs[0]), "y0": float(ys[0]), "dx": dx, "nx": len(xs), "ny": len(ys)}, + "z": z, + "f": f, + "v": volume, + "m": np.where(np.isfinite(f), 0.0, np.nan), + "best": best, + "trials": int(np.isfinite(f).sum()), + "preview": True, + } + + +def run_engine( + root: Path, + stations: list[dict[str, Any]], + ground: dict[str, Any] | None, + settings: dict[str, Any], + on_progress: Progress, + should_stop: StopFlag, +) -> dict[str, Any]: + """엔진 한 번 → {grid, z, f, v, m(넘파이 (ny, nx)), best, trials}.""" + return _preview_engine(root, stations, settings, on_progress, should_stop) + + +# --------------------------------------------------------------------------- +# 저장본 +# --------------------------------------------------------------------------- + + +def _flat(values: np.ndarray, digits: int) -> list[float | None]: + return [None if not math.isfinite(v) else round(float(v), digits) for v in values.ravel()] + + +def _nearest_chainage(stations: list[dict[str, Any]], x: float, y: float) -> float: + return min( + stations, key=lambda s: (float(s["center_x"]) - x) ** 2 + (float(s["center_y"]) - y) ** 2 + )["chainage_m"] + + +def pack( + raw: dict[str, Any], + stations: list[dict[str, Any]], + settings: dict[str, Any], + finger: str, + elapsed_s: float, +) -> dict[str, Any]: + """엔진 결과 → 저장본(JSON) — 요약은 best 에서.""" + grid = raw["grid"] + best = dict(raw["best"]) + best["ix"] = int(round((best["x"] - grid["x0"]) / grid["dx"])) + best["iy"] = int(round((best["y"] - grid["y0"]) / grid["dx"])) + f = np.asarray(raw["f"], dtype=float) + return { + "fingerprint": finger, + "engine": ENGINE_VERSION, + "preview": bool(raw.get("preview")), + "settings": settings, + "computed_at": datetime.now().isoformat(timespec="seconds"), + "elapsed_s": round(elapsed_s, 2), + "grid": grid, + "z": _flat(np.asarray(raw["z"], dtype=float), 2), + "f": _flat(f, 3), + "v": _flat(np.asarray(raw["v"], dtype=float), 1), + "m": _flat(np.asarray(raw["m"], dtype=float), 0), + "best": best, + "summary": { + "start_m": settings["start_m"], + "end_m": settings["end_m"], + "f_min": round(float(best["f"]), 3), + "volume": round(float(best["volume"]), 1), + "method": METHOD_NAMES[int(best.get("method") or 0)], + "chainage_m": float(_nearest_chainage(stations, best["x"], best["y"])), + "cells": int(np.isfinite(f).sum()), + "trials": int(raw.get("trials") or 0), + }, + } + + +def save(project_root: Path, key: str, result: dict[str, Any]) -> None: + folder = results_dir(project_root) + folder.mkdir(parents=True, exist_ok=True) + tmp = folder / f"{key}.json.tmp" + tmp.write_text(json.dumps(result, ensure_ascii=False, separators=(",", ":")), encoding="utf-8") + tmp.replace(folder / f"{key}.json") + + +def load(project_root: Path, key: str) -> dict[str, Any] | None: + path = results_dir(project_root) / f"{key}.json" + return json.loads(path.read_text(encoding="utf-8")) if path.is_file() else None + + +def saved_summaries(project_root: Path) -> list[dict[str, Any]]: + """저장본 요약(래스터 빼고) — 구간 목록 색 · 판정용.""" + out = [] + for path in sorted(results_dir(project_root).glob("*.json")): + doc = json.loads(path.read_text(encoding="utf-8")) + out.append( + { + "key": path.stem, + "fingerprint": doc["fingerprint"], + "settings": doc["settings"], + "preview": doc.get("preview", False), + **doc["summary"], + } + ) + return out + + +def analyze( + project_root: Path, + root: Path, + settings: dict[str, Any], + finger: str, + on_progress: Progress, + should_stop: StopFlag, +) -> dict[str, Any]: + """한 구간 3D 해석 → 저장본 쓰고 요약 돌려줌.""" + started = time.perf_counter() + stations = source.stations(root) + picked = section_stations(stations, settings["start_m"], settings["end_m"]) + if len(picked) < 2: + raise ValueError("구간 안 측점이 둘 미만입니다.") + ground = read_ground(project_root) + raw = run_engine(root, picked, ground, settings, on_progress, should_stop) + result = pack(raw, picked, settings, finger, time.perf_counter() - started) + save(project_root, section_key(settings["start_m"], settings["end_m"]), result) + return result["summary"] diff --git a/main.py b/main.py index 795369810..5e9bd4b07 100644 --- a/main.py +++ b/main.py @@ -80,6 +80,7 @@ from D04_PlanConfirm.D04_PlanConfirm_Link import dashboard_router as d04_link_da from D04_PlanConfirm.D04_PlanConfirm_Link import router as d04_plan_link_router from D04_PlanConfirm.D04_PlanConfirm_Router import router as d04_plan_confirm_router from E01_StabRegister.E01_StabRegister_Router import router as e01_stab_register_router +from E04_Stab3D.E04_Stab3D_Router import router as e04_stab_3d_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 from M02_MasterTemplete.M02_MasterTemplete_Router import router as m02_master_template_router @@ -486,6 +487,7 @@ app.include_router(d02_plan_base_router, dependencies=protected_with_company) app.include_router(d03_plan_analysis_router, dependencies=protected_with_company) app.include_router(d04_plan_confirm_router, dependencies=protected_with_company) app.include_router(d04_plan_link_router, dependencies=protected_with_company) +app.include_router(e04_stab_3d_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) app.include_router(b05_route_terrain_router, dependencies=protected_with_company) diff --git a/resources/tester/test_533_3_e04_router.py b/resources/tester/test_533_3_e04_router.py new file mode 100644 index 000000000..93be5dea0 --- /dev/null +++ b/resources/tester/test_533_3_e04_router.py @@ -0,0 +1,264 @@ +"""E04 3D 해석 길(PLAN 533-3) — 지문 같으면 안 셈 · force 면 다시 · 끊기 · 진행률 · 저장본 · 공유 403. + +./venv/Scripts/python.exe -m pytest resources/tester/test_533_3_e04_router.py -q +""" + +import asyncio +import json +import threading +import time + +import numpy as np +import pytest +from fastapi import FastAPI, HTTPException +from fastapi.testclient import TestClient + +from common_util import common_util_auth as auth +from E04_Stab3D import E04_Stab3D_Router as router_mod +from E04_Stab3D import E04_Stab3D_Run as run + +PID = "4b1f6f0c-0c52-4a49-9d2b-6c0c3f3e0a11" +STATIONS = [ + {"chainage_m": 20.0 * i, "label": f"No.{i}", "center_x": 100.0 + 20.0 * i, "center_y": 200.0} + for i in range(11) +] + + +@pytest.fixture +def project(tmp_path, monkeypatch): + src = tmp_path / "E01_StabRegister" / "source" + models = src / "B04_PreProcess" / "models" + models.mkdir(parents=True) + (src / "copy_manifest.json").write_text(json.dumps({"source_fingerprint": "a" * 40})) + xs = np.arange(0.0, 400.0, 2.0) + ys = np.arange(100.0, 300.0, 2.0) + xx, yy = np.meshgrid(xs, ys) + z = 500.0 + 0.4 * (yy - 200.0) + 0.002 * (xx - 200.0) ** 2 # 노선 옆 비탈 + np.savez(models / "dtm_classification.npz", x=xs, y=ys, z=z, valid_mask=np.ones_like(z, bool)) + monkeypatch.setattr(run.source, "stations", lambda root: STATIONS) + monkeypatch.setattr( + run.source, + "db_snapshot", + lambda root: { + "route_points": [ + { + "model_x": s["center_x"], + "model_y": 200.0, + "chainage_m": s["chainage_m"], + "sequence_num": i, + } + for i, s in enumerate(STATIONS) + ] + }, + ) + + async def fake_root(project_id): + return tmp_path + + monkeypatch.setattr(router_mod, "_project_root", fake_root) + router_mod._JOBS.clear() + return tmp_path + + +@pytest.fixture +def client(project): + app = FastAPI() + app.include_router(router_mod.router) + with TestClient(app) as c: + yield c + + +def _wait(client, done=lambda p: not p["running"], seconds=10.0): + end = time.time() + seconds + while time.time() < end: + body = client.get(f"/api/projects/{PID}/stab/progress3d").json() + if done(body): + return body + time.sleep(0.05) + raise AssertionError("진행 기다림 초과") + + +BODY = {"start_m": 0, "end_m": 100, "search": {"box_half_width_m": 20, "cell_m": 2}} + + +def test_auto_sections(): + sections = run.auto_sections(STATIONS) + assert sections == [{"start_m": 0.0, "end_m": 100.0}, {"start_m": 100.0, "end_m": 200.0}] + short = run.auto_sections(STATIONS[:8]) # 0~140 — 꼬리 40 m 는 앞에 붙임 + assert short == [{"start_m": 0.0, "end_m": 140.0}] + + +def test_analyze_cached_force_and_result(client, project, monkeypatch): + calls = [] + real = run.run_engine + + def counted(*args, **kwargs): + calls.append(1) + return real(*args, **kwargs) + + monkeypatch.setattr(run, "run_engine", counted) + first = client.post(f"/api/projects/{PID}/stab/analyze3d", json=BODY).json() + assert first == {"status": "accepted", "key": "0.00_100.00"} + done = _wait(client) + assert done["error"] is None and done["percent"] == 100 + assert done["summary"]["cells"] > 0 + # 같은 입력 = 셈하지 않음 + again = client.post(f"/api/projects/{PID}/stab/analyze3d", json=BODY).json() + assert again["status"] == "cached" and len(calls) == 1 + # force = 다시 셈 + assert ( + client.post(f"/api/projects/{PID}/stab/analyze3d", json={**BODY, "force": True}).json()[ + "status" + ] + == "accepted" + ) + _wait(client) + assert len(calls) == 2 + # 탐색 조건이 바뀌면 지문이 달라 셈 + changed = {**BODY, "search": {**BODY["search"], "dr_m": 0.5}} + assert ( + client.post(f"/api/projects/{PID}/stab/analyze3d", json=changed).json()["status"] + == "accepted" + ) + _wait(client) + assert len(calls) == 3 + + doc = client.get(f"/api/projects/{PID}/stab/result3d", params={"key": "0.00_100.00"}).json() + grid = doc["grid"] + assert len(doc["f"]) == len(doc["z"]) == len(doc["v"]) == grid["nx"] * grid["ny"] + f = np.array([np.nan if v is None else v for v in doc["f"]]) + best = doc["best"] + assert f[best["iy"] * grid["nx"] + best["ix"]] == pytest.approx(np.nanmin(f), abs=1e-3) + assert doc["summary"]["f_min"] == pytest.approx(np.nanmin(f), abs=1e-3) + assert doc["stale"] is False + + over = client.get(f"/api/projects/{PID}/stab/result3d").json() + assert [s["key"] for s in over["saved"]] == ["0.00_100.00"] + assert len(over["line"]) == len(STATIONS) and over["sections"][0]["end_m"] == 100.0 + # E02 지반을 바꾸면 저장본이 「옛 것」 + (project / "E02_StabGround").mkdir() + (project / "E02_StabGround" / "ground.json").write_text(json.dumps({"road_surcharge_kpa": 12})) + assert client.get(f"/api/projects/{PID}/stab/result3d").json()["saved"][0]["stale"] is True + assert ( + client.post(f"/api/projects/{PID}/stab/analyze3d", json=changed).json()["status"] + == "accepted" + ) + _wait(client) + + +def test_cancel_and_restart(client, monkeypatch): + started = threading.Event() + + def slow(root, stations, ground, settings, on_progress, should_stop): + started.set() + for i in range(500): + if should_stop(): + raise InterruptedError + on_progress(i // 5, f"{i}") + time.sleep(0.01) + raise AssertionError("끊기지 않음") + + monkeypatch.setattr(run, "run_engine", slow) + assert ( + client.post(f"/api/projects/{PID}/stab/analyze3d", json=BODY).json()["status"] == "accepted" + ) + assert started.wait(5) + # 도는 중 또 누르면 409 + assert client.post(f"/api/projects/{PID}/stab/analyze3d", json=BODY).status_code == 409 + assert client.post(f"/api/projects/{PID}/stab/analyze3d/cancel").json()["stopping"] is True + body = _wait(client) + assert body["cancelled"] is True and body["error"] is None + assert ( + client.get(f"/api/projects/{PID}/stab/result3d", params={"key": "0.00_100.00"}).status_code + == 404 + ) + # 다시 시작 + monkeypatch.setattr( + run, + "run_engine", + run._preview_engine.__wrapped__ + if hasattr(run._preview_engine, "__wrapped__") + else lambda *a: run._preview_engine(a[0], a[1], a[3], a[4], a[5]), + ) + assert ( + client.post(f"/api/projects/{PID}/stab/analyze3d", json=BODY).json()["status"] == "accepted" + ) + body = _wait(client) + assert body["cancelled"] is False and body["percent"] == 100 + + +def test_bad_requests(client): + url = f"/api/projects/{PID}/stab/analyze3d" + assert client.post(url, json={"start_m": 100, "end_m": 50}).status_code == 400 + assert client.post(url, json={"start_m": 0, "end_m": 900}).status_code == 400 + bad = {**BODY, "search": {"h_min_m": 10, "h_max_m": 5}} + assert client.post(url, json=bad).status_code == 400 + assert ( + client.get(f"/api/projects/{PID}/stab/result3d", params={"key": "../x"}).status_code == 400 + ) + + +class _Cur: + def __init__(self, rows): + self.rows, self.sql = rows, "" + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def execute(self, sql, args=None): + self.sql = sql + + async def fetchone(self): + return next((v for k, v in self.rows.items() if k in self.sql), None) + + +class _Pool: + def __init__(self, rows): + self.rows = rows + + def acquire(self): + pool = self + + class _Conn: + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + def cursor(self, *args): + return _Cur(pool.rows) + + return _Conn() + + +class _Req: + def __init__(self, method): + self.method = method + self.path_params = {"project_id": PID} + + +def test_shared_company_get_only(monkeypatch): + """main.py 가 E04 길을 프로젝트 권한 검사에 걸었는지 + 지정받은 회사 = GET 만(POST 403).""" + import main + + paths = { + "/api/projects/{project_id}/stab/analyze3d", + "/api/projects/{project_id}/stab/result3d", + } + hits = [r for r in main.app.routes if getattr(r, "path", None) in paths] + assert len(hits) == 2 + for route in hits: + calls = {d.call for d in route.dependant.dependencies} + assert auth.require_project_access in calls + + user = {"role": "USER", "company_id": 1} + pool = _Pool({"FROM projects": (2,), "plan_route_assignees": (1,)}) + monkeypatch.setattr(auth, "get_db_pool", lambda: pool) + assert asyncio.run(auth.require_project_access(_Req("GET"), user)) is user + with pytest.raises(HTTPException) as e: + asyncio.run(auth.require_project_access(_Req("POST"), user)) + assert e.value.status_code == 403