eomsangdonandClaude Opus 5
eec29a90df
보조 창이 노선 편집 [확인] 실측에서 둘을 잡음.
1. projects 에 design_speed_kph / terrain_type 컬럼이 없음(있는 것은 road_type 뿐).
매 요청마다 OperationalError(1054) 가 ERROR 로그로 남고 기본값 폴백으로 넘어갔음.
- 임도 종류는 projects.road_type, 설계속도·지형은 워크플로 stage 2 params 에서 읽음.
- 산식은 이미 있던 B05_Profile_Engine_Grade.legal_plan_radius_min_m 를 씀 —
내가 같은 표를 다시 짜 두었던 것을 지움(중복 제거).
2. 재확정 체인이 solve_route 400 으로 끊길 때 상태 코드만 남겨 원인을 못 짚었음.
본문(사유)까지 로그에 남김. 노선 편집 [확인]이 조용히 끊겨 배수유역·관은 새 노선으로
가고 종횡단만 옛 노선에 남는 어긋남이 굳었던 자리임.
recompute 의 상세 조회와 DB 두 건도 asyncio.gather 로 묶음(원격 DB 왕복 약 12ms/질의).
시험 409 통과·17 건너뜀.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
410 lines
20 KiB
Python
410 lines
20 KiB
Python
"""계획노선 편집 — 노선 두 벌(예상노선·계획노선)과 그 뒤 재계산의 서버 몫.
|
|
|
|
노선은 **세 벌**이다(2026-09-06 사용자 확정 → 같은 날 정정, PLAN 0-7).
|
|
|
|
- **예상노선**(원본) — 파일 업로드 자동 체인이 낸 것. `B05_Profile/route/expected_route.csv`.
|
|
⚠ 이것은 폴리라인이 아니라 **점 묶음**이고 규칙 없는 폴리라인과도 맞지 않는다
|
|
(2026-09-06 사용자 확인). 어떤 경로로도 고치지 않는다.
|
|
- **계획노선 초기본** — 위를 복사해 **폴리라인으로 바꾼 것**. `planned_route_initial.csv`.
|
|
**불변의 초기 데이터**이며, 유토곡선·3D 에 투영되는 선도 이것이다. 곡선은 지식DB
|
|
기준(별표2 Ⅰ.2.다 — 설계속도·지형별 최소곡선반지름, 내각 155° 이상은 생략)으로 끼운다.
|
|
- **계획노선**(수정본) — 초기본에서 시작해 사용자가 **노드를 잡아** 고친 것.
|
|
`planned_route.csv`. 설계 계통은 이것이 있으면 이것을 읽는다(`load_design_route`).
|
|
|
|
노선 초기화는 수정본 파일을 지우는 것이다 — 그러면 초기 폴리라인을 읽으므로 「초기본을
|
|
수정본으로 복사」와 결과가 같다.
|
|
|
|
[확인]을 눌렀을 때만 계산이 돈다. 재계산은 **초기 업로드 체인의 로직을 그대로 재사용**한다
|
|
(2026-09-06 사용자 제안) — `run_redesign_chain`이 배수유역 다시 분석 → 기본 관 저장 → 관
|
|
정착 계획선 재산출 → 종횡단 재생성 → 옛 측점 설계를 누가거리로 이월 → B05·B06 확정까지 한
|
|
줄로 돈다. 여기서 하는 일은 그 앞에 **노선을 갈아 끼우는 것**뿐이다.
|
|
|
|
시설 처리(2026-09-06 사용자 확정) — 횡단배수 지점은 노선 자리에 따라 생기고 없어지므로
|
|
저장분을 버리고 **새로 계산한 값**을 쓴다. 그 밖의 구조물은 **옛 측점값을 그대로** 이어받는다
|
|
(`structures.json`은 프로젝트 정본이라 손대지 않으면 그대로 남는다).
|
|
|
|
GET /api/projects/{id}/route/plan → 예상노선·계획노선 정점(사업지 좌표계)
|
|
POST /api/projects/{id}/route/replan → 고친 계획노선으로 갈아 끼우고 재계산
|
|
POST /api/projects/{id}/route/replan/reset → 계획노선을 예상노선으로 되돌리고 재계산
|
|
"""
|
|
|
|
import asyncio
|
|
import csv
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
|
from common_util.common_util_initial_snapshot import design_route_csv_path
|
|
from common_util.common_util_route_geometry import (
|
|
densify_route,
|
|
expected_route_csv_path,
|
|
planned_route_initial_path,
|
|
planned_route_working_path,
|
|
read_planned_route_csv,
|
|
write_route_csv,
|
|
)
|
|
from common_util.common_util_route_polyline import build_planned_polyline
|
|
from common_util.common_util_storage import resolve_stored_project_path
|
|
from config.config_db import get_db_pool
|
|
from config.config_system import (
|
|
ROUTE_DIRECT_LINK_CELL_FACTOR,
|
|
ROUTE_GRID_RES_M,
|
|
ROUTE_PLANNED_DENSIFY_SAFETY,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/projects", tags=["B05 Route Design"])
|
|
|
|
_PROJECT_PATH_MISSING = {"status": "error", "message": "프로젝트 저장 경로를 찾을 수 없습니다."}
|
|
|
|
|
|
class RouteVertexInput(BaseModel):
|
|
"""계획노선 정점 하나 — 사업지 좌표계(m)."""
|
|
|
|
x: float
|
|
y: float
|
|
|
|
|
|
class RouteReplanRequest(BaseModel):
|
|
"""고친 계획노선. 정점은 시점 → 종점 순서(사업지 좌표계 m)."""
|
|
|
|
vertices: list[RouteVertexInput] = Field(default_factory=list)
|
|
|
|
|
|
async def _project_paths(project_id: UUID) -> tuple[Path, str] | None:
|
|
"""(프로젝트 실경로, 저장소 상대경로). 없으면 None."""
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
stored = await get_project_storage_relative_path(connection, project_id)
|
|
if not stored:
|
|
return None
|
|
return Path(resolve_stored_project_path(stored)), str(stored)
|
|
|
|
|
|
def _vertices_of(path: Path) -> list[list[float]]:
|
|
"""CSV 정점을 [[x, y], …]로. 없거나 못 읽으면 빈 목록."""
|
|
if not path.is_file():
|
|
return []
|
|
route = read_planned_route_csv(path)
|
|
if route is None:
|
|
return []
|
|
return [[float(vertex.x), float(vertex.y)] for vertex in route.vertices]
|
|
|
|
|
|
def _ensure_expected_route(project_root: Path) -> str:
|
|
"""예상노선(원본) 정본이 없으면 만들어 둔다. 어디서 씨앗을 얻었는지 돌려준다.
|
|
|
|
새 프로젝트는 자동 체인이 만들어 두지만(2026-09-06), 그 전에 만들어진 프로젝트는
|
|
초기값 스냅샷 안에만 있거나 그마저 재확정 체인이 지운 뒤일 수 있다. 노선 초기화가
|
|
성립하려면 이 파일이 반드시 있어야 하므로 여기서 한 번 세워 둔다.
|
|
"""
|
|
target = expected_route_csv_path(project_root)
|
|
if target.is_file():
|
|
return "already"
|
|
for source, label in (
|
|
(design_route_csv_path(project_root), "snapshot"),
|
|
(planned_route_working_path(project_root), "working"),
|
|
):
|
|
vertices = _vertices_of(source)
|
|
if len(vertices) >= 2:
|
|
write_route_csv(target, [{"x": x, "y": y} for x, y in vertices])
|
|
logger.info("예상노선 정본을 %s 에서 세웠습니다: %s", label, target)
|
|
return label
|
|
return "none"
|
|
|
|
|
|
async def _min_plan_radius_m(project_id: UUID) -> float:
|
|
"""이 프로젝트에 적용할 법정 최소곡선반지름(m) — 임도 종류·설계속도·지형으로 고른다.
|
|
|
|
값의 출처는 지식DB(`01_임도/02_상세설계/평면선형.md`, 별표2 Ⅰ.2.다)이고 산식은 이미
|
|
`B05_Profile_Engine_Grade.legal_plan_radius_min_m` 에 있다 — 여기서 다시 짜지 않는다.
|
|
|
|
읽는 자리 — 임도 종류는 `projects.road_type`, 설계속도·지형은 **워크플로 stage 2 params**
|
|
(노선 풀기 요청이 남긴 값)다. `projects` 에는 설계속도·지형 칸이 없다(2026-09-06 확인:
|
|
있는 것은 `road_type` 뿐). 못 읽으면 가장 완화된 조건으로 떨어진다 — 막지 않고 위반
|
|
표시만 하기 때문이다.
|
|
"""
|
|
import aiomysql
|
|
|
|
from B05_Profile.B05_Profile_Engine_Grade import legal_plan_radius_min_m, resolve_design_speed
|
|
from common_util.common_util_workflow_state import get_workflow_state
|
|
|
|
grade_class, design_speed, terrain = "work", None, "special"
|
|
try:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
async with connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"SELECT road_type FROM projects WHERE id = %s", (str(project_id),)
|
|
)
|
|
row = await cursor.fetchone()
|
|
if row and row[0]:
|
|
grade_class = str(row[0])
|
|
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
|
state = await get_workflow_state(cursor, str(project_id))
|
|
stage2 = next(
|
|
(s for s in (state or {}).get("stages", []) if int(s.get("stage_no", -1)) == 2), None
|
|
)
|
|
params = (stage2 or {}).get("params") or {}
|
|
speed = params.get("design_speed_kph")
|
|
if isinstance(speed, (int, float)):
|
|
design_speed = int(speed)
|
|
if params.get("terrain_type") in ("normal", "special"):
|
|
terrain = str(params["terrain_type"])
|
|
except Exception: # noqa: BLE001 — 설정을 못 읽어도 폴리라인화는 이어 간다
|
|
logger.exception("최소곡선반지름 설정을 못 읽어 기본값을 씁니다: %s", project_id)
|
|
return legal_plan_radius_min_m(resolve_design_speed(grade_class, design_speed), terrain)
|
|
|
|
|
|
def _write_planned_polyline(path: Path, points: list[tuple[float, float]], radius_m: float) -> dict:
|
|
"""점 묶음을 폴리라인으로 바꿔 CSV 로 쓴다. 노드 요약을 돌려준다."""
|
|
result = build_planned_polyline(points, min_radius_m=radius_m)
|
|
write_route_csv(path, [{"x": x, "y": y} for x, y in result.vertices])
|
|
return {
|
|
"nodes": len(result.nodes),
|
|
"curves": result.curve_count,
|
|
"violations": result.violation_count,
|
|
"vertices": len(result.vertices),
|
|
}
|
|
|
|
|
|
def _ensure_planned_initial(project_root: Path, radius_m: float) -> dict | None:
|
|
"""계획노선 **초기 폴리라인**이 없으면 예상노선을 폴리라인화해 세운다."""
|
|
target = planned_route_initial_path(project_root)
|
|
if target.is_file():
|
|
return None
|
|
points = [(x, y) for x, y in _vertices_of(expected_route_csv_path(project_root))]
|
|
if len(points) < 2:
|
|
return None
|
|
summary = _write_planned_polyline(target, points, radius_m)
|
|
logger.info(
|
|
"계획노선 초기 폴리라인 생성: %s (노드 %d · 곡선 %d · 위반 %d · 정점 %d)",
|
|
target,
|
|
summary["nodes"],
|
|
summary["curves"],
|
|
summary["violations"],
|
|
summary["vertices"],
|
|
)
|
|
return summary
|
|
|
|
|
|
def _write_working_route(path: Path, vertices: list[tuple[float, float]]) -> int:
|
|
"""계획노선(수정본) CSV를 쓴다. 열 이름은 `read_planned_route_csv()`가 아는 것.
|
|
|
|
쓰기 전에 **조밀화**한다 — 정점이 성기면 B05 격자 탐색이 원좌표를 그대로 잇지 못하고
|
|
제 나름의 길을 찾아 사용자가 그린 선과 달라진다(`densify_route` 주석). 평면 형상은
|
|
바뀌지 않고 같은 선 위에 점만 더 찍힌다.
|
|
"""
|
|
dense = densify_route(
|
|
vertices,
|
|
ROUTE_DIRECT_LINK_CELL_FACTOR * ROUTE_GRID_RES_M * ROUTE_PLANNED_DENSIFY_SAFETY,
|
|
)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", encoding="utf-8", newline="") as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow(("sequence", "x", "y"))
|
|
writer.writerows((index, round(x, 4), round(y, 4)) for index, (x, y) in enumerate(dense))
|
|
return len(dense)
|
|
|
|
|
|
async def _recompute(project_id: UUID, project_root: Path, stored_path: str) -> dict[str, Any]:
|
|
"""노선을 갈아 끼운 뒤의 재계산 — 초기 업로드 체인과 같은 로직을 그대로 탄다."""
|
|
import aiomysql
|
|
|
|
from B03_FileInput.B03_FileInput_Service_Chain import (
|
|
_planned_route_points_in_project_crs,
|
|
run_redesign_chain,
|
|
)
|
|
from B04_PreProcess.B04_PreProcess_Service import find_surface_model_for_selection
|
|
from B05_Profile.B05_Profile_Repository import get_latest_route
|
|
from common_util.common_util_drainage_pipes import clear_pipe_points
|
|
from common_util.common_util_surface_confirmation import get_surface_confirmation_params
|
|
from common_util.common_util_workflow_state import get_workflow_state, start_stage
|
|
|
|
# 횡단배수 지점은 새 노선에서 새로 계산한다 — 저장분을 남기면 옛 자리가 살아난다.
|
|
await asyncio.to_thread(clear_pipe_points, stored_path)
|
|
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
selection = await get_surface_confirmation_params(connection, str(project_id))
|
|
latest = await get_latest_route(connection, project_id)
|
|
# 지금 노선이 쓰던 지표면 모델이 1순위다 — 확정 선택값으로 다시 찾으면 프로젝트마다
|
|
# 모델 조합이 달라 못 찾는 경우가 있다(용화: filter=classification/method=dtm 무매칭).
|
|
surface_model_id = (
|
|
int(latest["surface_model_id"]) if latest and latest.get("surface_model_id") else None
|
|
)
|
|
if surface_model_id is None:
|
|
try:
|
|
surface_model_id = await find_surface_model_for_selection(
|
|
connection, project_id, selection
|
|
)
|
|
except Exception:
|
|
surface_model_id = None
|
|
async with connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"SELECT route_start_m, route_end_m FROM projects WHERE id = %s",
|
|
(str(project_id),),
|
|
)
|
|
range_row = await cursor.fetchone()
|
|
if surface_model_id is None:
|
|
return {"error": "확정된 지표면 모델을 찾지 못해 노선을 다시 계산할 수 없습니다."}
|
|
route_range = (range_row[0], range_row[1]) if range_row else None
|
|
|
|
# 갈아 끼운 노선을 stage 2 제어점으로 세운다 — 재확정 체인이 여기서 BP·EP·경유점을
|
|
# 읽어 경로를 다시 푼다. 읽기는 설계 계통과 같은 한 곳(`load_design_route`)을 지난다.
|
|
points = await asyncio.to_thread(
|
|
_planned_route_points_in_project_crs, project_root, selection, route_range
|
|
)
|
|
if not points or len(points) < 2:
|
|
return {"error": "계획노선을 읽지 못했습니다(정점이 2개 미만)."}
|
|
|
|
# 수정본도 예상노선 정본도 없던 프로젝트 — 지금 읽은 것이 곧 예상노선이므로 여기서 세운다.
|
|
expected_path = expected_route_csv_path(project_root)
|
|
if not expected_path.is_file() and not planned_route_working_path(project_root).is_file():
|
|
await asyncio.to_thread(write_route_csv, expected_path, points)
|
|
logger.info("예상노선 정본을 원본 재판독으로 세웠습니다: %s", expected_path)
|
|
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
|
state = await get_workflow_state(cursor, str(project_id))
|
|
stage2 = next(
|
|
(s for s in (state or {}).get("stages", []) if int(s.get("stage_no", -1)) == 2), None
|
|
)
|
|
params = dict((stage2 or {}).get("params") or {})
|
|
params["points"] = {
|
|
"bp": points[0],
|
|
"ep": points[-1],
|
|
"cp": [{**point, "order": index} for index, point in enumerate(points[1:-1], start=1)],
|
|
}
|
|
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await start_stage(cursor, str(project_id), 2, params)
|
|
await connection.commit()
|
|
|
|
await run_redesign_chain(project_id, int(surface_model_id), selection)
|
|
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
made = await get_latest_route(connection, project_id)
|
|
return {
|
|
"route_id": int(made["id"]) if made else None,
|
|
"total_length_m": float(made.get("total_length_m") or 0.0) if made else None,
|
|
"vertex_count": len(points),
|
|
}
|
|
|
|
|
|
@router.get("/{project_id}/route/plan", response_model=None)
|
|
async def read_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse:
|
|
"""예상노선(점 묶음)·계획노선(폴리라인)·편집할 노드를 함께 돌려준다.
|
|
|
|
화면이 그리는 것은 셋이다 — 예상노선은 **점선**, 계획노선은 **실선**, 그리고 사용자가
|
|
잡아 옮기는 **노드**(꺾임점). 노드에는 그 자리에 끼운 반지름·내각·법정 위반 표시가
|
|
붙어 있어 화면이 그대로 보여 줄 수 있다(2026-09-06 사용자 지시).
|
|
"""
|
|
paths = await _project_paths(project_id)
|
|
if paths is None:
|
|
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
|
|
project_root, _ = paths
|
|
# 예상노선 정본이 먼저, 없으면 옛 자리(초기값 스냅샷)를 본다.
|
|
expected = await asyncio.to_thread(_vertices_of, expected_route_csv_path(project_root))
|
|
if not expected:
|
|
expected = await asyncio.to_thread(_vertices_of, design_route_csv_path(project_root))
|
|
|
|
radius_m = await _min_plan_radius_m(project_id)
|
|
await asyncio.to_thread(_ensure_planned_initial, project_root, radius_m)
|
|
working = await asyncio.to_thread(_vertices_of, planned_route_working_path(project_root))
|
|
initial = await asyncio.to_thread(_vertices_of, planned_route_initial_path(project_root))
|
|
planned = working or initial or expected
|
|
|
|
# 노드는 **원본 점**에서 뽑는다 — 폴리라인 정점에는 원호 위 점이 섞여 있어 편집 대상이
|
|
# 아니다. 고친 적이 있으면 그때 보낸 노드가 곧 수정본의 씨앗이므로 같은 규칙으로 다시 냄.
|
|
node_source = expected if not working else working
|
|
outline = await asyncio.to_thread(
|
|
build_planned_polyline, [(x, y) for x, y in node_source], min_radius_m=radius_m
|
|
)
|
|
return {
|
|
"status": "success",
|
|
"project_id": str(project_id),
|
|
"expected": expected,
|
|
"planned": planned,
|
|
"nodes": [node.as_dict() for node in outline.nodes],
|
|
"min_radius_m": round(radius_m, 2),
|
|
"curve_count": outline.curve_count,
|
|
"violation_count": outline.violation_count,
|
|
"edited": bool(working),
|
|
}
|
|
|
|
|
|
@router.post("/{project_id}/route/replan", response_model=None)
|
|
async def replan_route(
|
|
project_id: UUID, request: RouteReplanRequest
|
|
) -> dict[str, Any] | JSONResponse:
|
|
"""고친 계획노선으로 갈아 끼우고 배수유역부터 다시 계산한다(모달 [확인])."""
|
|
if len(request.vertices) < 2:
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"status": "error", "message": "계획노선은 정점이 2개 이상이어야 합니다."},
|
|
)
|
|
paths = await _project_paths(project_id)
|
|
if paths is None:
|
|
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
|
|
project_root, stored_path = paths
|
|
|
|
# 고치기 전에 예상노선(원본)·초기 폴리라인이 서 있는지 본다 — 초기화가 돌아갈 자리다.
|
|
await asyncio.to_thread(_ensure_expected_route, project_root)
|
|
radius_m = await _min_plan_radius_m(project_id)
|
|
await asyncio.to_thread(_ensure_planned_initial, project_root, radius_m)
|
|
# 화면이 보낸 것은 **노드(꺾임점)** 다 — 같은 R 규칙으로 다시 폴리라인을 만든다.
|
|
# 노드만 옮기면 선이 저절로 규칙을 지키는 것이 이 구조의 목적이다(2026-09-06 사용자).
|
|
nodes = [(vertex.x, vertex.y) for vertex in request.vertices]
|
|
summary = await asyncio.to_thread(
|
|
_write_planned_polyline, planned_route_working_path(project_root), nodes, radius_m
|
|
)
|
|
written = summary["vertices"]
|
|
logger.info(
|
|
"계획노선 갈아 끼움: project_id=%s 노드 %d → 정점 %d (곡선 %d · 위반 %d · R %.1fm)",
|
|
project_id,
|
|
len(nodes),
|
|
written,
|
|
summary["curves"],
|
|
summary["violations"],
|
|
radius_m,
|
|
)
|
|
result = await _recompute(project_id, project_root, stored_path)
|
|
if "error" in result:
|
|
return JSONResponse(
|
|
status_code=409, content={"status": "error", "message": result["error"]}
|
|
)
|
|
return {"status": "success", "project_id": str(project_id), **result}
|
|
|
|
|
|
@router.post("/{project_id}/route/replan/reset", response_model=None)
|
|
async def reset_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse:
|
|
"""계획노선을 예상노선으로 되돌리고 다시 계산한다(노선 초기화).
|
|
|
|
수정본 파일을 지우면 설계 계통이 원본을 읽으므로 「원본을 수정본으로 복사」와 같다.
|
|
"""
|
|
paths = await _project_paths(project_id)
|
|
if paths is None:
|
|
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
|
|
project_root, stored_path = paths
|
|
await asyncio.to_thread(_ensure_expected_route, project_root)
|
|
radius_m = await _min_plan_radius_m(project_id)
|
|
await asyncio.to_thread(_ensure_planned_initial, project_root, radius_m)
|
|
working_path = planned_route_working_path(project_root)
|
|
if working_path.is_file():
|
|
working_path.unlink()
|
|
logger.info("계획노선 초기화(초기 폴리라인으로): project_id=%s", project_id)
|
|
result = await _recompute(project_id, project_root, stored_path)
|
|
if "error" in result:
|
|
return JSONResponse(
|
|
status_code=409, content={"status": "error", "message": result["error"]}
|
|
)
|
|
return {"status": "success", "project_id": str(project_id), **result}
|