eomsangdonandClaude Opus 5.5
f127c0179e
- 새 B05_Profile_Engine_RouteInitial_Store(ALGO_VERSION · request_fingerprint · 읽기 · 쓰기 · 묶음 잠금) · 자리 B05_Profile/route_recommend/<키>.json · 덮어쓰기 없음 · 상한 없음 - 엔드포인트 몸통 compute_initial(찾기 → 셈 · 더함 · 덮개 수 꺼낼 때 다시 셈 · stored 표시) · precompute_defaults(기본 다섯 알약) - resolve_inputs 로 기본값 풀이 한 곳 · 머리글 「파일 안 씀」 정정 - 시험 test_224_3_recommend_store Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014utrjfoW831Vc5wJUd13kL
399 lines
18 KiB
Python
399 lines
18 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, Literal
|
|
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장).
|
|
|
|
작업본은 쓰지 않는다 — 추천 저장본(`..._Store` · PLAN 327)만. 적용은 받은 `nodes` 로
|
|
`POST /route/replan`([확인]).
|
|
"""
|
|
return await compute_initial(project_id, request)
|
|
|
|
|
|
async def compute_initial(
|
|
project_id: UUID, request: replan.RouteInitialRequest
|
|
) -> dict[str, Any] | JSONResponse:
|
|
"""같은 입력이면 저장본을 꺼내고 없으면 셈해 더한다(PLAN 327) · 덮개 수는 늘 지금 관으로.
|
|
|
|
응답 `stored` = 저장본에서 꺼냈나. 진행 중인 같은 묶음은 기다렸다 꺼낸다(두 번 셈 막음).
|
|
"""
|
|
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 (
|
|
cover_shortfall,
|
|
project_pipes,
|
|
)
|
|
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
|
|
is_default,
|
|
pool_lock,
|
|
read_stored,
|
|
request_fingerprint,
|
|
write_stored,
|
|
)
|
|
from common_util.common_util_surface_confirmation import (
|
|
analysis_terrain_params,
|
|
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))
|
|
selection = analysis_terrain_params(selection, request.terrain)
|
|
inputs = {
|
|
"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,
|
|
"with_profile": request.with_profile,
|
|
}
|
|
started = time.perf_counter()
|
|
try:
|
|
terrain, streams = await asyncio.to_thread(
|
|
load_terrain_and_streams, project_root, selection
|
|
)
|
|
route = [(x, y) for x, y in expected]
|
|
key, pool_key = await asyncio.to_thread(
|
|
functools.partial(request_fingerprint, route, terrain, streams, **inputs)
|
|
)
|
|
async with pool_lock(pool_key):
|
|
stored = await asyncio.to_thread(read_stored, project_root, key)
|
|
if stored is not None:
|
|
result = stored["result"]
|
|
else:
|
|
# 관 지점은 키 밖 — 덮개 수 없이 셈해 남기고 아래에서 지금 관으로 센다.
|
|
result = await asyncio.to_thread(
|
|
functools.partial(
|
|
generate_initial_route, route, terrain, streams, pipes=None, **inputs
|
|
)
|
|
)
|
|
await asyncio.to_thread(
|
|
write_stored,
|
|
project_root,
|
|
key,
|
|
pool_key,
|
|
request.model_dump(),
|
|
result,
|
|
is_default(
|
|
request.mode,
|
|
weights,
|
|
criteria.model_dump() if criteria else None,
|
|
request.terrain,
|
|
),
|
|
)
|
|
profile = result["metrics"].get("profile")
|
|
if request.with_profile and profile and "error" not in profile:
|
|
pipes = await asyncio.to_thread(project_pipes, project_root)
|
|
if pipes:
|
|
profile.update(cover_shortfall(profile, result["planned"], pipes))
|
|
except (FileNotFoundError, ValueError) as error:
|
|
return JSONResponse(status_code=409, content={"status": "error", "message": str(error)})
|
|
logger.info(
|
|
"초기 계획노선 %s(%s): project_id=%s %.1fs 연장 %.0fm 이격 평균 %.1fm",
|
|
request.mode,
|
|
"저장본" if stored is not None else "셈",
|
|
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,
|
|
"stored": stored is not None,
|
|
"current_initial": current,
|
|
}
|
|
|
|
|
|
async def precompute_defaults(project_id: UUID) -> None:
|
|
"""업로드 체인 끝 예약(PLAN 327-2) — 기본 다섯 알약을 미리 셈해 남김 · 있으면 꺼내기만."""
|
|
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import DEFAULT_MODES
|
|
|
|
started = time.perf_counter()
|
|
for mode in DEFAULT_MODES:
|
|
# 화면 기본과 같은 몸통 — 빈 제약 · 50/25/25 · 설계 지형 · 추적만 종단 끔(`withProfileOf`)
|
|
request = replan.RouteInitialRequest(mode=mode, with_profile=mode != "follow")
|
|
outcome = await compute_initial(project_id, request)
|
|
if isinstance(outcome, JSONResponse):
|
|
logger.warning(
|
|
"추천 미리 셈 멈춤(%s %s): project_id=%s", mode, outcome.status_code, project_id
|
|
)
|
|
return
|
|
logger.info("추천 미리 셈 끝: project_id=%s %.1fs", project_id, time.perf_counter() - started)
|
|
|
|
|
|
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, _dropped = 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,
|
|
terrain: Literal["las", "sheet"] | None = None,
|
|
) -> dict[str, Any] | None:
|
|
"""노선 정점의 균형 종단 한 벌(`route_profile`) — [확인] 체인 · 편집 창 종단 띠가 함께 씀.
|
|
|
|
`terrain` 은 편집 창 종단 띠만 바꿈(PLAN 133-1) — 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 (
|
|
analysis_terrain_params,
|
|
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))
|
|
selection = analysis_terrain_params(selection, terrain)
|
|
try:
|
|
surface, _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],
|
|
surface,
|
|
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
|
|
terrain: Literal["las", "sheet"] | 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, request.terrain
|
|
)
|
|
if profile is None:
|
|
return JSONResponse(
|
|
status_code=409,
|
|
content={"status": "error", "message": "이 노선으로 균형 종단을 세우지 못했습니다."},
|
|
)
|
|
return {"status": "success", "project_id": str(project_id), "profile": profile}
|