- 새 `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>
82 lines
3.6 KiB
Python
82 lines
3.6 KiB
Python
"""예정노선 다시 추적(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}
|