test(511-B B-5 화면 · 시간): 검증_sub3 B05 노선 — 계산 자리 route 「서버」(이 창이 받는 판만 · 판 안 고침) 초안 노선 190 점 · 측점 61 = 화면 셈 0 차 · 저장 노선 = 서버 안 부름 · 왕복 27 ms · Node 1 ms · 노선 저장 체인 시간 도구(run_511_b5_route_save_timing · 편집표 그대로 · 파이썬 짝 / Node 번갈아 · 정본 해시) — 요구 320-4

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-22 15:14:36 +09:00
parent 70a24c2416
commit 29e1e4be6d
2 changed files with 190 additions and 0 deletions
@@ -0,0 +1,114 @@
+# -*- coding: utf-8 -*-
+"""511-B B-5 노선 저장 체인 시간 · 값 — 검증 프로젝트의 지금 노드 · 편집표 그대로 저장(수동 실행).
+
+같은 노선이라 값은 그대로(새 노선 행 번호만 바뀜) · 초기값 안 건드림(`as_initial` 끔).
+갈래 = 파이썬 짝(Node 폴리라인 · 좌표 맞춤 끔) / Node 를 번갈아 N 번 · 저장 뒤 정본 파일 해시
+(노선 csv · 노드 · 곡선표 · 구조물 · 관 · 추가 측점 · 종단 · 횡단)가 처음과 같은지.
+쓰는 법: `./venv/Scripts/python.exe resources/tester/run_511_b5_route_save_timing.py <id> [N]`
+"""
+
+import asyncio
+import hashlib
+import json
+import logging
+import statistics
+import sys
+import time
+from pathlib import Path
+from uuid import UUID
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from config.config_db import close_db_pool, init_db_pool  # noqa: E402
+
+CANON = (
+    "B05_Profile/route/planned_route.csv",
+    "B05_Profile/route/planned_route_nodes.csv",
+    "B05_Profile/route/planned_route_curves.json",
+    "B05_Profile/route/route_main.geojson",
+    "B05_Profile/route/structures.json",
+    "B05_Profile/route/plus_stations.json",
+    "B04_PreProcess/drainage/edits/pipe_points.json",
+)
+
+
+def _hashes(root: Path) -> dict[str, str]:
+    files = [root / name for name in CANON]
+    files += sorted((root / "B06_Section").rglob("*.json"))
+    return {
+        str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()[:16]
+        for path in files
+        if path.is_file()
+    }
+
+
+async def _save(project_id: UUID, python: bool) -> float:
+    from B05_Profile import B05_Profile_Route_Polyline_Write as writer
+    from B05_Profile import B05_Profile_Router_Replan as replan
+    from B05_Profile.B05_Profile_Route_Save import RouteSaveRequest
+    from common_util import common_util_route_follow as follow
+
+    plan = await replan.read_route_plan(project_id)
+    edits = plan["node_edits"]
+    body = RouteSaveRequest(
+        vertices=[
+            {"x": n["x"], "y": n["y"], "curve": e["curve"], "radius_m": e["radius_m"]}
+            for n, e in zip(plan["nodes"], edits, strict=True)
+        ],
+        station_snap=bool(plan.get("station_snap")),
+        station_snap_step=plan.get("station_snap_step"),
+    )
+    real = (writer.node_polyline, follow.fill_structure_coordinates)
+    if python:
+        writer.node_polyline = lambda *a: None
+        follow.fill_structure_coordinates = follow._fill_structure_coordinates_py
+    try:
+        started = time.perf_counter()
+        result = await replan.replan_route(project_id, body)
+        seconds = time.perf_counter() - started
+    finally:
+        writer.node_polyline, follow.fill_structure_coordinates = real
+    status = getattr(result, "status_code", 200)
+    assert status == 200, getattr(result, "body", result)
+    return seconds
+
+
+async def main(project_id: UUID, rounds: int) -> None:
+    from B05_Profile import B05_Profile_Router_Replan as replan
+
+    logging.basicConfig(level=logging.WARNING, format="%(message)s")
+    await init_db_pool()
+    try:
+        root, _ = await replan._project_paths(project_id)
+        first = _hashes(root)
+        times: dict[str, list[float]] = {"python": [], "node": []}
+        for _ in range(rounds):
+            for kind in ("python", "node"):
+                times[kind].append(await _save(project_id, kind == "python"))
+                after = _hashes(root)
+                changed = sorted(
+                    k for k in first.keys() | after.keys() if first.get(k) != after.get(k)
+                )
+                print(f"{kind}: {times[kind][-1]:.2f} s · 바뀐 정본 {len(changed)} {changed[:5]}")
+        print(
+            json.dumps(
+                {
+                    k: {
+                        "median_s": round(statistics.median(v), 2),
+                        "runs": [round(x, 2) for x in v],
+                    }
+                    for k, v in times.items()
+                },
+                ensure_ascii=False,
+            )
+        )
+    finally:
+        from common_util.common_util_calc_runner import _SERVED
+
+        _SERVED.stop()
+        await close_db_pool()
+
+
+if __name__ == "__main__":
+    asyncio.run(main(UUID(sys.argv[1]), int(sys.argv[2]) if len(sys.argv) > 2 else 3))
@@ -243,3 +243,79 @@ def test_측점_계획_Node_파이썬_같음():
    node = node_plan_stations(groups)
    assert node is not None
    assert node == [plan_stations(*group) for group in groups]
