eomsangdonandClaude Opus 5.5
5569bb6750
- 꼭짓점 반복 보정을 pass_through 로 떼어 1차 종단(목표 = 지반고) · 균형 종단 반영(목표 = 균형 종단)이 함께 씀 - 측점 계획고 대 균형 종단 차 최대 0.092 m 였던 것을 없앰 - 「종단 반영」 도움말 — 켬으로 [적용]한 노선 그대로 [확인]하면 균형 종단이 첫 계획선 - 토글 단추 — 두 줄 알약 오른쪽 가운데 · 켬 글자색 토큰 바로잡음 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EYJJRCfHfdCGNhmXTjQKV7
258 lines
12 KiB
Python
258 lines
12 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 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(지반고 종단).
|
|
"""
|
|
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,
|
|
)
|
|
)
|
|
except (FileNotFoundError, ValueError):
|
|
logger.exception("균형 종단 셈 실패(지반고 종단으로): project_id=%s", project_id)
|
|
return None
|
|
logger.info(
|
|
"균형 종단 셈: project_id=%s 측점 %d 절 %.0f 성 %.0f %.1fs",
|
|
project_id,
|
|
len(profile["stations_m"]),
|
|
profile["cut_m3"],
|
|
profile["fill_m3"],
|
|
profile["seconds"],
|
|
)
|
|
return profile["stations_m"], profile["plan_m"]
|
|
|
|
|
|
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)
|