eomsangdonandClaude Opus 5
c322f53845
사용자 확정(2026-09-04) — 「다른 로직과 동일하게: 전처리 계산 마지막에 연산 후 영구저장, 진입 시 로딩, 조작은 캐시, 저장·확정 때 저장」. 예전에는 사용자가 B05에 처음 들어간 그 순간 브라우저가 17MB 규모를 만들어 첫 진입이 느렸음. 계산은 **재구현하지 않음** — 브라우저가 쓰는 TS 빌더(8,209줄)를 Node 로 그대로 돌림. - `B05_Profile_UI_Corridor_Envelope.ts` 신설 — 저장 형식·버전 해시·직렬화를 `_UI_Corridor.ts` 에서 그대로 떼어 브라우저·서버 공용으로 둠(내용 불변). - `B05_Profile_Corridor_Node.ts` 신설 — 입력 JSON 을 받아 저장본을 내놓는 진입점. `npm run build:corridor` 로 번들(335kB), `npm run build` 에 물림. - `B05_Profile_Corridor_Prebuild.py` 신설 — 상세·노선점을 모아 Node 실행 후 저장. 번들이 TS 원본보다 낡으면 스스로 다시 만듦(두 그림이 갈라지는 것을 막는 장치). - 체인 마지막(스냅샷 직전)에 호출. 실패는 비치명적 — 저장본이 없으면 브라우저 폴백. - 초기값 — 스냅샷이 코리도를 `initial_corridor.json` 으로 함께 뜨고, [초기화] 복원 때 **새 route id** 이름으로 되돌림(복원이 새 번호를 만들기 때문). - `window.__corridorSource` 디버그 훅 — 저장본을 썼는지 다시 만들었는지 화면 밖 확인용. 자체검증: 전체 시험 381 통과·17 건너뜀(옛 테스트 2건은 이름·경로 변경에 맞춰 갱신). 실경로 — 용화 프로젝트(route 139)로 사전 생성 성공, 브라우저가 그 저장본을 `source: "stored"`, 해시 `377c2a84` 로 **그대로 채택**(재빌드 0회, 전송 278ms). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
330 lines
14 KiB
Python
330 lines
14 KiB
Python
"""초기값 스냅샷 — 자동설계 체인이 만든 첫 결과를 그대로 떠 두고 [초기화]가 되돌린다.
|
|
|
|
CLAUDE.md 5장(조작·데이터 흐름 정책)의 **초기값** 층이다. 자동설계 체인이 끝난 직후
|
|
한 번 찍고, 그 뒤로는 읽기 전용이다 — 어떤 저장 경로도 이 폴더에 쓰지 않는다.
|
|
|
|
[초기화]가 재계산이 아니라 복원이어야 하는 이유: [저장]·[확정]이 새 행을 만들지 않고
|
|
최신 `routes` 행을 제자리 갱신하고 종·횡단 정본 파일도 덮어쓰므로, 초기 상태는 따로
|
|
떠 두지 않으면 남지 않는다. 재계산으로 되살리려 해도 `structures.json`·
|
|
`pipe_points.json` 편집분이 그대로 남아 초기값과 다른 결과가 나온다(2026-08-29).
|
|
"""
|
|
|
|
import csv
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import aiomysql
|
|
|
|
# 스냅샷 폴더는 워크플로우 단계가 아니므로 PROJECT_STORAGE_LAYOUT_V2에 넣지 않는다.
|
|
SNAPSHOT_DIRNAME = "initial_snapshot"
|
|
_DB_DUMP_NAME = "db.json"
|
|
# 설계가 쓰는 계획노선 정본 — shapefile로 온 노선도 여기서는 CSV 한 벌이다.
|
|
# 사업지(.prj) 좌표계로 옮기고 지표면 밖을 잘라 조밀화까지 끝낸 값이라, 설계 계통은
|
|
# 이 파일만 읽으면 매번 같은 노선을 본다(2026-09-03 사용자 확정).
|
|
DESIGN_ROUTE_CSV_NAME = "planned_route.csv"
|
|
# 초기 설계 체인이 도는 동안만 존재하는 마커(진입 차단 판정용).
|
|
DESIGNING_LOCK_NAME = "initial_design.lock"
|
|
# 초기 설계 체인이 실패로 끝났음을 남기는 마커.
|
|
DESIGN_FAILED_NAME = "initial_design.failed"
|
|
|
|
# 프로젝트 루트 기준 상대 경로 — 정본 파일이 사는 자리 전부.
|
|
#
|
|
# 배수유역은 `edits/`(관 지점 편집분)만 뜨다가 **폴더 통째**로 넓혔다(2026-09-04 사용자
|
|
# 확정: 「초기값 = 파일 입력 직후 결과 전부」). 관을 옮기면 세부유역(`04_detailed_basins`)이
|
|
# 다시 나뉘는데 그 산출물이 스냅샷 밖이라 [초기화]가 옛 유역도를 그대로 남겼다.
|
|
# 용량은 실측 8.6MB(스냅샷 전체 3.9MB → 약 12MB)로 감당할 만하다.
|
|
_FILE_TREES = (
|
|
"B05_Profile/route",
|
|
"B06_Section/longitudinal",
|
|
"B06_Section/cross_sections",
|
|
"B04_PreProcess/drainage",
|
|
)
|
|
|
|
# 배수유역을 폴더 통째로 넓히기 전(2026-09-04)에 찍힌 스냅샷이 갖고 있는 자리.
|
|
_LEGACY_DRAINAGE_TREE = "B04_PreProcess__drainage__edits"
|
|
|
|
# 코리도(3D 예상형상) 초기값 — 체인이 만든 저장본 한 벌(2026-09-04 사용자 확정).
|
|
# 작업본 파일명에는 route id가 박히는데 [초기화]는 **새 route id**를 만드므로, 스냅샷에는
|
|
# 번호 없는 이름으로 두었다가 복원 때 새 번호로 되돌린다.
|
|
_CORRIDOR_NAME = "initial_corridor.json"
|
|
|
|
# `routes.id`를 참조하는 표는 스키마상 이 넷이 전부다(001_create_schema.sql:539~556).
|
|
_CHILD_TABLES = ("route_points", "route_statistics", "longitudinal_sections", "cross_sections")
|
|
|
|
|
|
def snapshot_dir(project_root: Path) -> Path:
|
|
return Path(project_root) / SNAPSHOT_DIRNAME
|
|
|
|
|
|
def design_route_csv_path(project_root: Path) -> Path:
|
|
"""설계용 계획노선 CSV 정본의 자리."""
|
|
return snapshot_dir(Path(project_root)) / DESIGN_ROUTE_CSV_NAME
|
|
|
|
|
|
def designing_lock_path(project_root: Path) -> Path:
|
|
"""초기 설계 체인이 도는 동안만 존재하는 마커.
|
|
|
|
이게 있으면 B05·B06은 아직 들어갈 때가 아니다(CLAUDE.md 5장 — 계산 중 편집이 섞이면
|
|
초기값이 오염된다). 스냅샷 대상 4트리 **밖**인 프로젝트 루트에 두어 복원에 딸려
|
|
들어가지 않게 한다. `workflow_state`의 stage 2는 체인이 끝나도 IN_PROGRESS라 판정에
|
|
못 쓰고, route 존재 여부도 체인 3단계에서 이미 생겨 못 쓴다.
|
|
"""
|
|
return Path(project_root) / DESIGNING_LOCK_NAME
|
|
|
|
|
|
def is_designing(project_root: Path) -> bool:
|
|
return designing_lock_path(project_root).is_file()
|
|
|
|
|
|
def mark_designing(project_root: Path) -> None:
|
|
path = designing_lock_path(project_root)
|
|
try:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("designing", encoding="utf-8")
|
|
except OSError:
|
|
pass # 마커 실패가 체인을 막지는 않는다 — 문이 일찍 열릴 뿐이다.
|
|
|
|
|
|
def clear_designing(project_root: Path) -> None:
|
|
try:
|
|
designing_lock_path(project_root).unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def design_failed_path(project_root: Path) -> Path:
|
|
"""초기 설계 체인이 실패로 끝났음을 남기는 마커.
|
|
|
|
"스냅샷이 없다"는 사실만으로는 **실패한 프로젝트**와 스냅샷 기능 이전의 **옛
|
|
프로젝트**를 가를 수 없다. 앞은 [초기화]가 재계산으로 얼버무리면 안 되고(부분 결과는
|
|
분석 안 됨과 다르지 않다, 2026-09-02 사용자 확정), 뒤는 종전 재계산 폴백이 유일한
|
|
수단이다. 락과 같은 자리 — 스냅샷 대상 4트리 **밖**이라 복원에 딸려 들어가지 않는다.
|
|
"""
|
|
return Path(project_root) / DESIGN_FAILED_NAME
|
|
|
|
|
|
def is_design_failed(project_root: Path) -> bool:
|
|
return design_failed_path(project_root).is_file()
|
|
|
|
|
|
def read_design_failure(project_root: Path) -> str:
|
|
"""실패 사유를 읽는다. 마커가 없거나 못 읽으면 빈 문자열."""
|
|
try:
|
|
return design_failed_path(project_root).read_text(encoding="utf-8").strip()
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def mark_design_failed(project_root: Path, reason: str) -> None:
|
|
path = design_failed_path(project_root)
|
|
try:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(reason, encoding="utf-8")
|
|
except OSError:
|
|
pass # 마커 실패가 체인을 막지는 않는다 — 안내가 덜 정확해질 뿐이다.
|
|
|
|
|
|
def clear_design_failed(project_root: Path) -> None:
|
|
try:
|
|
design_failed_path(project_root).unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def discard_initial_snapshot(project_root: Path) -> bool:
|
|
"""초기값을 무효화한다 — 지표면·노선이 바뀌어 옛 초기값이 더는 기준이 아닐 때.
|
|
|
|
그 자리에서 다시 뜨지 않는다. 재설계 체인은 **사용자 입력을 유지한 채** 재계산하므로
|
|
그 결과를 찍으면 사용자 편집이 섞인 가짜 초기값이 된다. 지워 두면 다음 [초기화]가
|
|
재계산 폴백을 타면서 새 초기값을 만들고, 체인이 그때 촬영한다(2026-08-29 사용자 확정).
|
|
"""
|
|
target = snapshot_dir(Path(project_root))
|
|
if not target.exists():
|
|
return False
|
|
shutil.rmtree(target, ignore_errors=True)
|
|
return True
|
|
|
|
|
|
def has_initial_snapshot(project_root: Path) -> bool:
|
|
return (snapshot_dir(project_root) / _DB_DUMP_NAME).is_file()
|
|
|
|
|
|
def _copy_tree(source: Path, target: Path) -> None:
|
|
if not source.is_dir():
|
|
return
|
|
if target.exists():
|
|
shutil.rmtree(target)
|
|
shutil.copytree(source, target)
|
|
|
|
|
|
async def _dump_rows(
|
|
connection: aiomysql.Connection, table: str, route_id: int
|
|
) -> list[dict[str, Any]]:
|
|
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
|
# 표 이름은 이 모듈의 상수에서만 오므로 자리표시자 대상이 아니다.
|
|
await cursor.execute(f"SELECT * FROM {table} WHERE route_id = %s", (route_id,)) # noqa: S608
|
|
return list(await cursor.fetchall())
|
|
|
|
|
|
def _json_safe(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""TIMESTAMP 등 JSON이 모르는 값을 문자열로 낮춘다."""
|
|
return [
|
|
{key: (value if _is_json_native(value) else str(value)) for key, value in row.items()}
|
|
for row in rows
|
|
]
|
|
|
|
|
|
def _is_json_native(value: Any) -> bool:
|
|
return value is None or isinstance(value, (bool, int, float, str, list, dict))
|
|
|
|
|
|
async def save_initial_snapshot(
|
|
connection: aiomysql.Connection,
|
|
project_root: Path,
|
|
route_id: int,
|
|
design_route_points: list[dict[str, float]] | None = None,
|
|
) -> None:
|
|
"""자동설계 체인 성공 직후 한 번 부른다. 이미 있으면 덮어쓰지 않는다.
|
|
|
|
`design_route_points`는 체인이 이미 만들어 둔 계획노선 정점(사업지 좌표계·트림·조밀화
|
|
후)이다. 받으면 CSV 정본으로 함께 남긴다 — 노선이 shapefile로 왔더라도 설계 계통이
|
|
읽는 것은 이 CSV 한 벌이다.
|
|
"""
|
|
root = Path(project_root)
|
|
target = snapshot_dir(root)
|
|
if has_initial_snapshot(root):
|
|
return
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
if design_route_points:
|
|
_write_design_route_csv(target / DESIGN_ROUTE_CSV_NAME, design_route_points)
|
|
|
|
for tree in _FILE_TREES:
|
|
_copy_tree(root / tree, target / tree.replace("/", "__"))
|
|
|
|
# 코리도는 체인이 마지막에 만든다 — 있으면 함께 뜬다(없으면 브라우저 폴백 그대로).
|
|
from B05_Profile.B05_Profile_Router_Corridor import corridor_path
|
|
|
|
corridor = corridor_path(root, route_id)
|
|
if corridor.is_file():
|
|
shutil.copy2(corridor, target / _CORRIDOR_NAME)
|
|
|
|
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute("SELECT * FROM routes WHERE id = %s", (route_id,))
|
|
route = await cursor.fetchone()
|
|
if not route:
|
|
return
|
|
dump: dict[str, Any] = {"routes": _json_safe([dict(route)])}
|
|
for table in _CHILD_TABLES:
|
|
dump[table] = _json_safe(await _dump_rows(connection, table, route_id))
|
|
(target / _DB_DUMP_NAME).write_text(json.dumps(dump, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
def _write_design_route_csv(path: Path, points: list[dict[str, float]]) -> None:
|
|
"""계획노선 정점을 CSV로 적는다. 열 이름은 `read_planned_route_csv()`가 아는 것으로."""
|
|
with path.open("w", encoding="utf-8", newline="") as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow(("sequence", "x", "y"))
|
|
writer.writerows(
|
|
(index, round(point["x"], 4), round(point["y"], 4))
|
|
for index, point in enumerate(points)
|
|
)
|
|
|
|
|
|
def wipe_edited_masters(project_root: Path) -> list[str]:
|
|
"""사용자 편집 정본을 걷어낸다 — 재계산으로 **진짜 초기값**을 만들기 위한 사전 정리.
|
|
|
|
자동설계 체인은 계획노선 CSV와 config 기본값으로 도는 결정적 절차라, 이 둘만 없으면
|
|
원래 나왔어야 할 값이 나온다. 남겨 두면 `resolve_extra_stations`가 사용자가 편집한
|
|
관·구조물에서 측점을 다시 파생해 "초기값"이 오염된다(2026-08-29 실측에서 확인).
|
|
|
|
스냅샷 복원 경로에서는 부르지 않는다 — 거기서는 스냅샷본이 통째로 덮어쓴다.
|
|
"""
|
|
removed: list[str] = []
|
|
root = Path(project_root)
|
|
for rel in ("B05_Profile/route/structures.json", "B04_PreProcess/drainage/edits"):
|
|
path = root / rel
|
|
try:
|
|
if path.is_dir():
|
|
shutil.rmtree(path)
|
|
removed.append(rel)
|
|
elif path.is_file():
|
|
path.unlink()
|
|
removed.append(rel)
|
|
except OSError:
|
|
pass # 지우지 못하면 그만큼 초기값이 덜 깨끗할 뿐, 재계산은 계속한다.
|
|
return removed
|
|
|
|
|
|
def restore_snapshot_files(project_root: Path, route_id: int | None = None) -> None:
|
|
"""스냅샷 파일을 작업본 자리에 되돌린다. 스냅샷에 없던 트리는 비워 둔다.
|
|
|
|
배수유역 범위를 넓히기 전(2026-09-04)에 찍힌 스냅샷은 `edits/`만 갖고 있다 —
|
|
그런 프로젝트는 예전처럼 그 자리만 되돌린다. 넓힌 트리를 못 찾았다고 그냥 넘어가면
|
|
관 지점 편집분이 초기화 뒤에도 남는다.
|
|
"""
|
|
root = Path(project_root)
|
|
source = snapshot_dir(root)
|
|
for tree in _FILE_TREES:
|
|
stored = source / tree.replace("/", "__")
|
|
if not stored.is_dir() and tree == "B04_PreProcess/drainage":
|
|
_copy_tree(source / _LEGACY_DRAINAGE_TREE, root / "B04_PreProcess/drainage/edits")
|
|
continue
|
|
_copy_tree(stored, root / tree)
|
|
|
|
# 코리도는 복원으로 만든 **새 route id** 이름으로 되돌린다 — 이름이 어긋나면 브라우저가
|
|
# 저장본을 못 찾아 초기화 뒤 첫 진입마다 통째로 다시 만든다.
|
|
from B05_Profile.B05_Profile_Router_Corridor import corridor_path
|
|
|
|
corridor = source / _CORRIDOR_NAME
|
|
if route_id is not None and corridor.is_file():
|
|
target = corridor_path(root, int(route_id))
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(corridor, target)
|
|
|
|
|
|
async def restore_initial_snapshot(
|
|
connection: aiomysql.Connection, project_root: Path, project_id: str
|
|
) -> int | None:
|
|
"""`routes`와 자식 4표를 스냅샷 값으로 다시 세우고 새 route id를 돌려준다.
|
|
|
|
호출자가 기존 `routes` 행을 지운 **뒤에** 부른다(자식은 FK CASCADE로 함께 지워진다).
|
|
파일 복원은 트랜잭션 밖이라 `restore_snapshot_files()`를 따로 부른다.
|
|
"""
|
|
dump_path = snapshot_dir(Path(project_root)) / _DB_DUMP_NAME
|
|
if not dump_path.is_file():
|
|
return None
|
|
dump = json.loads(dump_path.read_text(encoding="utf-8"))
|
|
route = (dump.get("routes") or [None])[0]
|
|
if not route:
|
|
return None
|
|
|
|
route = dict(route)
|
|
route.pop("id", None)
|
|
route["project_id"] = project_id
|
|
new_id = await _insert_row(connection, "routes", route)
|
|
|
|
for table in _CHILD_TABLES:
|
|
for row in dump.get(table, []):
|
|
child = dict(row)
|
|
child.pop("id", None)
|
|
child["route_id"] = new_id
|
|
if "project_id" in child:
|
|
child["project_id"] = project_id
|
|
await _insert_row(connection, table, child)
|
|
return new_id
|
|
|
|
|
|
async def _insert_row(connection: aiomysql.Connection, table: str, row: dict[str, Any]) -> int:
|
|
columns = list(row.keys())
|
|
placeholders = ", ".join(["%s"] * len(columns))
|
|
names = ", ".join(f"`{name}`" for name in columns)
|
|
async with connection.cursor() as cursor:
|
|
# 표 이름은 상수, 열 이름은 스냅샷이 뜬 실제 스키마에서 온다.
|
|
await cursor.execute(
|
|
f"INSERT INTO {table} ({names}) VALUES ({placeholders})", # noqa: S608
|
|
tuple(row[name] for name in columns),
|
|
)
|
|
return int(cursor.lastrowid)
|