eomsangdonandClaude Opus 5.5
8f7bb8e7e6
- 새 E02_StabGround_Chain.py — 페이지 해석 함수 그대로 부름 · 지문 같은 단계 건너뜀 · 단계 실패 격리 · 페이지 _JOBS 꽂음 · 3D 진행률 = 구간 × 조건 수로 나눔 - E02 Router 길 셋(POST /stab/run · GET /stab/run/progress · POST /stab/run/cancel) · PUT 과 같은 저장 함수 - 3D 열쇠 = section_key(start, end, condition) 꼴(564-2 뒤 실제 연결) · 단계 상태 적기 자리 = 565-1 뒤 - 새 시험 test_564_1_e02_chain.py(차례 · 비율 · 건너뜀 · 끊기 · 실패 격리 · 3D 조각 · 길) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
289 lines
12 KiB
Python
289 lines
12 KiB
Python
"""E02 [해석 시작] 체인 (요구 390 1 · PLAN 564-1).
|
|
|
|
차례 = 2D → 3D(모든 구간 · 건기 · 우기) → 암반 → 광역 → 보고서. 계산 자리 = ③ 서버 단독.
|
|
각 페이지 해석 함수를 그대로 부름(그 파일 안 고침) · 단계마다 입력 지문이 저장본과 같으면 건너뜀
|
|
· 결과는 각 페이지 저장 자리에 영구 저장 · 한 단계가 실패해도 다음 단계로.
|
|
도는 동안 그 페이지 진행 표(`_JOBS`)에 같은 일감을 꽂음 → E03 · E04 · E07 화면이 「해석 중」
|
|
· 그 페이지 [해석] 은 409. [끊기] = 다음 확인 때 멈춤(3D 는 엔진 안 · 그 밖은 단계 사이).
|
|
|
|
진행 꼴 = {running, percent, stage, stage_name, steps:[{key, name, state, seconds}],
|
|
message, error, cancelled} · state = wait · run · done · skip · error · cancel
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from collections.abc import Awaitable, Callable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
#: 단계 — (열쇠, 이름, 비율 몫) · 몫 = 1.2 km 실측 시간 비(2D 1~3 분 · 3D 건기 · 우기 약 25 분)
|
|
STEPS: tuple[tuple[str, str, float], ...] = (
|
|
("2d", "2D 측점 해석", 10.0),
|
|
("3d", "3D 구간 해석", 84.0),
|
|
("rock", "암반 해석", 2.0),
|
|
("area", "광역 위험", 2.0),
|
|
("report", "보고서", 2.0),
|
|
)
|
|
CONDITIONS_3D = ("dry", "wet")
|
|
|
|
Report = Callable[[float, str], None] # (단계 안 비율 0~1, 글)
|
|
Stopped = Callable[[], bool]
|
|
Step = Callable[[Report, Stopped], Awaitable[str]] # → "done" | "skip"
|
|
|
|
_JOBS: dict[str, dict[str, Any]] = {}
|
|
_STOPS: dict[str, bool] = {}
|
|
_TASKS: set[asyncio.Task] = set()
|
|
|
|
|
|
def idle() -> dict[str, Any]:
|
|
return {
|
|
"running": False,
|
|
"percent": 0,
|
|
"stage": None,
|
|
"stage_name": None,
|
|
"steps": [{"key": k, "name": n, "state": "wait", "seconds": None} for k, n, _ in STEPS],
|
|
"message": None,
|
|
"error": None,
|
|
"cancelled": False,
|
|
}
|
|
|
|
|
|
def _mark_stage(project_id: str, key: str, state: str) -> None:
|
|
"""대시보드 단계 상태 — ponytail: 지금은 안 적음 · 565-1 함수가 들어오면 여기서 부름."""
|
|
|
|
|
|
async def run_chain(
|
|
job: dict[str, Any], steps: dict[str, Step], stopped: Stopped, project_id: str = ""
|
|
) -> None:
|
|
"""STEPS 차례로 돎 — job 을 제자리에서 고침(진행 GET 이 그대로 읽음)."""
|
|
total = sum(w for _, _, w in STEPS)
|
|
base = 0.0
|
|
errors: list[str] = []
|
|
for index, (key, name, weight) in enumerate(STEPS):
|
|
row = job["steps"][index]
|
|
if stopped():
|
|
job.update(cancelled=True, message="끊음")
|
|
row["state"] = "cancel"
|
|
break
|
|
job.update(stage=key, stage_name=name, message=name)
|
|
row["state"] = "run"
|
|
_mark_stage(project_id, key, "run")
|
|
|
|
def report(frac: float, message: str, base: float = base, weight: float = weight) -> None:
|
|
frac = min(1.0, max(0.0, float(frac)))
|
|
job.update(percent=min(99, int(100 * (base + weight * frac) / total)), message=message)
|
|
|
|
started = time.perf_counter()
|
|
try:
|
|
row["state"] = await steps[key](report, stopped)
|
|
except InterruptedError:
|
|
row["state"] = "cancel"
|
|
job.update(cancelled=True, message="끊음")
|
|
break
|
|
except Exception as error: # noqa: BLE001 — 단계 실패 격리 · 까닭은 그대로 올림
|
|
logger.exception("E02 체인 %s 실패: project_id=%s", key, project_id)
|
|
row["state"] = "error"
|
|
errors.append(f"{name}: {getattr(error, 'detail', None) or error}")
|
|
finally:
|
|
row["seconds"] = round(time.perf_counter() - started, 1)
|
|
_mark_stage(project_id, key, row["state"])
|
|
base += weight
|
|
job["percent"] = min(99, int(100 * base / total))
|
|
job["error"] = " · ".join(errors) or None
|
|
if not job["cancelled"]:
|
|
job.update(percent=100, message="끝")
|
|
job.update(stage=None, stage_name=None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 실제 단계 — 페이지 함수 부르기(늦은 import · E03 · E06 라우터가 E02 라우터를 import 함)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _plug(jobs: dict[str, dict[str, Any]], pid: str, extra: dict[str, Any]) -> dict[str, Any]:
|
|
"""그 페이지 진행 표에 체인 일감을 꽂음 — 페이지 화면 「해석 중」 · 페이지 [해석] 409."""
|
|
job = {"running": True, "percent": 0, "message": "한 번에 해석", "error": None, **extra}
|
|
jobs[pid] = job
|
|
return job
|
|
|
|
|
|
def build_steps(project_id: UUID, project_root: Path) -> dict[str, Step]:
|
|
from E01_StabRegister import E01_StabRegister_Source as source
|
|
from E03_StabSection import E03_StabSection_Router as e03
|
|
from E04_Stab3D import E04_Stab3D_Router as e04
|
|
from E04_Stab3D import E04_Stab3D_Run as run3d
|
|
from E05_StabReport import E05_StabReport_Build as e05_build
|
|
from E05_StabReport import E05_StabReport_Router as e05
|
|
from E05_StabReport import E05_StabReport_Store as e05_store
|
|
from E06_StabRock import E06_StabRock_Router as e06
|
|
from E07_StabArea import E07_StabArea_Router as e07
|
|
|
|
pid = str(project_id)
|
|
root = project_root
|
|
|
|
async def step_2d(report: Report, _stopped: Stopped) -> str:
|
|
plan = await asyncio.to_thread(e03.plan_run, root, e03._ground(root))
|
|
if e03.read_derived(root, e03.KIND_SUMMARY, plan["fp"]):
|
|
return "skip"
|
|
page = _plug(e03._JOBS, pid, {"computed": 0})
|
|
|
|
def relay(percent: int, message: str) -> None:
|
|
page.update(percent=int(percent), message=message)
|
|
report(percent / 100, f"2D {message}")
|
|
|
|
try:
|
|
page["computed"] = await asyncio.to_thread(e03.run_all, root, plan, relay)
|
|
page.update(percent=100, message="끝")
|
|
finally:
|
|
page["running"] = False
|
|
return "done"
|
|
|
|
def search_settings() -> dict[str, Any]:
|
|
"""3D 탐색 설정 = 가장 최근 저장본 설정 · 없으면 기본값(구간 · 조건 빼고)."""
|
|
files = sorted(run3d.results_dir(root).glob("*.json"), key=lambda p: p.stat().st_mtime_ns)
|
|
saved = run3d.load(root, files[-1].stem) if files else None
|
|
fields = e04.SearchSettings.model_fields
|
|
values = {k: v for k, v in ((saved or {}).get("settings") or {}).items() if k in fields}
|
|
return e04.SearchSettings(**values).model_dump()
|
|
|
|
async def step_3d(report: Report, stopped: Stopped) -> str:
|
|
src = e04._source_root(root)
|
|
stations = await asyncio.to_thread(source.stations, src)
|
|
sections = run3d.auto_sections(stations)
|
|
search = await asyncio.to_thread(search_settings)
|
|
units = [(s, c) for s in sections for c in CONDITIONS_3D]
|
|
computed = 0
|
|
for index, (section, condition) in enumerate(units):
|
|
if stopped():
|
|
raise InterruptedError
|
|
settings = {
|
|
"start_m": round(section["start_m"], 2),
|
|
"end_m": round(section["end_m"], 2),
|
|
"condition": condition,
|
|
**search,
|
|
}
|
|
key = run3d.section_key(settings["start_m"], settings["end_m"], condition)
|
|
label = f"3D {index + 1} / {len(units)} 구간"
|
|
finger = await asyncio.to_thread(e04._current_fingerprint, root, src, settings)
|
|
saved = await asyncio.to_thread(run3d.load, root, key)
|
|
if saved and saved["fingerprint"] == finger:
|
|
report((index + 1) / len(units), label)
|
|
continue
|
|
page = _plug(e04._JOBS, pid, {**e04._IDLE, "running": True, "key": key})
|
|
e04._STOPS[pid] = False
|
|
|
|
def relay(percent: int, message: str, index: int = index, page=page) -> None:
|
|
page.update(percent=max(0, min(99, int(percent))), message=message)
|
|
report((index + percent / 100) / len(units), label)
|
|
|
|
try:
|
|
page["summary"] = await asyncio.to_thread(
|
|
run3d.analyze,
|
|
root,
|
|
src,
|
|
settings,
|
|
finger,
|
|
relay,
|
|
lambda: stopped() or e04._STOPS.get(pid, False),
|
|
)
|
|
page.update(percent=100, message="끝")
|
|
except InterruptedError:
|
|
page.update(cancelled=True, message="끊음")
|
|
raise
|
|
finally:
|
|
page["running"] = False
|
|
computed += 1
|
|
return "done" if computed else "skip"
|
|
|
|
async def step_rock(_report: Report, _stopped: Stopped) -> str:
|
|
out = await asyncio.to_thread(e06._run, root)
|
|
return "skip" if out["status"] == "fresh" else "done"
|
|
|
|
async def step_area(report: Report, _stopped: Stopped) -> str:
|
|
src = e07._source_root(root)
|
|
saved = await asyncio.to_thread(e07.load, root)
|
|
base = (saved or {}).get("settings") or e07.defaults(root)
|
|
settings = e07.AreaRequest(**base).model_dump(exclude={"force"})
|
|
finger = e07.fingerprint(source.manifest(src)["source_fingerprint"], settings)
|
|
if saved and saved["fingerprint"] == finger:
|
|
return "skip"
|
|
page = _plug(e07._JOBS, pid, {})
|
|
|
|
def relay(percent: int, message: str) -> None:
|
|
page.update(percent=max(0, min(99, int(percent))), message=message)
|
|
report(percent / 100, f"광역 {message}")
|
|
|
|
try:
|
|
await asyncio.to_thread(e07.run, root, src, settings, finger, relay)
|
|
page.update(percent=100, message="끝")
|
|
finally:
|
|
page["running"] = False
|
|
return "done"
|
|
|
|
async def step_report(_report: Report, _stopped: Stopped) -> str:
|
|
async with e05._lock(project_id):
|
|
book, meta = e05_store.load(root)
|
|
fresh = await asyncio.to_thread(e05_build.analysis_fingerprint, root)
|
|
if book is not None and meta and meta.get("fingerprint") == fresh:
|
|
return "skip"
|
|
await asyncio.to_thread(e05_store.refresh, root)
|
|
return "done"
|
|
|
|
return {
|
|
"2d": step_2d,
|
|
"3d": step_3d,
|
|
"rock": step_rock,
|
|
"area": step_area,
|
|
"report": step_report,
|
|
}
|
|
|
|
|
|
def busy(project_id: UUID) -> bool:
|
|
"""체인 · 페이지 해석 하나라도 도는지 — 한 프로젝트에 체인 하나만."""
|
|
from E03_StabSection import E03_StabSection_Router as e03
|
|
from E04_Stab3D import E04_Stab3D_Router as e04
|
|
from E07_StabArea import E07_StabArea_Router as e07
|
|
|
|
pid = str(project_id)
|
|
return any(
|
|
(jobs.get(pid) or {}).get("running") for jobs in (_JOBS, e03._JOBS, e04._JOBS, e07._JOBS)
|
|
)
|
|
|
|
|
|
def start(project_id: UUID, project_root: Path) -> dict[str, Any]:
|
|
"""뒤에서 체인 시작 — 부르는 쪽이 busy 를 먼저 봄."""
|
|
pid = str(project_id)
|
|
job = {**idle(), "running": True, "message": "시작"}
|
|
_JOBS[pid] = job
|
|
_STOPS[pid] = False
|
|
steps = build_steps(project_id, project_root)
|
|
|
|
async def work() -> None:
|
|
try:
|
|
await run_chain(job, steps, lambda: _STOPS.get(pid, False), pid)
|
|
finally:
|
|
job["running"] = False
|
|
|
|
task = asyncio.create_task(work(), name=f"E02 chain {pid}")
|
|
_TASKS.add(task)
|
|
task.add_done_callback(_TASKS.discard)
|
|
return job
|
|
|
|
|
|
def progress(project_id: UUID) -> dict[str, Any]:
|
|
return dict(_JOBS.get(str(project_id)) or idle())
|
|
|
|
|
|
def cancel(project_id: UUID) -> bool:
|
|
pid = str(project_id)
|
|
running = bool((_JOBS.get(pid) or {}).get("running"))
|
|
if running:
|
|
_STOPS[pid] = True
|
|
return running
|