eomsangdonandClaude Opus 5.5
0e0c1be1c0
- 격자 원점 = 좌표 정수 m(GRID_ORIGIN_XY · 옛 「도로 시작점」 지시를 바꿈 · 사용자 확정) - 흐름 방향 저장본 drainage/descent_<입력 지문>_<범위>.npz — 입력 지문 = 도엽 등고선 지문 + 남긴 등고 단 + 셀 + 계산 규칙 · 덮으면 읽어 잘라 씀 · 최근 2벌 - 도엽(하천 · 등고선) 사업지 좌표 저장본 drainage/00_sheet_metric.pkl — 원본 자국 · 좌표계가 같으면 재투영 안 함 - 검증_sub3 실측: 저장본 읽기 = 지은 것과 칸마다 같음(적색 · 해석 영역 · 화살표 · 귀속 차이 0) · 108 → 10.8 s(붐빔 · 340 기준 약 6 s) · 원점 바뀜으로 적색 460,035 → 459,792칸(−0.05 %) · 관 자리 같음 - 시험 test_347_1_descent_store · 실측 bench 저장본 갈래 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XWh5wNNycMkbwMpACz1rte
102 lines
3.9 KiB
Python
102 lines
3.9 KiB
Python
"""342-3 · 347-1 배수유역 빠른 길 실측 — 검증_sub3 도엽 · 노선으로 preview_stages 를 돌려
|
|
시간과 결과(적색 · 흐름 방향 · 도로 귀속 · 강도)를 tmp/342/<이름>.npz 에 남기고 기준과 대조한다.
|
|
|
|
./venv/Scripts/python.exe resources/tester/bench_342_watershed.py base # 기준 저장
|
|
./venv/Scripts/python.exe resources/tester/bench_342_watershed.py new base # 새로 돌려 대조
|
|
./venv/Scripts/python.exe resources/tester/bench_342_watershed.py s1 base store # 흐름 방향 저장본
|
|
|
|
pytest 대상 아님(DB · 프로젝트 파일이 있어야 돎). 분석 결과는 저장 안 함 — 도엽 좌표 저장본
|
|
(drainage/00_sheet_metric.pkl)만 프로젝트에 남고 · 흐름 방향 저장본은 tmp/342/descent 에 둠.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from uuid import UUID
|
|
|
|
import numpy as np
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT))
|
|
PID = UUID("1f93a515-df0a-4975-a8ba-daaa7672b123") # 검증_sub3
|
|
OUT_DIR = ROOT / "tmp" / "342"
|
|
|
|
|
|
async def _prepare():
|
|
from config.config_db import close_db_pool, init_db_pool
|
|
|
|
await init_db_pool()
|
|
try:
|
|
from B04_PreProcess.B04_PreProcess_Router_Watershed import _prepare as prepare
|
|
|
|
started = time.perf_counter()
|
|
prepared = await prepare(PID)
|
|
return prepared, time.perf_counter() - started
|
|
finally:
|
|
await close_db_pool()
|
|
|
|
|
|
def main() -> None:
|
|
name = sys.argv[1]
|
|
base = sys.argv[2] if len(sys.argv) > 2 and sys.argv[2] != "-" else None
|
|
store = len(sys.argv) > 3 and sys.argv[3] == "store"
|
|
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(name)s:%(message)s",
|
|
handlers=[logging.FileHandler(OUT_DIR / f"{name}.log", "w", encoding="utf-8")],
|
|
)
|
|
prepared, prepare_s = asyncio.run(_prepare())
|
|
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import preview_stages
|
|
|
|
started = time.perf_counter()
|
|
pv = preview_stages(
|
|
prepared["vertices"],
|
|
prepared["contours"],
|
|
prepared["streams"],
|
|
(OUT_DIR / "descent", prepared["contour_key"]) if store else None,
|
|
)
|
|
preview_s = time.perf_counter() - started
|
|
spec = pv.spec
|
|
red = pv.flow.reaches_road & pv.flow.analyzed
|
|
result = {
|
|
"spec": np.array([spec.x_min, spec.y_max, spec.cell_m, spec.n_rows, spec.n_cols]),
|
|
"domain": pv.domain,
|
|
"red": red,
|
|
"direction": pv.flow.direction,
|
|
"road_slot": pv.routing.road_slot,
|
|
"strength": np.array(pv.strength_profile),
|
|
"pipes": np.array([p.chainage_m for p in pv.pipes]),
|
|
}
|
|
np.savez_compressed(OUT_DIR / f"{name}.npz", **result)
|
|
report = {
|
|
"prepare_s": round(prepare_s, 2),
|
|
"preview_s": round(preview_s, 2),
|
|
"total_s": round(prepare_s + preview_s, 2),
|
|
"rounds": pv.expand_rounds,
|
|
"grid": [spec.n_rows, spec.n_cols],
|
|
"red_cells": int(red.sum()),
|
|
}
|
|
if base:
|
|
ref = np.load(OUT_DIR / f"{base}.npz")
|
|
same_spec = bool(np.array_equal(ref["spec"], result["spec"]))
|
|
report["same_spec"] = same_spec
|
|
if same_spec:
|
|
for key in ("domain", "red", "direction", "road_slot"):
|
|
report[f"diff_{key}"] = int((ref[key] != result[key]).sum())
|
|
else:
|
|
report["base_spec"] = ref["spec"].tolist()
|
|
report["new_spec"] = result["spec"].tolist()
|
|
report["red_cells_base"] = int(ref["red"].sum())
|
|
report["same_strength"] = bool(np.array_equal(ref["strength"], result["strength"]))
|
|
report["same_pipes"] = bool(np.array_equal(ref["pipes"], result["pipes"]))
|
|
print(json.dumps(report, ensure_ascii=False))
|
|
(OUT_DIR / f"{name}.json").write_text(json.dumps(report, ensure_ascii=False, indent=1), "utf-8")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|