diff --git a/common_util/common_util_initial_snapshot.py b/common_util/common_util_initial_snapshot.py index 07b8dfc9c..1ee7ca833 100644 --- a/common_util/common_util_initial_snapshot.py +++ b/common_util/common_util_initial_snapshot.py @@ -169,6 +169,49 @@ def _copy_tree(source: Path, target: Path) -> None: shutil.copytree(source, target) +_DRAINAGE_TREE = "B04_PreProcess/drainage" + + +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).""" + if not source.is_dir(): + return + if target.exists(): + shutil.rmtree(target) + top = str(source) + shutil.copytree( + source, + target, + ignore=lambda folder, names: [n for n in names if folder == top and _drainage_shared(n)], + ) + + +def _restore_drainage(stored: Path, live: Path) -> None: + """노선 몫만 갈음 — 공유 캐시는 제자리에 둠(지우면 다음 배수 계산이 처음부터 · 765-6) · + 캐시 든 옛 사본도 캐시 칸은 건너뜀.""" + if live.is_dir(): + for path in live.iterdir(): + if _drainage_shared(path.name): + continue + if path.is_dir(): + shutil.rmtree(path) + else: + path.unlink() + live.mkdir(parents=True, exist_ok=True) + for path in stored.iterdir(): + if _drainage_shared(path.name): + continue + if path.is_dir(): + shutil.copytree(path, live / path.name) + else: + shutil.copy2(path, live / path.name) + + async def _dump_rows( connection: aiomysql.Connection, table: str, route_id: int ) -> list[dict[str, Any]]: @@ -211,7 +254,8 @@ async def save_initial_snapshot( _write_design_route_csv(target / DESIGN_ROUTE_CSV_NAME, design_route_points) for tree in _FILE_TREES: - _copy_tree(root / tree, target / tree.replace("/", "__")) + save = _save_drainage if tree == _DRAINAGE_TREE else _copy_tree + save(root / tree, target / tree.replace("/", "__")) # 코리도는 체인이 마지막에 만든다 — 있으면 함께 뜬다(없으면 브라우저 폴백 그대로). from B05_Profile.B05_Profile_Router_Corridor import corridor_path @@ -294,6 +338,9 @@ 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) # 코리도는 복원으로 만든 **새 route id** 이름으로 되돌린다 — 이름이 어긋나면 브라우저가 diff --git a/common_util/common_util_project_copy.py b/common_util/common_util_project_copy.py index 1c3dc4a47..4331904d8 100644 --- a/common_util/common_util_project_copy.py +++ b/common_util/common_util_project_copy.py @@ -158,18 +158,68 @@ def _fix_files( print("프로젝트 id 치환:", replace_project_id(new_root, source_id, new_id)) print("노선 · 지표면 id 파일:", rekey_id_files(new_root, model_map, route_map)) - snapshot = new_root / "initial_snapshot" / "db.json" - if snapshot.is_file(): - dump = json.loads(snapshot.read_text(encoding="utf-8")) - changed = False - for route in dump.get("routes", []): - old_model = route.get("surface_model_id") - if old_model in model_map: - route["surface_model_id"] = model_map[old_model] - changed = True - if changed: - snapshot.write_text(json.dumps(dump, ensure_ascii=False), encoding="utf-8") - print("초기값 스냅샷 지표면 id 교정:", changed) + print( + "초기값 스냅샷 지표면 id 교정:", _fix_dump_models(new_root / "initial_snapshot", model_map) + ) + print("설계 벌 묶음 지표면 id 교정:", _fix_design_sets(new_root, model_map)) + + +def _fix_dump_models(folder: Path, model_map: dict) -> bool: + """`folder/db.json` 노선 행의 지표면 id → 새 id · 바꿨으면 참.""" + path = folder / "db.json" + if not path.is_file(): + return False + dump = json.loads(path.read_text(encoding="utf-8")) + changed = False + for route in dump.get("routes", []): + old_model = route.get("surface_model_id") + if old_model in model_map: + route["surface_model_id"] = model_map[old_model] + changed = True + if changed: + path.write_text(json.dumps(dump, ensure_ascii=False), encoding="utf-8") + return changed + + +def _fix_sampling(root: Path, model_map: dict) -> bool: + """`B06_Section/sampling.json` 의 지표면 id → 새 id(코리도는 안 건드림).""" + path = root / "B06_Section" / "sampling.json" + if not path.is_file(): + return False + data = json.loads(path.read_text(encoding="utf-8")) + if data.get("surface_model_id") not in model_map: + return False + data["surface_model_id"] = model_map[data["surface_model_id"]] + path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + return True + + +def _fix_design_sets(root: Path, model_map: dict) -> dict[str, list[str]]: + """쉬는 설계 벌 묶음(765-6) 안 지표면 id → 새 id — db.json · 그 벌 초기값 · B06 표본 설정 · + meta.json. 노선 id 는 꺼낼 때 늘 새로 받아(묶음 안에서만 짝) 안 고침 → 묶음 → 고친 자리.""" + from common_util.common_util_design_sets import sets_dir + + report: dict[str, list[str]] = {} + base = sets_dir(root) + for bundle in sorted(base.iterdir()) if base.is_dir() else []: + fixed = [ + name + for name, done in ( + ("db.json", _fix_dump_models(bundle, model_map)), + ("initial_snapshot", _fix_dump_models(bundle / "initial_snapshot", model_map)), + ("sampling", _fix_sampling(bundle / "files", model_map)), + ) + if done + ] + meta_path = bundle / "meta.json" + if meta_path.is_file(): + meta = json.loads(meta_path.read_text(encoding="utf-8")) + if meta.get("model_id") in model_map: + meta["model_id"] = model_map[meta["model_id"]] + meta_path.write_text(json.dumps(meta, ensure_ascii=False), encoding="utf-8") + fixed.append("meta.json") + report[bundle.name] = fixed + return report def _write_manifest(root: Path, manifest: dict[str, Any]) -> None: diff --git a/resources/tester/test_765_design_sets.py b/resources/tester/test_765_design_sets.py index 3df33836f..d9fbdac66 100644 --- a/resources/tester/test_765_design_sets.py +++ b/resources/tester/test_765_design_sets.py @@ -517,3 +517,53 @@ def test_비우기에_묶음_포함(tmp_path: Path) -> None: removed = purge_output_folders(tmp_path, keep_preprocess=True) assert SETS_DIR in removed and not (tmp_path / SETS_DIR).exists() assert (tmp_path / "B04_PreProcess" / "models").is_dir() + + +# ── 765-6 초기값 사본 = 캐시 뺌 · 복제가 묶음 안 지표면 id 고침 ──────────── + + +def test_초기값_사본은_배수_캐시를_빼고_복원은_캐시를_안_지움(tmp_path: Path) -> None: + from common_util import common_util_initial_snapshot as snap + + _seed_files(tmp_path) + drainage = tmp_path / "B04_PreProcess" / "drainage" + (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) + saved = sorted(p.name for p in target.iterdir()) + assert saved == ["01_primary_region.geojson", "edits", "manifest.json"] # 캐시 없음 + + # 작업본이 바뀜 · 캐시는 새로 지어짐 · 옛 사본(캐시 듦) 꼴도 섞음 + (drainage / "01_primary_region.geojson").write_text("edited", encoding="utf-8") + (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) + 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 # 초기값에 없던 노선 몫 = 지움 + assert live["descent_ab_cd.npz"].read_text(encoding="utf-8") == "d2" # 캐시 그대로 + assert live["00_sheet_metric.pkl"].read_text(encoding="utf-8") == "s" # 옛 사본 캐시 안 덮음 + assert (live["rainfall_stations"] / "a.json").is_file() + assert (live["edits"] / "pipe_points.json").is_file() + + +def test_복제가_묶음_안_지표면_id_고침(tmp_path: Path) -> None: + from common_util.common_util_project_copy import _fix_design_sets + + db = _seed_db() + _seed_files(tmp_path) + asyncio.run(sets.park_active_set(_Conn(db), tmp_path, PID, FP)) + bundle = tmp_path / SETS_DIR / "las" + corridor = bundle / "files" / "B05_Profile" / "corridor" / "corridor_0011.json" + report = _fix_design_sets(tmp_path, {5: 905}) + assert sorted(report["las"]) == ["db.json", "meta.json", "sampling"] + dump = json.loads((bundle / "db.json").read_text(encoding="utf-8")) + assert {r["surface_model_id"] for r in dump["routes"]} == {905} + assert [r["id"] for r in dump["routes"]] == [10, 11] # 노선 id 는 꺼낼 때 새로 + assert sets.read_meta(tmp_path, "las")["model_id"] == 905 + sampling = bundle / "files" / "B06_Section" / "sampling.json" + 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 를 안 봄