fix(765-6): 초기값 사본 · 복원의 캐시 규칙을 B05 비용면에도 — 사본에 안 넣고 복원이 안 지움

- 검증_sub5 실측: 초기값 사본 B05_Profile__route 4.4~5.6 MB 의 대부분이 추천 비용면(cost_surface_*) — 배수 캐시와 같은 공유 캐시
- 트리마다 공유 캐시 앞머리(_TREE_CACHES) · 뜨기 = 뺌 · 복원 = 설계 파일만 갈음(예상노선은 그대로 뜸)
- 시험 test_765_design_sets 하나 덧

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-11-03 16:00:07 +09:00
co-authored by Claude Opus 5.5
parent 9b12d4ff5d
commit c89c9223e3
2 changed files with 40 additions and 22 deletions
+15 -20
View File
@@ -169,16 +169,15 @@ def _copy_tree(source: Path, target: Path) -> None:
shutil.copytree(source, target)
_DRAINAGE_TREE = "B04_PreProcess/drainage"
#: 트리 안 공유 캐시(이름 앞머리) — 사본에 안 넣고 복원이 안 지움(765-6 · 지우면 다음 셈이 처음부터)
_TREE_CACHES = {
"B04_PreProcess/drainage": DRAINAGE_SHARED_PREFIXES,
"B05_Profile/route": ("cost_surface_",), # 추천 비용면
}
def _drainage_shared(name: str) -> bool:
return name.startswith(DRAINAGE_SHARED_PREFIXES)
def _save_drainage(source: Path, target: Path) -> None:
"""배수 폴더 중 노선 몫만 뜸 — 공유 캐시(하강장 · 도엽 좌표 · 강우)는 뺌(765-6 ·
약 105 → 20 MB)."""
def _save_tree(source: Path, target: Path, shared: tuple[str, ...]) -> None:
"""트리를 뜸 — 맨 위 공유 캐시는 뺌(배수 약 105 → 20 MB)."""
if not source.is_dir():
return
if target.exists():
@@ -187,16 +186,15 @@ def _save_drainage(source: Path, target: Path) -> None:
shutil.copytree(
source,
target,
ignore=lambda folder, names: [n for n in names if folder == top and _drainage_shared(n)],
ignore=lambda folder, names: [n for n in names if folder == top and n.startswith(shared)],
)
def _restore_drainage(stored: Path, live: Path) -> None:
"""노선 몫만 갈음 — 공유 캐시는 제자리에 둠(지우면 다음 배수 계산이 처음부터 · 765-6) ·
캐시 든 옛 사본도 캐시 칸은 건너뜀."""
def _restore_tree(stored: Path, live: Path, shared: tuple[str, ...]) -> None:
"""설계 파일만 갈음 — 공유 캐시는 제자리 · 캐시 든 옛 사본도 캐시 칸은 건너뜀."""
if live.is_dir():
for path in live.iterdir():
if _drainage_shared(path.name):
if path.name.startswith(shared):
continue
if path.is_dir():
shutil.rmtree(path)
@@ -204,7 +202,7 @@ def _restore_drainage(stored: Path, live: Path) -> None:
path.unlink()
live.mkdir(parents=True, exist_ok=True)
for path in stored.iterdir():
if _drainage_shared(path.name):
if path.name.startswith(shared):
continue
if path.is_dir():
shutil.copytree(path, live / path.name)
@@ -254,8 +252,7 @@ async def save_initial_snapshot(
_write_design_route_csv(target / DESIGN_ROUTE_CSV_NAME, design_route_points)
for tree in _FILE_TREES:
save = _save_drainage if tree == _DRAINAGE_TREE else _copy_tree
save(root / tree, target / tree.replace("/", "__"))
_save_tree(root / tree, target / tree.replace("/", "__"), _TREE_CACHES.get(tree, ()))
# 코리도는 체인이 마지막에 만든다 — 있으면 함께 뜬다(없으면 브라우저 폴백 그대로).
from B05_Profile.B05_Profile_Router_Corridor import corridor_path
@@ -338,10 +335,8 @@ def restore_snapshot_files(project_root: Path, route_id: int | None = None) -> N
else: # 배수 폴더 없는 스냅샷 — 관 편집은 비움(초기값에 없던 것)
shutil.rmtree(edits, ignore_errors=True)
continue
if tree == _DRAINAGE_TREE:
_restore_drainage(stored, root / tree)
continue
_copy_tree(stored, root / tree)
if stored.is_dir():
_restore_tree(stored, root / tree, _TREE_CACHES.get(tree, ()))
# 코리도는 복원으로 만든 **새 route id** 이름으로 되돌린다 — 이름이 어긋나면 브라우저가
# 저장본을 못 찾아 초기화 뒤 첫 진입마다 통째로 다시 만든다.
+25 -2
View File
@@ -530,7 +530,7 @@ def test_초기값_사본은_배수_캐시를_빼고_복원은_캐시를_안_지
(drainage / "rainfall_stations").mkdir()
(drainage / "rainfall_stations" / "a.json").write_text("st", encoding="utf-8")
target = tmp_path / "initial_snapshot" / "B04_PreProcess__drainage"
snap._save_drainage(drainage, target)
snap._save_tree(drainage, target, snap.DRAINAGE_SHARED_PREFIXES)
saved = sorted(p.name for p in target.iterdir())
assert saved == ["01_primary_region.geojson", "edits", "manifest.json"] # 캐시 없음
@@ -539,7 +539,7 @@ def test_초기값_사본은_배수_캐시를_빼고_복원은_캐시를_안_지
(drainage / "04_detailed_basins.geojson").write_text("new", encoding="utf-8")
(drainage / "descent_ab_cd.npz").write_text("d2", encoding="utf-8")
(target / "00_sheet_metric.pkl").write_text("old-cache", encoding="utf-8")
snap._restore_drainage(target, drainage)
snap._restore_tree(target, drainage, snap.DRAINAGE_SHARED_PREFIXES)
live = {p.name: p for p in drainage.iterdir()}
assert live["01_primary_region.geojson"].read_text(encoding="utf-8") == "{}"
assert "04_detailed_basins.geojson" not in live # 초기값에 없던 노선 몫 = 지움
@@ -567,3 +567,26 @@ def test_복제가_묶음_안_지표면_id_고침(tmp_path: Path) -> None:
assert json.loads(sampling.read_text(encoding="utf-8")) == {"surface_model_id": 905}
assert corridor.is_file() # 코리도는 안 건드림
assert sets.bundle_ready(tmp_path, "las", FP) # 지문은 id 를 안 봄
def test_초기값_사본은_비용면도_빼고_복원이_안_지움(tmp_path: Path) -> None:
from common_util import common_util_initial_snapshot as snap
_seed_files(tmp_path)
asyncio.run(_save(snap, tmp_path))
stored = tmp_path / "initial_snapshot" / "B05_Profile__route"
assert not any(p.name.startswith("cost_surface_") for p in stored.iterdir())
assert (stored / "expected_route.csv").is_file() # 예상노선은 그대로 뜸
route = tmp_path / "B05_Profile" / "route"
(route / "cost_surface_x_implicit.npz").write_text("c2", encoding="utf-8")
(route / "plus_stations.json").write_text("[]", encoding="utf-8")
snap.restore_snapshot_files(tmp_path)
assert (route / "cost_surface_x_implicit.npz").read_text(encoding="utf-8") == "c2"
assert not (route / "plus_stations.json").exists() # 초기값에 없던 설계 파일 = 지움
assert (route / "structures.json").read_text(encoding="utf-8") == '{"items": [1]}'
async def _save(snap, root: Path) -> None:
(root / "initial_snapshot").rename(root / "_old_snapshot") # 시드 사본 비켜 둠
db = _seed_db()
await snap.save_initial_snapshot(_Conn(db), root, 11)