feat(B05): 예정노선 다시 추적 POST /route/follow · 고친 예정노선 expected_route_edit.csv (PLAN 626-2 · 요구 530 D3 · D4 · D8)
- 새 `B05_Profile_Router_RouteFollow.py` — 고친 예정노선(TM · 경위도) → 평면선(노드 · 곡선 · 편집표 · GET /route/plan 칸 이름) · 저장 0 - 같은 함수 — `Route_Polyline_Write.followed_table` 을 체인 초기 폴리라인(`_ensure_planned_initial`)과 함께 씀 · 셈 몫 `polyline_table` 을 쓰기에서 나눔 - [저장] — 노선 초안 `expected`(없음 그대로 · [] 지움 · 점 묶음 씀) → `B05_Profile/route/expected_route_edit.csv` · GET /route/plan 덧칸 `expected_edited` - [초기화] — 고친 예정노선 지움 · `expected_route.csv` · 초기값 · 추천 저장본 · `/route/replan` · `as_initial` 그대로 - 시험 — 새 `test_626_2_route_follow`(follow = 체인 초기선 · 저장 0 · 경위도 · edit csv 씀 · 지움 · initial_snapshot 지문 · 초기화 지움) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -146,6 +146,33 @@ def _write_planned_polyline(
|
||||
섞여 있어 다시 단순화하면 꺾임점이 조금씩 지워지고, 그것을 반복하면 [확인]을 누를
|
||||
때마다 선이 깎인다(2026-09-07 실측: 정점 136 → 134 → 119). 낳은 값을 그대로 보관한다.
|
||||
"""
|
||||
result, table = polyline_table(
|
||||
points,
|
||||
radius_m,
|
||||
simplify=simplify,
|
||||
curve_flags=curve_flags,
|
||||
radii=radii,
|
||||
station_snap=station_snap,
|
||||
station_snap_step=station_snap_step,
|
||||
locks=locks,
|
||||
arcs=arcs,
|
||||
)
|
||||
return _write_table(path, result, table)
|
||||
|
||||
|
||||
def polyline_table(
|
||||
points: list[tuple[float, float]],
|
||||
radius_m: float,
|
||||
*,
|
||||
simplify: bool = True,
|
||||
curve_flags: list[bool] | None = None,
|
||||
radii: list[float | None] | None = None,
|
||||
station_snap: bool | None = None,
|
||||
station_snap_step: float | None = None,
|
||||
locks: list[str | None] | None = None,
|
||||
arcs: list[float | None] | None = None,
|
||||
) -> tuple[PlannedPolyline, dict[str, Any]]:
|
||||
"""`_write_planned_polyline` 의 셈 몫 — (폴리라인, 곡선표) · 파일 안 씀(626-2)."""
|
||||
edited = curve_flags is not None or radii is not None
|
||||
made = (
|
||||
node_polyline(points, radius_m, curve_flags, radii, locks, arcs)
|
||||
@@ -163,8 +190,6 @@ def _write_planned_polyline(
|
||||
radii=radii,
|
||||
)
|
||||
edits = node_edits(points, curve_flags, radii, locks, arcs) if edited else None
|
||||
write_route_csv(path, [{"x": x, "y": y} for x, y in result.vertices])
|
||||
write_route_csv(_nodes_path(path), [{"x": node.x, "y": node.y} for node in result.nodes])
|
||||
table: dict[str, Any] = {
|
||||
"min_radius_m": round(radius_m, 3),
|
||||
"curves": [c.as_dict() for c in result.curves],
|
||||
@@ -175,6 +200,27 @@ def _write_planned_polyline(
|
||||
table["station_snap_step"] = float(station_snap_step)
|
||||
if edits is not None:
|
||||
table["edits"] = edits
|
||||
return result, table
|
||||
|
||||
|
||||
def followed_table(
|
||||
points: list[tuple[float, float]], radius_m: float
|
||||
) -> tuple[PlannedPolyline, dict[str, Any]]:
|
||||
"""예정노선 추적(383-1) — 점 묶음 → 곡선 MC 를 일반측점에 맞춘 폴리라인 + 곡선표(스냅 켬).
|
||||
|
||||
체인 초기 폴리라인(`_ensure_planned_initial`) · `/route/follow`(626-2)가 같이 씀 · 못 맞추면
|
||||
지금 선.
|
||||
"""
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Station import snapped_initial
|
||||
|
||||
fit = snapped_initial(points, radius_m)
|
||||
shape = {"simplify": False, "curve_flags": fit[1], "radii": fit[2]} if fit else {}
|
||||
return polyline_table(fit[0] if fit else points, radius_m, station_snap=True, **shape)
|
||||
|
||||
|
||||
def _write_table(path: Path, result: PlannedPolyline, table: dict[str, Any]) -> dict:
|
||||
write_route_csv(path, [{"x": x, "y": y} for x, y in result.vertices])
|
||||
write_route_csv(_nodes_path(path), [{"x": node.x, "y": node.y} for node in result.nodes])
|
||||
_curves_path(path).write_text(json.dumps(table, ensure_ascii=False), encoding="utf-8")
|
||||
return {
|
||||
"nodes": len(result.nodes),
|
||||
|
||||
@@ -75,6 +75,8 @@ class RouteDraftInput(BaseModel):
|
||||
stationSnap: bool = False # noqa: N815
|
||||
stationSnapStep: float | None = None # noqa: N815 — 390-5
|
||||
terrain: Literal["las", "sheet"] | None = None
|
||||
#: 고친 예정노선(626-2) — 없음 = 그대로 · [] = 지움 · 점 묶음 = `expected_route_edit.csv`
|
||||
expected: list[tuple[float, float]] | None = None
|
||||
|
||||
|
||||
class RouteSaveRequest(RouteReplanRequest):
|
||||
|
||||
@@ -272,6 +272,7 @@ async def _reset_route_design(project_id: UUID, session: dict[str, Any] | None)
|
||||
restore_snapshot_files,
|
||||
wipe_edited_masters,
|
||||
)
|
||||
from common_util.common_util_route_geometry import expected_route_edit_path
|
||||
from common_util.common_util_surface_confirmation import get_surface_confirmation_params
|
||||
|
||||
pool = get_db_pool()
|
||||
@@ -352,6 +353,8 @@ async def _reset_route_design(project_id: UUID, session: dict[str, Any] | None)
|
||||
"B05 초기화: 편집 정본 제거 %s (project_id=%s)", removed, project_id
|
||||
)
|
||||
await run_auto_design_chain(project_id, surface_model_id=surface_model_id)
|
||||
if project_root: # 고친 예정노선(626-2)은 초기값이 아님 — 스냅샷에 섞여 있어도 지움
|
||||
expected_route_edit_path(project_root).unlink(missing_ok=True)
|
||||
|
||||
async with pool.acquire() as connection:
|
||||
latest = await get_latest_route(connection, project_id)
|
||||
|
||||
@@ -20,6 +20,7 @@
|
||||
GET /api/projects/{id}/route/plan → 예상노선·계획노선 정점(사업지 좌표계)
|
||||
POST /api/projects/{id}/route/replan → 노선 초안 저장(노선 저장 체인)
|
||||
POST /api/projects/{id}/route/initial → 초기 계획노선 갈래 하나를 만들어 돌려줌(추천 저장본만 씀)
|
||||
POST /api/projects/{id}/route/follow → 고친 예정노선 다시 추적(저장 0 · 626-2 · 나눈 파일)
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
@@ -35,13 +36,14 @@ from fastapi.responses import JSONResponse
|
||||
|
||||
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial import read_initial_choice, write_initial_choice
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Station import snapped_initial
|
||||
from B05_Profile.B05_Profile_Engine_Sections import load_station_snap, load_station_snap_step
|
||||
from B05_Profile.B05_Profile_Route_Polyline_Write import ( # noqa: F401 — 부르는 곳이 `replan.` 으로 씀
|
||||
_curves_path,
|
||||
_nodes_path,
|
||||
_read_curves,
|
||||
_write_planned_polyline,
|
||||
_write_table,
|
||||
followed_table,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Route_Save import RouteSaveRequest
|
||||
from B05_Profile.B05_Profile_Schema import ( # noqa: F401 — 부르는 곳이 `replan.` 으로 씀
|
||||
@@ -54,6 +56,7 @@ from B05_Profile.B05_Profile_Schema import ( # noqa: F401 — 부르는 곳이
|
||||
from common_util.common_util_initial_snapshot import design_route_csv_path
|
||||
from common_util.common_util_route_geometry import (
|
||||
expected_route_csv_path,
|
||||
expected_route_edit_path,
|
||||
planned_route_initial_path,
|
||||
planned_route_working_path,
|
||||
read_planned_route_csv,
|
||||
@@ -195,6 +198,15 @@ async def _road_settings(project_id: UUID) -> tuple[str, int | None, str]:
|
||||
return grade_class, design_speed, terrain
|
||||
|
||||
|
||||
def write_expected_edit(project_root: Path, points: list[tuple[float, float]]) -> None:
|
||||
"""고친 예정노선 → `expected_route_edit.csv`(626-2 · 점 2 미만 = 지움)."""
|
||||
target = expected_route_edit_path(project_root)
|
||||
if len(points) < 2:
|
||||
target.unlink(missing_ok=True)
|
||||
return
|
||||
write_route_csv(target, [{"x": x, "y": y} for x, y in points])
|
||||
|
||||
|
||||
def _ensure_planned_initial(project_root: Path, radius_m: float) -> dict | None:
|
||||
"""계획노선 **초기 폴리라인**이 없거나 낡았으면 예상노선을 폴리라인화해 세운다.
|
||||
|
||||
@@ -211,11 +223,7 @@ def _ensure_planned_initial(project_root: Path, radius_m: float) -> dict | None:
|
||||
if len(points) < 2:
|
||||
return None
|
||||
# 383-1 — 곡선 MC = 일반측점(기본 제약) + [측점 스냅] 켬 · 못 맞추면 지금 선 + 켬
|
||||
fit = snapped_initial(points, radius_m)
|
||||
shape = {"simplify": False, "curve_flags": fit[1], "radii": fit[2]} if fit else {}
|
||||
summary = _write_planned_polyline(
|
||||
target, fit[0] if fit else points, radius_m, station_snap=True, **shape
|
||||
)
|
||||
summary = _write_table(target, *followed_table(points, radius_m))
|
||||
logger.info(
|
||||
"계획노선 초기 폴리라인 생성: %s (노드 %d · 곡선 %d · 위반 %d · 정점 %d)",
|
||||
target,
|
||||
@@ -426,6 +434,10 @@ async def read_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse:
|
||||
"status": "success",
|
||||
"project_id": str(project_id),
|
||||
"expected": expected,
|
||||
# 고친 예정노선(626-2 · [저장] 때) — 없으면 빈 목록 · 원본 `expected` 는 그대로
|
||||
"expected_edited": await asyncio.to_thread(
|
||||
_vertices_of, expected_route_edit_path(project_root)
|
||||
),
|
||||
"planned": planned,
|
||||
"nodes": [node.as_dict() for node in outline.nodes],
|
||||
# 곡선 성분 — 화면이 **곡선 시작·끝점**을 손잡이로 그리고 반지름 칸을 띄우는 재료다
|
||||
@@ -580,6 +592,10 @@ async def _save_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, Any
|
||||
return JSONResponse(
|
||||
status_code=409, content={"status": "error", "message": result["error"]}
|
||||
)
|
||||
# 고친 예정노선(626-2) — 초안에 오면만 · [] = 지움 · `expected_route.csv` 는 안 씀
|
||||
expected = getattr(draft, "expected", None)
|
||||
if expected is not None:
|
||||
await asyncio.to_thread(write_expected_edit, project_root, expected)
|
||||
choice = request.initial
|
||||
current = (
|
||||
await asyncio.to_thread(
|
||||
@@ -611,5 +627,8 @@ async def _save_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, Any
|
||||
}
|
||||
|
||||
|
||||
# `POST /route/initial` 은 나눈 파일에서 같은 `router` 에 붙는다 — 끝에서 불러 등록.
|
||||
from B05_Profile import B05_Profile_Router_RouteInitial # noqa: E402, F401
|
||||
# `POST /route/initial` · `/route/follow` 는 나눈 파일에서 같은 `router` 에 붙음 — 끝에서 등록.
|
||||
from B05_Profile import ( # noqa: E402
|
||||
B05_Profile_Router_RouteFollow, # noqa: F401
|
||||
B05_Profile_Router_RouteInitial, # noqa: F401
|
||||
)
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
"""예정노선 다시 추적(626-2 · 요구 530) — 고친 예정노선 → 평면선(노드 · 곡선 · 편집표) · 저장 0.
|
||||
|
||||
계산 자리 ③ 서버 단독 — 체인 초기 폴리라인(`_ensure_planned_initial`)과 같은 함수
|
||||
`followed_table`(지형 안 씀 · 순수 기하). 화면은 받은 노드를 편집기 한 걸음으로 싣고 페이지 [저장]
|
||||
길로 보냄(고친 예정노선 = 초안 `expected` → `expected_route_edit.csv`). `/route/replan` ·
|
||||
`as_initial` · `expected_route.csv` · 초기값 · 추천 저장본은 안 건드림.
|
||||
|
||||
POST /api/projects/{id}/route/follow
|
||||
본문 {points: [[x, y], …], crs: "tm" | "lonlat"} ("lonlat" = [경도, 위도])
|
||||
→ {status, planned, nodes, curves, node_edits, station_snap, min_radius_m,
|
||||
curve_count, violation_count} (GET /route/plan 칸 이름 · 사업지 TM)
|
||||
400 점 2 미만 · 404 프로젝트 없음 · 409 좌표계 모름(lonlat)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Any, Literal
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi.responses import JSONResponse
|
||||
from pydantic import BaseModel
|
||||
|
||||
from B05_Profile.B05_Profile_Repository import get_surface_crs_epsg
|
||||
from B05_Profile.B05_Profile_Route_Polyline_Write import followed_table
|
||||
from B05_Profile.B05_Profile_Router_Replan import (
|
||||
_PROJECT_PATH_MISSING,
|
||||
_plan_criteria,
|
||||
_project_paths,
|
||||
router,
|
||||
)
|
||||
from common_util.common_util_crs import lonlat_to_tm
|
||||
from config.config_db import get_db_pool
|
||||
|
||||
|
||||
class RouteFollowRequest(BaseModel):
|
||||
points: list[tuple[float, float]]
|
||||
crs: Literal["tm", "lonlat"] = "tm"
|
||||
|
||||
|
||||
def follow_route_line(points: list[tuple[float, float]], radius_m: float) -> dict[str, Any]:
|
||||
"""점 묶음(TM) → 예정노선 추적 평면선 — 파일 꼴(csv 4자리)과 같은 값."""
|
||||
result, table = followed_table(points, radius_m)
|
||||
return {
|
||||
"planned": [[round(x, 4), round(y, 4)] for x, y in result.vertices],
|
||||
"nodes": [node.as_dict() for node in result.nodes],
|
||||
"curves": table["curves"],
|
||||
"node_edits": table.get("edits"),
|
||||
"station_snap": table["station_snap"],
|
||||
"min_radius_m": round(radius_m, 2),
|
||||
"curve_count": result.curve_count,
|
||||
"violation_count": result.violation_count,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/{project_id}/route/follow", response_model=None)
|
||||
async def follow_route(project_id: UUID, body: RouteFollowRequest) -> dict[str, Any] | JSONResponse:
|
||||
"""고친 예정노선을 다시 추적한 평면선 — 아무것도 안 씀(626-2)."""
|
||||
if len(body.points) < 2:
|
||||
return JSONResponse(
|
||||
status_code=400,
|
||||
content={"status": "error", "message": "예정노선은 점이 2개 이상이어야 합니다."},
|
||||
)
|
||||
if await _project_paths(project_id) is None:
|
||||
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
|
||||
points = [(float(x), float(y)) for x, y in body.points]
|
||||
if body.crs == "lonlat":
|
||||
async with get_db_pool().acquire() as connection:
|
||||
epsg = await get_surface_crs_epsg(connection, project_id, 0)
|
||||
if epsg is None:
|
||||
return JSONResponse(
|
||||
status_code=409,
|
||||
content={
|
||||
"status": "error",
|
||||
"message": "사업지 좌표계를 몰라 경위도를 바꿀 수 없습니다.",
|
||||
},
|
||||
)
|
||||
points = [lonlat_to_tm(lon, lat, epsg) for lon, lat in points]
|
||||
radius_m = (await _plan_criteria(project_id))[0]
|
||||
line = await asyncio.to_thread(follow_route_line, points, radius_m)
|
||||
return {"status": "success", "project_id": str(project_id), **line}
|
||||
@@ -224,6 +224,11 @@ def expected_route_csv_path(project_root: Path) -> Path:
|
||||
return Path(project_root) / "B05_Profile" / "route" / "expected_route.csv"
|
||||
|
||||
|
||||
def expected_route_edit_path(project_root: Path) -> Path:
|
||||
"""고친 예정노선 CSV 자리(626-2) — 페이지 [저장] 때만 씀 · [초기화]면 지움 · 원본 그대로."""
|
||||
return Path(project_root) / "B05_Profile" / "route" / "expected_route_edit.csv"
|
||||
|
||||
|
||||
def write_route_csv(path: Path, points: list[dict[str, float]]) -> None:
|
||||
"""노선 정점을 CSV로 적는다. 열 이름은 `read_planned_route_csv()`가 아는 것."""
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
@@ -0,0 +1,221 @@
|
||||
"""626-2 예정노선 다시 추적 `POST /route/follow` · 고친 예정노선 `expected_route_edit.csv`.
|
||||
|
||||
요구 530 D3 · D4 · D8. 못박는 것
|
||||
- follow = 체인 초기 폴리라인(`_ensure_planned_initial` 파일)과 같은 입력이면 같은 선 · 노드 ·
|
||||
곡선 · 편집표.
|
||||
- follow 는 아무것도 안 씀(프로젝트 폴더 그대로) · 경위도 입력 = TM 입력과 같은 선.
|
||||
- [저장](노선 초안 `expected`) = `expected_route_edit.csv` 씀 · `expected_route.csv` ·
|
||||
초기 폴리라인 · `initial_snapshot/` 그대로 · 없음 = 그대로 · [] = 지움 ·
|
||||
GET `/route/plan` `expected_edited`.
|
||||
- [초기화] = 고친 예정노선 지움(스냅샷에 섞여 있어도).
|
||||
- follow 길은 `/route/replan` · `as_initial` 을 안 부름.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import csv
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
|
||||
from B05_Profile import B05_Profile_Router_Replan as replan
|
||||
from B05_Profile import B05_Profile_Router_RouteFollow as follow
|
||||
from common_util.common_util_crs import tm_to_lonlat
|
||||
from common_util.common_util_initial_snapshot import snapshot_dir
|
||||
from common_util.common_util_route_geometry import (
|
||||
expected_route_csv_path,
|
||||
expected_route_edit_path,
|
||||
planned_route_initial_path,
|
||||
write_route_csv,
|
||||
)
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
RADIUS_M = 30.0
|
||||
EPSG = 5186
|
||||
# 예정노선 점 묶음 꼴(3 m 간격 · 굽은 산길) — 사업지 TM
|
||||
LINE = [(200000.0 + 3 * i, 450000.0 + 40 * math.sin(i / 20)) for i in range(160)]
|
||||
|
||||
|
||||
def _patch(monkeypatch, root: Path) -> None:
|
||||
async def paths(_pid):
|
||||
return root, "stored"
|
||||
|
||||
async def criteria(_pid):
|
||||
return RADIUS_M, 0.0, 0.0
|
||||
|
||||
monkeypatch.setattr(replan, "_project_paths", paths)
|
||||
monkeypatch.setattr(replan, "_plan_criteria", criteria)
|
||||
monkeypatch.setattr(follow, "_project_paths", paths)
|
||||
monkeypatch.setattr(follow, "_plan_criteria", criteria)
|
||||
|
||||
|
||||
def _rows(path: Path) -> list[list[float]]:
|
||||
with path.open(encoding="utf-8") as file:
|
||||
return [[float(r["x"]), float(r["y"])] for r in csv.DictReader(file)]
|
||||
|
||||
|
||||
def _tree(root: Path) -> dict[str, str]:
|
||||
return {
|
||||
str(p.relative_to(root)): hashlib.sha1(p.read_bytes()).hexdigest()
|
||||
for p in sorted(root.rglob("*"))
|
||||
if p.is_file()
|
||||
}
|
||||
|
||||
|
||||
def test_follow_은_체인_초기선과_같은_선(tmp_path: Path) -> None:
|
||||
write_route_csv(expected_route_csv_path(tmp_path), [{"x": x, "y": y} for x, y in LINE])
|
||||
assert replan._ensure_planned_initial(tmp_path, RADIUS_M)
|
||||
target = planned_route_initial_path(tmp_path)
|
||||
table = json.loads(replan._curves_path(target).read_text(encoding="utf-8"))
|
||||
# 같은 입력 = 체인이 읽는 예정노선 파일 점(csv 4자리)
|
||||
same = [tuple(p) for p in replan._vertices_of(expected_route_csv_path(tmp_path))]
|
||||
line = follow.follow_route_line(same, RADIUS_M)
|
||||
assert line["planned"] == _rows(target)
|
||||
assert [[round(n["x"], 4), round(n["y"], 4)] for n in line["nodes"]] == _rows(
|
||||
replan._nodes_path(target)
|
||||
)
|
||||
assert line["curves"] == table["curves"] and line["curves"]
|
||||
assert line["node_edits"] == table.get("edits")
|
||||
assert line["station_snap"] is True and table["station_snap"] is True
|
||||
|
||||
|
||||
def test_follow_은_안_씀_경위도도_같은_선(tmp_path: Path, monkeypatch) -> None:
|
||||
_patch(monkeypatch, tmp_path)
|
||||
(tmp_path / "keep.txt").write_text("x", encoding="utf-8")
|
||||
before = _tree(tmp_path)
|
||||
body = follow.RouteFollowRequest(points=LINE)
|
||||
tm = asyncio.run(follow.follow_route(uuid4(), body))
|
||||
assert tm["status"] == "success" and len(tm["nodes"]) >= 3
|
||||
|
||||
class _Pool:
|
||||
@asynccontextmanager
|
||||
async def acquire(self):
|
||||
yield None
|
||||
|
||||
async def epsg(*_a):
|
||||
return EPSG
|
||||
|
||||
monkeypatch.setattr(follow, "get_db_pool", lambda: _Pool())
|
||||
monkeypatch.setattr(follow, "get_surface_crs_epsg", epsg)
|
||||
lonlat = [tm_to_lonlat(x, y, EPSG) for x, y in LINE]
|
||||
geo = asyncio.run(
|
||||
follow.follow_route(uuid4(), follow.RouteFollowRequest(points=lonlat, crs="lonlat"))
|
||||
)
|
||||
assert len(geo["nodes"]) == len(tm["nodes"])
|
||||
assert all(math.dist(a, b) < 0.01 for a, b in zip(geo["planned"], tm["planned"], strict=True))
|
||||
assert _tree(tmp_path) == before # 저장 0
|
||||
short = asyncio.run(follow.follow_route(uuid4(), follow.RouteFollowRequest(points=LINE[:1])))
|
||||
assert short.status_code == 400
|
||||
|
||||
|
||||
def _save(root: Path, expected) -> None:
|
||||
nodes = [LINE[0], LINE[80], LINE[-1]]
|
||||
draft = {"nodes": nodes, "curveOn": [True] * 3, "stationSnap": True, "expected": expected}
|
||||
body = replan.RouteSaveRequest(vertices=[], draft=draft)
|
||||
out = asyncio.run(replan.replan_route(uuid4(), body))
|
||||
assert out["status"] == "success"
|
||||
|
||||
|
||||
def test_저장은_고친_예정노선만_씀(tmp_path: Path, monkeypatch) -> None:
|
||||
_patch(monkeypatch, tmp_path)
|
||||
|
||||
async def recompute(*_a, **_k):
|
||||
return {"route_id": 1}
|
||||
|
||||
monkeypatch.setattr(replan, "_recompute", recompute)
|
||||
write_route_csv(expected_route_csv_path(tmp_path), [{"x": x, "y": y} for x, y in LINE])
|
||||
replan._ensure_planned_initial(tmp_path, RADIUS_M)
|
||||
(snapshot_dir(tmp_path) / "B05_Profile__route").mkdir(parents=True)
|
||||
(snapshot_dir(tmp_path) / "B05_Profile__route" / "planned_route.csv").write_text("x")
|
||||
keep = [expected_route_csv_path(tmp_path), planned_route_initial_path(tmp_path)]
|
||||
fixed = {p: p.read_bytes() for p in keep}
|
||||
shot = _tree(snapshot_dir(tmp_path))
|
||||
|
||||
edited = [list(p) for p in LINE[::10]]
|
||||
_save(tmp_path, edited)
|
||||
assert _rows(expected_route_edit_path(tmp_path)) == [
|
||||
[round(x, 4), round(y, 4)] for x, y in edited
|
||||
]
|
||||
assert asyncio.run(replan.read_route_plan(uuid4()))["expected_edited"] == _rows(
|
||||
expected_route_edit_path(tmp_path)
|
||||
)
|
||||
_save(tmp_path, None) # 없음 = 그대로
|
||||
assert expected_route_edit_path(tmp_path).is_file()
|
||||
_save(tmp_path, []) # 지움
|
||||
assert not expected_route_edit_path(tmp_path).exists()
|
||||
assert asyncio.run(replan.read_route_plan(uuid4()))["expected_edited"] == []
|
||||
assert {p: p.read_bytes() for p in keep} == fixed
|
||||
assert _tree(snapshot_dir(tmp_path)) == shot # 초기값 그대로
|
||||
|
||||
|
||||
def test_초기화는_고친_예정노선_지움(tmp_path: Path, monkeypatch) -> None:
|
||||
from B05_Profile import B05_Profile_Router_Corridor as corridor
|
||||
from B05_Profile import B05_Profile_Router_Lifecycle as lifecycle
|
||||
from common_util import common_util_initial_snapshot as snapshot
|
||||
|
||||
class _Cursor:
|
||||
rowcount = 1
|
||||
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *_):
|
||||
return False
|
||||
|
||||
async def execute(self, *_a):
|
||||
pass
|
||||
|
||||
class _Connection:
|
||||
def cursor(self, *_a, **_k):
|
||||
return _Cursor()
|
||||
|
||||
async def begin(self):
|
||||
pass
|
||||
|
||||
async def commit(self):
|
||||
pass
|
||||
|
||||
async def rollback(self):
|
||||
pass
|
||||
|
||||
class _Pool:
|
||||
@asynccontextmanager
|
||||
async def acquire(self):
|
||||
yield _Connection()
|
||||
|
||||
def _async(value):
|
||||
async def run(*_a, **_k):
|
||||
return value
|
||||
|
||||
return run
|
||||
|
||||
# 추천 [적용]이 고친 예정노선째 스냅샷을 뜬 경우도 — 복원 뒤 지움
|
||||
stored = snapshot_dir(tmp_path) / "B05_Profile__route"
|
||||
stored.mkdir(parents=True)
|
||||
(stored / "expected_route_edit.csv").write_text("sequence,x,y\n", encoding="utf-8")
|
||||
(snapshot_dir(tmp_path) / "db.json").write_text(json.dumps({"routes": [{"id": 1}]}))
|
||||
write_route_csv(expected_route_edit_path(tmp_path), [{"x": 0, "y": 0}, {"x": 1, "y": 1}])
|
||||
monkeypatch.setattr(lifecycle, "get_db_pool", lambda: _Pool())
|
||||
monkeypatch.setattr(lifecycle, "get_project_storage_relative_path", _async("p"))
|
||||
monkeypatch.setattr(lifecycle, "resolve_stored_project_path", lambda _p: str(tmp_path))
|
||||
monkeypatch.setattr(lifecycle, "get_latest_route", _async({"id": 9}))
|
||||
monkeypatch.setattr(lifecycle, "start_stage", _async(None))
|
||||
monkeypatch.setattr(snapshot, "restore_initial_snapshot", _async(9))
|
||||
monkeypatch.setattr(corridor, "prune_corridor_files", lambda *_a: 0)
|
||||
response = asyncio.run(lifecycle.reset_route_design(uuid4(), None))
|
||||
assert json.loads(response.body)["restored"] is True
|
||||
assert not expected_route_edit_path(tmp_path).exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["B05_Profile_Router_RouteFollow.py"])
|
||||
def test_follow_길은_replan_as_initial_안_부름(name: str) -> None:
|
||||
text = (ROOT / "B05_Profile" / name).read_text(encoding="utf-8")
|
||||
code = text.split('"""', 2)[2] # 머리글 뺀 몸통
|
||||
assert "replan_route" not in code and "as_initial" not in code
|
||||
assert "expected_route_csv_path" not in code and "write_" not in code
|
||||
Reference in New Issue
Block a user