+
+
+# ── 화면 — 계산 자리 route 「서버」 노선 = 화면 셈 노선(검증_sub3 · Playwright · 저장 안 함) ──
+
+VERIFY = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"  # 검증_sub3(OWNERS.md)
+
+JS_ROUTE = r"""async () => {
+  const urls = performance.getEntriesByType('resource').map((e) => e.name);
+  const page = await import(urls.find((n) => /design_pipeline_page\.ts/.test(n)));
+  const pid = localStorage.getItem('frd_current_project_id');
+  const pipe = page.pagePipeline(pid);
+  await page.reconcilePagePipeline(pid);
+  await pipe.settled();
+  const calls = (window.__aisloCalcServer?.calls ?? []).filter((c) => c.id === 'route.polyline');
+  const plain = (v) => JSON.parse(JSON.stringify(v ?? null));
+  return { route: plain(pipe.output('route')), stations: plain(pipe.output('stations')),
+           calls: calls.map((c) => ({ round: c.roundMs, node: c.nodeMs, same: c.same })),
+           runs: (window.__aisloPipeline?.runs ?? []).slice(-1).map((r) => r.outcome) };
+}"""
+
+
+@pytest.fixture(scope="module")
+def screens():
+    from test_511_b1_calc_node_stages import _UP, BASE, JS_DRAFT, _started
+
+    if not BASE or _UP is None or _UP != _started(BASE.replace("5178", "8005")):
+        pytest.skip("이 폴더 서버 없음")
+    sync = pytest.importorskip("playwright.sync_api")
+    cred = json.loads((ROOT / "tmp" / "aislo_cred.json").read_text(encoding="utf-8"))
+    shots = {}
+    with sync.sync_playwright() as play:
+        chrome = play.chromium.launch(channel="msedge", headless=True)
+        try:
+            for mode in ("both", "server"):
+                context = chrome.new_context(viewport={"width": 1600, "height": 1000})
+                login = {k: cred[k] for k in ("email", "password")}
+                r = context.request.post(
+                    f"{BASE}/api/auth/login/request", data=login, timeout=60000
+                )
+                assert r.ok, f"로그인 실패 {r.status}"  # 비밀번호는 안 찍음
+                if mode == "server":  # 판은 안 고침 — 이 창이 받는 판만 route = server
+
+                    def board(route):
+                        body = route.fetch().json()
+                        body["sites"]["route"] = "server"
+                        route.fulfill(json=body)
+
+                    context.route("**/api/dashboard/calc-sites", board)
+                page = context.new_page()
+                page.add_init_script(
+                    f"localStorage.setItem('frd_current_project_id', '{VERIFY}');"
+                    " performance.setResourceTimingBufferSize(20000);"
+                )
+                page.goto(f"{BASE}/#/b05-profile", wait_until="domcontentloaded")
+                page.wait_for_function("() => !!window.__aisloPipeline", timeout=90000)
+                page.wait_for_timeout(9000)
+                shots[mode] = {"saved": page.evaluate(JS_ROUTE)}
+                page.evaluate(JS_DRAFT)  # 교각점 하나 옮김(세션 초안 · 창과 함께 버림)
+                page.wait_for_timeout(3000)
+                shots[mode]["draft"] = page.evaluate(JS_ROUTE)
+                context.close()
+        finally:
+            chrome.close()
+    (ROOT / "tmp" / "511b5_screens.json").write_text(json.dumps(shots), encoding="utf-8")
+    return shots
+
+
+@pytest.mark.parametrize("case", ["saved", "draft"])
+def test_화면_서버_노선_화면_셈과_같음(screens, case):
+    both, server = screens["both"][case], screens["server"][case]
+    assert both["runs"] == server["runs"] == ["done"]
+    assert len(both["route"]["line"]) > 10
+    assert both["route"] == server["route"] and both["stations"] == server["stations"]
+    assert both["calls"] == []
+    # 저장 노선(초안 없음) = 서버 안 부름 · 초안 = server 판만 서버 노선
+    assert bool(server["calls"]) is (case == "draft")