eomsangdonandClaude Opus 5.5
0deeb52e9a
- 확폭은 노선 곡선표로(후보 · 편집 띠 · [확인]) · 없으면 종전 실측 반경 - 반폭에서 비탈이 안 닫히면 지표면 끝까지 넓힌 샘플로 다시 셈(본 설계 extend_unclosed 와 같은 뜻) - 암반 경계 기본 오프셋(본 설계 기본 설계와 같음) - 균형 목표 = 누가토량 끝값(절토 다짐 − 성토) ±500㎥ · 고른 선은 보간 말고 바로 셈해 λ 를 더 좁힘 - 후보 줄 세우기는 보간값 그대로(빠름) · 화면에 낼 선만 바로 셈 - 추천 창 한 줄: 불균형 % → 누가토량 ±㎥ Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019EMfGERKUg5nMMNQsyVZ2j
318 lines
14 KiB
Python
318 lines
14 KiB
Python
"""초기 계획노선 갈래 하나 — `POST /api/projects/{id}/route/initial`(PLAN 20 · 23장).
|
|
|
|
`B05_Profile_Router_Replan`(700줄)에서 나눈 길. 같은 `router` 에 붙고 그 모듈 끝에서 불려
|
|
등록된다 — main · 시험이 `Router_Replan.router` 하나만 실어도 이 길이 함께 실림.
|
|
"""
|
|
|
|
import asyncio
|
|
import functools
|
|
import logging
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from B05_Profile import B05_Profile_Router_Replan as replan
|
|
from B05_Profile.B05_Profile_Engine_RouteInitial import read_initial_choice
|
|
from common_util.common_util_initial_snapshot import design_route_csv_path
|
|
from common_util.common_util_route_geometry import expected_route_csv_path
|
|
from config.config_db import get_db_pool
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@replan.router.post("/{project_id}/route/initial", response_model=None)
|
|
async def initial_route(
|
|
project_id: UUID, request: replan.RouteInitialRequest
|
|
) -> dict[str, Any] | JSONResponse:
|
|
"""초기 계획노선 갈래 하나를 만들어 비교값과 함께 돌려준다(PLAN 20장).
|
|
|
|
**파일을 쓰지 않는다** — 화면 캐시용. 적용은 받은 `nodes` 로 `POST /route/replan`([확인]).
|
|
"""
|
|
from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct
|
|
from B05_Profile.B05_Profile_Engine_RouteInitial import (
|
|
generate_initial_route,
|
|
load_terrain_and_streams,
|
|
)
|
|
from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import project_pipes
|
|
from common_util.common_util_surface_confirmation import get_surface_confirmation_params
|
|
|
|
weights = request.weights.model_dump() if request.weights else None
|
|
if request.mode == "weighted" and weights and abs(sum(weights.values()) - 100.0) > 0.5:
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"status": "error", "message": "비중 세 값의 합은 100 이어야 합니다."},
|
|
)
|
|
try:
|
|
paths = await replan._project_paths(project_id)
|
|
except LookupError: # 없는 프로젝트 — 저장소 조회가 None 대신 던짐
|
|
paths = None
|
|
if paths is None:
|
|
return JSONResponse(status_code=404, content=replan._PROJECT_PATH_MISSING)
|
|
project_root, _ = paths
|
|
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))
|
|
if len(expected) < 2:
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"status": "error", "message": "예상노선이 없어 초기노선을 만들 수 없습니다."},
|
|
)
|
|
radius_m, _, _ = await replan._plan_criteria(project_id)
|
|
grade_class, design_speed, terrain_type = await replan._road_settings(project_id)
|
|
# 비포장 기준 — 포장 예외(18 %)는 포장 여부가 정해진 뒤 종단 설계가 따진다(사용자 확정 대기).
|
|
max_grade = legal_grade_limit_pct(grade_class, terrain_type, False, design_speed) / 100.0
|
|
# 화면 제약 칸(PLAN 20장 확정) — 비운 칸은 위 프로젝트 기본값 그대로, 채운 칸만 겹쳐 쓴다.
|
|
criteria = request.criteria
|
|
if criteria and criteria.max_grade_pct is not None:
|
|
max_grade = criteria.max_grade_pct / 100.0
|
|
if criteria and criteria.min_radius_m is not None:
|
|
radius_m = criteria.min_radius_m
|
|
stream_offset_m = criteria.stream_offset_m if criteria else None
|
|
grade_smooth_m = criteria.grade_smooth_m if criteria else None
|
|
target_pct = criteria.target_grade_pct if criteria else None
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
selection = await get_surface_confirmation_params(connection, str(project_id))
|
|
started = time.perf_counter()
|
|
try:
|
|
terrain, streams = await asyncio.to_thread(
|
|
load_terrain_and_streams, project_root, selection
|
|
)
|
|
pipes = (
|
|
await asyncio.to_thread(project_pipes, project_root) if request.with_profile else None
|
|
)
|
|
result = await asyncio.to_thread(
|
|
functools.partial(
|
|
generate_initial_route,
|
|
[(x, y) for x, y in expected],
|
|
terrain,
|
|
streams,
|
|
mode=request.mode,
|
|
weights=weights,
|
|
max_grade=max_grade,
|
|
min_radius_m=radius_m,
|
|
stream_offset_m=stream_offset_m,
|
|
grade_smooth_m=grade_smooth_m,
|
|
target_grade=None if target_pct is None else target_pct / 100.0,
|
|
design_speed=design_speed or 20,
|
|
pipes=pipes,
|
|
with_profile=request.with_profile,
|
|
)
|
|
)
|
|
except (FileNotFoundError, ValueError) as error:
|
|
return JSONResponse(status_code=409, content={"status": "error", "message": str(error)})
|
|
logger.info(
|
|
"초기 계획노선 %s: project_id=%s %.1fs 연장 %.0fm 이격 평균 %.1fm",
|
|
request.mode,
|
|
project_id,
|
|
time.perf_counter() - started,
|
|
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),
|
|
**result,
|
|
"current_initial": current,
|
|
}
|
|
|
|
|
|
def _set_base_pvi(path: Path, stations: list[float], plan: list[float]) -> int:
|
|
"""종단 정본의 1차 변화점(`base_pvi`) 계획고를 `stations` · `plan` 직선 보간으로 바꿔 쓴다."""
|
|
import json
|
|
|
|
import numpy as np
|
|
|
|
from B05_Profile.B05_Profile_Engine_Grade_Alignment import AlignmentPolicy
|
|
from B05_Profile.B05_Profile_Engine_Grade_Profile import pass_through
|
|
from common_util.common_util_json import atomic_write_json
|
|
|
|
longitudinal = json.loads(path.read_text(encoding="utf-8"))
|
|
alignment = longitudinal["profile_alignment"]
|
|
base = alignment["base_pvi"]
|
|
base_s = np.array([float(item["chainage_m"]) for item in base])
|
|
# 1차 종단과 같은 맞춤 — 종단곡선을 넣어도 측점마다 균형 종단 계획고를 지나게.
|
|
base_z = pass_through(
|
|
base_s, np.interp(base_s, stations, plan), AlignmentPolicy.from_dict(alignment["policy"])
|
|
)
|
|
for item, value in zip(base, base_z, strict=True):
|
|
item["elevation_m"] = round(float(value), 4)
|
|
atomic_write_json(path, longitudinal)
|
|
return len(base)
|
|
|
|
|
|
async def apply_profile_target(
|
|
project_id: UUID, route_id: int, target: tuple[list[float], list[float]]
|
|
) -> None:
|
|
"""[확인] 체인의 1차 종단을 추천 균형 종단(누가거리 · 계획고)으로 갈아 끼운다(PLAN 74-4).
|
|
|
|
1차 종단(`design_ground_following_profile`)은 측점마다 계획고 = 지반고로 맞춘다 — 그 맞춤 대신
|
|
변화점(`base_pvi`)을 균형 종단에 놓고 `rebuild_alignment_profile` 로 계획선을 다시 낸다.
|
|
구조물 · 관 측점은 앞뒤 20 m 측점 사이 직선 보간. 실패는 비치명적 — 지반고 종단이 남는다.
|
|
"""
|
|
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
|
from B05_Profile.B05_Profile_Repository import update_longitudinal_grade_summary
|
|
from B05_Profile.B05_Profile_Router import _apply_alignment_edits
|
|
from B06_Section.B06_Section_Repository import get_longitudinal_section
|
|
from common_util.common_util_storage import resolve_stored_project_path
|
|
|
|
pool = get_db_pool()
|
|
try:
|
|
async with pool.acquire() as connection:
|
|
row = await get_longitudinal_section(connection, project_id, route_id)
|
|
stored_path = await get_project_storage_relative_path(connection, project_id)
|
|
root = Path(resolve_stored_project_path(stored_path))
|
|
relative = str(row["longitudinal_file_path"])
|
|
count = await asyncio.to_thread(_set_base_pvi, root / relative, *target)
|
|
_alignment, summary = await asyncio.to_thread(_apply_alignment_edits, root, relative, {})
|
|
async with pool.acquire() as connection:
|
|
await connection.begin()
|
|
await update_longitudinal_grade_summary(
|
|
connection, route_id=route_id, grade_summary=summary
|
|
)
|
|
await connection.commit()
|
|
logger.info(
|
|
"균형 종단 반영: project_id=%s route_id=%s 변화점 %d", project_id, route_id, count
|
|
)
|
|
except Exception: # noqa: BLE001 — 지반고 종단이 남는다
|
|
logger.exception("균형 종단 반영 실패(지반고 종단 유지): project_id=%s", project_id)
|
|
|
|
|
|
async def balanced_target(
|
|
project_id: UUID,
|
|
project_root: Path,
|
|
points: list[dict[str, float]],
|
|
criteria: Any,
|
|
) -> tuple[list[float], list[float]] | None:
|
|
"""확정할 노선의 균형 종단(누가거리 · 계획고) — 추천 창과 같은 코드 · 같은 상한(PLAN 74-4).
|
|
|
|
화면 값을 받지 않고 노선 정점으로 다시 셈한다(CLAUDE.md 5). 못 세우면 None(지반고 종단).
|
|
확폭은 측점 생성과 같은 곡선표(`_load_design_curves`)로.
|
|
"""
|
|
from B05_Profile.B05_Profile_Engine_Sections import _load_design_curves
|
|
|
|
curves = await asyncio.to_thread(_load_design_curves, project_root)
|
|
profile = await balanced_profile_of(project_id, project_root, points, criteria, curves)
|
|
return None if profile is None else (profile["stations_m"], profile["plan_m"])
|
|
|
|
|
|
async def balanced_profile_of(
|
|
project_id: UUID,
|
|
project_root: Path,
|
|
points: list[dict[str, float]],
|
|
criteria: Any,
|
|
design_curves: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any] | None:
|
|
"""노선 정점의 균형 종단 한 벌(`route_profile`) — [확인] 체인 · 편집 창 종단 띠가 함께 씀."""
|
|
from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct
|
|
from B05_Profile.B05_Profile_Engine_RouteInitial import load_terrain_and_streams
|
|
from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import route_profile
|
|
from common_util.common_util_surface_confirmation import get_surface_confirmation_params
|
|
|
|
grade_class, design_speed, terrain_type = await replan._road_settings(project_id)
|
|
max_grade = legal_grade_limit_pct(grade_class, terrain_type, False, design_speed) / 100.0
|
|
if criteria is not None and criteria.max_grade_pct is not None:
|
|
max_grade = criteria.max_grade_pct / 100.0
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
selection = await get_surface_confirmation_params(connection, str(project_id))
|
|
try:
|
|
terrain, _streams = await asyncio.to_thread(
|
|
load_terrain_and_streams, project_root, selection
|
|
)
|
|
profile = await asyncio.to_thread(
|
|
functools.partial(
|
|
route_profile,
|
|
[(p["x"], p["y"]) for p in points],
|
|
terrain,
|
|
max_grade=max_grade,
|
|
design_speed=design_speed or 20,
|
|
design_curves=design_curves,
|
|
)
|
|
)
|
|
except (FileNotFoundError, ValueError):
|
|
logger.exception("균형 종단 셈 실패(지반고 종단으로): project_id=%s", project_id)
|
|
return None
|
|
logger.info(
|
|
"균형 종단 셈: project_id=%s 측점 %d 절 %.0f 성 %.0f 누가 %.0f %.1fs",
|
|
project_id,
|
|
len(profile["stations_m"]),
|
|
profile["cut_m3"],
|
|
profile["fill_m3"],
|
|
profile["mass_end_m3"],
|
|
profile["seconds"],
|
|
)
|
|
return profile
|
|
|
|
|
|
async def refresh_initial_snapshot(
|
|
project_id: UUID, project_root: Path, route_id: int | None
|
|
) -> None:
|
|
"""노선 [확인] 결과로 초기값(initial_snapshot)을 다시 뜬다(PLAN 74-4) — 실패는 비치명적."""
|
|
from common_util.common_util_initial_snapshot import (
|
|
discard_initial_snapshot,
|
|
save_initial_snapshot,
|
|
)
|
|
|
|
if route_id is None:
|
|
return
|
|
try:
|
|
# 스냅샷 안 `planned_route.csv` 는 예상노선(원본) 자리 — 확정 노선 아닌 예상노선을 둔다.
|
|
expected = await asyncio.to_thread(
|
|
replan._vertices_of, expected_route_csv_path(project_root)
|
|
)
|
|
points = [{"x": x, "y": y} for x, y in expected] or None
|
|
await asyncio.to_thread(discard_initial_snapshot, project_root)
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
await save_initial_snapshot(connection, project_root, int(route_id), points)
|
|
logger.info("노선 [확인] 초기값 다시 뜸: project_id=%s route_id=%s", project_id, route_id)
|
|
except Exception: # noqa: BLE001 — 계산 결과는 그대로 남는다
|
|
logger.exception("노선 [확인] 초기값 스냅샷 실패: project_id=%s", project_id)
|
|
|
|
|
|
class ProfilePreviewRequest(BaseModel):
|
|
"""편집 창 종단 띠(PLAN 75-1) — 지금 노드 · 지금 추천 제약."""
|
|
|
|
vertices: list[replan.RouteVertexInput] = Field(..., min_length=2)
|
|
criteria: replan.RouteInitialCriteria | None = None
|
|
|
|
|
|
@replan.router.post("/{project_id}/route/profile-preview", response_model=None)
|
|
async def profile_preview(
|
|
project_id: UUID, request: ProfilePreviewRequest
|
|
) -> dict[str, Any] | JSONResponse:
|
|
"""지금 노드의 균형 종단 — [확인]과 같은 폴리라인(같은 R · CSV 자릿수) · 같은 셈.
|
|
|
|
파일 · DB 를 쓰지 않는다. 「종단 반영」 켬일 때 편집 창 띠가 노드를 놓을 때마다 부른다.
|
|
"""
|
|
from common_util.common_util_route_polyline import build_planned_polyline
|
|
|
|
paths = await replan._project_paths(project_id)
|
|
if paths is None:
|
|
return JSONResponse(status_code=404, content=replan._PROJECT_PATH_MISSING)
|
|
project_root, _ = paths
|
|
radius_m, _, _ = await replan._plan_criteria(project_id)
|
|
built = build_planned_polyline(
|
|
[(v.x, v.y) for v in request.vertices],
|
|
min_radius_m=radius_m,
|
|
simplify=False,
|
|
curve_flags=[bool(v.curve) for v in request.vertices],
|
|
radii=[v.radius_m for v in request.vertices],
|
|
)
|
|
# [확인]은 CSV(소수 4자리)로 적은 뒤 읽는다 — 같은 점을 셈하게 같은 자릿수로.
|
|
points = [{"x": round(float(x), 4), "y": round(float(y), 4)} for x, y in built.vertices]
|
|
curves = [curve.as_dict() for curve in built.curves]
|
|
profile = await balanced_profile_of(project_id, project_root, points, request.criteria, curves)
|
|
if profile is None:
|
|
return JSONResponse(
|
|
status_code=409,
|
|
content={"status": "error", "message": "이 노선으로 균형 종단을 세우지 못했습니다."},
|
|
)
|
|
return {"status": "success", "project_id": str(project_id), "profile": profile}
|