fix(505-1 B05): 추천 노선 미리 셈 = 전처리 체인에서만 · 프로젝트 접근은 저장본 읽기만

- B05 열기(GET /route/plan) 미리 셈 예약 걷음 · 체인 쪽(업로드 · B04 재확정 · B01 사용 범위) 그대로
- 추천 창 기본 갈래(기본 다섯 · 저장 제약 · 종단 켬 · 설계 지형) = 저장본만 · 없으면 409 「전처리를 다시 하면 생깁니다」(판 글 · 토스트)
- 사용자가 바꾼 제약 · 비중 · 종단 끔 · 다른 분석 지형 · [재계산] = 그 자리 셈
- 화면이 실은 분석 지형이 설계 지형과 같으면 설계 지형으로 봄
- 저장본 꺼내기 path.touch 걷음 · 초기 종단 지문 노선 파일 칸 걷음(셈이 안 읽음) · 초기 종단 지문 함수 캐시
- B04 _prepare 재투영 · 피클 읽기 · followed_pipes 스레드로
- 시험 test_309_3 새로 · 224_3 · 364_4 · 476_1 새 규칙에 맞춤

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-21 13:14:33 +09:00
co-authored by Claude Opus 5.5
parent 163ab72881
commit d42726030e
12 changed files with 226 additions and 66 deletions
@@ -238,13 +238,23 @@ async def _prepare(project_id: UUID) -> dict[str, Any] | JSONResponse:
)
# 설계 계통과 같은 프로젝트 좌표계로 맞춘다. 도엽 재투영도 이 좌표계로 간다.
source_crs = resolve_project_crs(project_root, route_crs_input=planned.crs_input, db_epsg=epsg)
to_lonlat_transformer = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True)
to_metric_transformer = Transformer.from_crs("EPSG:4326", source_crs, always_xy=True)
def sheets() -> tuple[Transformer, list[dict[str, Any]], list[dict[str, Any]], str]:
source_crs = resolve_project_crs(
project_root, route_crs_input=planned.crs_input, db_epsg=epsg
)
to_metric_transformer = Transformer.from_crs("EPSG:4326", source_crs, always_xy=True)
return (
Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True),
*_metric_sheets(
_sheet_dir(stored_path),
drainage_dir(stored_path),
source_crs,
to_metric_transformer,
),
)
streams, contour_features, contour_key = _metric_sheets(
_sheet_dir(stored_path), drainage_dir(stored_path), source_crs, to_metric_transformer
)
# 재투영(저장본 없으면 3.8 s) · 피클 읽기는 스레드로 — 이벤트 루프를 안 막음(505-1)
to_lonlat_transformer, streams, contour_features, contour_key = await asyncio.to_thread(sheets)
return {
"route_source": route_file.name,
"vertices": planned.vertices,
@@ -17,6 +17,7 @@ curve_count, straightened, max_grade_pct, curve_length_m, warnings, with_profile
from __future__ import annotations
import functools
import hashlib
import json
import math
@@ -185,8 +186,9 @@ def profile_summary(longitudinal: dict[str, Any]) -> dict[str, Any]:
}
@functools.cache
def source_fingerprint() -> str:
"""초기 종단 코드 지문 — `SOURCE_GLOBS` 내용(줄끝 LF)."""
"""초기 종단 코드 지문 — `SOURCE_GLOBS` 내용(줄끝 LF) · 프로세스에 한 번(505-1)."""
root = Path(__file__).resolve().parents[1]
digest = hashlib.blake2b(digest_size=8)
for path in sorted({p for pattern in SOURCE_GLOBS for p in root.glob(pattern)}):
@@ -195,17 +197,6 @@ def source_fingerprint() -> str:
return digest.hexdigest()
def file_marks(paths: list[Path]) -> list[str]:
"""지문에 드는 파일 자국 — 내용 해시(작은 정본) · 없음 「-」."""
marks = []
for path in paths:
try:
marks.append(hashlib.blake2b(path.read_bytes(), digest_size=8).hexdigest())
except OSError:
marks.append("-")
return marks
def stat_marks(paths: list[Path]) -> list[str]:
"""큰 자료(지형 npz · 도엽) 자국 — 이름 · 크기 · 수정시각."""
marks = []
@@ -48,7 +48,7 @@ SOURCE_GLOBS = (
"common_util/common_util_route_polyline.py",
)
STORE_DIR = "B05_Profile/route_recommend"
# 종류(갈래 · 지형) 마다 남길 벌 수 — 꺼낼 때 수정시각을 새로 해 최근에 쓴 것이 남음
# 종류(갈래 · 지형) 마다 남길 벌 수 — 최근에 적은 것이 남음(꺼내기는 읽기만 · 505-1)
KEEP_PER_KIND = 2
# 다른 판 · 소스 지문 저장본을 이 날수 지나면 미리 셈 때 지움(420-2)
STALE_DAYS = 30
@@ -147,19 +147,13 @@ def _current(data: Any) -> bool:
def read_stored(project_root: Path, key: str) -> dict[str, Any] | None:
"""저장본 한 벌 — 없거나 못 읽거나 판 · 소스 지문이 다르면 None."""
"""저장본 한 벌 — 없거나 못 읽거나 판 · 소스 지문이 다르면 None · 파일을 안 건드림(505-1)."""
path = store_dir(project_root) / f"{key}.json"
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
return None
if not _current(data):
return None
try:
path.touch() # 최근에 쓴 것 = 정리 때 남김
except OSError:
pass
return data
return data if _current(data) else None
def write_stored(
+3 -1
View File
@@ -218,7 +218,9 @@ async def _place_pipes(
prepared = await _prepare(project_id)
streams = [] if isinstance(prepared, JSONResponse) else prepared["streams"] # 도엽 없음
pairs, crossing, vertices = followed_pipes(old_pipes, old_xy, points, streams)
pairs, crossing, vertices = await asyncio.to_thread(
followed_pipes, old_pipes, old_xy, points, streams
)
await asyncio.to_thread(
save_pipe_points,
stored_path,
+1 -4
View File
@@ -448,10 +448,7 @@ async def read_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse:
if paths is None:
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
project_root, _ = paths
# B05 열 때 추천 저장본이 없거나 판이 다르면 뒤에서 미리 셈(364-4 · 있으면 꺼내기만)
from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
schedule_precompute(project_id)
# 추천 미리 셈은 전처리 체인에서만(505-1) — B05 열기는 저장본을 안 셈 · 안 씀
# 예상노선 정본이 먼저, 없으면 옛 자리(초기값 스냅샷)를 본다.
expected = await asyncio.to_thread(_vertices_of, expected_route_csv_path(project_root))
if not expected:
+26 -13
View File
@@ -25,6 +25,8 @@ logger = logging.getLogger(__name__)
# 미리 셈 진행(364-4) — 프로젝트마다 하나만 돎 · 도는 동안 또 오면 끝난 뒤 한 번 더(입력이 바뀜)
_RUNNING: dict[str, dict[str, Any]] = {}
# 미리 셈 갈래인데 저장본이 없을 때(505-1) — 프로젝트 접근은 셈하지 않음
NOT_PREPARED = "추천 저장본이 없습니다 — 전처리를 다시 하면 생깁니다."
@replan.router.post("/{project_id}/route/initial", response_model=None)
@@ -40,12 +42,14 @@ async def initial_route(
async def compute_initial(
project_id: UUID, request: replan.RouteInitialRequest
project_id: UUID, request: replan.RouteInitialRequest, precompute: bool = False
) -> dict[str, Any] | JSONResponse:
"""같은 입력이면 저장본을 꺼내고 없으면 셈해 더한다(PLAN 327) · 덮개 수는 늘 지금 관으로.
응답 `stored` = 저장본에서 꺼냈나. 진행 중인 같은 묶음은 기다렸다 꺼낸다(두 번 셈 막음).
`recompute` = [재계산] — 묶음 잠금 안에서 이 묶음 저장본 · 메모리 묶음을 버리고 새로 셈(420).
미리 셈 갈래(기본 · 저장 제약 · 종단 켬 · 설계 지형)는 `precompute`(전처리 체인)만 셈 — 화면
요청은 저장본만 · 없으면 409 `NOT_PREPARED`(505-1). 사용자가 바꾼 값 · [재계산]은 그 자리 셈.
"""
from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct
from B05_Profile.B05_Profile_Engine_RouteInitial import (
@@ -69,6 +73,7 @@ async def compute_initial(
from B05_Profile.B05_Profile_Router_RouteInitial_Initial import initial_profile_of
from common_util.common_util_surface_confirmation import (
analysis_terrain_params,
design_terrain,
get_surface_confirmation_params,
)
@@ -110,7 +115,9 @@ async def compute_initial(
pool = get_db_pool()
async with pool.acquire() as connection:
selection = await get_surface_confirmation_params(connection, str(project_id))
selection = analysis_terrain_params(selection, request.terrain)
# 화면은 늘 분석 지형을 실음 — 설계 지형과 같으면 설계 지형(미리 셈 갈래 판정 · 505-1)
picked = None if request.terrain == design_terrain(selection) else request.terrain
selection = analysis_terrain_params(selection, picked)
inputs = {
"mode": request.mode,
"weights": weights,
@@ -124,6 +131,13 @@ async def compute_initial(
# 균형 종단 · 횡단 토공 표준 횡단면 = 프로젝트 저장 패널값(420)
"standard": await project_standard(project_id),
}
criteria_dump = criteria.model_dump() if criteria else None
may_compute = (
precompute
or request.recompute
or not request.with_profile
or not is_default(request.mode, weights, criteria_dump, picked, current)
)
started = time.perf_counter()
try:
terrain, streams = await asyncio.to_thread(
@@ -139,6 +153,10 @@ async def compute_initial(
logger.info("추천 [재계산]: project_id=%s 저장본 %d 지움", project_id, removed)
forget_pool_on_standard(pool_key, inputs["standard"])
stored = await asyncio.to_thread(read_stored, project_root, key)
if stored is None and not may_compute:
return JSONResponse(
status_code=409, content={"status": "error", "message": NOT_PREPARED}
)
if stored is not None:
result = stored["result"]
else:
@@ -157,13 +175,7 @@ async def compute_initial(
pool_key,
request.model_dump(exclude={"recompute"}),
result,
is_default(
request.mode,
weights,
criteria.model_dump() if criteria else None,
request.terrain,
current,
),
is_default(request.mode, weights, criteria_dump, picked, current),
)
# 측점 맞춘 선 · 맞추기 전 선(351-2) 둘 다 덮개 수를 지금 관으로
shown = [result, *([result["unsnapped"]] if result.get("unsnapped") else [])]
@@ -180,7 +192,7 @@ async def compute_initial(
result = {
**result,
"initial_profile": await initial_profile_of( # 476
project_id, project_root, paths[1], result, request, selection, key
project_id, project_root, paths[1], result, request, selection, key, may_compute
),
}
logger.info(
@@ -263,7 +275,7 @@ async def precompute_defaults(project_id: UUID) -> None:
criteria=criteria,
weights=weights if mode == "weighted" else None,
)
outcome = await compute_initial(project_id, request)
outcome = await compute_initial(project_id, request, precompute=True)
if isinstance(outcome, JSONResponse):
logger.warning(
"추천 미리 셈 멈춤(%s %s): project_id=%s", mode, outcome.status_code, project_id
@@ -304,8 +316,9 @@ async def initial_progress(project_id: UUID) -> dict[str, Any]:
def schedule_precompute(project_id: UUID) -> None:
"""추천 미리 셈 예약 — 업로드 · 재확정 체인 끝(351-1) · B05 열 때(364-4). 기다리지 않음 ·
이미 돌면 끝난 뒤 한 번 더 · 실패는 로그만. 저장본이 다 있으면 꺼내기만(갈래마다 0.2 s 안팎)."""
"""추천 미리 셈 예약 — 업로드 · 재확정 체인 끝(351-1)만 · B05 열기는 안 부름(505-1).
기다리지 않음 · 이미 돌면 끝난 뒤 한 번 더 · 실패는 로그만. 저장본이 다 있으면 꺼내기만
(갈래마다 0.2 s 안팎)."""
try:
from B03_FileInput.B03_FileInput_Router_Helpers import _schedule_background_task
@@ -29,9 +29,11 @@ async def initial_profile_of(
request: Any,
selection: dict[str, Any],
key: str,
compute: bool = True,
) -> dict[str, Any]:
"""추천 후보 초기 종단(476 · `B05_Profile_Engine_InitialProfile` 머리) — [적용] 체인과 같은
입력 · 같은 조각 · 지문 같으면 저장본 꺼냄. 실패 = `{error}`."""
입력 · 같은 조각 · 지문 같으면 저장본 꺼냄. 실패 = `{error}`. `compute` 거짓 = 저장본만
(505-1)."""
from dataclasses import asdict
import aiomysql
@@ -43,7 +45,11 @@ async def initial_profile_of(
from B05_Profile.B05_Profile_Engine_Sections import _MODELS_SUBDIR
from B05_Profile.B05_Profile_Repository import get_latest_route
from B05_Profile.B05_Profile_Route_Save import section_seed
from B05_Profile.B05_Profile_Router_RouteInitial import balanced_profile_of, project_standard
from B05_Profile.B05_Profile_Router_RouteInitial import (
NOT_PREPARED,
balanced_profile_of,
project_standard,
)
from B06_Section.B06_Section_Repository import (
get_latest_section_options,
get_route_generation_source,
@@ -88,7 +94,6 @@ async def initial_profile_of(
crs = f"EPSG:{epsg}" if epsg is not None else None
surface = {k: selection.get(k) for k in ("source_filter", "method", "smooth")}
models = project_root / _MODELS_SUBDIR
route_path = str(latest.get("route_data_path") or "")
mark = ip.fingerprint(
{
"key": key,
@@ -98,8 +103,8 @@ async def initial_profile_of(
"surface": surface,
"models": ip.stat_marks(list(models.glob(f"*_{surface['source_filter']}*.npz"))),
"streams": ip.stat_marks([_sheet_dir(stored_path) / STREAM_FILE]),
# 초기화(492) = 저장 관 · 구조물 · 추가 측점을 안 씀 — 노선 파일만
"files": ip.file_marks([project_root / route_path]),
# 저장 노선 파일은 안 넣음(505-1) — 초기화(492) 셈은 노선 파일 · 관 · 구조물 ·
# 추가 측점을 안 읽음(선 = 후보 노드) · 넣으면 노선 [저장]마다 다시 셈
"options": asdict(options),
"grade": grade_options.as_dict() if grade_options else None,
"crs": crs,
@@ -112,6 +117,8 @@ async def initial_profile_of(
stored = await asyncio.to_thread(ip.read_initial, project_root, key, mark)
if stored is not None:
return stored
if not compute:
return {"error": NOT_PREPARED}
prepared = await _prepare(project_id)
streams = [] if isinstance(prepared, JSONResponse) else prepared["streams"]
facilities = await asyncio.to_thread(ip.reset_facilities, line, streams)
@@ -545,11 +545,10 @@ export function openRecommendModal(params: RecommendParams): void {
} — [적용]하면 바로 저장 · 새 초기값이 됩니다.`;
} catch (error) {
if (closed || mine !== seq) return;
stateEl.textContent = "셈하지 못했습니다.";
showToast(
error instanceof Error ? error.message : "초기노선을 셈하지 못했습니다.",
"error",
);
// 서버 사유를 판에 남김 — 저장본 없음이면 「전처리를 다시 하면 생깁니다」(505-1)
const message = error instanceof Error ? error.message : "초기노선을 셈하지 못했습니다.";
stateEl.textContent = message;
showToast(message, "error");
} finally {
if (mine === seq) window.clearInterval(ticker);
if (!closed && mine === seq) {
@@ -196,15 +196,16 @@ def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls:
return calls
def _ask(**fields) -> dict:
def _ask(precompute: bool = True, **fields) -> dict:
"""`precompute` 참 = 전처리 체인 셈 · 거짓 = 화면 요청(기본 갈래는 저장본만 · 505-1)."""
request = replan.RouteInitialRequest(**fields)
return asyncio.run(router.compute_initial(uuid4(), request))
return asyncio.run(router.compute_initial(uuid4(), request, precompute))
def test_second_request_computes_nothing(wired: _Calls) -> None:
first = _ask(mode="earthwork")
assert wired.count == 1 and first["stored"] is False
second = _ask(mode="earthwork")
second = _ask(False, mode="earthwork")
assert wired.count == 1 and second["stored"] is True
assert second["planned"] == first["planned"]
# 덮개 수는 꺼낼 때마다 지금 관으로 · 저장본에는 없음
@@ -226,7 +227,8 @@ def test_same_pool_in_flight_waits(wired: _Calls) -> None:
request = replan.RouteInitialRequest(mode="gentle")
project = uuid4()
return await asyncio.gather(
router.compute_initial(project, request), router.compute_initial(project, request)
router.compute_initial(project, request, True),
router.compute_initial(project, request, True),
)
results = asyncio.run(both())
@@ -245,6 +247,6 @@ def test_precompute_defaults_then_hits(wired: _Calls) -> None:
assert all(item["default"] for item in saved)
# 화면 첫 누름(비중은 화면이 50/25/25 를 채워 보냄 · 다섯 갈래 종단 켬 · 420) — 셈 없이 저장본
assert all(item["request"]["with_profile"] for item in saved)
hit = _ask(mode="weighted", weights={"expected": 50, "earthwork": 25, "grade": 25})
follow = _ask(mode="follow")
hit = _ask(False, mode="weighted", weights={"expected": 50, "earthwork": 25, "grade": 25})
follow = _ask(False, mode="follow")
assert hit["stored"] and follow["stored"] and wired.count == 5
@@ -0,0 +1,141 @@
"""요구 309-3 · 310 · PLAN 505-1 — 추천 노선 미리 셈 = 전처리 체인에서만.
./venv/Scripts/python.exe -m pytest resources/tester/test_309_3_recommend_preprocess_only.py -q
B05 열기(GET /route/plan) · 노선 [저장] · [적용] 뒤 다시 세움 = 미리 셈 0 · 저장본 읽기만(파일
안 건드림) · 미리 셈 갈래가 저장본에 없으면 409 「전처리를 다시 하면 생김」 · 사용자가 바꾼 값 ·
[재계산]은 그 자리 셈 · 체인 셈(`run_precompute`)이 만들면 화면이 꺼냄 · 초기 종단 지문에 노선
파일 없음.
"""
from __future__ import annotations
import asyncio
import json
import os
import re
import sys
from pathlib import Path
from uuid import uuid4
ROOT = Path(__file__).resolve().parents[2]
TESTER = Path(__file__).resolve().parent
for path in (ROOT, TESTER):
if str(path) not in sys.path:
sys.path.insert(0, str(path))
import helper_472_1_paths as P # noqa: E402
from test_224_2_station_snap import _NODES, _patch_replan, _working # noqa: E402
from test_224_3_recommend_store import _ask, wired # noqa: E402, F401 — 가짜 세계 fixture
from test_476_1_recommend_initial import gather # noqa: E402, F401
from B05_Profile import B05_Profile_Engine_InitialProfile as ip # noqa: E402
from B05_Profile import B05_Profile_Router_RouteInitial as router # noqa: E402
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import ( # noqa: E402
DEFAULT_MODES,
read_stored,
store_dir,
write_stored,
)
def test_B05_열기_미리_셈_0(tmp_path: Path, monkeypatch) -> None:
import B03_FileInput.B03_FileInput_Router_Helpers as helpers
scheduled: list[str] = []
monkeypatch.setattr(
helpers,
"_schedule_background_task",
lambda coroutine, task_name: (scheduled.append(task_name), coroutine.close()),
)
replan = _patch_replan(monkeypatch, tmp_path)
replan._write_planned_polyline(_working(tmp_path), _NODES, 30.0, simplify=False)
body = asyncio.run(replan.read_route_plan(uuid4()))
assert body["status"] == "success" and scheduled == []
assert not store_dir(tmp_path).exists()
def test_미리_셈_부르는_곳은_체인만() -> None:
"""노선 [저장] · [적용] · B05 열기 길에는 미리 셈 호출이 없음 — 체인 둘 · B01 사용 범위만."""
callers = set()
for path in ROOT.glob("B0*/*.py"):
text = path.read_text(encoding="utf-8")
if re.search(r"\b(schedule_precompute|run_precompute|refresh_expected_route)\(", text):
callers.add(path.name)
assert callers == {
"B03_FileInput_Service_Chain.py", # 업로드 체인 · B04 재확정 체인
"B01_Dashboard_Router.py", # 사용 범위 바꿈 = 예상노선 다시 씀
"B05_Profile_Router_RouteInitial.py", # 정의
}
def test_화면_기본_갈래_저장본_없으면_409_셈_0(wired) -> None: # noqa: F811
for mode in DEFAULT_MODES:
response = _ask(False, mode=mode)
assert response.status_code == 409
assert "전처리를 다시 하면 생깁니다" in json.loads(response.body)["message"]
assert wired.count == 0 and not list(store_dir(wired.root).glob("*.json"))
def test_체인이_만들면_화면이_꺼냄(wired) -> None: # noqa: F811
asyncio.run(router.run_precompute(uuid4())) # 체인 끝 예약이 도는 한 줄
assert wired.count == 5
for mode in DEFAULT_MODES:
assert _ask(False, mode=mode)["stored"] is True
assert wired.count == 5
def test_화면이_실은_분석_지형이_설계_지형이면_기본_갈래(wired) -> None: # noqa: F811
"""화면은 늘 분석 지형을 실음 — 설계 지형(가짜 세계 = 라이다)과 같으면 저장본만 · 다르면 셈."""
assert _ask(False, mode="gentle", terrain="las").status_code == 409
assert wired.count == 0
assert _ask(False, mode="gentle", terrain="sheet")["stored"] is False
assert wired.count == 1
def test_사용자가_바꾼_값_재계산은_셈(wired) -> None: # noqa: F811
assert _ask(False, mode="earthwork", criteria={"min_radius_m": 15.0})["stored"] is False
assert _ask(False, mode="earthwork", with_profile=False)["stored"] is False
assert wired.count == 2
assert _ask(False, mode="gentle", recompute=True)["stored"] is False
assert wired.count == 3
def test_초기_종단_화면은_저장본만(wired, monkeypatch) -> None: # noqa: F811
import B05_Profile.B05_Profile_Router_RouteInitial_Initial as gatherer
flags: list[bool] = []
async def fake(project_id, root, stored, result, request, selection, key, compute=True):
flags.append(compute)
return {"pvi_count": 3}
monkeypatch.setattr(gatherer, "initial_profile_of", fake)
_ask(True, mode="earthwork") # 체인
_ask(False, mode="earthwork") # 화면 기본 갈래
_ask(False, mode="earthwork", criteria={"min_radius_m": 15.0}) # 사용자가 바꾼 값
assert flags == [True, False, True]
def test_꺼내기는_파일_안_건드림(tmp_path: Path) -> None:
write_stored(tmp_path, "k", "p", {"mode": "gentle"}, {"nodes": []}, True)
path = store_dir(tmp_path) / "k.json"
os.utime(path, ns=(1_000_000_000, 1_000_000_000))
assert read_stored(tmp_path, "k") is not None
assert path.stat().st_mtime_ns == 1_000_000_000
def test_초기_종단_지문에_노선_파일_없음(gather, monkeypatch) -> None: # noqa: F811
first = gather["run"]()
assert "error" not in first, first
assert gather["run"]("k2", False) == {"error": router.NOT_PREPARED} # 저장본 없음 = 셈 안 함
def broken(*args, **kwargs):
raise RuntimeError("다시 셈")
monkeypatch.setattr(ip, "initial_longitudinal", broken)
route = gather["root"] / P.ROUTE
doc = json.loads(route.read_text("utf-8"))
doc["properties"] = {**(doc.get("properties") or {}), "saved": "노선 [저장]"}
route.write_text(json.dumps(doc), "utf-8") # 노선 [저장] = 노선 파일만 바뀜
assert gather["run"]("k1", False) == first
@@ -65,7 +65,8 @@ def test_실패해도_진행_비움(monkeypatch) -> None:
def test_b05_열기_예약_모달_진행_배선() -> None:
replan = (ROOT / "B05_Profile/B05_Profile_Router_Replan.py").read_text(encoding="utf-8")
plan = replan[replan.index("async def read_route_plan") :]
assert "schedule_precompute(project_id)" in plan[: plan.index("\n@router")]
# 505-1 — B05 열기는 미리 셈 안 부름(전처리 체인에서만)
assert "schedule_precompute(" not in plan[: plan.index("\n@router")]
source = (ROOT / "B05_Profile/B05_Profile_Router_RouteInitial.py").read_text(encoding="utf-8")
assert '"/{project_id}/route/initial/progress"' in source
# 체인 · B01 범위도 같은 한 줄(run_precompute)로
@@ -88,7 +89,8 @@ def test_모달이_여는_갈래_먼저(monkeypatch) -> None:
async def paths(_pid):
return (ROOT, None)
async def compute(_pid, request):
async def compute(_pid, request, precompute=False):
assert precompute # 체인 셈 = 저장본 없어도 셈(505-1)
asked.append(request.mode)
assert request.with_profile # 다섯 갈래 모두 종단 켬(420)
return {}
@@ -334,9 +334,11 @@ def gather(tmp_path, monkeypatch):
candidate = _candidate(root)
selection = {"source_filter": "classification", "method": "dtm", "smooth": True}
def run(key: str = "k1") -> dict:
def run(key: str = "k1", compute: bool = True) -> dict:
return asyncio.run(
gatherer.initial_profile_of(uuid4(), root, "p", candidate, request, selection, key)
gatherer.initial_profile_of(
uuid4(), root, "p", candidate, request, selection, key, compute
)
)
return {"run": run, "root": root, "state": state, "pipes": pipes_file}
@@ -392,7 +394,7 @@ def test_추천_응답에_initial_profile(wired, monkeypatch):
seen: list = []
async def fake(project_id, root, stored, result, request, selection, key):
async def fake(project_id, root, stored, result, request, selection, key, compute=True):
seen.append(key)
return {"pvi_count": 3}