feat(노선추천): 327-1 추천 결과 영구 저장 — 입력 지문 키 · 같은 입력이면 저장본 · 바뀐 입력만 셈해 더함
- 새 B05_Profile_Engine_RouteInitial_Store(ALGO_VERSION · request_fingerprint · 읽기 · 쓰기 · 묶음 잠금) · 자리 B05_Profile/route_recommend/<키>.json · 덮어쓰기 없음 · 상한 없음 - 엔드포인트 몸통 compute_initial(찾기 → 셈 · 더함 · 덮개 수 꺼낼 때 다시 셈 · stored 표시) · precompute_defaults(기본 다섯 알약) - resolve_inputs 로 기본값 풀이 한 곳 · 머리글 「파일 안 씀」 정정 - 시험 test_224_3_recommend_store Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014utrjfoW831Vc5wJUd13kL
This commit is contained in:
Claude Opus 5.5
@@ -3,8 +3,9 @@
|
||||
예상노선 추종 · 영선 노선 · 토공 최소 · 기울기 순한 · 안전 우선 · 계류 횡단 최소 ·
|
||||
공사비 어림 최소 · 비중.
|
||||
|
||||
계산 자리 = 서버 단독(③). 결과는 화면 캐시용 — 이 모듈은 파일을 쓰지 않는다. 적용은 화면이
|
||||
받은 노드로 기존 `POST /route/replan`([확인])을 부른다.
|
||||
계산 자리 = 서버 단독(③). 이 모듈은 파일을 쓰지 않는다 — 결과는 라우터가 입력 지문 키로
|
||||
`B05_Profile/route_recommend/` 에 남긴다(`..._Store` · PLAN 327). 적용은 화면이 받은 노드로
|
||||
기존 `POST /route/replan`([확인])을 부른다.
|
||||
|
||||
갈래(`mode`):
|
||||
· follow — 예상노선 추적 = 전처리 초기값(`planned_route_initial.csv`)과 같은 규칙 · 같은 선.
|
||||
@@ -182,6 +183,20 @@ def write_initial_choice(
|
||||
return choice
|
||||
|
||||
|
||||
def resolve_inputs(
|
||||
max_grade: float,
|
||||
stream_offset_m: float | None,
|
||||
grade_smooth_m: float | None,
|
||||
target_grade: float | None,
|
||||
) -> tuple[float, float, float]:
|
||||
"""(계류 이격, 기울기 잴 지반 폭, 영선 목표 기울기) — 비운 값은 기본값 · 저장 키와 같은 풀이."""
|
||||
stream_offset = STREAM_OFFSET_M if stream_offset_m is None else float(stream_offset_m)
|
||||
grade_smooth = GRADE_SMOOTH_M if grade_smooth_m is None else float(grade_smooth_m)
|
||||
# 영선 노선 목표 기울기(비율) — 없으면 상한 · 상한 넘게 넣어도 상한(제약 안에서만)
|
||||
goal_grade = max_grade if target_grade is None else min(float(target_grade), max_grade)
|
||||
return stream_offset, grade_smooth, goal_grade
|
||||
|
||||
|
||||
def resolve_weights(mode: str, weights: dict[str, float] | None) -> dict[str, float]:
|
||||
"""갈래의 비중(합 100). weighted 가 아니면 갈래가 정한 값."""
|
||||
if mode != "weighted":
|
||||
@@ -240,10 +255,9 @@ def generate_initial_route(
|
||||
if len(expected) < 2:
|
||||
raise ValueError("예상노선 정점이 2개 미만입니다.")
|
||||
used = resolve_weights(mode, weights)
|
||||
stream_offset = STREAM_OFFSET_M if stream_offset_m is None else float(stream_offset_m)
|
||||
grade_smooth = GRADE_SMOOTH_M if grade_smooth_m is None else float(grade_smooth_m)
|
||||
# 영선 노선 목표 기울기(비율) — 없으면 상한 · 상한 넘게 넣어도 상한(제약 안에서만)
|
||||
goal_grade = max_grade if target_grade is None else min(float(target_grade), max_grade)
|
||||
stream_offset, grade_smooth, goal_grade = resolve_inputs(
|
||||
max_grade, stream_offset_m, grade_smooth_m, target_grade
|
||||
)
|
||||
|
||||
def measure(
|
||||
outline: Any, *, bound: float | None, follow: bool = False, weighted: bool = False
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
"""노선 추천 후보의 균형 종단(PLAN 63-4 · 63-2 안 A).
|
||||
|
||||
후보 노선 하나에 종단을 세워 절 · 성토를 **횡단 ㎥** 로 맞춘 계획선과 그 토량을 낸다.
|
||||
편집 창 [노선 추천]의 점수 · 미리 보기에만 쓴다 — 파일 · 캐시 · initial_snapshot · 확정 종단을
|
||||
바꾸지 않는다. 확정 체인의 종단 규칙(1차 계획고 = 지반고 · 2026-09-02 사용자 확정)은 그대로다.
|
||||
편집 창 [노선 추천]의 점수 · 미리 보기에만 쓴다 — 작업본 파일 · initial_snapshot · 확정 종단을
|
||||
바꾸지 않는다(추천 결과째 저장본에 남는 것은 라우터 몫 · PLAN 327). 확정 체인의 종단 규칙
|
||||
(1차 계획고 = 지반고 · 2026-09-02 사용자 확정)은 그대로다.
|
||||
|
||||
계산 자리 = 서버 단독(③) — 화면은 받은 측점 · 지반고 · 계획고를 그리기만 한다.
|
||||
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
"""계획노선 추천 결과 영구 저장(PLAN 327 · 요구 224-3).
|
||||
|
||||
같은 입력이면 같은 결과 — 셈한 추천을 입력 지문 키로 프로젝트 폴더에 남기고, 같은 입력이 오면
|
||||
셈하지 않고 꺼낸다. 바뀐 입력만 새로 셈해 더한다(덮어쓰기 없음 · 상한 없음).
|
||||
|
||||
층 = 서버 계산 캐시(데이터 3층 밖 · 사용자 편집값 아님 — 자동저장 금지와 무관). 자리
|
||||
`B05_Profile/route_recommend/<키>.json` — initial_snapshot 대상(`B05_Profile/route`) 밖이라
|
||||
[초기화]에 안 휩쓸리고, 자료 교체(`purge_output_folders`) · 프로젝트 삭제 때 폴더째 지워진다.
|
||||
|
||||
키 = 후보 묶음 지문(`_pool_key` — 지형 · 예상노선 · 계류 · 상한 · 최소 R · 이격 · 고름 · 목표
|
||||
기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · `ALGO_VERSION`. 관 지점은 뺀다 — 덮개 수는 꺼낼 때
|
||||
다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을 올린다(옛 파일은 다른 키라 안 쓰임).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import json
|
||||
from contextlib import asynccontextmanager
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, AsyncIterator
|
||||
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial import (
|
||||
DEFAULT_WEIGHTS,
|
||||
Terrain,
|
||||
_pool_key,
|
||||
resolve_inputs,
|
||||
resolve_weights,
|
||||
)
|
||||
from common_util.common_util_json import atomic_write_json
|
||||
|
||||
ALGO_VERSION = 1
|
||||
STORE_DIR = "B05_Profile/route_recommend"
|
||||
# 업로드 체인 끝에 미리 셈할 기본 다섯 알약(화면 차례) · 추적은 묶음이 선 뒤라 `similar` 가 채워짐
|
||||
DEFAULT_MODES = ("gradeline", "earthwork", "gentle", "weighted", "follow")
|
||||
|
||||
# 묶음 지문마다 잠금 — 진행 중 같은 묶음(같은 키 포함)은 기다렸다 꺼냄(두 번 셈 막음)
|
||||
_LOCKS: dict[str, asyncio.Lock] = {}
|
||||
|
||||
|
||||
def store_dir(project_root: Path) -> Path:
|
||||
return Path(project_root) / STORE_DIR
|
||||
|
||||
|
||||
def request_fingerprint(
|
||||
expected: list[tuple[float, float]],
|
||||
terrain: Terrain,
|
||||
streams: list[list[tuple[float, float]]],
|
||||
*,
|
||||
mode: str,
|
||||
weights: dict[str, float] | None,
|
||||
max_grade: float,
|
||||
min_radius_m: float,
|
||||
stream_offset_m: float | None,
|
||||
grade_smooth_m: float | None,
|
||||
target_grade: float | None,
|
||||
with_profile: bool,
|
||||
design_speed: int,
|
||||
) -> tuple[str, str]:
|
||||
"""(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키."""
|
||||
pool = _pool_key(
|
||||
expected,
|
||||
terrain,
|
||||
streams,
|
||||
max_grade,
|
||||
min_radius_m,
|
||||
*resolve_inputs(max_grade, stream_offset_m, grade_smooth_m, target_grade),
|
||||
).hex()
|
||||
used = resolve_weights(mode, weights)
|
||||
tail = (mode, sorted(used.items()), bool(with_profile), int(design_speed), ALGO_VERSION)
|
||||
key = hashlib.blake2b((pool + repr(tail)).encode(), digest_size=16).hexdigest()
|
||||
return key, pool
|
||||
|
||||
|
||||
def is_default(
|
||||
mode: str, weights: dict[str, float] | None, criteria: dict[str, Any] | None, terrain: Any
|
||||
) -> bool:
|
||||
"""기본값 결과 — 빈 제약 · 50/25/25 · 설계 지형. 표시일 뿐 지우는 데 쓰지 않음."""
|
||||
empty = not any(value is not None for value in (criteria or {}).values())
|
||||
plain = mode != "weighted" or resolve_weights(mode, weights) == DEFAULT_WEIGHTS
|
||||
return empty and plain and terrain is None
|
||||
|
||||
|
||||
def read_stored(project_root: Path, key: str) -> dict[str, Any] | None:
|
||||
"""저장본 한 벌 — 없거나 못 읽거나 판이 다르면 None."""
|
||||
try:
|
||||
data = json.loads((store_dir(project_root) / f"{key}.json").read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
if not isinstance(data, dict) or data.get("version") != ALGO_VERSION:
|
||||
return None
|
||||
return data
|
||||
|
||||
|
||||
def write_stored(
|
||||
project_root: Path,
|
||||
key: str,
|
||||
pool: str,
|
||||
request: dict[str, Any],
|
||||
result: dict[str, Any],
|
||||
default: bool,
|
||||
) -> bool:
|
||||
"""새 키만 적는다 — 이미 있으면 그대로 두고 False(덮어쓰기 없음)."""
|
||||
path = store_dir(project_root) / f"{key}.json"
|
||||
if path.exists():
|
||||
return False
|
||||
atomic_write_json(
|
||||
path,
|
||||
{
|
||||
"version": ALGO_VERSION,
|
||||
"key": key,
|
||||
"pool": pool,
|
||||
"default": default,
|
||||
"created_at": datetime.now().isoformat(timespec="seconds"),
|
||||
"request": request,
|
||||
"result": result,
|
||||
},
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def pool_lock(pool: str) -> AsyncIterator[None]:
|
||||
# ponytail: 잠금은 지우지 않음 — 입력 묶음 수만큼(작음) · 다중 프로세스면 파일 잠금으로
|
||||
async with _LOCKS.setdefault(pool, asyncio.Lock()):
|
||||
yield
|
||||
@@ -26,7 +26,7 @@
|
||||
GET /api/projects/{id}/route/plan → 예상노선·계획노선 정점(사업지 좌표계)
|
||||
POST /api/projects/{id}/route/replan → 고친 계획노선으로 갈아 끼우고 재계산
|
||||
POST /api/projects/{id}/route/replan/reset → 계획노선을 예상노선으로 되돌리고 재계산
|
||||
POST /api/projects/{id}/route/initial → 초기 계획노선 갈래 하나를 만들어 돌려줌(파일 안 씀)
|
||||
POST /api/projects/{id}/route/initial → 초기 계획노선 갈래 하나를 만들어 돌려줌(추천 저장본만 씀)
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
||||
@@ -30,14 +30,35 @@ async def initial_route(
|
||||
) -> dict[str, Any] | JSONResponse:
|
||||
"""초기 계획노선 갈래 하나를 만들어 비교값과 함께 돌려준다(PLAN 20장).
|
||||
|
||||
**파일을 쓰지 않는다** — 화면 캐시용. 적용은 받은 `nodes` 로 `POST /route/replan`([확인]).
|
||||
작업본은 쓰지 않는다 — 추천 저장본(`..._Store` · PLAN 327)만. 적용은 받은 `nodes` 로
|
||||
`POST /route/replan`([확인]).
|
||||
"""
|
||||
return await compute_initial(project_id, request)
|
||||
|
||||
|
||||
async def compute_initial(
|
||||
project_id: UUID, request: replan.RouteInitialRequest
|
||||
) -> dict[str, Any] | JSONResponse:
|
||||
"""같은 입력이면 저장본을 꺼내고 없으면 셈해 더한다(PLAN 327) · 덮개 수는 늘 지금 관으로.
|
||||
|
||||
응답 `stored` = 저장본에서 꺼냈나. 진행 중인 같은 묶음은 기다렸다 꺼낸다(두 번 셈 막음).
|
||||
"""
|
||||
from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial import (
|
||||
generate_initial_route,
|
||||
load_terrain_and_streams,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import project_pipes
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import (
|
||||
cover_shortfall,
|
||||
project_pipes,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
|
||||
is_default,
|
||||
pool_lock,
|
||||
read_stored,
|
||||
request_fingerprint,
|
||||
write_stored,
|
||||
)
|
||||
from common_util.common_util_surface_confirmation import (
|
||||
analysis_terrain_params,
|
||||
get_surface_confirmation_params,
|
||||
@@ -81,37 +102,62 @@ async def initial_route(
|
||||
async with pool.acquire() as connection:
|
||||
selection = await get_surface_confirmation_params(connection, str(project_id))
|
||||
selection = analysis_terrain_params(selection, request.terrain)
|
||||
inputs = {
|
||||
"mode": request.mode,
|
||||
"weights": weights,
|
||||
"max_grade": max_grade,
|
||||
"min_radius_m": radius_m,
|
||||
"stream_offset_m": stream_offset_m,
|
||||
"grade_smooth_m": grade_smooth_m,
|
||||
"target_grade": None if target_pct is None else target_pct / 100.0,
|
||||
"design_speed": design_speed or 20,
|
||||
"with_profile": request.with_profile,
|
||||
}
|
||||
started = time.perf_counter()
|
||||
try:
|
||||
terrain, streams = await asyncio.to_thread(
|
||||
load_terrain_and_streams, project_root, selection
|
||||
)
|
||||
pipes = (
|
||||
await asyncio.to_thread(project_pipes, project_root) if request.with_profile else None
|
||||
)
|
||||
result = await asyncio.to_thread(
|
||||
functools.partial(
|
||||
generate_initial_route,
|
||||
[(x, y) for x, y in expected],
|
||||
terrain,
|
||||
streams,
|
||||
mode=request.mode,
|
||||
weights=weights,
|
||||
max_grade=max_grade,
|
||||
min_radius_m=radius_m,
|
||||
stream_offset_m=stream_offset_m,
|
||||
grade_smooth_m=grade_smooth_m,
|
||||
target_grade=None if target_pct is None else target_pct / 100.0,
|
||||
design_speed=design_speed or 20,
|
||||
pipes=pipes,
|
||||
with_profile=request.with_profile,
|
||||
)
|
||||
route = [(x, y) for x, y in expected]
|
||||
key, pool_key = await asyncio.to_thread(
|
||||
functools.partial(request_fingerprint, route, terrain, streams, **inputs)
|
||||
)
|
||||
async with pool_lock(pool_key):
|
||||
stored = await asyncio.to_thread(read_stored, project_root, key)
|
||||
if stored is not None:
|
||||
result = stored["result"]
|
||||
else:
|
||||
# 관 지점은 키 밖 — 덮개 수 없이 셈해 남기고 아래에서 지금 관으로 센다.
|
||||
result = await asyncio.to_thread(
|
||||
functools.partial(
|
||||
generate_initial_route, route, terrain, streams, pipes=None, **inputs
|
||||
)
|
||||
)
|
||||
await asyncio.to_thread(
|
||||
write_stored,
|
||||
project_root,
|
||||
key,
|
||||
pool_key,
|
||||
request.model_dump(),
|
||||
result,
|
||||
is_default(
|
||||
request.mode,
|
||||
weights,
|
||||
criteria.model_dump() if criteria else None,
|
||||
request.terrain,
|
||||
),
|
||||
)
|
||||
profile = result["metrics"].get("profile")
|
||||
if request.with_profile and profile and "error" not in profile:
|
||||
pipes = await asyncio.to_thread(project_pipes, project_root)
|
||||
if pipes:
|
||||
profile.update(cover_shortfall(profile, result["planned"], pipes))
|
||||
except (FileNotFoundError, ValueError) as error:
|
||||
return JSONResponse(status_code=409, content={"status": "error", "message": str(error)})
|
||||
logger.info(
|
||||
"초기 계획노선 %s: project_id=%s %.1fs 연장 %.0fm 이격 평균 %.1fm",
|
||||
"초기 계획노선 %s(%s): project_id=%s %.1fs 연장 %.0fm 이격 평균 %.1fm",
|
||||
request.mode,
|
||||
"저장본" if stored is not None else "셈",
|
||||
project_id,
|
||||
time.perf_counter() - started,
|
||||
result["metrics"]["length_m"],
|
||||
@@ -122,10 +168,28 @@ async def initial_route(
|
||||
"status": "success",
|
||||
"project_id": str(project_id),
|
||||
**result,
|
||||
"stored": stored is not None,
|
||||
"current_initial": current,
|
||||
}
|
||||
|
||||
|
||||
async def precompute_defaults(project_id: UUID) -> None:
|
||||
"""업로드 체인 끝 예약(PLAN 327-2) — 기본 다섯 알약을 미리 셈해 남김 · 있으면 꺼내기만."""
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import DEFAULT_MODES
|
||||
|
||||
started = time.perf_counter()
|
||||
for mode in DEFAULT_MODES:
|
||||
# 화면 기본과 같은 몸통 — 빈 제약 · 50/25/25 · 설계 지형 · 추적만 종단 끔(`withProfileOf`)
|
||||
request = replan.RouteInitialRequest(mode=mode, with_profile=mode != "follow")
|
||||
outcome = await compute_initial(project_id, request)
|
||||
if isinstance(outcome, JSONResponse):
|
||||
logger.warning(
|
||||
"추천 미리 셈 멈춤(%s %s): project_id=%s", mode, outcome.status_code, project_id
|
||||
)
|
||||
return
|
||||
logger.info("추천 미리 셈 끝: project_id=%s %.1fs", project_id, time.perf_counter() - started)
|
||||
|
||||
|
||||
def _set_base_pvi(path: Path, stations: list[float], plan: list[float]) -> int:
|
||||
"""종단 정본의 1차 변화점(`base_pvi`) 계획고를 `stations` · `plan` 직선 보간으로 바꿔 쓴다."""
|
||||
import json
|
||||
|
||||
@@ -0,0 +1,231 @@
|
||||
"""계획노선 추천 결과 영구 저장(PLAN 327 · 요구 224-3) — `B05_Profile_Engine_RouteInitial_Store`.
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_224_3_recommend_store.py -q
|
||||
|
||||
같은 입력 = 같은 키 · 하나만 바뀌어도 다른 키 · 두 번째 요청 셈 0회 · 덮어쓰기 없음 ·
|
||||
[초기화](스냅샷 복원) 뒤에도 남음 · 자료 교체 때 지워짐 · 덮개 수는 꺼낼 때 지금 관으로.
|
||||
엔드포인트 몸통(`compute_initial`)은 DB · 지형 읽기만 갈아 끼우고 그대로 돌린다.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import shutil
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from B05_Profile import B05_Profile_Router_Replan as replan
|
||||
from B05_Profile import B05_Profile_Router_RouteInitial as router
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial import STREAM_OFFSET_M, Terrain
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
|
||||
DEFAULT_MODES,
|
||||
request_fingerprint,
|
||||
store_dir,
|
||||
write_stored,
|
||||
)
|
||||
from common_util.common_util_initial_snapshot import (
|
||||
_FILE_TREES,
|
||||
restore_snapshot_files,
|
||||
snapshot_dir,
|
||||
)
|
||||
from common_util.common_util_project_reset import purge_output_folders
|
||||
|
||||
|
||||
def _terrain(lift: float = 0.0) -> Terrain:
|
||||
x = np.arange(0.0, 40.0, 2.0)
|
||||
y = np.arange(0.0, 20.0, 2.0)
|
||||
xx, yy = np.meshgrid(x, y)
|
||||
return Terrain(x, y, 0.1 * xx + lift, np.ones(xx.shape, dtype=bool))
|
||||
|
||||
|
||||
EXPECTED = [(0.0, 10.0), (20.0, 10.0), (38.0, 10.0)]
|
||||
STREAMS = [[(5.0, 0.0), (5.0, 18.0)]]
|
||||
BASE = {
|
||||
"mode": "weighted",
|
||||
"weights": None,
|
||||
"max_grade": 0.14,
|
||||
"min_radius_m": 12.0,
|
||||
"stream_offset_m": None,
|
||||
"grade_smooth_m": None,
|
||||
"target_grade": None,
|
||||
"with_profile": True,
|
||||
"design_speed": 20,
|
||||
}
|
||||
|
||||
|
||||
def _key(expected=EXPECTED, terrain=None, streams=STREAMS, **changes) -> str:
|
||||
return request_fingerprint(expected, terrain or _terrain(), streams, **{**BASE, **changes})[0]
|
||||
|
||||
|
||||
def test_same_input_same_key() -> None:
|
||||
assert _key() == _key()
|
||||
# 비운 값 = 기본값 · 비중 None = 50/25/25 — 화면이 채워 보내도 같은 키
|
||||
assert _key() == _key(stream_offset_m=STREAM_OFFSET_M)
|
||||
assert _key() == _key(weights={"expected": 50.0, "earthwork": 25.0, "grade": 25.0})
|
||||
# 비중 갈래가 아니면 비중은 갈래가 정함 — 보낸 비중은 키에 안 듦
|
||||
assert _key(mode="earthwork") == _key(mode="earthwork", weights={"expected": 1.0})
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"change",
|
||||
[
|
||||
{"mode": "earthwork"},
|
||||
{"weights": {"expected": 40.0, "earthwork": 30.0, "grade": 30.0}},
|
||||
{"max_grade": 0.12},
|
||||
{"min_radius_m": 15.0},
|
||||
{"stream_offset_m": 50.0},
|
||||
{"grade_smooth_m": 3.0},
|
||||
{"target_grade": 0.08},
|
||||
{"with_profile": False},
|
||||
{"design_speed": 30},
|
||||
{"expected": [(0.0, 10.0), (20.0, 11.0), (38.0, 10.0)]},
|
||||
{"terrain": _terrain(lift=0.5)},
|
||||
{"streams": []},
|
||||
],
|
||||
ids=lambda change: next(iter(change)),
|
||||
)
|
||||
def test_one_change_other_key(change: dict) -> None:
|
||||
assert _key(**change) != _key()
|
||||
|
||||
|
||||
def test_no_overwrite(tmp_path: Path) -> None:
|
||||
assert write_stored(tmp_path, "k", "p", {}, {"n": 1}, True)
|
||||
assert not write_stored(tmp_path, "k", "p", {}, {"n": 2}, False)
|
||||
data = json.loads((store_dir(tmp_path) / "k.json").read_text(encoding="utf-8"))
|
||||
assert data["result"] == {"n": 1} and data["default"] is True
|
||||
assert set(data) == {"version", "key", "pool", "default", "created_at", "request", "result"}
|
||||
|
||||
|
||||
def test_survives_reset_and_goes_with_data_swap(tmp_path: Path) -> None:
|
||||
(tmp_path / "B05_Profile/route").mkdir(parents=True)
|
||||
(tmp_path / "B05_Profile/route/initial_choice.json").write_text("{}", encoding="utf-8")
|
||||
write_stored(tmp_path, "k", "p", {}, {}, True)
|
||||
# 스냅샷 = 작업본 트리 사본(`save_initial_snapshot` 과 같은 자리)
|
||||
for tree in _FILE_TREES:
|
||||
source = tmp_path / tree
|
||||
if source.is_dir():
|
||||
shutil.copytree(source, snapshot_dir(tmp_path) / tree.replace("/", "__"))
|
||||
restore_snapshot_files(tmp_path)
|
||||
assert (store_dir(tmp_path) / "k.json").is_file()
|
||||
purge_output_folders(tmp_path) # 자료 교체
|
||||
assert not store_dir(tmp_path).exists()
|
||||
|
||||
|
||||
class _Calls:
|
||||
def __init__(self) -> None:
|
||||
self.count = 0
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls:
|
||||
"""`compute_initial` 의 DB · 파일 읽기를 갈아 끼움 · 셈은 횟수만 세는 가짜."""
|
||||
import B05_Profile.B05_Profile_Engine_Grade as grade
|
||||
import B05_Profile.B05_Profile_Engine_RouteInitial as engine
|
||||
import B05_Profile.B05_Profile_Engine_RouteInitial_Profile as profile
|
||||
import common_util.common_util_surface_confirmation as surface
|
||||
|
||||
calls = _Calls()
|
||||
|
||||
async def project_paths(_pid):
|
||||
return tmp_path, "x"
|
||||
|
||||
async def plan_criteria(_pid):
|
||||
return 12.0, 12.0, 5.0
|
||||
|
||||
async def road_settings(_pid):
|
||||
return "간선", 20, "일반"
|
||||
|
||||
async def surface_params(_connection, _pid):
|
||||
return {}
|
||||
|
||||
class _Pool:
|
||||
@asynccontextmanager
|
||||
async def acquire(self):
|
||||
yield None
|
||||
|
||||
def generate(expected, terrain, streams, *, pipes, **_inputs):
|
||||
assert pipes is None # 관은 키 밖 — 덮개 없이 셈해 남김
|
||||
calls.count += 1
|
||||
return {
|
||||
"planned": [[x, y] for x, y in expected],
|
||||
"metrics": {
|
||||
"length_m": 38.0,
|
||||
"offset_avg_m": 0.0,
|
||||
"profile": {
|
||||
"stations_m": [0.0, 38.0],
|
||||
"plan_m": [1.0, 1.0],
|
||||
"ground_m": [0.0, 0.0],
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
monkeypatch.setattr(replan, "_project_paths", project_paths)
|
||||
monkeypatch.setattr(replan, "_plan_criteria", plan_criteria)
|
||||
monkeypatch.setattr(replan, "_road_settings", road_settings)
|
||||
monkeypatch.setattr(replan, "_vertices_of", lambda _path: [list(p) for p in EXPECTED])
|
||||
monkeypatch.setattr(router, "get_db_pool", lambda: _Pool())
|
||||
monkeypatch.setattr(surface, "get_surface_confirmation_params", surface_params)
|
||||
monkeypatch.setattr(surface, "analysis_terrain_params", lambda selection, _t: selection)
|
||||
monkeypatch.setattr(grade, "legal_grade_limit_pct", lambda *_a: 14.0)
|
||||
monkeypatch.setattr(engine, "load_terrain_and_streams", lambda *_a: (_terrain(), STREAMS))
|
||||
monkeypatch.setattr(engine, "generate_initial_route", generate)
|
||||
# 덮개 여유 2 m 관 하나가 노선 위 — 계획고 − 지반고 1 m 라 모자람 1
|
||||
monkeypatch.setattr(profile, "project_pipes", lambda _root: [(19.0, 10.0, 2.0)])
|
||||
calls.root = tmp_path
|
||||
return calls
|
||||
|
||||
|
||||
def _ask(**fields) -> dict:
|
||||
request = replan.RouteInitialRequest(**fields)
|
||||
return asyncio.run(router.compute_initial(uuid4(), request))
|
||||
|
||||
|
||||
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")
|
||||
assert wired.count == 1 and second["stored"] is True
|
||||
assert second["planned"] == first["planned"]
|
||||
# 덮개 수는 꺼낼 때마다 지금 관으로 · 저장본에는 없음
|
||||
assert second["metrics"]["profile"]["cover_short"] == 1
|
||||
files = list(store_dir(wired.root).glob("*.json"))
|
||||
assert len(files) == 1
|
||||
saved = json.loads(files[0].read_text(encoding="utf-8"))
|
||||
assert "cover_short" not in saved["result"]["metrics"]["profile"]
|
||||
assert saved["default"] is True and saved["request"]["mode"] == "earthwork"
|
||||
# 제약 하나 바꾸면 파일 하나 더 · 옛 파일 그대로 · 기본값 표시 꺼짐
|
||||
changed = _ask(mode="earthwork", criteria={"min_radius_m": 15.0})
|
||||
assert changed["stored"] is False and wired.count == 2
|
||||
assert len(list(store_dir(wired.root).glob("*.json"))) == 2
|
||||
assert json.loads(files[0].read_text(encoding="utf-8")) == saved
|
||||
|
||||
|
||||
def test_same_pool_in_flight_waits(wired: _Calls) -> None:
|
||||
async def both() -> list[dict]:
|
||||
request = replan.RouteInitialRequest(mode="gentle")
|
||||
project = uuid4()
|
||||
return await asyncio.gather(
|
||||
router.compute_initial(project, request), router.compute_initial(project, request)
|
||||
)
|
||||
|
||||
results = asyncio.run(both())
|
||||
assert wired.count == 1
|
||||
assert sorted(result["stored"] for result in results) == [False, True]
|
||||
|
||||
|
||||
def test_precompute_defaults_then_hits(wired: _Calls) -> None:
|
||||
asyncio.run(router.precompute_defaults(uuid4()))
|
||||
assert wired.count == len(DEFAULT_MODES) == 5
|
||||
saved = [
|
||||
json.loads(path.read_text(encoding="utf-8"))
|
||||
for path in store_dir(wired.root).glob("*.json")
|
||||
]
|
||||
assert sorted(item["request"]["mode"] for item in saved) == sorted(DEFAULT_MODES)
|
||||
assert all(item["default"] for item in saved)
|
||||
# 화면 첫 누름(비중은 화면이 50/25/25 를 채워 보냄 · 추적은 종단 끔) — 셈 없이 저장본
|
||||
hit = _ask(mode="weighted", weights={"expected": 50, "earthwork": 25, "grade": 25})
|
||||
follow = _ask(mode="follow", with_profile=False)
|
||||
assert hit["stored"] and follow["stored"] and wired.count == 5
|
||||
Reference in New Issue
Block a user