eomsangdonandClaude Opus 5.5
8a50a155ab
- 검증_sub5 실측: _load_db 3.1~5.2 s(원격 DB 에 횡단 · 정점 줄마다 INSERT) → 0.5 s - test_637_lidar 줄 길이 맞춤 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
474 lines
19 KiB
Python
474 lines
19 KiB
Python
"""지형마다 설계 한 벌(765 · 요구 605 · 606-1) — 활성 벌은 제자리(DB · 파일) · 쉬는 벌은 묶음.
|
|
|
|
묶음 = `<프로젝트>/design_sets/<las|sheet>/`
|
|
db.json 노선과 자식 넷 · 단계 2 이후 행 · projects.status
|
|
files/ 벌 몫 파일(프로젝트 뿌리 기준 상대 경로 그대로)
|
|
initial_snapshot/ 그 벌의 초기값([초기화])
|
|
meta.json 지형 · 지문(필터 · 표현 · 평활 · 모델 파일 내용) · 옮긴 자리 · 못 지은 까닭
|
|
벌 몫 = B05 노선 정본 · 코리도 · B06~B09 · 배수 결과 중 노선 몫 · 초기값.
|
|
공유(안 옮김 · 안 지움) = B05 비용면 · 예정노선 · 추천 고름 · 배수 공유 캐시
|
|
(도엽 좌표 · 하강장 · 강우) · 그 밖 전부.
|
|
떠 두기 = 벌 몫을 묶음으로 옮기고(같은 볼륨 = 이름 바꿈) DB 노선 지움 · 단계 2 이후 처음.
|
|
꺼내 끼우기 = 새 노선 id 로 넣고 자식 다시 잇기 · 코리도 이름(`rekey_id_files`) · 묶음 지움.
|
|
지형 칩 = 맞바꿈(`swap_design_set`) — 실패 = `save_guard` 로 통째 되돌림 · 묶음 없음 · 지문 어긋남 =
|
|
그 자리에서 자동 설계 체인으로 한 번 지음. 계산 자리 = ③ 서버 단독.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import shutil
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import aiomysql
|
|
|
|
from common_util.common_util_initial_snapshot import (
|
|
DRAINAGE_SHARED_PREFIXES,
|
|
SNAPSHOT_DIRNAME,
|
|
_json_safe,
|
|
)
|
|
from common_util.common_util_surface_confirmation import (
|
|
AnalysisTerrain,
|
|
analysis_terrain_params,
|
|
analysis_terrain_sources,
|
|
design_terrain,
|
|
get_surface_confirmation_params,
|
|
)
|
|
|
|
SETS_DIRNAME = "design_sets"
|
|
MODELS_DIR = Path("B04_PreProcess") / "models"
|
|
TERRAINS: tuple[AnalysisTerrain, AnalysisTerrain] = ("las", "sheet")
|
|
#: 통째로 벌 몫인 폴더
|
|
_SET_DIRS = (
|
|
"B05_Profile/corridor",
|
|
"B06_Section",
|
|
"B07_DesignDetail",
|
|
"B08_Quantity",
|
|
"B09_Estimation",
|
|
SNAPSHOT_DIRNAME,
|
|
)
|
|
#: 공유를 뺀 나머지가 벌 몫인 폴더 → 공유 이름 앞머리
|
|
_SPLIT_DIRS = {
|
|
"B05_Profile/route": ("cost_surface_", "expected_route", "initial_choice"),
|
|
"B04_PreProcess/drainage": DRAINAGE_SHARED_PREFIXES,
|
|
}
|
|
_ROUTE_CHILDREN = ("route_points", "route_statistics", "longitudinal_sections", "cross_sections")
|
|
_STAGE_COLUMNS = (
|
|
"state",
|
|
"progress_percent",
|
|
"params",
|
|
"message",
|
|
"started_at",
|
|
"completed_at",
|
|
)
|
|
#: 벌이 되는 단계(노선 · 횡단 · 상세 · 수량 · 내역)
|
|
FIRST_SET_STAGE = 2
|
|
|
|
|
|
def sets_dir(root: Path) -> Path:
|
|
return Path(root) / SETS_DIRNAME
|
|
|
|
|
|
def bundle_dir(root: Path, terrain: str) -> Path:
|
|
return sets_dir(root) / terrain
|
|
|
|
|
|
def set_entries(root: Path) -> list[str]:
|
|
"""지금 제자리에 있는 벌 몫(프로젝트 뿌리 기준 상대 경로 · 폴더 또는 파일)."""
|
|
root = Path(root)
|
|
out = [name for name in _SET_DIRS if (root / name).exists()]
|
|
for folder, shared in _SPLIT_DIRS.items():
|
|
base = root / folder
|
|
if base.is_dir():
|
|
out += [
|
|
f"{folder}/{path.name}"
|
|
for path in sorted(base.iterdir())
|
|
if not path.name.startswith(shared)
|
|
]
|
|
return out
|
|
|
|
|
|
def _bundle_path(bundle: Path, relative: str) -> Path:
|
|
if relative == SNAPSHOT_DIRNAME:
|
|
return bundle / SNAPSHOT_DIRNAME
|
|
return bundle / "files" / relative
|
|
|
|
|
|
def _move(source: Path, target: Path) -> None:
|
|
if target.is_dir():
|
|
shutil.rmtree(target)
|
|
elif target.exists():
|
|
target.unlink()
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(source), str(target))
|
|
|
|
|
|
def read_meta(root: Path, terrain: str) -> dict[str, Any]:
|
|
try:
|
|
return json.loads((bundle_dir(root, terrain) / "meta.json").read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return {}
|
|
|
|
|
|
def _write_json(path: Path, value: Any) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(value, ensure_ascii=False, default=str), encoding="utf-8")
|
|
|
|
|
|
def note_unbuilt(root: Path, terrain: str, reason: str) -> None:
|
|
"""그 지형 벌을 못 지음 — 묶음 비우고 meta.json 에 까닭만."""
|
|
bundle = bundle_dir(root, terrain)
|
|
shutil.rmtree(bundle, ignore_errors=True)
|
|
_write_json(bundle / "meta.json", {"terrain": terrain, "built": False, "reason": reason})
|
|
|
|
|
|
def discard_sets(root: Path) -> bool:
|
|
"""묶음 모두 버림(전처리 다시 · 입력 바뀜)."""
|
|
target = sets_dir(root)
|
|
if not target.exists():
|
|
return False
|
|
shutil.rmtree(target, ignore_errors=True)
|
|
return True
|
|
|
|
|
|
def other_terrain(params: dict[str, Any], root: Path) -> AnalysisTerrain | None:
|
|
"""설계 지형 말고 고를 수 있는 지형 — 없으면 None."""
|
|
design = design_terrain(params)
|
|
sources = analysis_terrain_sources(params, Path(root) / MODELS_DIR)
|
|
return next((name for name in sources if name != design), None)
|
|
|
|
|
|
# ── 지문 ───────────────────────────────────────────────────────────────────
|
|
|
|
_FINGERPRINT_KEYS = ("terrain", "source_filter", "method", "smooth", "model_digest")
|
|
|
|
|
|
async def set_fingerprint(
|
|
connection: aiomysql.Connection, project_id: str, root: Path, terrain: str
|
|
) -> dict[str, Any] | None:
|
|
"""그 지형 벌이 기댄 입력 — 지형 · 필터 · 표현 · 평활 · 지표면 모델 id · 모델 파일 내용 지문.
|
|
그 지형 모델이 없으면 None."""
|
|
from B04_PreProcess.B04_PreProcess_Repository import list_surface_models
|
|
from common_util.common_util_derived_store import file_digest
|
|
|
|
params = await get_surface_confirmation_params(connection, project_id)
|
|
target = analysis_terrain_params(params, terrain) # type: ignore[arg-type]
|
|
if design_terrain(target) != terrain:
|
|
return None
|
|
for model in await list_surface_models(connection, UUID(project_id)):
|
|
generation = model.get("generation_params") or {}
|
|
if (
|
|
model["status"] in ("COMPLETE", "CONFIRMED")
|
|
and str(model["model_type"]).lower() == str(target["method"]).lower()
|
|
and str(generation.get("source_filter", "")).lower()
|
|
== str(target["source_filter"]).lower()
|
|
):
|
|
path = model.get("model_file_path")
|
|
digest = await asyncio.to_thread(file_digest, Path(root) / path if path else None)
|
|
return {
|
|
"terrain": terrain,
|
|
"source_filter": str(target["source_filter"]),
|
|
"method": str(target["method"]),
|
|
"smooth": bool(target.get("smooth")),
|
|
"model_id": int(model["id"]),
|
|
"model_digest": digest,
|
|
}
|
|
return None
|
|
|
|
|
|
def bundle_ready(root: Path, terrain: str, fingerprint: dict[str, Any] | None) -> bool:
|
|
"""묶음이 있고 지금 입력과 같은 지문인가(모델 id 는 안 봄 — 복제 · 갈음에 바뀜)."""
|
|
meta = read_meta(root, terrain)
|
|
if not fingerprint or not meta.get("built"):
|
|
return False
|
|
if not (bundle_dir(root, terrain) / "db.json").is_file():
|
|
return False
|
|
return all(meta.get(key) == fingerprint.get(key) for key in _FINGERPRINT_KEYS)
|
|
|
|
|
|
# ── DB ─────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
async def _select(cursor: Any, sql: str, args: Any) -> list[dict[str, Any]]:
|
|
await cursor.execute(sql, args)
|
|
return _json_safe([dict(row) for row in await cursor.fetchall()])
|
|
|
|
|
|
async def _dump_db(connection: aiomysql.Connection, project_id: str) -> dict[str, Any]:
|
|
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
|
dump: dict[str, Any] = {
|
|
"routes": await _select(
|
|
cursor, "SELECT * FROM routes WHERE project_id=%s ORDER BY id", (project_id,)
|
|
),
|
|
"stages": await _select(
|
|
cursor,
|
|
"SELECT stage_no, " + ", ".join(_STAGE_COLUMNS) + " FROM project_workflow_stages"
|
|
" WHERE project_id=%s AND stage_no >= %s ORDER BY stage_no",
|
|
(project_id, FIRST_SET_STAGE),
|
|
),
|
|
"project_status": (
|
|
await _select(cursor, "SELECT status FROM projects WHERE id=%s", (project_id,))
|
|
or [{}]
|
|
)[0].get("status"),
|
|
}
|
|
ids = [row["id"] for row in dump["routes"]] or [-1]
|
|
marks = ",".join(["%s"] * len(ids))
|
|
for table in _ROUTE_CHILDREN:
|
|
dump[table] = await _select(
|
|
cursor, f"SELECT * FROM {table} WHERE route_id IN ({marks}) ORDER BY id", ids
|
|
)
|
|
return dump
|
|
|
|
|
|
async def _clear_db(connection: aiomysql.Connection, project_id: str) -> None:
|
|
"""노선(자식은 CASCADE) 지움 · 단계 2 이후 처음 — 한 트랜잭션."""
|
|
from common_util.common_util_workflow_state import reset_stages_after_input_change
|
|
|
|
await connection.begin()
|
|
try:
|
|
async with connection.cursor() as cursor:
|
|
await cursor.execute("DELETE FROM routes WHERE project_id=%s", (project_id,))
|
|
await reset_stages_after_input_change(cursor, project_id, FIRST_SET_STAGE)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
|
|
|
|
async def _insert(cursor: Any, table: str, row: dict[str, Any]) -> int:
|
|
names = ", ".join(f"`{name}`" for name in row)
|
|
marks = ", ".join(["%s"] * len(row))
|
|
await cursor.execute(f"INSERT INTO {table} ({names}) VALUES ({marks})", tuple(row.values()))
|
|
return int(cursor.lastrowid)
|
|
|
|
|
|
async def _insert_many(cursor: Any, table: str, rows: list[dict[str, Any]]) -> None:
|
|
"""한 표 여러 줄을 한 번에(원격 DB 왕복 줄 · 줄마다 넣으면 칩 5 s)."""
|
|
if not rows:
|
|
return
|
|
columns = list(rows[0])
|
|
names = ", ".join(f"`{name}`" for name in columns)
|
|
marks = ", ".join(["%s"] * len(columns))
|
|
await cursor.executemany(
|
|
f"INSERT INTO {table} ({names}) VALUES ({marks})",
|
|
[tuple(row.get(name) for name in columns) for row in rows],
|
|
)
|
|
|
|
|
|
async def _load_db(
|
|
connection: aiomysql.Connection, project_id: str, dump: dict[str, Any]
|
|
) -> dict[int, int]:
|
|
"""노선 새 id 로 넣고 자식 다시 잇기 · 단계 · 상태 되씀 — 한 트랜잭션 → 옛 id → 새 id."""
|
|
route_map: dict[int, int] = {}
|
|
await connection.begin()
|
|
try:
|
|
async with connection.cursor() as cursor:
|
|
for route in dump.get("routes", []):
|
|
row = {**route, "project_id": project_id}
|
|
old = int(row.pop("id"))
|
|
route_map[old] = await _insert(cursor, "routes", row)
|
|
for table in _ROUTE_CHILDREN:
|
|
rows = []
|
|
for child in dump.get(table, []):
|
|
row = dict(child)
|
|
row.pop("id", None)
|
|
row["route_id"] = route_map[int(row["route_id"])]
|
|
if "project_id" in row:
|
|
row["project_id"] = project_id
|
|
rows.append(row)
|
|
await _insert_many(cursor, table, rows)
|
|
for stage in dump.get("stages", []):
|
|
sets = ", ".join(f"`{name}`=%s" for name in _STAGE_COLUMNS)
|
|
await cursor.execute(
|
|
f"UPDATE project_workflow_stages SET {sets}"
|
|
" WHERE project_id=%s AND stage_no=%s",
|
|
[stage.get(name) for name in _STAGE_COLUMNS] + [project_id, stage["stage_no"]],
|
|
)
|
|
if dump.get("project_status") is not None:
|
|
await cursor.execute(
|
|
"UPDATE projects SET status=%s WHERE id=%s",
|
|
(dump["project_status"], project_id),
|
|
)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
return route_map
|
|
|
|
|
|
# ── 떠 두기 · 꺼내 끼우기 ───────────────────────────────────────────────────
|
|
|
|
|
|
def _clear_files(root: Path) -> None:
|
|
for relative in set_entries(root):
|
|
path = Path(root) / relative
|
|
if path.is_dir():
|
|
shutil.rmtree(path, ignore_errors=True)
|
|
else:
|
|
path.unlink(missing_ok=True)
|
|
|
|
|
|
async def clear_active_set(connection: aiomysql.Connection, root: Path, project_id: str) -> None:
|
|
"""제자리 벌 비움(못 지은 벌 · 반쯤 지은 결과) — 공유 캐시는 그대로."""
|
|
await _clear_db(connection, project_id)
|
|
await asyncio.to_thread(_clear_files, root)
|
|
|
|
|
|
def _park_files(root: Path, work: Path) -> list[str]:
|
|
entries = set_entries(root)
|
|
for relative in entries:
|
|
_move(Path(root) / relative, _bundle_path(work, relative))
|
|
return entries
|
|
|
|
|
|
async def park_active_set(
|
|
connection: aiomysql.Connection,
|
|
root: Path,
|
|
project_id: str,
|
|
fingerprint: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""제자리 벌을 `design_sets/<지형>/` 으로 떠 두고 제자리를 비움 → meta."""
|
|
root = Path(root)
|
|
terrain = str(fingerprint["terrain"])
|
|
dump = await _dump_db(connection, project_id)
|
|
work = sets_dir(root) / f".{terrain}_new"
|
|
await asyncio.to_thread(shutil.rmtree, work, True)
|
|
_write_json(work / "db.json", dump)
|
|
entries = await asyncio.to_thread(_park_files, root, work)
|
|
meta = {
|
|
**fingerprint,
|
|
"built": True,
|
|
"entries": entries,
|
|
"routes": [row["id"] for row in dump["routes"]],
|
|
"parked_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
_write_json(work / "meta.json", meta)
|
|
await _clear_db(connection, project_id)
|
|
bundle = bundle_dir(root, terrain)
|
|
await asyncio.to_thread(shutil.rmtree, bundle, True)
|
|
work.rename(bundle)
|
|
return meta
|
|
|
|
|
|
def _unpark_files(root: Path, bundle: Path, entries: list[str], route_map: dict) -> None:
|
|
from common_util.common_util_project_copy import rekey_id_files
|
|
|
|
for relative in entries:
|
|
source = _bundle_path(bundle, relative)
|
|
if source.exists():
|
|
_move(source, Path(root) / relative)
|
|
rekey_id_files(Path(root), {}, route_map)
|
|
shutil.rmtree(bundle, ignore_errors=True)
|
|
|
|
|
|
async def load_set(
|
|
connection: aiomysql.Connection, root: Path, project_id: str, terrain: str
|
|
) -> dict[int, int] | None:
|
|
"""묶음을 제자리로 꺼내 끼움(제자리는 비어 있어야 함) → 옛 노선 id → 새 id ·
|
|
묶음 없음 = None."""
|
|
bundle = bundle_dir(root, terrain)
|
|
try:
|
|
dump = json.loads((bundle / "db.json").read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return None
|
|
meta = read_meta(root, terrain)
|
|
route_map = await _load_db(connection, project_id, dump)
|
|
await asyncio.to_thread(_unpark_files, root, bundle, list(meta.get("entries") or []), route_map)
|
|
return route_map
|
|
|
|
|
|
# ── 지형 바꿈 ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
async def switch_terrain(connection: aiomysql.Connection, project_id: UUID, terrain: str) -> int:
|
|
"""설계 지형(stage 1 확정값 · 확정 모델)을 바꿈 — 한 트랜잭션 → 확정 모델 id."""
|
|
from B04_PreProcess.B04_PreProcess_Service import switch_design_terrain
|
|
|
|
await connection.begin()
|
|
try:
|
|
switched = await switch_design_terrain(connection, project_id, terrain) # type: ignore[arg-type]
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
if switched:
|
|
return int(switched[0])
|
|
async with connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"SELECT id FROM surface_models WHERE project_id=%s AND status='CONFIRMED'",
|
|
(str(project_id),),
|
|
)
|
|
row = await cursor.fetchone()
|
|
if not row:
|
|
raise LookupError("확정 지표면 모델이 없습니다.")
|
|
return int(row[0])
|
|
|
|
|
|
async def swap_design_set(project_id: UUID, root: Path, target: str) -> dict[str, Any]:
|
|
"""지형 칩 = 맞바꿈 — 지금 벌 떠 둠 → 지형 바꿈 → 고른 벌 꺼내 끼움(없음 · 묵음 = 한 번 지음).
|
|
실패 = DB 행 · 벌 몫 · 묶음을 떠 둔 대로 되돌리고 예외 그대로. 잠금은 부르는 쪽."""
|
|
from B03_FileInput.B03_FileInput_Service_Chain import run_auto_design_chain
|
|
from common_util import common_util_save_guard as guard
|
|
from common_util.common_util_initial_snapshot import (
|
|
clear_design_failed,
|
|
clear_designing,
|
|
is_design_failed,
|
|
read_design_failure,
|
|
)
|
|
from config.config_db import get_db_pool
|
|
|
|
started = time.perf_counter()
|
|
pid, root = str(project_id), Path(root)
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
now = design_terrain(await get_surface_confirmation_params(connection, pid))
|
|
if now == target:
|
|
return {"changed": False, "design": now}
|
|
here = await set_fingerprint(connection, pid, root, now) or {"terrain": now}
|
|
there = await set_fingerprint(connection, pid, root, target)
|
|
if there is None:
|
|
raise LookupError("고른 지형의 지표면 모델이 없습니다.")
|
|
rows = await guard.snapshot_rows(connection, pid, surface=True)
|
|
ready = bundle_ready(root, target, there)
|
|
# 되돌림 자리 = 지금 벌 몫 + 꺼낼 벌 몫(끼운 뒤 생기는 파일도 지우게) + 묶음
|
|
scope = tuple(
|
|
dict.fromkeys(
|
|
[*set_entries(root), *(read_meta(root, target).get("entries") or []), SETS_DIRNAME]
|
|
)
|
|
)
|
|
files = await asyncio.to_thread(guard.snapshot_storage, root, scope)
|
|
had_failed = is_design_failed(root)
|
|
had_bundle = bundle_dir(root, now).exists()
|
|
try:
|
|
async with pool.acquire() as connection:
|
|
await park_active_set(connection, root, pid, here)
|
|
model_id = await switch_terrain(connection, project_id, target)
|
|
loaded = await load_set(connection, root, pid, target) if ready else None
|
|
if loaded is None:
|
|
await asyncio.to_thread(shutil.rmtree, bundle_dir(root, target), True)
|
|
summary = await run_auto_design_chain(project_id, model_id, inner=True)
|
|
if summary is None:
|
|
raise RuntimeError(read_design_failure(root) or "고른 지형 설계를 짓지 못했습니다.")
|
|
except BaseException:
|
|
files.restore()
|
|
if not had_bundle: # 되돌림은 파일만 — 떠 둔 빈 폴더도 걷음
|
|
shutil.rmtree(bundle_dir(root, now), ignore_errors=True)
|
|
async with pool.acquire() as connection:
|
|
await guard.restore_rows(connection, pid, rows)
|
|
if not had_failed:
|
|
clear_design_failed(root)
|
|
raise
|
|
finally:
|
|
files.discard()
|
|
clear_designing(root) # 지음 길(inner 체인)이 건 마커
|
|
return {
|
|
"changed": True,
|
|
"design": target,
|
|
"built": loaded is None,
|
|
"seconds": round(time.perf_counter() - started, 2),
|
|
}
|