feat(B05): 초기노선 갈래 logic 이름 받음 · 현재 초기값을 [확인] 때 프로젝트에 적음
- POST /route/initial 이 logic(또는 mode) 을 받음 · 응답에 logic 칸
- POST /route/replan 의 initial {logic, weights} 을 성공 때만 initial_choice.json 에 적음(자동저장 없음)
- GET /route/plan · POST /route/initial · POST /route/replan 응답에 current_initial
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TULoa94ZFL26KU6ZqVpjkF
This commit is contained in:
@@ -27,6 +27,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
@@ -133,6 +134,31 @@ def load_terrain_and_streams(
|
||||
return Terrain(np.asarray(x), np.asarray(y), np.asarray(z), np.asarray(valid)), streams
|
||||
|
||||
|
||||
def _choice_path(project_root: Path) -> Path:
|
||||
return project_root / "B05_Profile" / "route" / "initial_choice.json"
|
||||
|
||||
|
||||
def read_initial_choice(project_root: Path) -> dict[str, Any]:
|
||||
"""현재 초기값(갈래 · 비중) — [확인] 때 적어 둔 것. 없거나 못 읽으면 추종."""
|
||||
try:
|
||||
data = json.loads(_choice_path(project_root).read_text(encoding="utf-8"))
|
||||
logic = data["logic"] if data.get("logic") in MODES else "follow"
|
||||
return {"logic": logic, "weights": resolve_weights(logic, data.get("weights"))}
|
||||
except (OSError, ValueError, KeyError, TypeError, AttributeError):
|
||||
return {"logic": "follow", "weights": resolve_weights("follow", None)}
|
||||
|
||||
|
||||
def write_initial_choice(
|
||||
project_root: Path, logic: str, weights: dict[str, float] | None
|
||||
) -> dict[str, Any]:
|
||||
"""[확인] 때만 부름 — 작업본 한 칸. 다시 열기 · 다른 PC 에서도 같은 초기값."""
|
||||
choice = {"logic": logic, "weights": resolve_weights(logic, weights)}
|
||||
path = _choice_path(project_root)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(choice, ensure_ascii=False), encoding="utf-8")
|
||||
return choice
|
||||
|
||||
|
||||
def resolve_weights(mode: str, weights: dict[str, float] | None) -> dict[str, float]:
|
||||
"""갈래의 비중(합 100). weighted 가 아니면 갈래가 정한 값."""
|
||||
if mode != "weighted":
|
||||
@@ -192,6 +218,7 @@ def generate_initial_route(
|
||||
)
|
||||
return {
|
||||
"mode": mode,
|
||||
"logic": mode,
|
||||
"weights": used,
|
||||
"planned": [[round(x, 4), round(y, 4)] for x, y in outline.vertices],
|
||||
"nodes": [node.as_dict() for node in outline.nodes],
|
||||
|
||||
@@ -41,9 +41,10 @@ from uuid import UUID
|
||||
|
||||
from fastapi import APIRouter
|
||||
from fastapi.responses import JSONResponse
|
||||
from pydantic import BaseModel, Field
|
||||
from pydantic import AliasChoices, BaseModel, Field
|
||||
|
||||
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 common_util.common_util_initial_snapshot import design_route_csv_path
|
||||
from common_util.common_util_route_geometry import (
|
||||
densify_route,
|
||||
@@ -96,7 +97,9 @@ class RouteInitialWeights(BaseModel):
|
||||
class RouteInitialRequest(BaseModel):
|
||||
"""초기 계획노선 갈래 — follow(예상노선 추종) · earthwork · gentle · weighted."""
|
||||
|
||||
mode: Literal["follow", "earthwork", "gentle", "weighted"] = "follow"
|
||||
mode: Literal["follow", "earthwork", "gentle", "weighted"] = Field(
|
||||
"follow", validation_alias=AliasChoices("logic", "mode")
|
||||
)
|
||||
weights: RouteInitialWeights | None = None
|
||||
|
||||
|
||||
@@ -104,6 +107,8 @@ class RouteReplanRequest(BaseModel):
|
||||
"""고친 계획노선. 정점은 시점 → 종점 순서(사업지 좌표계 m)."""
|
||||
|
||||
vertices: list[RouteVertexInput] = Field(default_factory=list)
|
||||
initial: RouteInitialRequest | None = None
|
||||
"""이 노드를 낳은 초기값(갈래 · 비중) — 성공하면 프로젝트에 적어 둔다(현재 초기값)."""
|
||||
|
||||
|
||||
async def _project_paths(project_id: UUID) -> tuple[Path, str] | None:
|
||||
@@ -496,6 +501,7 @@ async def read_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse:
|
||||
"curve_count": len(saved_curves) if saved_curves else outline.curve_count,
|
||||
"violation_count": outline.violation_count,
|
||||
"edited": bool(working),
|
||||
"current_initial": await asyncio.to_thread(read_initial_choice, project_root),
|
||||
}
|
||||
|
||||
|
||||
@@ -561,7 +567,23 @@ async def replan_route(
|
||||
return JSONResponse(
|
||||
status_code=409, content={"status": "error", "message": result["error"]}
|
||||
)
|
||||
return {"status": "success", "project_id": str(project_id), **result}
|
||||
choice = request.initial
|
||||
current = (
|
||||
await asyncio.to_thread(
|
||||
write_initial_choice,
|
||||
project_root,
|
||||
choice.mode,
|
||||
choice.weights.model_dump() if choice.weights else None,
|
||||
)
|
||||
if choice
|
||||
else await asyncio.to_thread(read_initial_choice, project_root)
|
||||
)
|
||||
return {
|
||||
"status": "success",
|
||||
"project_id": str(project_id),
|
||||
**result,
|
||||
"current_initial": current,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/{project_id}/route/replan/reset", response_model=None)
|
||||
@@ -656,4 +678,10 @@ async def initial_route(
|
||||
result["metrics"]["length_m"],
|
||||
result["metrics"]["offset_avg_m"],
|
||||
)
|
||||
return {"status": "success", "project_id": str(project_id), **result}
|
||||
current = await asyncio.to_thread(read_initial_choice, project_root)
|
||||
return {
|
||||
"status": "success",
|
||||
"project_id": str(project_id),
|
||||
**result,
|
||||
"current_initial": current,
|
||||
}
|
||||
|
||||
@@ -13,8 +13,10 @@ from B05_Profile.B05_Profile_Engine_RouteInitial import (
|
||||
Terrain,
|
||||
generate_initial_route,
|
||||
grade_line,
|
||||
read_initial_choice,
|
||||
resolve_weights,
|
||||
stream_lines,
|
||||
write_initial_choice,
|
||||
)
|
||||
|
||||
GRADE = 0.14
|
||||
@@ -124,3 +126,24 @@ def test_weights_and_stream_filter() -> None:
|
||||
def test_unknown_mode_rejected(hill: Terrain) -> None:
|
||||
with pytest.raises(ValueError):
|
||||
_run(hill, "shortest")
|
||||
|
||||
|
||||
def test_initial_choice_roundtrip(tmp_path) -> None:
|
||||
"""[확인] 때 적은 현재 초기값을 다시 읽음 · 없거나 깨지면 추종."""
|
||||
assert read_initial_choice(tmp_path)["logic"] == "follow"
|
||||
write_initial_choice(tmp_path, "weighted", {"expected": 60, "earthwork": 20, "grade": 20})
|
||||
assert read_initial_choice(tmp_path) == {
|
||||
"logic": "weighted",
|
||||
"weights": {"expected": 60.0, "earthwork": 20.0, "grade": 20.0},
|
||||
}
|
||||
(tmp_path / "B05_Profile" / "route" / "initial_choice.json").write_text("{", encoding="utf-8")
|
||||
assert read_initial_choice(tmp_path)["logic"] == "follow"
|
||||
|
||||
|
||||
def test_request_accepts_logic_name() -> None:
|
||||
from B05_Profile.B05_Profile_Router_Replan import RouteInitialRequest, RouteReplanRequest
|
||||
|
||||
assert RouteInitialRequest.model_validate({"logic": "gentle"}).mode == "gentle"
|
||||
assert RouteInitialRequest.model_validate({"mode": "earthwork"}).mode == "earthwork"
|
||||
replan = RouteReplanRequest.model_validate({"vertices": [], "initial": {"logic": "weighted"}})
|
||||
assert replan.initial is not None and replan.initial.mode == "weighted"
|
||||
|
||||
Reference in New Issue
Block a user