fix(노선추천): 351-1 저장 빈틈 넷 — 재확정 체인 끝 미리 셈 · 기본값 · 저장된 제약 결과는 「최근 2벌」 정리에서 따로 셈 · 저장된 제약으로도 미리 셈 · B01 사용 범위 바꾸면 예상노선 다시 씀 + 미리 셈
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ji4nuyLVHcxrXWHTdb5nRD
This commit is contained in:
@@ -2,6 +2,7 @@
|
||||
|
||||
import html
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Path, Query, Request, Response
|
||||
from pymysql.err import IntegrityError
|
||||
@@ -378,9 +379,24 @@ async def dashboard_update_project(
|
||||
if member_ids is not None:
|
||||
# 만든 사람은 늘 참여자로 남는다.
|
||||
await set_project_members(project_id, [int(project["user_id"]), *member_ids])
|
||||
if _range_of(project) != _range_of(data):
|
||||
# 사용 범위가 바뀌면 추천 시점 · 종점(예상노선)도 따라감 · 미리 셈(351-1)
|
||||
from B03_FileInput.B03_FileInput_Router_Helpers import _schedule_background_task
|
||||
from B05_Profile.B05_Profile_Router_RouteInitial import refresh_expected_route
|
||||
|
||||
_schedule_background_task(
|
||||
refresh_expected_route(UUID(project_id)), task_name=f"예상노선 다시 씀 {project_id}"
|
||||
)
|
||||
return {"status": "success"}
|
||||
|
||||
|
||||
def _range_of(row: dict[str, Any]) -> tuple[float | None, float | None]:
|
||||
return tuple( # type: ignore[return-value]
|
||||
None if row.get(key) is None else float(row[key])
|
||||
for key in ("route_start_m", "route_end_m")
|
||||
)
|
||||
|
||||
|
||||
@router.delete("/projects/{project_id}")
|
||||
async def dashboard_delete_project(
|
||||
project_id: str,
|
||||
|
||||
@@ -342,15 +342,9 @@ async def run_auto_design_chain(
|
||||
|
||||
# 노선 추천 기본 다섯 알약 미리 셈(PLAN 327-2) — 예약만 · 체인이 [초기화] 요청 안에서도
|
||||
# 불리므로 기다리지 않는다. 같은 입력이면 저장본이 있어 셈 없이 끝난다.
|
||||
try:
|
||||
from B03_FileInput.B03_FileInput_Router_Helpers import _schedule_background_task
|
||||
from B05_Profile.B05_Profile_Router_RouteInitial import precompute_defaults
|
||||
from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
|
||||
|
||||
_schedule_background_task(
|
||||
precompute_defaults(project_id), task_name=f"추천 노선 미리 셈 {project_id}"
|
||||
)
|
||||
except Exception: # noqa: BLE001 — 미리 셈이 체인을 막지는 않는다
|
||||
logger.exception("추천 노선 미리 셈 예약 실패: project_id=%s", project_id)
|
||||
schedule_precompute(project_id)
|
||||
|
||||
return {
|
||||
"route_id": route_id,
|
||||
@@ -663,6 +657,10 @@ async def run_redesign_chain(
|
||||
return f"종횡단 확정 실패({sections_result.status_code})"
|
||||
marks.append(("B06 확정(서버 재계산 포함)", time.perf_counter()))
|
||||
_log_steps("재확정 체인", marks)
|
||||
# 지형이 바뀌면 추천 키도 바뀜 — 첫 누름이 저장본이게 미리 셈(351-1)
|
||||
from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
|
||||
|
||||
schedule_precompute(project_id)
|
||||
logger.info(
|
||||
"재확정 체인 완료: project_id=%s route %s→%s (설계 %d건 이월)",
|
||||
project_id,
|
||||
|
||||
@@ -10,6 +10,7 @@
|
||||
키 = 후보 묶음 지문(`_pool_key` — 지형 · 예상노선 · 계류 · 상한 · 최소 R · 이격 · 고름 · 목표
|
||||
기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · `ALGO_VERSION`. 관 지점은 뺀다 — 덮개 수는 꺼낼 때
|
||||
다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을 올린다(옛 파일은 다른 키라 안 쓰임).
|
||||
기본값 · 저장된 제약 결과(`default`)는 「최근 벌」을 따로 센다(351-1).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -77,12 +78,26 @@ def request_fingerprint(
|
||||
|
||||
|
||||
def is_default(
|
||||
mode: str, weights: dict[str, float] | None, criteria: dict[str, Any] | None, terrain: Any
|
||||
mode: str,
|
||||
weights: dict[str, float] | None,
|
||||
criteria: dict[str, Any] | None,
|
||||
terrain: Any,
|
||||
saved: dict[str, Any] | None = None,
|
||||
) -> 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
|
||||
"""지킬 결과 — 기본값(빈 제약 · 50/25/25 · 설계 지형) 또는 저장된 제약(`initial_choice`) 그대로.
|
||||
|
||||
모달이 여는 값이라 「최근 2벌」 정리에서 따로 센다(351-1 · 사용자 시도에 안 밀림).
|
||||
"""
|
||||
if terrain is not None:
|
||||
return False
|
||||
filled = {key: value for key, value in (criteria or {}).items() if value is not None}
|
||||
used = resolve_weights(mode, weights)
|
||||
if not filled and (mode != "weighted" or used == DEFAULT_WEIGHTS):
|
||||
return True
|
||||
saved = saved or {}
|
||||
return filled == (saved.get("criteria") or {}) and (
|
||||
mode != "weighted" or used == saved.get("weights")
|
||||
)
|
||||
|
||||
|
||||
def read_stored(project_root: Path, key: str) -> dict[str, Any] | None:
|
||||
@@ -125,23 +140,26 @@ def write_stored(
|
||||
"result": result,
|
||||
},
|
||||
)
|
||||
prune_stored(project_root, request)
|
||||
prune_stored(project_root, request, default)
|
||||
return True
|
||||
|
||||
|
||||
def _kind(request: Any) -> tuple[Any, Any]:
|
||||
def _kind(request: Any, default: Any) -> tuple[Any, Any, bool]:
|
||||
request = request if isinstance(request, dict) else {}
|
||||
return request.get("mode"), request.get("terrain")
|
||||
return request.get("mode"), request.get("terrain"), bool(default)
|
||||
|
||||
|
||||
def prune_stored(project_root: Path, request: dict[str, Any]) -> int:
|
||||
"""같은 종류(갈래 · 지형) 저장본을 최근 `KEEP_PER_KIND` 벌만 남기고 지운다. 지운 수."""
|
||||
kind = _kind(request)
|
||||
def prune_stored(project_root: Path, request: dict[str, Any], default: bool = False) -> int:
|
||||
"""같은 종류(갈래 · 지형 · 기본값 여부) 저장본을 최근 `KEEP_PER_KIND` 벌만 남기고 지운다.
|
||||
|
||||
기본값 · 저장된 제약 결과(`default`)는 따로 셈 — 사용자 시도가 밀어내지 않음(351-1). 지운 수.
|
||||
"""
|
||||
kind = _kind(request, default)
|
||||
same: list[tuple[int, Path]] = []
|
||||
for path in store_dir(project_root).glob("*.json"):
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
if _kind(data.get("request")) == kind:
|
||||
if _kind(data.get("request"), data.get("default")) == kind:
|
||||
same.append((path.stat().st_mtime_ns, path))
|
||||
except (OSError, ValueError, AttributeError):
|
||||
continue
|
||||
|
||||
@@ -76,6 +76,7 @@ async def compute_initial(
|
||||
if paths is None:
|
||||
return JSONResponse(status_code=404, content=replan._PROJECT_PATH_MISSING)
|
||||
project_root, _ = paths
|
||||
current = await asyncio.to_thread(read_initial_choice, project_root)
|
||||
expected = await asyncio.to_thread(replan._vertices_of, expected_route_csv_path(project_root))
|
||||
if not expected:
|
||||
expected = await asyncio.to_thread(replan._vertices_of, design_route_csv_path(project_root))
|
||||
@@ -144,6 +145,7 @@ async def compute_initial(
|
||||
weights,
|
||||
criteria.model_dump() if criteria else None,
|
||||
request.terrain,
|
||||
current,
|
||||
),
|
||||
)
|
||||
profile = result["metrics"].get("profile")
|
||||
@@ -162,7 +164,6 @@ async def compute_initial(
|
||||
result["metrics"]["length_m"],
|
||||
result["metrics"]["offset_avg_m"],
|
||||
)
|
||||
current = await asyncio.to_thread(read_initial_choice, project_root)
|
||||
return {
|
||||
"status": "success",
|
||||
"project_id": str(project_id),
|
||||
@@ -173,22 +174,99 @@ async def compute_initial(
|
||||
|
||||
|
||||
async def precompute_defaults(project_id: UUID) -> None:
|
||||
"""업로드 체인 끝 예약(PLAN 327-2) — 기본 다섯 알약을 미리 셈해 남김 · 있으면 꺼내기만."""
|
||||
"""체인 끝 예약(PLAN 327-2 · 351-1) — 기본 다섯 알약 + 저장된 제약(initial_choice)의 다섯.
|
||||
|
||||
모달은 저장된 제약 · 비중으로 열리므로 그 값으로도 미리 셈한다. 있으면 꺼내기만.
|
||||
"""
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial import DEFAULT_WEIGHTS
|
||||
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
|
||||
variants: list[tuple[Any, Any]] = [(None, None)]
|
||||
try:
|
||||
paths = await replan._project_paths(project_id)
|
||||
except LookupError:
|
||||
paths = None
|
||||
if paths is not None:
|
||||
saved = await asyncio.to_thread(read_initial_choice, paths[0])
|
||||
weights = saved["weights"] if saved["logic"] == "weighted" else None
|
||||
if saved["criteria"] or (weights and weights != DEFAULT_WEIGHTS):
|
||||
variants.append((saved["criteria"], weights))
|
||||
for criteria, weights in variants:
|
||||
for mode in DEFAULT_MODES:
|
||||
# 화면과 같은 몸통 — 설계 지형 · 비중은 비중 갈래만 · 추적만 종단 끔(`withProfileOf`)
|
||||
request = replan.RouteInitialRequest(
|
||||
mode=mode,
|
||||
with_profile=mode != "follow",
|
||||
criteria=criteria,
|
||||
weights=weights if mode == "weighted" else None,
|
||||
)
|
||||
return
|
||||
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 schedule_precompute(project_id: UUID) -> None:
|
||||
"""추천 미리 셈 예약 — 업로드 · 재확정 체인 끝(351-1). 기다리지 않음 · 실패는 로그만."""
|
||||
try:
|
||||
from B03_FileInput.B03_FileInput_Router_Helpers import _schedule_background_task
|
||||
|
||||
_schedule_background_task(
|
||||
precompute_defaults(project_id), task_name=f"추천 노선 미리 셈 {project_id}"
|
||||
)
|
||||
except Exception: # noqa: BLE001 — 미리 셈이 체인을 막지는 않는다
|
||||
logger.exception("추천 노선 미리 셈 예약 실패: project_id=%s", project_id)
|
||||
|
||||
|
||||
async def refresh_expected_route(project_id: UUID) -> None:
|
||||
"""B01 노선 사용 범위를 바꾼 뒤(351-1) — 예상노선을 새 범위로 다시 쓰고 추천 미리 셈.
|
||||
|
||||
체인을 아직 안 돈 프로젝트(예상노선 없음)는 건너뜀 — 체인이 새 범위로 적는다. 설계 노선
|
||||
(수정본 · 초기 폴리라인)은 그대로 — 추천의 시점 · 종점만 따라간다.
|
||||
"""
|
||||
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
||||
from common_util.common_util_route_geometry import load_design_route, write_route_csv
|
||||
from common_util.common_util_storage import resolve_stored_project_path
|
||||
from common_util.common_util_surface_confirmation import get_surface_confirmation_params
|
||||
|
||||
pool = get_db_pool()
|
||||
async with pool.acquire() as connection:
|
||||
stored_path = await get_project_storage_relative_path(connection, project_id)
|
||||
surface = await get_surface_confirmation_params(connection, str(project_id))
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"SELECT route_start_m, route_end_m FROM projects WHERE id = %s",
|
||||
(str(project_id),),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not stored_path:
|
||||
return
|
||||
root = Path(resolve_stored_project_path(stored_path))
|
||||
path = expected_route_csv_path(root)
|
||||
if not path.is_file():
|
||||
return
|
||||
planned = await asyncio.to_thread(
|
||||
functools.partial(
|
||||
load_design_route,
|
||||
root,
|
||||
surface,
|
||||
(row[0], row[1]) if row else None,
|
||||
from_source=True,
|
||||
)
|
||||
)
|
||||
if planned is None:
|
||||
logger.warning("예상노선 다시 쓰기 건너뜀(노선 못 읽음): project_id=%s", project_id)
|
||||
return
|
||||
points = [{"x": v.x, "y": v.y} for v in planned.vertices]
|
||||
await asyncio.to_thread(write_route_csv, path, points)
|
||||
logger.info("예상노선 다시 씀(사용 범위): project_id=%s 점 %d", project_id, len(points))
|
||||
await precompute_defaults(project_id)
|
||||
|
||||
|
||||
def _set_base_pvi(path: Path, stations: list[float], plan: list[float]) -> int:
|
||||
"""종단 정본의 1차 변화점(`base_pvi`) 계획고를 `stations` · `plan` 직선 보간으로 바꿔 쓴다."""
|
||||
import json
|
||||
|
||||
@@ -265,6 +265,8 @@ def load_design_route(
|
||||
project_root: Path,
|
||||
surface_params: dict[str, Any] | None = None,
|
||||
route_range: tuple[float | None, float | None] | None = None,
|
||||
*,
|
||||
from_source: bool = False,
|
||||
) -> PlannedRoute | None:
|
||||
"""설계가 쓸 계획노선 한 벌을 만든다 — 읽기·좌표계 변환·트림·조밀화를 여기서 끝낸다.
|
||||
|
||||
@@ -279,6 +281,9 @@ def load_design_route(
|
||||
|
||||
`route_range`(시작·종료 누가거리 m)를 주면 **서피스 트림보다 먼저** 그 구간만 남긴다
|
||||
(2026-09-04 사용자 지시). 순서가 바뀌면 사용자가 정한 시점이 서피스 트림에 밀린다.
|
||||
|
||||
`from_source` 면 CSV 정본을 건너뛰고 올린 원본을 다시 읽는다 — 사용 범위를 바꾼 뒤
|
||||
예상노선을 다시 쓸 때(351-1).
|
||||
"""
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Extent import project_epsg_from_prj
|
||||
from common_util.common_util_initial_snapshot import design_route_csv_path
|
||||
@@ -293,7 +298,7 @@ def load_design_route(
|
||||
# 지표면·노선이 바뀌면 `discard_initial_snapshot()`이 폴더째 지우므로 이 경로가 저절로
|
||||
# 닫히고 원본 재판독으로 되돌아간다. 트림 **전** 원본이 필요한 호출(도엽 범위 —
|
||||
# surface_params 없음)은 여기를 타지 않는다.
|
||||
if surface_params:
|
||||
if surface_params and not from_source:
|
||||
# 읽는 순서 — 수정본 → **초기 폴리라인** → 예상노선(점 묶음) → 초기값 스냅샷.
|
||||
# 초기 폴리라인이 예상노선보다 앞선다: 예상노선은 점 묶음이라 그대로 이으면
|
||||
# 규칙 없는 선이 된다(2026-09-06 사용자 지시).
|
||||
|
||||
@@ -297,8 +297,11 @@ def test_추천_저장본_종류당_최근_두_벌(tmp_path: Path) -> None:
|
||||
assert read_stored(tmp_path, "a") is not None # 꺼내면 최근으로
|
||||
time.sleep(0.02)
|
||||
write_stored(tmp_path, "c", "p", request, {"n": 3}, False)
|
||||
time.sleep(0.02)
|
||||
write_stored(tmp_path, "d", "p", request, {"n": 4}, False)
|
||||
names = sorted(path.stem for path in store_dir(tmp_path).glob("*.json"))
|
||||
assert names == ["a", "c", "g"]
|
||||
# 기본값(a)은 따로 셈 — 사용자 시도 b · c · d 중 옛 b 만 지움(351-1)
|
||||
assert names == ["a", "c", "d", "g"]
|
||||
|
||||
|
||||
# ── 횡단 상세 — GET 은 안 지움 · ETag(R26) ───────────────────────────────────────
|
||||
|
||||
@@ -0,0 +1,194 @@
|
||||
"""351-1 노선 추천 저장 빈틈 — 재확정 체인 · 기본값 지킴 · 저장된 제약 · B01 사용 범위.
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_351_1_recommend_gaps.py -q
|
||||
|
||||
체인 뒤 첫 누름 = 저장본(기본값 · 저장된 제약 둘 다) · 「최근 2벌」 정리가 기본값을 안 밀어냄 ·
|
||||
사용 범위를 바꾸면 예상노선을 원본에서 다시 자르고 미리 셈.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import time
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
|
||||
from B05_Profile import B05_Profile_Router_RouteInitial as router
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial import write_initial_choice
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
|
||||
DEFAULT_MODES,
|
||||
is_default,
|
||||
store_dir,
|
||||
write_stored,
|
||||
)
|
||||
from common_util.common_util_route_geometry import (
|
||||
expected_route_csv_path,
|
||||
load_design_route,
|
||||
planned_route_working_path,
|
||||
write_route_csv,
|
||||
)
|
||||
from test_224_3_recommend_store import _ask, wired # noqa: F401 — 픽스처
|
||||
|
||||
|
||||
def _saved(root: Path) -> list[dict]:
|
||||
return [json.loads(p.read_text(encoding="utf-8")) for p in store_dir(root).glob("*.json")]
|
||||
|
||||
|
||||
def test_저장된_제약으로도_미리_셈_첫_누름_저장본(wired) -> None: # noqa: F811
|
||||
criteria = {"min_radius_m": 15.0, "max_grade_pct": 12.0}
|
||||
write_initial_choice(
|
||||
wired.root, "weighted", {"expected": 40, "earthwork": 30, "grade": 30}, criteria
|
||||
)
|
||||
asyncio.run(router.precompute_defaults(uuid4()))
|
||||
assert wired.count == 2 * len(DEFAULT_MODES)
|
||||
assert all(item["default"] for item in _saved(wired.root))
|
||||
# 모달은 저장된 제약 · 비중으로 열림 — 셈 없이 저장본
|
||||
hit = _ask(
|
||||
mode="weighted", weights={"expected": 40, "earthwork": 30, "grade": 30}, criteria=criteria
|
||||
)
|
||||
gentle = _ask(mode="gentle", criteria=criteria)
|
||||
plain = _ask(mode="earthwork")
|
||||
assert hit["stored"] and gentle["stored"] and plain["stored"]
|
||||
assert wired.count == 10
|
||||
# 다시 불러도 셈 0
|
||||
asyncio.run(router.precompute_defaults(uuid4()))
|
||||
assert wired.count == 10
|
||||
|
||||
|
||||
def test_저장된_제약_없으면_기본값만(wired) -> None: # noqa: F811
|
||||
write_initial_choice(wired.root, "gentle", None, None)
|
||||
asyncio.run(router.precompute_defaults(uuid4()))
|
||||
assert wired.count == len(DEFAULT_MODES)
|
||||
|
||||
|
||||
def test_기본값은_최근_두_벌_정리에서_안_밀림(tmp_path: Path) -> None:
|
||||
request = {"mode": "earthwork", "terrain": None}
|
||||
write_stored(tmp_path, "d", "p", request, {}, True)
|
||||
for name in ("a", "b", "c"):
|
||||
time.sleep(0.02)
|
||||
write_stored(tmp_path, name, "p", request, {}, False)
|
||||
names = sorted(path.stem for path in store_dir(tmp_path).glob("*.json"))
|
||||
assert names == ["b", "c", "d"]
|
||||
|
||||
|
||||
def test_기본값_판정() -> None:
|
||||
saved = {
|
||||
"criteria": {"min_radius_m": 15.0},
|
||||
"weights": {"expected": 40.0, "earthwork": 30.0, "grade": 30.0},
|
||||
}
|
||||
assert is_default("gentle", None, None, None)
|
||||
assert is_default("gentle", None, {"min_radius_m": None}, None)
|
||||
assert not is_default("gentle", None, None, "las")
|
||||
assert not is_default("gentle", None, {"min_radius_m": 15.0}, None)
|
||||
assert is_default("gentle", None, {"min_radius_m": 15.0}, None, saved)
|
||||
assert not is_default("gentle", None, {"min_radius_m": 16.0}, None, saved)
|
||||
weights = {"expected": 40.0, "earthwork": 30.0, "grade": 30.0}
|
||||
assert is_default("weighted", weights, {"min_radius_m": 15.0}, None, saved)
|
||||
assert not is_default("weighted", None, {"min_radius_m": 15.0}, None, saved)
|
||||
|
||||
|
||||
def test_두_체인_끝에서_미리_셈_예약() -> None:
|
||||
source = Path("B03_FileInput/B03_FileInput_Service_Chain.py").read_text(encoding="utf-8")
|
||||
redesign = source[source.index("async def run_redesign_chain(") :]
|
||||
upload = source[: source.index("async def run_redesign_chain(")]
|
||||
assert "schedule_precompute(project_id)" in upload
|
||||
# 재확정 체인 성공 끝(완료 로그 앞)
|
||||
done = redesign.index("재확정 체인 완료")
|
||||
assert "schedule_precompute(project_id)" in redesign[:done]
|
||||
b01 = Path("B01_Dashboard/B01_Dashboard_Router.py").read_text(encoding="utf-8")
|
||||
assert "refresh_expected_route(" in b01
|
||||
|
||||
|
||||
def test_사용_범위_비교() -> None:
|
||||
from decimal import Decimal
|
||||
|
||||
from B01_Dashboard.B01_Dashboard_Router import _range_of
|
||||
|
||||
assert _range_of({"route_start_m": Decimal("10.0"), "route_end_m": None}) == (10.0, None)
|
||||
assert _range_of({"route_start_m": 10.0, "route_end_m": None}) == _range_of(
|
||||
{"route_start_m": Decimal("10"), "route_end_m": None}
|
||||
)
|
||||
assert _range_of({"route_start_m": 10.0}) != _range_of({"route_start_m": 20.0})
|
||||
|
||||
|
||||
def _source_project(root: Path) -> None:
|
||||
write_route_csv(
|
||||
root / "B03_FileInput/input/route.csv",
|
||||
[{"x": float(x), "y": 0.0} for x in range(0, 101, 10)],
|
||||
)
|
||||
# 설계 정본(수정본) · 예상노선이 따로 있음 — 원본에서 다시 읽어야 함
|
||||
write_route_csv(planned_route_working_path(root), [{"x": 0.0, "y": 5.0}, {"x": 9.0, "y": 5.0}])
|
||||
write_route_csv(expected_route_csv_path(root), [{"x": 0.0, "y": 0.0}, {"x": 100.0, "y": 0.0}])
|
||||
|
||||
|
||||
def test_원본에서_다시_읽기(tmp_path: Path) -> None:
|
||||
_source_project(tmp_path)
|
||||
surface = {"source_filter": "csf", "method": "dtm", "smooth": False}
|
||||
stored = load_design_route(tmp_path, surface, (20.0, 80.0))
|
||||
assert [v.y for v in stored.vertices] == [5.0, 5.0] # 정본이 먼저
|
||||
fresh = load_design_route(tmp_path, surface, (20.0, 80.0), from_source=True)
|
||||
xs = [round(v.x, 3) for v in fresh.vertices]
|
||||
assert xs[0] == 20.0 and xs[-1] == 80.0
|
||||
|
||||
|
||||
def test_사용_범위_바꾸면_예상노선_다시_쓰고_미리_셈(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
import B03_FileInput.B03_FileInput_Repository as repo
|
||||
import common_util.common_util_storage as storage
|
||||
import common_util.common_util_surface_confirmation as surface
|
||||
|
||||
_source_project(tmp_path)
|
||||
calls: list[str] = []
|
||||
|
||||
class _Cursor:
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *_):
|
||||
return None
|
||||
|
||||
async def execute(self, *_):
|
||||
return None
|
||||
|
||||
async def fetchone(self):
|
||||
return (30.0, 70.0)
|
||||
|
||||
class _Connection:
|
||||
def cursor(self):
|
||||
return _Cursor()
|
||||
|
||||
class _Acquire:
|
||||
async def __aenter__(self):
|
||||
return _Connection()
|
||||
|
||||
async def __aexit__(self, *_):
|
||||
return None
|
||||
|
||||
class _Pool:
|
||||
def acquire(self):
|
||||
return _Acquire()
|
||||
|
||||
async def storage_path(*_):
|
||||
return "x"
|
||||
|
||||
async def surface_params(*_):
|
||||
return {"source_filter": "csf", "method": "dtm", "smooth": False}
|
||||
|
||||
async def precompute(_pid):
|
||||
calls.append("precompute")
|
||||
|
||||
monkeypatch.setattr(router, "get_db_pool", lambda: _Pool())
|
||||
monkeypatch.setattr(repo, "get_project_storage_relative_path", storage_path)
|
||||
monkeypatch.setattr(storage, "resolve_stored_project_path", lambda _p: str(tmp_path))
|
||||
monkeypatch.setattr(surface, "get_surface_confirmation_params", surface_params)
|
||||
monkeypatch.setattr(router, "precompute_defaults", precompute)
|
||||
asyncio.run(router.refresh_expected_route(uuid4()))
|
||||
rows = expected_route_csv_path(tmp_path).read_text(encoding="utf-8").splitlines()
|
||||
assert rows[1].split(",")[1:] == ["30.0", "0.0"] and rows[-1].split(",")[1:] == ["70.0", "0.0"]
|
||||
assert calls == ["precompute"]
|
||||
# 체인 전(예상노선 없음)은 건너뜀
|
||||
expected_route_csv_path(tmp_path).unlink()
|
||||
asyncio.run(router.refresh_expected_route(uuid4()))
|
||||
assert calls == ["precompute"] and not expected_route_csv_path(tmp_path).exists()
|
||||
Reference in New Issue
Block a user