From a64d0fc57eb32f6fa20414a213592ad1cd04676b Mon Sep 17 00:00:00 2001 From: umsangdon Date: Tue, 3 Nov 2026 15:42:28 +0900 Subject: [PATCH] =?UTF-8?q?feat(765-2):=20=EC=A7=80=ED=98=95=EB=A7=88?= =?UTF-8?q?=EB=8B=A4=20=EC=84=A4=EA=B3=84=20=ED=95=9C=20=EB=B2=8C=20?= =?UTF-8?q?=EB=AC=B6=EC=9D=8C=20=EB=AA=A8=EB=93=88=20=E2=80=94=20=EB=96=A0?= =?UTF-8?q?=20=EB=91=90=EA=B8=B0=20=C2=B7=20=EA=BA=BC=EB=82=B4=20=EB=81=BC?= =?UTF-8?q?=EC=9A=B0=EA=B8=B0(=EC=83=88=20=EB=85=B8=EC=84=A0=20id=20=C2=B7?= =?UTF-8?q?=20=EC=9E=90=EC=8B=9D=20=EB=8B=A4=EC=8B=9C=20=EC=9E=87=EA=B8=B0?= =?UTF-8?q?=20=C2=B7=20=EC=BD=94=EB=A6=AC=EB=8F=84=20=EC=9D=B4=EB=A6=84)?= =?UTF-8?q?=20=C2=B7=20=EC=A7=80=EB=AC=B8=20=C2=B7=20=EB=A7=9E=EB=B0=94?= =?UTF-8?q?=EA=BF=88(=EC=8B=A4=ED=8C=A8=20=3D=20=ED=86=B5=EC=A7=B8=20?= =?UTF-8?q?=EB=90=98=EB=8F=8C=EB=A6=BC)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - common_util_design_sets: 활성 벌 = 제자리 · 쉬는 벌 = design_sets//(db.json · files/ · initial_snapshot/ · meta.json) - 벌 몫 = B05 노선 정본 · 코리도 · B06~B09 · 배수 노선 몫 · 초기값 · 공유 캐시(비용면 · 예정노선 · 하강장 · 도엽 좌표 · 강우)는 안 옮김 · 안 지움 - 옮김 = 같은 볼륨 이름 바꿈 · 단계 2 이후 행 · projects.status 함께 - 묶음 없음 · 지문 어긋남 = 그 자리에서 자동 설계 체인 한 번(inner = 마커 · 미리 셈은 부르는 쪽) - 시험 test_765_design_sets 6 (왕복 값 같음 · 새 id · 묵은 벌 · 실패 되돌림) Co-Authored-By: Claude Opus 5.5 --- B03_FileInput/B03_FileInput_Service_Chain.py | 11 +- common_util/common_util_design_sets.py | 458 +++++++++++++++++++ common_util/common_util_initial_snapshot.py | 4 + resources/tester/test_765_design_sets.py | 413 +++++++++++++++++ 4 files changed, 882 insertions(+), 4 deletions(-) create mode 100644 common_util/common_util_design_sets.py create mode 100644 resources/tester/test_765_design_sets.py diff --git a/B03_FileInput/B03_FileInput_Service_Chain.py b/B03_FileInput/B03_FileInput_Service_Chain.py index 34a4eceed..29381950a 100644 --- a/B03_FileInput/B03_FileInput_Service_Chain.py +++ b/B03_FileInput/B03_FileInput_Service_Chain.py @@ -155,10 +155,12 @@ async def _prepare_drainage_pipes_and_reprofile( async def run_auto_design_chain( - project_id: UUID, surface_model_id: int | None = None + project_id: UUID, surface_model_id: int | None = None, *, inner: bool = False ) -> dict[str, Any] | None: """B05 기본 경로 계산·확정 → B06 기본 횡단 설계 확정을 기본값으로 이어 실행한다. + `inner` = 두 벌 짓기 · 지형 칩 안에서 부름(765) — 진입 막음 마커 · 추천 미리 셈은 부르는 쪽. + WF1 자동 확정 직후 같은 백그라운드 태스크에서 호출된다. 어떤 단계가 실패해도 예외를 밖으로 던지지 않는다 — 로그와 각 단계의 workflow 상태 기록으로 남긴다. @@ -359,12 +361,13 @@ async def run_auto_design_chain( mark_design_failed(project_root, f"초기 설계 처리 중 오류: {exc}") finally: # 성공·실패·중단 어느 쪽이든 문은 연다 — 마커가 남으면 영영 못 들어간다. - if project_root is not None: + if project_root is not None and not inner: clear_designing(project_root) # 성공 · 실패 · 건너뜀 어느 쪽이든 추천 미리 셈 예약(763-3) — 추천은 지형 · 예정노선만 씀 - from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute + if not inner: + from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute - schedule_precompute(project_id) + schedule_precompute(project_id) async def _ensure_initial_polyline( diff --git a/common_util/common_util_design_sets.py b/common_util/common_util_design_sets.py new file mode 100644 index 000000000..83a3dc8d0 --- /dev/null +++ b/common_util/common_util_design_sets.py @@ -0,0 +1,458 @@ +"""지형마다 설계 한 벌(765 · 요구 605 · 606-1) — 활성 벌은 제자리(DB · 파일) · 쉬는 벌은 묶음. + +묶음 = `<프로젝트>/design_sets//` + 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 _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: + 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 + await _insert(cursor, table, row) + 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), + } diff --git a/common_util/common_util_initial_snapshot.py b/common_util/common_util_initial_snapshot.py index b6ee4fe88..07b8dfc9c 100644 --- a/common_util/common_util_initial_snapshot.py +++ b/common_util/common_util_initial_snapshot.py @@ -38,6 +38,10 @@ DESIGN_FAILED_NAME = "initial_design.failed" # 확정: 「초기값 = 파일 입력 직후 결과 전부」). 관을 옮기면 세부유역(`04_detailed_basins`)이 # 다시 나뉘는데 그 산출물이 스냅샷 밖이라 [초기화]가 옛 유역도를 그대로 남겼다. # 용량은 실측 8.6MB(스냅샷 전체 3.9MB → 약 12MB)로 감당할 만하다. +#: 배수 폴더 중 노선과 무관한 공유 캐시(이름 앞머리 · 도엽 좌표 · 하강장 · 강우) — 초기값 사본 · +#: 설계 벌에 안 넣고 복원 · 맞바꿈이 안 지움(765-6) +DRAINAGE_SHARED_PREFIXES = ("00_sheet_metric", "descent_", "rainfall_") + _FILE_TREES = ( "B05_Profile/route", "B06_Section/longitudinal", diff --git a/resources/tester/test_765_design_sets.py b/resources/tester/test_765_design_sets.py new file mode 100644 index 000000000..0cca50bba --- /dev/null +++ b/resources/tester/test_765_design_sets.py @@ -0,0 +1,413 @@ +# -*- coding: utf-8 -*- +"""765-2 지형마다 설계 한 벌 — 떠 두기 · 꺼내 끼우기 · 맞바꿈(`common_util_design_sets`). + +DB = 메모리 sqlite 를 aiomysql 꼴로 감싼 가짜(노선 · 자식 넷 · 단계 · 프로젝트 · 지표면 모델) · +storage = 시험 안 임시 폴더. 지킴: + ① 떠 둔 뒤 꺼내면 값 같음 — 노선 · 자식 · 단계 · 상태 · 벌 몫 파일 · 초기값 · 코리도 + ② 꺼낼 때 새 노선 id · 자식 다시 잇기 · 코리도 이름 새 id + ③ 공유 캐시(비용면 · 예정노선 · 배수 하강장 · 도엽 좌표 · 강우)는 안 옮기고 안 지움 + ④ 지문 어긋난(묵은) 벌은 버리고 그 자리에서 지음 · 실패면 통째 되돌림 +""" + +from __future__ import annotations + +import asyncio +import json +import re +import sqlite3 +import sys +from pathlib import Path +from uuid import uuid4 + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import aiomysql # noqa: E402 + +from common_util import common_util_design_sets as sets # noqa: E402 + +PID = str(uuid4()) + +_SCHEMA = """ +CREATE TABLE projects (id TEXT PRIMARY KEY, status TEXT, name TEXT); +CREATE TABLE surface_models (id INTEGER PRIMARY KEY, project_id TEXT, status TEXT); +CREATE TABLE project_workflow_stages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT, stage_no INT, stage_key TEXT, + state TEXT, progress_percent INT, params TEXT, message TEXT, started_at TEXT, completed_at TEXT); +CREATE TABLE routes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT, surface_model_id INT, status TEXT, + total_length_m REAL, route_data_path TEXT, created_at TEXT); +CREATE TABLE route_points ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + route_id INT REFERENCES routes(id) ON DELETE CASCADE, chainage_m REAL, sequence_num INT); +CREATE TABLE route_statistics ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + route_id INT REFERENCES routes(id) ON DELETE CASCADE, cut_volume_m3 REAL); +CREATE TABLE longitudinal_sections ( + id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT, + route_id INT REFERENCES routes(id) ON DELETE CASCADE, data TEXT); +CREATE TABLE cross_sections ( + id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT, + route_id INT REFERENCES routes(id) ON DELETE CASCADE, chainage_m REAL, data TEXT); +""" + + +class _Cursor: + def __init__(self, db: sqlite3.Connection, dict_rows: bool) -> None: + self.db, self.dict_rows, self.rows, self.lastrowid, self.rowcount = db, dict_rows, [], 0, 0 + + async def __aenter__(self): + return self + + async def __aexit__(self, *_): + return False + + @staticmethod + def _sql(sql: str) -> str: + return re.sub(r"%s", "?", sql) + + async def execute(self, sql: str, args=()) -> None: + cur = self.db.execute(self._sql(sql), tuple(args or ())) + self.lastrowid, self.rowcount = cur.lastrowid, cur.rowcount + names = [d[0] for d in cur.description or []] + rows = cur.fetchall() + self.rows = [dict(zip(names, r)) for r in rows] if self.dict_rows else rows + + async def executemany(self, sql: str, seq) -> None: + self.db.executemany(self._sql(sql), [tuple(a) for a in seq]) + + async def fetchall(self): + return self.rows + + async def fetchone(self): + return self.rows[0] if self.rows else None + + +class _Conn: + def __init__(self, db: sqlite3.Connection) -> None: + self.db = db + + def cursor(self, kind=None): + return _Cursor(self.db, kind is aiomysql.DictCursor) + + async def begin(self): + pass + + async def commit(self): + self.db.commit() + + async def rollback(self): + self.db.rollback() + + +class _Pool: + def __init__(self, conn: _Conn) -> None: + self.conn = conn + + def acquire(self): + conn = self.conn + + class _Ctx: + async def __aenter__(self): + return conn + + async def __aexit__(self, *_): + return False + + return _Ctx() + + +def _seed_db() -> sqlite3.Connection: + db = sqlite3.connect(":memory:") + db.execute("PRAGMA foreign_keys = ON") + db.executescript(_SCHEMA) + db.execute("INSERT INTO projects VALUES (?, 'WF3_ANALYZING', 'p')", (PID,)) + db.executemany( + "INSERT INTO surface_models VALUES (?, ?, ?)", + [(5, PID, "CONFIRMED"), (6, PID, "COMPLETE")], + ) + for no in range(7): + db.execute( + "INSERT INTO project_workflow_stages (project_id, stage_no, stage_key, state," + " progress_percent, params, message) VALUES (?, ?, ?, ?, ?, ?, ?)", + (PID, no, f"S{no}", "IN_PROGRESS", 50, json.dumps({"no": no, "k": "las"}), f"m{no}"), + ) + for rid in (10, 11): + db.execute( + "INSERT INTO routes (id, project_id, surface_model_id, status, total_length_m," + " route_data_path, created_at) VALUES (?, ?, 5, 'CONFIRMED', ?, 'r.geojson', 't')", + (rid, PID, rid * 100.0), + ) + for seq in range(3): + db.execute( + "INSERT INTO route_points (route_id, chainage_m, sequence_num) VALUES (?, ?, ?)", + (rid, seq * 20.0, seq), + ) + db.execute( + "INSERT INTO cross_sections (project_id, route_id, chainage_m, data)" + " VALUES (?, ?, ?, ?)", + (PID, rid, seq * 20.0, json.dumps({"cut": seq})), + ) + db.execute("INSERT INTO route_statistics (route_id, cut_volume_m3) VALUES (?, 7)", (rid,)) + db.execute( + "INSERT INTO longitudinal_sections (project_id, route_id, data) VALUES (?, ?, '{}')", + (PID, rid), + ) + db.commit() + return db + + +_SET_FILES = { + "B05_Profile/route/planned_route.csv": "x,y\n1,2\n", + "B05_Profile/route/structures.json": '{"items": [1]}', + "B05_Profile/route/route_main.geojson": "{}", + "B05_Profile/corridor/corridor_0011.json": '{"route": 11}', + "B06_Section/longitudinal/longitudinal.json": '{"z": [1, 2]}', + "B06_Section/sampling.json": '{"surface_model_id": 5}', + "B04_PreProcess/drainage/01_primary_region.geojson": "{}", + "B04_PreProcess/drainage/edits/pipe_points.json": "[1]", + "B04_PreProcess/drainage/manifest.json": "{}", + "initial_snapshot/db.json": '{"routes": []}', +} +_SHARED_FILES = { + "B05_Profile/route/cost_surface_x_implicit.npz": "c", + "B05_Profile/route/expected_route.csv": "e", + "B05_Profile/route/initial_choice.json": "{}", + "B04_PreProcess/drainage/descent_ab_cd.npz": "d", + "B04_PreProcess/drainage/00_sheet_metric.pkl": "s", + "B04_PreProcess/drainage/rainfall_table.json": "r", + "B04_PreProcess/models/dtm_x.npz": "m", +} + + +def _seed_files(root: Path) -> None: + for relative, text in {**_SET_FILES, **_SHARED_FILES}.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + + +def _tree(root: Path) -> dict[str, str]: + return { + p.relative_to(root).as_posix(): p.read_text(encoding="utf-8") + for p in sorted(root.rglob("*")) + if p.is_file() and SETS_DIR not in p.relative_to(root).parts + } + + +SETS_DIR = sets.SETS_DIRNAME +FP = { + "terrain": "las", + "source_filter": "classification", + "method": "implicit", + "smooth": False, + "model_id": 5, + "model_digest": "aa", +} + + +def _rows(db: sqlite3.Connection, sql: str, args=()) -> list[tuple]: + return db.execute(sql, args).fetchall() + + +def _no_ids(db: sqlite3.Connection) -> dict[str, list]: + """id 를 뺀 값(자식은 노선 차례로) — 왕복 비교용.""" + order = {rid: i for i, (rid,) in enumerate(_rows(db, "SELECT id FROM routes ORDER BY id"))} + out = { + "routes": _rows( + db, + "SELECT project_id, surface_model_id, status, total_length_m, route_data_path," + " created_at FROM routes ORDER BY id", + ), + "stages": _rows( + db, + "SELECT stage_no, state, progress_percent, params, message FROM" + " project_workflow_stages ORDER BY stage_no", + ), + "status": _rows(db, "SELECT status FROM projects"), + } + for table, cols in ( + ("route_points", "chainage_m, sequence_num"), + ("route_statistics", "cut_volume_m3"), + ("longitudinal_sections", "project_id, data"), + ("cross_sections", "project_id, chainage_m, data"), + ): + out[table] = sorted( + (order[r[0]], *r[1:]) + for r in _rows(db, f"SELECT route_id, {cols} FROM {table} ORDER BY id") + ) + return out + + +def test_벌_몫과_공유를_가름(tmp_path: Path) -> None: + _seed_files(tmp_path) + entries = sets.set_entries(tmp_path) + assert "B06_Section" in entries and "initial_snapshot" in entries + assert "B05_Profile/route/structures.json" in entries + assert "B04_PreProcess/drainage/edits" in entries + joined = " ".join(entries) + for shared in ("cost_surface_", "expected_route", "initial_choice", "descent_", "00_sheet"): + assert shared not in joined + assert "rainfall_" not in joined and "models" not in joined + + +def test_떠_두고_꺼내면_값_같음_새_id(tmp_path: Path) -> None: + db = _seed_db() + conn = _Conn(db) + _seed_files(tmp_path) + before_db, before_files = _no_ids(db), _tree(tmp_path) + + meta = asyncio.run(sets.park_active_set(conn, tmp_path, PID, FP)) + assert meta["routes"] == [10, 11] and meta["built"] + assert _rows(db, "SELECT COUNT(*) FROM routes") == [(0,)] + assert _rows(db, "SELECT COUNT(*) FROM cross_sections") == [(0,)] + states = _rows(db, "SELECT stage_no, state, params FROM project_workflow_stages ORDER BY 1") + assert states[:2] == [(0, "IN_PROGRESS", before_db["stages"][0][3])] + [ + (1, "IN_PROGRESS", before_db["stages"][1][3]) + ] + assert all(s[1] == "NOT_STARTED" and s[2] is None for s in states[2:]) + left = _tree(tmp_path) + assert set(left) == set(_SHARED_FILES) # 공유만 남음 · 그대로 + assert all(left[k] == v for k, v in _SHARED_FILES.items()) + bundle = tmp_path / SETS_DIR / "las" + assert (bundle / "db.json").is_file() and (bundle / "initial_snapshot" / "db.json").is_file() + assert (bundle / "files" / "B06_Section" / "sampling.json").is_file() + assert sets.bundle_ready(tmp_path, "las", FP) + + route_map = asyncio.run(sets.load_set(conn, tmp_path, PID, "las")) + assert set(route_map) == {10, 11} + assert not set(route_map.values()) & {10, 11} # 새 id + assert _no_ids(db) == before_db # 노선 · 자식 · 단계 · 상태 같음 + after = _tree(tmp_path) + moved = before_files.pop("B05_Profile/corridor/corridor_0011.json") + new_corridor = f"B05_Profile/corridor/corridor_{route_map[11]:04d}.json" + assert after.pop(new_corridor) == moved # 코리도 = 새 id 이름 + assert after == before_files + assert not bundle.exists() # 꺼낸 묶음은 지움 + + +def test_지문_어긋나면_묵은_벌(tmp_path: Path) -> None: + db = _seed_db() + _seed_files(tmp_path) + asyncio.run(sets.park_active_set(_Conn(db), tmp_path, PID, FP)) + assert sets.bundle_ready(tmp_path, "las", FP) + assert sets.bundle_ready(tmp_path, "las", {**FP, "model_id": 99}) # 복제 · 갈음 id 는 안 봄 + assert not sets.bundle_ready(tmp_path, "las", {**FP, "model_digest": "bb"}) + assert not sets.bundle_ready(tmp_path, "las", {**FP, "source_filter": "csf"}) + assert not sets.bundle_ready(tmp_path, "las", None) + assert not sets.bundle_ready(tmp_path, "sheet", {**FP, "terrain": "sheet"}) + sets.note_unbuilt(tmp_path, "sheet", "수치지형도 모델 없음") + assert sets.read_meta(tmp_path, "sheet") == { + "terrain": "sheet", + "built": False, + "reason": "수치지형도 모델 없음", + } + assert not sets.bundle_ready(tmp_path, "sheet", {**FP, "terrain": "sheet"}) + + +def _swap_world(monkeypatch, tmp_path: Path, *, sheet_fp: dict, chain): + """las 활성 · sheet 묶음(지문 = sheet_fp) — 칩으로 sheet.""" + import B03_FileInput.B03_FileInput_Service_Chain as chain_mod + import config.config_db as config_db + + db = _seed_db() + conn = _Conn(db) + _seed_files(tmp_path) + sheet = {**FP, "terrain": "sheet", "source_filter": "sheet_laplace", "model_id": 6} + # sheet 벌을 미리 떠 둠: 노선 하나(길이 777) + asyncio.run(sets.park_active_set(conn, tmp_path, PID, FP)) + db.execute( + "INSERT INTO routes (project_id, surface_model_id, status, total_length_m) VALUES" + " (?, 6, 'CONFIRMED', 777)", + (PID,), + ) + db.commit() + (tmp_path / "B06_Section").mkdir(parents=True, exist_ok=True) + (tmp_path / "B06_Section" / "sheet.json").write_text("sheet", encoding="utf-8") + asyncio.run(sets.park_active_set(conn, tmp_path, PID, sheet_fp)) + asyncio.run(sets.load_set(conn, tmp_path, PID, "las")) + state = {"design": "las"} + + async def params(_c, _pid): + return { + "source_filter": "sheet_laplace" if state["design"] == "sheet" else "classification" + } + + async def fingerprint(_c, _pid, _root, terrain): + return sheet if terrain == "sheet" else FP + + async def switch(_c, _pid, terrain): + state["design"] = terrain + db.execute( + "UPDATE surface_models SET status = CASE id WHEN 6 THEN 'CONFIRMED' ELSE 'COMPLETE' END" + ) + return 6 + + monkeypatch.setattr(config_db, "get_db_pool", lambda: _Pool(conn)) + monkeypatch.setattr(sets, "get_surface_confirmation_params", params) + monkeypatch.setattr(sets, "set_fingerprint", fingerprint) + monkeypatch.setattr(sets, "switch_terrain", switch) + monkeypatch.setattr(chain_mod, "run_auto_design_chain", chain) + return db, state + + +def test_맞바꿈_묶음_있으면_꺼냄(monkeypatch, tmp_path: Path) -> None: + called = [] + + async def chain(*a, **k): + called.append(a) + + sheet_fp = {**FP, "terrain": "sheet", "source_filter": "sheet_laplace", "model_id": 6} + db, _ = _swap_world(monkeypatch, tmp_path, sheet_fp=sheet_fp, chain=chain) + result = asyncio.run(sets.swap_design_set(PID, tmp_path, "sheet")) # type: ignore[arg-type] + assert result["changed"] and not result["built"] and not called + assert _rows(db, "SELECT total_length_m FROM routes") == [(777.0,)] + assert (tmp_path / "B06_Section" / "sheet.json").read_text(encoding="utf-8") == "sheet" + assert sets.bundle_ready(tmp_path, "las", FP) # 떠난 벌 살림 + assert not (tmp_path / SETS_DIR / "sheet").exists() + + +def test_맞바꿈_묵은_벌은_버리고_지음(monkeypatch, tmp_path: Path) -> None: + called = [] + holder: dict = {} + + async def chain(pid, model_id, inner=False): + called.append((model_id, inner)) + holder["db"].execute( + "INSERT INTO routes (project_id, surface_model_id, status, total_length_m) VALUES" + " (?, 6, 'CONFIRMED', 555)", + (PID,), + ) + return {"route_id": 1} + + stale = {**FP, "terrain": "sheet", "source_filter": "sheet_laplace", "model_digest": "old"} + db, _ = _swap_world(monkeypatch, tmp_path, sheet_fp=stale, chain=chain) + holder["db"] = db + result = asyncio.run(sets.swap_design_set(PID, tmp_path, "sheet")) # type: ignore[arg-type] + assert result["built"] and called == [(6, True)] + assert _rows(db, "SELECT total_length_m FROM routes") == [(555.0,)] # 묵은 777 버림 + assert not (tmp_path / "B06_Section" / "sheet.json").exists() + assert sets.bundle_ready(tmp_path, "las", FP) + + +def test_맞바꿈_실패면_통째_되돌림(monkeypatch, tmp_path: Path) -> None: + async def chain(*_a, **_k): + return None # 지음 실패 + + stale = {**FP, "terrain": "sheet", "source_filter": "sheet_laplace", "model_digest": "old"} + db, _ = _swap_world(monkeypatch, tmp_path, sheet_fp=stale, chain=chain) + before_db, before_files = _no_ids(db), _tree(tmp_path) + route_ids = _rows(db, "SELECT id FROM routes ORDER BY id") + sheet_meta = sets.read_meta(tmp_path, "sheet") + models = _rows(db, "SELECT id, status FROM surface_models ORDER BY id") + with pytest.raises(RuntimeError): + asyncio.run(sets.swap_design_set(PID, tmp_path, "sheet")) # type: ignore[arg-type] + assert _no_ids(db) == before_db + assert _rows(db, "SELECT id FROM routes ORDER BY id") == route_ids + assert _rows(db, "SELECT id, status FROM surface_models ORDER BY id") == models + assert _tree(tmp_path) == before_files + assert sets.read_meta(tmp_path, "sheet") == sheet_meta # 묶음도 그대로 + assert not (tmp_path / SETS_DIR / "las").exists()