eomsangdonandClaude Opus 5.5
bdaca728fe
- 검증_sub5 실측: 칩으로 수치지형도 벌 → [초기화] → 칩 라이다 뒤 예상노선이 수치지형도로 자른 것(643 점 · 2136.2 m)으로 남음 · 라이다로 자른 것 = 415 점 · 1377.4 m - 예상노선은 지형마다 벌이 있어도 하나(공유 · 추천이 씀) — 초기값 사본에 안 넣고 복원이 안 건드림(옛 사본 안 것도 건너뜀) - 시험 test_765_design_sets 에 덮지 않음 검사 덧 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
595 lines
25 KiB
Python
595 lines
25 KiB
Python
# -*- 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()
|
|
|
|
|
|
# ── 765-3 전처리 체인 = 두 벌 ─────────────────────────────────────────────
|
|
|
|
|
|
def _chain_world(monkeypatch, tmp_path: Path, *, other, built_other=True):
|
|
import B03_FileInput.B03_FileInput_Repository as repo
|
|
import B03_FileInput.B03_FileInput_Service_Chain as chain
|
|
import B05_Profile.B05_Profile_Router_RouteInitial as router
|
|
import common_util.common_util_storage as storage
|
|
import config.config_db as config_db
|
|
|
|
log: list = []
|
|
expected = tmp_path / "B05_Profile" / "route" / "expected_route.csv"
|
|
|
|
async def auto(pid, model_id=None, inner=False):
|
|
log.append(("auto", model_id, inner))
|
|
expected.parent.mkdir(parents=True, exist_ok=True)
|
|
expected.write_text(str(model_id), encoding="utf-8") # 체인이 처음 쓰는 예상노선
|
|
return {"route_id": 1} if (built_other or len(log) > 2) else None
|
|
|
|
async def switch(_c, _pid, terrain):
|
|
log.append(("switch", terrain))
|
|
return 6 if terrain == "sheet" else 5
|
|
|
|
async def park(_c, _root, _pid, fingerprint):
|
|
log.append(("park", fingerprint["terrain"]))
|
|
|
|
async def clear(_c, _root, _pid):
|
|
log.append("clear")
|
|
|
|
async def fingerprint(_c, _pid, _root, terrain):
|
|
return {**FP, "terrain": terrain}
|
|
|
|
async def params(_c, _pid):
|
|
return {"source_filter": "classification"}
|
|
|
|
monkeypatch.setattr(config_db, "get_db_pool", lambda: _Pool(_Conn(sqlite3.connect(":memory:"))))
|
|
monkeypatch.setattr(repo, "get_project_storage_relative_path", lambda *_a: _done("p"))
|
|
monkeypatch.setattr(storage, "resolve_stored_project_path", lambda _p: str(tmp_path))
|
|
import common_util.common_util_surface_confirmation as confirmation
|
|
|
|
monkeypatch.setattr(confirmation, "get_surface_confirmation_params", params)
|
|
monkeypatch.setattr(sets, "other_terrain", lambda _p, _r: other)
|
|
monkeypatch.setattr(sets, "switch_terrain", switch)
|
|
monkeypatch.setattr(sets, "park_active_set", park)
|
|
monkeypatch.setattr(sets, "clear_active_set", clear)
|
|
monkeypatch.setattr(sets, "set_fingerprint", fingerprint)
|
|
monkeypatch.setattr(chain, "run_auto_design_chain", auto)
|
|
monkeypatch.setattr(router, "schedule_precompute", lambda pid: log.append("precompute"))
|
|
return chain, log, expected
|
|
|
|
|
|
async def _done(value):
|
|
return value
|
|
|
|
|
|
def test_전처리_체인_다른_지형_먼저_떠_두고_설계_지형(monkeypatch, tmp_path: Path) -> None:
|
|
chain, log, expected = _chain_world(monkeypatch, tmp_path, other="sheet")
|
|
(tmp_path / SETS_DIR / "las").mkdir(parents=True) # 옛 묶음 = 버림
|
|
summary = asyncio.run(chain.run_preprocess_chain(PID, 5)) # type: ignore[arg-type]
|
|
assert summary == {"route_id": 1}
|
|
assert log == [
|
|
("switch", "sheet"),
|
|
("auto", 6, True),
|
|
("park", "sheet"),
|
|
("switch", "las"),
|
|
("auto", 5, True),
|
|
"precompute",
|
|
]
|
|
assert expected.read_text(encoding="utf-8") == "5" # 예상노선 = 설계 지형 벌이 씀
|
|
assert not (tmp_path / SETS_DIR / "las").exists()
|
|
assert not (tmp_path / "initial_design.lock").exists()
|
|
|
|
|
|
def test_전처리_체인_다른_지형_없으면_한_벌(monkeypatch, tmp_path: Path) -> None:
|
|
chain, log, _ = _chain_world(monkeypatch, tmp_path, other=None)
|
|
asyncio.run(chain.run_preprocess_chain(PID, 5)) # type: ignore[arg-type]
|
|
assert log == [("auto", 5, True), "precompute"]
|
|
meta = sets.read_meta(tmp_path, "sheet")
|
|
assert meta["built"] is False and "sheet" in meta["reason"]
|
|
|
|
|
|
def test_전처리_체인_다른_벌_실패해도_설계_지형(monkeypatch, tmp_path: Path) -> None:
|
|
chain, log, _ = _chain_world(monkeypatch, tmp_path, other="sheet", built_other=False)
|
|
summary = asyncio.run(chain.run_preprocess_chain(PID, 5)) # type: ignore[arg-type]
|
|
assert summary == {"route_id": 1}
|
|
assert log == [
|
|
("switch", "sheet"),
|
|
("auto", 6, True),
|
|
"clear",
|
|
("switch", "las"),
|
|
("auto", 5, True),
|
|
"precompute",
|
|
]
|
|
assert sets.read_meta(tmp_path, "sheet")["built"] is False
|
|
|
|
|
|
def test_비우기에_묶음_포함(tmp_path: Path) -> None:
|
|
from common_util.common_util_project_reset import purge_output_folders
|
|
|
|
(tmp_path / SETS_DIR / "sheet" / "files").mkdir(parents=True)
|
|
(tmp_path / "B04_PreProcess" / "models").mkdir(parents=True)
|
|
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_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"] # 캐시 없음
|
|
|
|
# 작업본이 바뀜 · 캐시는 새로 지어짐 · 옛 사본(캐시 듦) 꼴도 섞음
|
|
(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_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 # 초기값에 없던 노선 몫 = 지움
|
|
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 를 안 봄
|
|
|
|
|
|
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_", "expected_route")) for p in stored.iterdir())
|
|
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")
|
|
(route / "expected_route.csv").write_text("other-terrain", encoding="utf-8")
|
|
snap.restore_snapshot_files(tmp_path)
|
|
assert (route / "cost_surface_x_implicit.npz").read_text(encoding="utf-8") == "c2"
|
|
# 예상노선 = 공유 — 다른 벌 [초기화]가 덮지 않음
|
|
assert (route / "expected_route.csv").read_text(encoding="utf-8") == "other-terrain"
|
|
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)
|