eomsangdonandClaude Sonnet 5.5
79d33601f2
- 목록 판정 면 칸 · 결과 방법 줄 · 면 확인 · 모형 끝 면 배지 - 해석 설정 탐색 한계(모형 끝 / 비탈 끝 + 2H / 비탈 끝 + 정한 m) — [해석] 이 저장 + 그 값으로 셈 - Router 설정 저장본 · 지문 · 일꾼 kwargs · 단면 · 가정 · 확률 길 Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
573 lines
24 KiB
Python
573 lines
24 KiB
Python
"""E03 2D 측점 해석 API (요구 345 · 347 · 351 · PLAN 532-3).
|
|
|
|
- `POST /api/projects/{id}/stab/analyze2d` — 전 측점 해석(뒤에서 · 바뀐 측점만) · 같은 지문 = 안 셈
|
|
- `GET …/stab/progress` — 진행률
|
|
- `GET …/stab/result2d` — 요약 + 측점 목록(판정 · 대표 단면 · 벽 수 · 지배 비탈) + 횡단 없는 측점
|
|
- `GET …/stab/section/{station_key}` — 그 측점 지층 폴리곤 · 지하수선 · 임계면 · 후보 · 절편 · 격자
|
|
- `PUT …/stab/representative` — 대표 단면 켬/끔 → 작업본 `E03_StabSection/representative.json`
|
|
- `GET …/stab/corridor3d` — 복사본 노선면(1 m 솎음 · 3D 색칠 보기용)
|
|
|
|
입력 = E01 복사본(측점 · 횡단 · DTM) + E02 지반 조건(작업본 · 없으면 초기값)
|
|
계산 자리 = ③ 서버 단독.
|
|
결과 = 파생 저장본 `_derived/stab2d_station/<측점 지문>` · `_derived/stab2d_summary/<전체 지문>` —
|
|
측점 지문 = 엔진 코드 · 복사 지문 · 물성 · 조건 · 하중 · 지하수 갈래 · 그 측점 지층 · 벽.
|
|
공유 회사(보기만)는 `require_project_access` 가 POST · PUT 를 막음.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import numpy as np
|
|
import shapely
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
from scipy.spatial import cKDTree
|
|
|
|
from common_util.common_util_auth import verify_session
|
|
from common_util.common_util_derived_store import (
|
|
derived_dir,
|
|
derived_key,
|
|
read_derived,
|
|
write_derived,
|
|
)
|
|
from common_util.common_util_json import atomic_write_json
|
|
from common_util.common_util_station_key import chainage_key
|
|
from common_util.common_util_surface_sampler import DtmGridSampler
|
|
from E01_StabRegister import E01_StabRegister_Source as src
|
|
from E02_StabGround import E02_StabGround_Model as ground_model
|
|
from E02_StabGround import E02_StabGround_Router as e02
|
|
from E03_StabSection import E03_StabSection_View as view
|
|
from E03_StabSection import E03_StabSection_Walls as walls_mod
|
|
from E03_StabSection import E03_StabSection_Worker as worker
|
|
from E03_StabSection.E03_StabSection_Engine_Model import (
|
|
SEARCH_REACH,
|
|
ground_from_cross,
|
|
section_model_from_b06,
|
|
widen_ground,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/projects", tags=["E03 Stab Section"])
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
KIND_STATION = worker.KIND_STATION
|
|
KIND_SUMMARY = "stab2d_summary"
|
|
REPRESENTATIVE = Path("E03_StabSection") / "representative.json"
|
|
SETTINGS = Path("E03_StabSection") / "settings.json"
|
|
ENGINE_FILES = (
|
|
"E03_StabSection_Types.py",
|
|
"E03_StabSection_Engine_Model.py",
|
|
"E03_StabSection_Engine_LEM.py",
|
|
"E03_StabSection_Engine_Search.py",
|
|
"E03_StabSection_Engine_Station.py",
|
|
)
|
|
#: 3D 보기 노선면 솎음 간격(m)
|
|
CORRIDOR_STEP_M = 1.0
|
|
TERRAIN_MARGIN_M = 80.0 # 노선면 둘레 지형 띠 폭(B05 지형 띠 여유와 같음)
|
|
TERRAIN_STEP_M = 2.0 # 지형 칸(이보다 촘촘히 안 함)
|
|
TERRAIN_MAX_CELLS = 250_000 # 응답 크기 상한 — 넘으면 칸을 키움
|
|
#: 프로세스 수 — None = 코어 − 2 · 1 = 서버 안에서 차례로(시험)
|
|
WORKERS: int | None = None
|
|
|
|
# 진행 — 프로젝트 → {running, percent, message, error, done, total} · 프로세스 메모리
|
|
_JOBS: dict[str, dict[str, Any]] = {}
|
|
_TASKS: set[asyncio.Task] = set()
|
|
|
|
|
|
class RepresentativeRequest(BaseModel):
|
|
overrides: dict[str, bool | None]
|
|
|
|
|
|
class AnalyzeRequest(BaseModel):
|
|
#: 2D 탐색 한계 — "model_end" · "2h" · 수(비탈 끝 + m) · None = 지금 저장값 그대로
|
|
search_reach: str | float | None = None
|
|
|
|
|
|
def _clean_reach(value: Any) -> str | float:
|
|
"""탐색 한계 값 검사 — 모르는 값은 기본."""
|
|
if value in ("model_end", "2h"):
|
|
return value
|
|
try:
|
|
reach = float(value)
|
|
except (TypeError, ValueError):
|
|
return SEARCH_REACH
|
|
return reach if math.isfinite(reach) and reach > 0 else SEARCH_REACH
|
|
|
|
|
|
def search_reach_of(root: Path) -> str | float:
|
|
"""작업본 해석 설정의 탐색 한계 — 없으면 서버 기본(`SEARCH_REACH`)."""
|
|
try:
|
|
data = json.loads((root / SETTINGS).read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return SEARCH_REACH
|
|
return _clean_reach(data.get("search_reach"))
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def engine_hash() -> str:
|
|
"""엔진 코드 지문 — 코드가 바뀌면 모든 측점을 다시 셈."""
|
|
digest = hashlib.sha1()
|
|
for name in ENGINE_FILES:
|
|
digest.update((Path(__file__).parent / name).read_bytes())
|
|
return digest.hexdigest()[:12]
|
|
|
|
|
|
def _ground(root: Path) -> dict[str, Any]:
|
|
"""E02 지반 조건 — 작업본 · 없으면 초기값(옛 정본 = 빠진 지하수 칸 기본값 · E02 화면과 같음)."""
|
|
ground = e02._read_working(root) or e02.ensure_initial(root)
|
|
return ground_model.upgrade(ground, ground_model.load_defaults())
|
|
|
|
|
|
def _layers(resolved: dict[str, Any] | None) -> list | None:
|
|
"""E02 측점 적용 지층 → 엔진 지층(위부터 [물성 id, 원지반 아래 바닥 깊이 | None]).
|
|
|
|
없으면 None(엔진 기본 = B06 라벨)."""
|
|
if not resolved or not resolved.get("upper"):
|
|
return None
|
|
if resolved.get("lower"):
|
|
return [[resolved["upper"], float(resolved["soil_thickness_m"])], [resolved["lower"], None]]
|
|
return [[resolved["upper"], None]]
|
|
|
|
|
|
@lru_cache(maxsize=4)
|
|
def _walls(source: str, _copy_fp: str) -> dict[str, dict[str, Any]]:
|
|
"""측점별 벽(532-4 · Node 한 번 약 2 s) — 복사본은 동결이라 복사 지문으로 기억."""
|
|
return walls_mod.walls_by_station(source)
|
|
|
|
|
|
def plan_run(root: Path, ground: dict[str, Any]) -> dict[str, Any]:
|
|
"""측점마다 입력 · 지문 + 전체 지문(판정 기준 포함) — 복사본이 없으면 400."""
|
|
try:
|
|
source = src.source_dir(root)
|
|
copy_fp = src.manifest(source)["source_fingerprint"]
|
|
stations = src.stations(source)
|
|
except (OSError, KeyError) as error:
|
|
raise HTTPException(status_code=400, detail="원본 복사본이 없습니다.") from error
|
|
seismic = ground.get("seismic") or {}
|
|
conditions = ["dry", "wet"] + (["seismic"] if seismic.get("enabled") else [])
|
|
shared = {
|
|
"materials": ground.get("materials"),
|
|
"conditions": conditions,
|
|
"seismic_kh": float(seismic.get("kh") or 0.0) if seismic.get("enabled") else 0.0,
|
|
"road_surcharge_kpa": float(ground.get("road_surcharge_kpa") or 0.0),
|
|
"groundwater": ground.get("groundwater"),
|
|
}
|
|
reach = search_reach_of(root)
|
|
if reach != SEARCH_REACH:
|
|
shared["search_reach"] = reach # 기본값이면 빼서 옛 지문 그대로
|
|
resolved = {chainage_key(r["chainage_m"]): r for r in ground_model.resolve_stations(ground)}
|
|
try:
|
|
found = _walls(str(source), copy_fp)
|
|
except RuntimeError as error:
|
|
raise HTTPException(status_code=500, detail=str(error)) from error
|
|
items = []
|
|
for station in stations:
|
|
key = chainage_key(station["chainage_m"])
|
|
layers = _layers(resolved.get(key))
|
|
entry = found.get(key)
|
|
parts = [engine_hash(), copy_fp, shared, key, layers] + ([entry] if entry else [])
|
|
items.append(
|
|
{
|
|
"key": key,
|
|
"chainage_m": station["chainage_m"],
|
|
"label": station.get("label"),
|
|
"ground_type": station.get("ground_type"),
|
|
"center": [station.get("center_x"), station.get("center_y")],
|
|
"layers": layers,
|
|
"walls": entry,
|
|
"fp": derived_key(*parts), # 벽 없는 측점 = 종전 지문 꼴
|
|
}
|
|
)
|
|
factors = ground.get("safety_factors") or {}
|
|
overall = derived_key([item["fp"] for item in items], factors)
|
|
return {
|
|
"source": source,
|
|
"ground": ground,
|
|
"shared": shared,
|
|
"items": items,
|
|
"factors": factors,
|
|
"fp": overall,
|
|
}
|
|
|
|
|
|
def _station_input(source: Path, item: dict[str, Any], sampler: Any) -> dict[str, Any]:
|
|
"""측점 하나 엔진 입력 — 횡단 · 설계 · 넓힌 지반선(DTM 밖은 표본 그대로)."""
|
|
cross = src.cross_section(source, item["chainage_m"])
|
|
try:
|
|
ground = widen_ground(cross, sampler) if sampler else ground_from_cross(cross)
|
|
except (KeyError, TypeError, ValueError):
|
|
ground = ground_from_cross(cross)
|
|
return {
|
|
"cross": cross,
|
|
"design": walls_mod.design_with_walls(cross["design"], item.get("walls")),
|
|
"walls": walls_mod.walls_of(item.get("walls")),
|
|
"ground": [list(p) for p in ground],
|
|
}
|
|
|
|
|
|
def _sampler(source: Path) -> Any:
|
|
try:
|
|
return DtmGridSampler.from_npz(src.terrain_grid_path(source))
|
|
except (OSError, FileNotFoundError, ValueError):
|
|
logger.warning("E03 지형 격자 없음 — 횡단 표본 폭으로 해석: %s", source)
|
|
return None
|
|
|
|
|
|
def _meta(item: dict[str, Any], design: dict[str, Any]) -> dict[str, Any]:
|
|
keys = ("key", "chainage_m", "label", "ground_type", "center", "fp")
|
|
walls = len(walls_mod.walls_of(item.get("walls")) or [])
|
|
return {
|
|
**{k: item[k] for k in keys},
|
|
"section_mode": design.get("section_mode"),
|
|
"walls": walls,
|
|
}
|
|
|
|
|
|
def _run_worker(job: dict[str, Any], emit: Any) -> None:
|
|
"""일꾼 프로세스(`E03_StabSection_Worker`) — 한 줄마다 emit · 실패면 RuntimeError.
|
|
|
|
stderr 는 파일로(파이프로 두면 경고가 쌓여 일꾼이 멈춤)."""
|
|
with tempfile.TemporaryDirectory(prefix="aislo_e03_") as folder:
|
|
path = Path(folder) / "job.json"
|
|
path.write_text(json.dumps(job, ensure_ascii=False), encoding="utf-8")
|
|
log = Path(folder) / "stderr.txt"
|
|
with log.open("w", encoding="utf-8") as errors:
|
|
process = subprocess.Popen(
|
|
[sys.executable, "-m", "E03_StabSection.E03_StabSection_Worker", str(path)],
|
|
cwd=PROJECT_ROOT,
|
|
stdout=subprocess.PIPE,
|
|
stderr=errors,
|
|
text=True,
|
|
encoding="utf-8",
|
|
)
|
|
for line in process.stdout:
|
|
if line.startswith("{"):
|
|
emit(json.loads(line).get("done"))
|
|
code = process.wait()
|
|
if code != 0:
|
|
tail = log.read_text(encoding="utf-8", errors="replace").strip().splitlines()[-1:]
|
|
raise RuntimeError(f"해석 일꾼 실패: {tail}")
|
|
|
|
|
|
def run_all(root: Path, plan: dict[str, Any], report: Any) -> int:
|
|
"""지문이 없는 측점만 셈(일꾼 프로세스 병렬) → 측점 저장본 · 요약 저장본. 센 측점 수."""
|
|
source, shared, items = plan["source"], plan["shared"], plan["items"]
|
|
todo = [it for it in items if read_derived(root, KIND_STATION, it["fp"]) is None]
|
|
if todo:
|
|
sampler = _sampler(source)
|
|
job_items = []
|
|
for it in todo:
|
|
data = _station_input(source, it, sampler)
|
|
job_items.append({"fp": it["fp"], "meta": _meta(it, data["design"]), **data})
|
|
job_items[-1]["layers"] = it["layers"]
|
|
job = {
|
|
"root": str(root),
|
|
"keep": max(3 * len(items), 6),
|
|
"workers": WORKERS or max(1, (os.cpu_count() or 2) - 2),
|
|
"conditions_ground": plan["ground"],
|
|
"kwargs": {
|
|
"materials": shared["materials"],
|
|
"conditions": list(shared["conditions"]),
|
|
"seismic_kh": shared["seismic_kh"],
|
|
"road_surcharge_kpa": shared["road_surcharge_kpa"],
|
|
"search_reach": shared.get("search_reach", SEARCH_REACH),
|
|
},
|
|
"items": job_items,
|
|
}
|
|
report(2, f"측점 {len(todo)} 개 해석 시작")
|
|
done = [0]
|
|
|
|
def emit(_key: str | None) -> None:
|
|
done[0] += 1
|
|
report(2 + 95 * done[0] // len(todo), f"측점 {done[0]} / {len(todo)}")
|
|
|
|
if WORKERS == 1:
|
|
worker.run_job(job, emit)
|
|
else:
|
|
_run_worker(job, emit)
|
|
rows = []
|
|
for it in items:
|
|
saved = read_derived(root, KIND_STATION, it["fp"])
|
|
value = saved["value"] if saved else {"meta": _meta(it, {}), "out": None, "error": "없음"}
|
|
rows.append(view.station_row(value["meta"], value["out"], plan["factors"], value["error"]))
|
|
summary = {
|
|
"fingerprint": plan["fp"],
|
|
"computed_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
|
|
"conditions": shared["conditions"],
|
|
"method": view.PRIMARY,
|
|
"factors": plan["factors"],
|
|
"summary": view.summarize(rows, shared["conditions"]),
|
|
"stations": rows,
|
|
}
|
|
write_derived(root, KIND_SUMMARY, plan["fp"], summary)
|
|
return len(todo)
|
|
|
|
|
|
def _latest_summary(root: Path, fp: str) -> tuple[dict[str, Any] | None, bool]:
|
|
"""(요약, 지금 입력과 같은지) — 같은 지문이 없으면 가장 최근 것(옛 결과)."""
|
|
saved = read_derived(root, KIND_SUMMARY, fp)
|
|
if saved:
|
|
return saved["value"], True
|
|
files = sorted(
|
|
derived_dir(root, KIND_SUMMARY).glob("*.json"),
|
|
key=lambda p: p.stat().st_mtime_ns,
|
|
reverse=True,
|
|
)
|
|
for path in files:
|
|
old = read_derived(root, KIND_SUMMARY, path.stem)
|
|
if old:
|
|
return old["value"], False
|
|
return None, False
|
|
|
|
|
|
def _overrides(root: Path) -> dict[str, bool]:
|
|
path = root / REPRESENTATIVE
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return {}
|
|
return {k: bool(v) for k, v in (data.get("overrides") or {}).items() if v is not None}
|
|
|
|
|
|
@router.post("/{project_id}/stab/analyze2d", response_model=None)
|
|
async def analyze2d(project_id: UUID, req: AnalyzeRequest | None = None) -> dict[str, Any]:
|
|
"""전 측점 해석(뒤에서) — 지문이 같으면 다시 안 셈(`fresh`) · 진행은 `GET …/stab/progress`.
|
|
|
|
본문 `search_reach` 가 있으면 해석 설정에 저장한 뒤 그 값으로 셈."""
|
|
root, _ = await e02._project(project_id)
|
|
if req is not None and req.search_reach is not None:
|
|
reach = _clean_reach(req.search_reach)
|
|
if reach != search_reach_of(root):
|
|
atomic_write_json(root / SETTINGS, {"search_reach": reach})
|
|
plan = plan_run(root, _ground(root))
|
|
if read_derived(root, KIND_SUMMARY, plan["fp"]):
|
|
return {"status": "fresh", "fingerprint": plan["fp"]}
|
|
key = str(project_id)
|
|
if (_JOBS.get(key) or {}).get("running"):
|
|
raise HTTPException(status_code=409, detail="이미 해석 중입니다.")
|
|
job = {"running": True, "percent": 0, "message": "시작", "error": None, "computed": 0}
|
|
_JOBS[key] = job
|
|
|
|
def report(percent: int, message: str) -> None:
|
|
job.update(percent=int(percent), message=message)
|
|
|
|
async def work() -> None:
|
|
try:
|
|
job["computed"] = await asyncio.to_thread(run_all, root, plan, report)
|
|
job.update(percent=100, message="끝")
|
|
except Exception as error: # noqa: BLE001 — 진행률에 까닭을 그대로 올림
|
|
logger.exception("E03 2D 해석 실패: project_id=%s", project_id)
|
|
job["error"] = str(getattr(error, "detail", None) or error)
|
|
finally:
|
|
job["running"] = False
|
|
|
|
task = asyncio.create_task(work(), name=f"E03 analyze2d {project_id}")
|
|
_TASKS.add(task)
|
|
task.add_done_callback(_TASKS.discard)
|
|
return {"status": "accepted", "fingerprint": plan["fp"], **job}
|
|
|
|
|
|
@router.get("/{project_id}/stab/progress", response_model=None)
|
|
async def progress(project_id: UUID) -> dict[str, Any]:
|
|
await e02._project(project_id)
|
|
job = _JOBS.get(str(project_id))
|
|
return dict(job) if job else {"running": False, "percent": 0, "message": "", "error": None}
|
|
|
|
|
|
@router.get("/{project_id}/stab/result2d", response_model=None)
|
|
async def result2d(
|
|
project_id: UUID, session: dict[str, Any] = Depends(verify_session)
|
|
) -> dict[str, Any]:
|
|
"""요약 + 측점 목록 — 결과가 없으면 복사본 측점만(status none)."""
|
|
root, project = await e02._project(project_id)
|
|
ground = _ground(root)
|
|
plan = plan_run(root, ground)
|
|
summary, fresh = _latest_summary(root, plan["fp"])
|
|
editable = e02._editable(session, project)
|
|
running = bool((_JOBS.get(str(project_id)) or {}).get("running"))
|
|
missing = src.missing_stations(plan["source"]) # 횡단 없어 해석에서 빠진 측점(화면 띠)
|
|
if not summary:
|
|
rows = [view.station_row(_meta(item, {}), None, plan["factors"]) for item in plan["items"]]
|
|
view.mark_representative(rows, _overrides(root))
|
|
rows = [{**r, **view.governing_row(r)} for r in rows]
|
|
return {
|
|
"status": "none",
|
|
"fresh": False,
|
|
"editable": editable,
|
|
"running": running,
|
|
"conditions": plan["shared"]["conditions"],
|
|
"factors": plan["factors"],
|
|
"search_reach": plan["shared"].get("search_reach", SEARCH_REACH),
|
|
"stations": rows,
|
|
"missing_stations": missing,
|
|
}
|
|
rows = summary["stations"]
|
|
view.mark_representative(rows, _overrides(root))
|
|
rows = [{**r, **view.governing_row(r)} for r in rows]
|
|
return {
|
|
"status": "ok",
|
|
"fresh": fresh,
|
|
"editable": editable,
|
|
"running": running,
|
|
**{k: v for k, v in summary.items() if k != "stations"},
|
|
"search_reach": plan["shared"].get("search_reach", SEARCH_REACH),
|
|
"stations": rows,
|
|
"missing_stations": missing,
|
|
}
|
|
|
|
|
|
@router.get("/{project_id}/stab/section/{station_key}", response_model=None)
|
|
async def section(project_id: UUID, station_key: str) -> dict[str, Any]:
|
|
"""측점 하나 — 해석에 쓴 지반선 · 지층으로 모형을 다시 세워(빠름) 결과와 함께."""
|
|
root, _ = await e02._project(project_id)
|
|
ground = _ground(root)
|
|
plan = plan_run(root, ground)
|
|
summary, fresh = _latest_summary(root, plan["fp"])
|
|
row = next((r for r in (summary or {}).get("stations", []) if r["key"] == station_key), None)
|
|
saved = read_derived(root, KIND_STATION, row["fp"]) if row else None
|
|
if not saved or not saved["value"]["out"]:
|
|
raise HTTPException(status_code=404, detail="그 측점 해석 결과가 없습니다.")
|
|
value = saved["value"]
|
|
cross = src.cross_section(plan["source"], row["chainage_m"])
|
|
entry = next((it["walls"] for it in plan["items"] if it["key"] == station_key), None)
|
|
design = walls_mod.design_with_walls(cross["design"], entry)
|
|
kw = {
|
|
"materials": plan["shared"]["materials"],
|
|
"layers": value["inputs"]["layers"],
|
|
"ground": [tuple(p) for p in value["inputs"]["ground"]],
|
|
"road_surcharge_kpa": plan["shared"]["road_surcharge_kpa"],
|
|
"walls": walls_mod.walls_of(entry),
|
|
"search_reach": plan["shared"].get("search_reach", SEARCH_REACH),
|
|
}
|
|
dry = section_model_from_b06(cross, design, **kw)
|
|
water, rules = {}, {}
|
|
for cond in summary["conditions"]:
|
|
# 수위선 = E02 고른 갈래(r_u · 없음 = 선 없음)
|
|
model = section_model_from_b06(cross, design, wet=cond == "wet", **kw)
|
|
gw = ground_model.station_groundwater(ground, cond, model)
|
|
water[cond] = gw.get("line") if gw.get("mode") == "line" else None
|
|
rules[cond] = {"mode": gw.get("mode"), "ru": gw.get("ru")}
|
|
out = view.section_view(row, value["out"], dry, water)
|
|
out["water_rule"] = rules
|
|
out["fresh"] = fresh
|
|
return out
|
|
|
|
|
|
@router.put("/{project_id}/stab/representative", response_model=None)
|
|
async def put_representative(project_id: UUID, payload: RepresentativeRequest) -> dict[str, Any]:
|
|
"""대표 단면 켬/끔(None = 자동) — 아는 측점 열쇠만."""
|
|
root, _ = await e02._project(project_id)
|
|
known = {item["key"] for item in plan_run(root, _ground(root))["items"]}
|
|
unknown = sorted(set(payload.overrides) - known)
|
|
if unknown:
|
|
raise HTTPException(status_code=422, detail=f"모르는 측점: {unknown[:5]}")
|
|
overrides = {k: v for k, v in sorted(payload.overrides.items()) if v is not None}
|
|
atomic_write_json(root / REPRESENTATIVE, {"overrides": overrides})
|
|
return {"overrides": overrides}
|
|
|
|
|
|
def _terrain(dtm: str, origin: list[float], strips: list[np.ndarray]) -> dict[str, Any] | None:
|
|
"""설계 지형 격자(E 복사본 DTM) — 노선면 둘레 띠만 · 노면 발자국 안 = None(깎기 지형이 노면을
|
|
덮지 않게 · 가장자리 틈 한 칸 허용). 꼴 = E04 Grid + z(행 우선 iy·nx+ix) · 리본과 같은 원점."""
|
|
try:
|
|
sampler = DtmGridSampler.open_cached(dtm)
|
|
except (OSError, ValueError, KeyError):
|
|
logger.warning("E03 3D 지형 격자 못 엶: %s", dtm)
|
|
return None
|
|
pts = np.vstack([p.reshape(-1, 3)[:, :2] for p in strips])
|
|
lo = pts.min(axis=0) - TERRAIN_MARGIN_M
|
|
hi = pts.max(axis=0) + TERRAIN_MARGIN_M
|
|
area = float(np.prod(hi - lo))
|
|
dx = max(TERRAIN_STEP_M, math.ceil(math.sqrt(area / TERRAIN_MAX_CELLS) * 10) / 10)
|
|
nx, ny = (np.floor((hi - lo) / dx).astype(int) + 1).tolist()
|
|
xx, yy = np.meshgrid(lo[0] + np.arange(nx) * dx, lo[1] + np.arange(ny) * dx)
|
|
flat = np.column_stack([xx.ravel(), yy.ravel()])
|
|
near = np.isfinite(cKDTree(pts).query(flat, distance_upper_bound=TERRAIN_MARGIN_M)[0])
|
|
# 발자국 = 리본마다 첫 열 + 끝 열(거꾸로) 다각형의 합집합
|
|
rings = [np.vstack([p[:, 0, :2], p[::-1, -1, :2]]) for p in strips if len(p) >= 2]
|
|
foot = shapely.union_all([shapely.make_valid(shapely.Polygon(r)) for r in rings])
|
|
hole = shapely.contains_xy(foot, flat[:, 0], flat[:, 1])
|
|
z, valid = sampler.sample_xy(flat + np.asarray(origin[:2]))
|
|
keep = near & valid & ~hole & np.isfinite(z)
|
|
local = np.round(z - origin[2], 2)
|
|
return {
|
|
"grid": {
|
|
"x0": round(float(lo[0]), 3),
|
|
"y0": round(float(lo[1]), 3),
|
|
"dx": dx,
|
|
"nx": nx,
|
|
"ny": ny,
|
|
},
|
|
"z": [float(v) if k else None for v, k in zip(local.tolist(), keep.tolist())],
|
|
}
|
|
|
|
|
|
@lru_cache(maxsize=2)
|
|
def _corridor3d(
|
|
path: str, _stamp: tuple[int, int], dtm: str | None, _dtm_stamp: tuple[int, int] | None
|
|
) -> dict[str, Any]:
|
|
data = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
ribbons = []
|
|
strips = []
|
|
origin = None
|
|
for ribbon in data.get("ribbons") or []:
|
|
cols = int(ribbon["colCount"])
|
|
chain = np.asarray(ribbon["chainages"], dtype=float)
|
|
pos = np.frombuffer(base64.b64decode(ribbon["positionsBase64"]), dtype="<f8")
|
|
pos = pos.reshape(len(chain), cols, 3)
|
|
if origin is None and len(pos):
|
|
origin = pos[0, 0].tolist()
|
|
# 솎음 — 1 m 마다 첫 줄 + 끝 줄
|
|
bins = np.floor(np.round(chain, 6) / CORRIDOR_STEP_M)
|
|
keep = np.flatnonzero(np.r_[True, np.diff(bins) > 0])
|
|
if keep[-1] != len(chain) - 1:
|
|
keep = np.r_[keep, len(chain) - 1]
|
|
local = pos[keep] - np.asarray(origin)
|
|
strips.append(local)
|
|
ribbons.append(
|
|
{
|
|
"kind": ribbon["kind"],
|
|
"side": ribbon["side"],
|
|
"colCount": cols,
|
|
"chainages": np.round(chain[keep], 3).tolist(),
|
|
"positions": np.round(local, 3).reshape(-1).tolist(),
|
|
}
|
|
)
|
|
terrain = _terrain(dtm, origin, strips) if dtm and origin and strips else None
|
|
return {"origin": origin, "ribbons": ribbons, "terrain": terrain}
|
|
|
|
|
|
@router.get("/{project_id}/stab/corridor3d", response_model=None)
|
|
async def corridor3d(project_id: UUID) -> dict[str, Any]:
|
|
"""복사본 노선면 — 원점 빼고 1 m 솎음(모델 x · y · z) + 설계 지형 격자(노면 구멍)."""
|
|
root, _ = await e02._project(project_id)
|
|
source = src.source_dir(root)
|
|
path = src.corridor_path(source)
|
|
if not path:
|
|
raise HTTPException(status_code=404, detail="복사본에 노선면이 없습니다.")
|
|
stat = path.stat()
|
|
try:
|
|
dtm = src.terrain_grid_path(source)
|
|
dtm_stat = dtm.stat()
|
|
dtm_key = (str(dtm), (dtm_stat.st_mtime_ns, dtm_stat.st_size))
|
|
except (OSError, FileNotFoundError):
|
|
dtm_key = (None, None)
|
|
return await asyncio.to_thread(
|
|
_corridor3d, str(path), (stat.st_mtime_ns, stat.st_size), *dtm_key
|
|
)
|