feat(346-3): 노선 저장 서버 길 — 배수유역 뗌 · 관 지킴 · 측점 열쇠 이월 · 초기값은 추천 [적용] 때만
- POST /route/replan = 노선 저장 체인(B05_Profile_Route_Save) — 풀기 → 구조물 · 관 좌표 따라감 → 새 세류선 교차점 기본 관 · 옛 노선 밖 오목 지점 보충 → 종횡단 한 번 → 계획고 옮김(바뀐 구간만 새 지반) → 횡단 이월(측점 열쇠 grid · pipe · structure · end · CARRY 일곱 + 구조물 조정) → B06 확정 · 3D - 관 정본 안 지움 · 배수유역 체인에서 뗌(관 정본 새 노선 지문 → B04 분석 낡음) - 요청에 노선 초안(RouteDraft 모양 draft) · as_initial · 응답에 관 · 구조물 · 버린 횡단 수 · 초기값 여부 · 시간 - 초기값 = 자동설계 + 추천 [적용](as_initial)만 · 손 편집 저장은 안 바꿈 - 실패 = 노선 · 관 · 구조물 · 종횡단 파일 · 새 노선 행 · stage 2 params 되살림 - 시험 test_346_3_route_save · test_133_2 새 체인으로 · 시간 재기 도구 run_346_3_route_save_timing Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ji4nuyLVHcxrXWHTdb5nRD
This commit is contained in:
Claude Opus 5.5
@@ -392,6 +392,53 @@ async def _ensure_initial_polyline(
|
||||
logger.exception("초기 계획노선 폴리라인 생성 실패(계속 진행): %s", project_id)
|
||||
|
||||
|
||||
def redesign_solve_request(
|
||||
params: dict[str, Any], selection: dict[str, Any], surface_model_id: int
|
||||
) -> Any:
|
||||
"""stage 2 params(사용자 저장분) + 새 확정 지표면 → as_planned 풀기 요청(노선 저장도 씀)."""
|
||||
from B05_Profile.B05_Profile_Schema import RouteSolveRequest
|
||||
|
||||
points = params.get("points") or {}
|
||||
options = params.get("options") or {}
|
||||
return RouteSolveRequest(
|
||||
filter_key=str(selection.get("source_filter") or params.get("filter_key")),
|
||||
method=str(selection.get("method") or params.get("method") or "dtm"),
|
||||
smooth=bool(selection.get("smooth", params.get("smooth", False))),
|
||||
surface_model_id=surface_model_id,
|
||||
# 사용자가 고친 계획노선을 **그대로** 쓴다 — 다시 풀면 탐색 제약(종단경사·
|
||||
# 최소곡선반지름·회피지역)에 걸려 체인이 통째로 멈춘다(2026-09-06 실측:
|
||||
# 「세그먼트 1 (BP → CP1) 경로 탐색 실패」로 종횡단만 옛 노선에 남았음).
|
||||
# PLAN 0-3 에서 자동탐색은 접기로 했고 노선은 사용자가 직접 고친다.
|
||||
algorithm="as_planned",
|
||||
bp=points["bp"],
|
||||
ep=points["ep"],
|
||||
cp=points.get("cp") or [],
|
||||
ap=points.get("ap") or [],
|
||||
fp=points.get("fp") or [],
|
||||
station_interval_m=params.get("station_interval_m"),
|
||||
cross_half_width_m=params.get("cross_half_width_m"),
|
||||
cross_sample_interval_m=params.get("cross_sample_interval_m"),
|
||||
long_sample_interval_m=params.get("long_sample_interval_m"),
|
||||
**{key: options.get(key) for key in ("grade_class",) if options.get(key)},
|
||||
paved=bool(options.get("paved", False)),
|
||||
terrain_type=str(options.get("terrain_type") or "special"),
|
||||
main_direction=str(options.get("main_direction") or "auto"),
|
||||
min_curve_radius_m=options.get("min_curve_radius_m"),
|
||||
max_uphill_grade=options.get("max_uphill_grade"),
|
||||
max_downhill_grade=options.get("max_downhill_grade"),
|
||||
min_uphill_grade=options.get("min_uphill_grade"),
|
||||
min_downhill_grade=options.get("min_downhill_grade"),
|
||||
weights=options.get("weights"),
|
||||
allow_avoid_pass_through=bool(options.get("allow_avoid_pass_through", False)),
|
||||
max_grade_pct=params.get("max_grade_pct"),
|
||||
min_vertical_radius_m=params.get("min_vertical_radius_m"),
|
||||
min_tangent_length_m=params.get("min_tangent_length_m"),
|
||||
balance_segment_length_m=params.get("balance_segment_length_m"),
|
||||
start_elevation_offset_m=params.get("start_elevation_offset_m"),
|
||||
end_elevation_offset_m=params.get("end_elevation_offset_m"),
|
||||
)
|
||||
|
||||
|
||||
async def run_redesign_chain(
|
||||
project_id: UUID,
|
||||
surface_model_id: int,
|
||||
@@ -421,7 +468,6 @@ async def run_redesign_chain(
|
||||
confirm_latest_route,
|
||||
solve_route_record,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Schema import RouteSolveRequest
|
||||
from B06_Section.B06_Section_Repository import (
|
||||
get_cross_section_designs,
|
||||
get_latest_section_options,
|
||||
@@ -478,7 +524,6 @@ async def run_redesign_chain(
|
||||
)
|
||||
params = (stage2 or {}).get("params") or {}
|
||||
points = params.get("points") or {}
|
||||
options = params.get("options") or {}
|
||||
if not points.get("bp") or not points.get("ep"):
|
||||
logger.warning(
|
||||
"재확정 체인 중단(stage 2 params에 제어점 없음): project_id=%s", project_id
|
||||
@@ -488,43 +533,7 @@ async def run_redesign_chain(
|
||||
marks.append(("준비·사용자입력 회수", time.perf_counter()))
|
||||
|
||||
# 2) B05 재계산 — 지표면 관련 값만 새 확정 선택으로 교체, 나머지는 사용자 저장분.
|
||||
request = RouteSolveRequest(
|
||||
filter_key=str(selection.get("source_filter") or params.get("filter_key")),
|
||||
method=str(selection.get("method") or params.get("method") or "dtm"),
|
||||
smooth=bool(selection.get("smooth", params.get("smooth", False))),
|
||||
surface_model_id=surface_model_id,
|
||||
# 사용자가 고친 계획노선을 **그대로** 쓴다 — 다시 풀면 탐색 제약(종단경사·
|
||||
# 최소곡선반지름·회피지역)에 걸려 체인이 통째로 멈춘다(2026-09-06 실측:
|
||||
# 「세그먼트 1 (BP → CP1) 경로 탐색 실패」로 종횡단만 옛 노선에 남았음).
|
||||
# PLAN 0-3 에서 자동탐색은 접기로 했고 노선은 사용자가 직접 고친다.
|
||||
algorithm="as_planned",
|
||||
bp=points["bp"],
|
||||
ep=points["ep"],
|
||||
cp=points.get("cp") or [],
|
||||
ap=points.get("ap") or [],
|
||||
fp=points.get("fp") or [],
|
||||
station_interval_m=params.get("station_interval_m"),
|
||||
cross_half_width_m=params.get("cross_half_width_m"),
|
||||
cross_sample_interval_m=params.get("cross_sample_interval_m"),
|
||||
long_sample_interval_m=params.get("long_sample_interval_m"),
|
||||
**{key: options.get(key) for key in ("grade_class",) if options.get(key)},
|
||||
paved=bool(options.get("paved", False)),
|
||||
terrain_type=str(options.get("terrain_type") or "special"),
|
||||
main_direction=str(options.get("main_direction") or "auto"),
|
||||
min_curve_radius_m=options.get("min_curve_radius_m"),
|
||||
max_uphill_grade=options.get("max_uphill_grade"),
|
||||
max_downhill_grade=options.get("max_downhill_grade"),
|
||||
min_uphill_grade=options.get("min_uphill_grade"),
|
||||
min_downhill_grade=options.get("min_downhill_grade"),
|
||||
weights=options.get("weights"),
|
||||
allow_avoid_pass_through=bool(options.get("allow_avoid_pass_through", False)),
|
||||
max_grade_pct=params.get("max_grade_pct"),
|
||||
min_vertical_radius_m=params.get("min_vertical_radius_m"),
|
||||
min_tangent_length_m=params.get("min_tangent_length_m"),
|
||||
balance_segment_length_m=params.get("balance_segment_length_m"),
|
||||
start_elevation_offset_m=params.get("start_elevation_offset_m"),
|
||||
end_elevation_offset_m=params.get("end_elevation_offset_m"),
|
||||
)
|
||||
request = redesign_solve_request(params, selection, surface_model_id)
|
||||
# 종횡단은 관을 정한 뒤 한 번만 만든다(342-2) — 여기서 만들면 아래 4c 가 통째로 덮는 헛일.
|
||||
solve_result: Any = await solve_route_record(project_id, request, generate_sections=False)
|
||||
if isinstance(solve_result, JSONResponse):
|
||||
|
||||
@@ -0,0 +1,496 @@
|
||||
"""노선 저장 서버 길(346-3) — 노선 초안을 정본으로: 노선 → 풀기 → 관 · 구조물 따라감 →
|
||||
종 · 횡단 한 번 → 계획고 옮김 → 횡단 이월 → B06 확정(서버 재계산 · 3D). 배수유역은 체인에서 뗌.
|
||||
|
||||
API — `POST /api/projects/{id}/route/replan`(`B05_Profile_Router_Replan.replan_route`):
|
||||
요청 `RouteSaveRequest` = 옛 칸(`vertices` · `initial` · `balanced_profile` · `terrain` ·
|
||||
`station_snap`) +
|
||||
`draft` — 계약 `RouteDraft`(`A00_Common/design/design_pipeline_types.ts`) 그대로
|
||||
`{rev, nodes: [[x, y]…], curveOn: [bool…], curveRadius: [m | null…], curveLock, curveArc,
|
||||
stationSnap, terrain}` · 오면 `vertices` · `station_snap` · `terrain` 대신 이것.
|
||||
`as_initial` — 추천 노선 [적용](초기값 다시 뜸) · 손으로 고친 노선 저장은 false(초기값 그대로).
|
||||
응답 200 `{status: "success", project_id, route_id, total_length_m, vertex_count, current_initial,
|
||||
initial_refreshed, pipes: {kept, crossing, concave}, structures: {moved, far}, cross_dropped,
|
||||
drainage_stale: true, seconds}`.
|
||||
오류 400(노드 둘 미만 · 초안 칸 길이 어긋남) · 404(프로젝트 없음) ·
|
||||
409 `{status: "error", message}`(체인 실패 — 노선 · 관 · 구조물 · 종횡단 파일 · 새 노선 행을
|
||||
옛것으로 되살린 뒤).
|
||||
|
||||
규칙:
|
||||
- 관 정본을 지우지 않음 — 좌표를 따라감 · 새 세류선 교차점 기본 관 · 오목 지점 보충만 더함
|
||||
(`B05_Profile_Route_Save_Parts`) · 배수유역 저장분은 노선 지문이 달라 「낡음」(B04 `_resolve`).
|
||||
- 계획고 = 바뀐 구간만 새 지반 첫 계획선 · 그 밖 사용자 계획고 지킴
|
||||
(`B05_Profile_Engine_Grade_Carry`) · 추천 「종단 반영」 켬이면 균형 종단
|
||||
(`apply_profile_target`).
|
||||
- 초기값은 여기서 안 뜸 — 부르는 쪽이 `as_initial` 일 때만(자동설계 체인 직후 + 추천 [적용]).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import shutil
|
||||
import tempfile
|
||||
import time
|
||||
from dataclasses import asdict
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
from uuid import UUID
|
||||
|
||||
import numpy as np
|
||||
from fastapi.responses import JSONResponse
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from B05_Profile.B05_Profile_Route_Save_Parts import (
|
||||
add_crossing_pipes,
|
||||
carry_cross_inputs,
|
||||
concave_chainages,
|
||||
fill_concave_pipes,
|
||||
follow_pipes,
|
||||
station_keys,
|
||||
vertices_of,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Schema import RouteReplanRequest, RouteVertexInput
|
||||
from config.config_db import get_db_pool
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class RouteDraftInput(BaseModel):
|
||||
"""노선 초안 — 계약 `RouteDraft` 모양 그대로(칸 이름 camelCase)."""
|
||||
|
||||
rev: int = 0
|
||||
nodes: list[tuple[float, float]] = Field(default_factory=list)
|
||||
curveOn: list[bool] = Field(default_factory=list) # noqa: N815 — 계약 칸 이름
|
||||
curveRadius: list[float | None] = Field(default_factory=list) # noqa: N815
|
||||
curveLock: list[str | None] = Field(default_factory=list) # noqa: N815
|
||||
curveArc: list[float | None] = Field(default_factory=list) # noqa: N815
|
||||
stationSnap: bool = False # noqa: N815
|
||||
terrain: Literal["las", "sheet"] | None = None
|
||||
|
||||
|
||||
class RouteSaveRequest(RouteReplanRequest):
|
||||
"""노선 저장 요청 — 옛 [노선 반영] 칸 + 초안(`draft`) · 추천 [적용](`as_initial`)."""
|
||||
|
||||
draft: RouteDraftInput | None = None
|
||||
as_initial: bool = False
|
||||
|
||||
def resolved(self) -> RouteReplanRequest:
|
||||
"""초안이 오면 노드 · 곡선 · 스냅 · 지형을 옛 칸으로 편다. 칸 길이가 어긋나면 ValueError."""
|
||||
if self.draft is None:
|
||||
return self
|
||||
draft = self.draft
|
||||
count = len(draft.nodes)
|
||||
for name in ("curveOn", "curveRadius"):
|
||||
values = getattr(draft, name)
|
||||
if values and len(values) != count:
|
||||
raise ValueError(
|
||||
f"노선 초안 {name} 길이({len(values)})가 노드 수({count})와 다릅니다."
|
||||
)
|
||||
vertices = [
|
||||
RouteVertexInput(
|
||||
x=x,
|
||||
y=y,
|
||||
curve=draft.curveOn[index] if draft.curveOn else True,
|
||||
radius_m=draft.curveRadius[index] if draft.curveRadius else None,
|
||||
)
|
||||
for index, (x, y) in enumerate(draft.nodes)
|
||||
]
|
||||
return self.model_copy(
|
||||
update={
|
||||
"vertices": vertices,
|
||||
"station_snap": draft.stationSnap,
|
||||
"terrain": draft.terrain,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class _Backup:
|
||||
"""체인이 고치는 파일을 손에 쥠 — 실패하면 그대로 되돌림(노선 · 관 · 구조물 · 종횡단)."""
|
||||
|
||||
def __init__(self, project_root: Path, paths: list[Path], folders: list[Path]) -> None:
|
||||
self.held = tempfile.mkdtemp(prefix="route_save_", dir=str(project_root))
|
||||
self.files = {path: path.read_bytes() if path.is_file() else None for path in paths}
|
||||
self.folders: dict[Path, Path | None] = {}
|
||||
for index, folder in enumerate(folders):
|
||||
copy = Path(self.held) / f"dir{index}"
|
||||
self.folders[folder] = shutil.copytree(folder, copy) if folder.is_dir() else None
|
||||
|
||||
def restore(self) -> None:
|
||||
for path, saved in self.files.items():
|
||||
if saved is None:
|
||||
path.unlink(missing_ok=True)
|
||||
else:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(saved)
|
||||
for folder, copy in self.folders.items():
|
||||
shutil.rmtree(folder, ignore_errors=True)
|
||||
if copy is not None:
|
||||
shutil.copytree(copy, folder)
|
||||
|
||||
def drop(self) -> None:
|
||||
shutil.rmtree(self.held, ignore_errors=True)
|
||||
|
||||
|
||||
class _Fail(RuntimeError):
|
||||
"""체인 한 단계 실패 — 사유를 그대로 화면까지(까닭을 덮지 않음)."""
|
||||
|
||||
|
||||
def _reason(response: JSONResponse) -> str:
|
||||
try:
|
||||
return bytes(response.body).decode("utf-8", "replace")[:300]
|
||||
except Exception: # noqa: BLE001 — 사유는 로그 · 응답용
|
||||
return f"status={response.status_code}"
|
||||
|
||||
|
||||
def _read_json(path: Path) -> dict[str, Any]:
|
||||
try:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
|
||||
|
||||
def _xy(project_root: Path, route_data_path: str | None) -> list[list[float]]:
|
||||
from B05_Profile.B05_Profile_Engine_Sections import _load_route_polyline
|
||||
|
||||
if not route_data_path:
|
||||
return []
|
||||
try:
|
||||
return [[x, y] for x, y, _ in _load_route_polyline(project_root, route_data_path)]
|
||||
except (OSError, ValueError):
|
||||
return []
|
||||
|
||||
|
||||
def _ground_along(project_root: Path, selection: dict[str, Any], vertices: list) -> tuple:
|
||||
"""새 노선 지반 종단(2 m) — 오목 지점 찾기. 지표면을 못 열면 빈 배열(보충 건너뜀)."""
|
||||
from shapely.geometry import LineString
|
||||
|
||||
from B05_Profile.B05_Profile_Route_Save_Parts import CONCAVE_STEP_M
|
||||
from common_util.common_util_surface_sampler import build_surface_sampler
|
||||
|
||||
try:
|
||||
sampler = build_surface_sampler(
|
||||
project_root / "B04_PreProcess" / "models",
|
||||
str(selection.get("source_filter")),
|
||||
str(selection.get("method") or "dtm"),
|
||||
bool(selection.get("smooth")),
|
||||
)
|
||||
except (FileNotFoundError, KeyError, OSError, ValueError):
|
||||
logger.warning("노선 저장: 지표면을 못 열어 오목 지점 보충을 건너뜀")
|
||||
return np.zeros(0), np.zeros(0)
|
||||
line = LineString([(vertex.x, vertex.y) for vertex in vertices])
|
||||
chainage = np.arange(0.0, line.length + 1e-9, CONCAVE_STEP_M)
|
||||
xy = np.array([line.interpolate(float(s)).coords[0] for s in chainage], dtype=np.float64)
|
||||
z, valid = sampler.sample_xy(xy)
|
||||
return chainage[valid], np.asarray(z, dtype=np.float64)[valid]
|
||||
|
||||
|
||||
async def _place_pipes(
|
||||
project_id: UUID,
|
||||
project_root: Path,
|
||||
stored_path: str,
|
||||
selection: dict[str, Any],
|
||||
old_pipes: list,
|
||||
old_xy: list[list[float]],
|
||||
points: list[dict[str, float]],
|
||||
) -> tuple[list, dict[str, int]]:
|
||||
"""관 따라감 → 새 교차점 기본 관 → 오목 지점 보충 → 정본 저장(새 노선 지문)."""
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import _base_pipes
|
||||
from B04_PreProcess.B04_PreProcess_Router_Watershed import _prepare
|
||||
from common_util.common_util_drainage_pipes import route_signature, save_pipe_points
|
||||
|
||||
prepared = await _prepare(project_id)
|
||||
if isinstance(prepared, JSONResponse): # 도엽 · 노선 파일 없음 — 따라감만
|
||||
vertices, streams = vertices_of([(p["x"], p["y"]) for p in points]), []
|
||||
else:
|
||||
vertices, streams = prepared["vertices"], prepared["streams"]
|
||||
pairs = follow_pipes(old_pipes, vertices)
|
||||
total = float(vertices[-1].chainage_m)
|
||||
crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total)
|
||||
chainage, ground = await asyncio.to_thread(_ground_along, project_root, selection, vertices)
|
||||
concave = fill_concave_pipes(pairs, concave_chainages(chainage, ground), vertices, old_xy)
|
||||
await asyncio.to_thread(
|
||||
save_pipe_points,
|
||||
stored_path,
|
||||
route_signature(vertices),
|
||||
[pipe for _, pipe in pairs],
|
||||
vertices,
|
||||
)
|
||||
return pairs, {"kept": len(old_pipes), "crossing": crossing, "concave": concave}
|
||||
|
||||
|
||||
async def _carry_profile(
|
||||
project_root: Path,
|
||||
relative: str,
|
||||
route_id: int,
|
||||
old_alignment: dict[str, Any],
|
||||
old_xy: list[list[float]],
|
||||
new_xy: list[list[float]],
|
||||
) -> None:
|
||||
"""바뀐 구간만 새 지반 첫 계획선 · 그 밖 사용자 계획고 · 종단곡선 R 은 길이 차만큼 밀어 지킴."""
|
||||
from B05_Profile.B05_Profile_Engine_Grade_Carry import carry_profile_edits, changed_span
|
||||
from B05_Profile.B05_Profile_Repository import update_longitudinal_grade_summary
|
||||
from B05_Profile.B05_Profile_Router import _apply_alignment_edits
|
||||
from common_util.common_util_json import atomic_write_json
|
||||
|
||||
if not old_alignment.get("base_pvi"):
|
||||
return # 옛 선형 없음 — 새 지반 첫 계획선 그대로
|
||||
path = project_root / relative
|
||||
longitudinal = _read_json(path)
|
||||
alignment = longitudinal.get("profile_alignment") or {}
|
||||
span = changed_span([tuple(p) for p in old_xy], [tuple(p) for p in new_xy])
|
||||
if span is None: # 같은 노선 — 옛 계획선 통째
|
||||
base, edits = old_alignment["base_pvi"], old_alignment.get("edits") or {}
|
||||
else:
|
||||
base, edits = carry_profile_edits(
|
||||
old_alignment["base_pvi"], old_alignment.get("edits"), span, alignment["base_pvi"]
|
||||
)
|
||||
alignment["base_pvi"] = base
|
||||
await asyncio.to_thread(atomic_write_json, path, longitudinal)
|
||||
_alignment, summary, _dropped = await asyncio.to_thread(
|
||||
_apply_alignment_edits, project_root, relative, edits
|
||||
)
|
||||
pool = get_db_pool()
|
||||
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()
|
||||
|
||||
|
||||
async def _carry_cross(
|
||||
project_id: UUID,
|
||||
project_root: Path,
|
||||
route_id: int,
|
||||
relative: str,
|
||||
old_long: dict[str, Any],
|
||||
old_designs: list[dict[str, Any]],
|
||||
old_pipes: list,
|
||||
old_structures: list,
|
||||
pairs: list,
|
||||
interval_m: float,
|
||||
standard: dict[str, Any] | None,
|
||||
) -> int:
|
||||
"""측점 열쇠로 옛 횡단 입력을 새 측점에 얹고 전 측점을 새 계획고로 셈 · 버린 수 돌려줌."""
|
||||
from B05_Profile.B05_Profile_Structures_Repository import load_structures
|
||||
from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_designs
|
||||
from B06_Section.B06_Section_Router_Confirm import _rowless_station_chainages
|
||||
from B06_Section.B06_Section_Router_Design import (
|
||||
CARRY_INPUT_KEYS,
|
||||
STRUCTURE_ADJUST_KEYS,
|
||||
compute_default_designs,
|
||||
)
|
||||
|
||||
new_long = _read_json(project_root / relative)
|
||||
_, new_structures = load_structures(str(project_root))
|
||||
old_keys = station_keys(
|
||||
old_long.get("stations") or [], list(enumerate(old_pipes)), old_structures, interval_m
|
||||
)
|
||||
new_keys = station_keys(new_long.get("stations") or [], pairs, new_structures, interval_m)
|
||||
designs = {
|
||||
round(float(row["chainage_m"]), 3): row["design"]
|
||||
for row in old_designs
|
||||
if isinstance(row.get("design"), dict)
|
||||
}
|
||||
inputs, adjusts, dropped = carry_cross_inputs(
|
||||
old_keys, new_keys, designs, CARRY_INPUT_KEYS, STRUCTURE_ADJUST_KEYS
|
||||
)
|
||||
chainages = await asyncio.to_thread(_rowless_station_chainages, project_root, relative, [])
|
||||
entries = await asyncio.to_thread(
|
||||
compute_default_designs, project_root, relative, chainages, standard, inputs
|
||||
)
|
||||
entries = [(at, {**design, **adjusts.get(round(at, 3), {})}) for at, design in entries]
|
||||
pool = get_db_pool()
|
||||
async with pool.acquire() as connection:
|
||||
await connection.begin()
|
||||
try:
|
||||
await merge_cross_section_designs(
|
||||
connection, route_id=route_id, entries=entries, replace=True, project_id=project_id
|
||||
)
|
||||
await connection.commit()
|
||||
except Exception:
|
||||
await connection.rollback()
|
||||
raise
|
||||
return dropped
|
||||
|
||||
|
||||
async def _rollback(project_id: UUID, old_route_id: int, params: dict[str, Any]) -> None:
|
||||
"""새 노선 행(종횡단 함께 지워짐)을 지우고 stage 2 params 를 옛것으로 — 최신 노선 = 옛 노선."""
|
||||
from common_util.common_util_workflow_state import start_stage
|
||||
|
||||
pool = get_db_pool()
|
||||
async with pool.acquire() as connection:
|
||||
await connection.begin()
|
||||
try:
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"DELETE FROM routes WHERE project_id = %s AND id > %s",
|
||||
(str(project_id), old_route_id),
|
||||
)
|
||||
await start_stage(cursor, str(project_id), 2, params)
|
||||
await connection.commit()
|
||||
except Exception:
|
||||
await connection.rollback()
|
||||
logger.exception("노선 저장 되살림(DB) 실패: project_id=%s", project_id)
|
||||
|
||||
|
||||
async def run_route_save_chain(
|
||||
project_id: UUID,
|
||||
project_root: Path,
|
||||
stored_path: str,
|
||||
surface_model_id: int,
|
||||
selection: dict[str, Any],
|
||||
old_route: dict[str, Any],
|
||||
points: list[dict[str, float]],
|
||||
previous_params: dict[str, Any],
|
||||
profile_target: tuple[list[float], list[float]] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""노선 저장 체인 — 성공이면 요약 dict · 실패면 `{"error": 사유}`(파일 · 행은 되살린 뒤)."""
|
||||
import aiomysql
|
||||
|
||||
from B03_FileInput.B03_FileInput_Service_Chain import _log_steps, redesign_solve_request
|
||||
from B05_Profile.B05_Profile_Engine_Sections import _STAGE_SUBDIR
|
||||
from B05_Profile.B05_Profile_Router import (
|
||||
_section_options,
|
||||
confirm_latest_route,
|
||||
solve_route_record,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Structures_Repository import load_structures, structures_file_path
|
||||
from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route
|
||||
from B06_Section.B06_Section_Repository import (
|
||||
get_cross_section_designs,
|
||||
get_latest_section_options,
|
||||
get_longitudinal_section,
|
||||
)
|
||||
from B06_Section.B06_Section_Router import regenerate_route_sections
|
||||
from B06_Section.B06_Section_Router_Confirm import confirm_sections
|
||||
from B06_Section.B06_Section_Router_Design import stored_standard_cross_section
|
||||
from B06_Section.B06_Section_Schema import SectionConfirmRequest
|
||||
from common_util.common_util_drainage_pipes import pipe_points_path, read_pipe_points_file
|
||||
from common_util.common_util_initial_snapshot import clear_designing, mark_designing
|
||||
from common_util.common_util_workflow_state import get_workflow_state
|
||||
|
||||
marks = [("시작", time.perf_counter())]
|
||||
old_route_id = int(old_route["id"])
|
||||
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))
|
||||
old_long_row = await get_longitudinal_section(connection, project_id, old_route_id)
|
||||
old_designs = await get_cross_section_designs(connection, old_route_id)
|
||||
old_section_options = await get_latest_section_options(connection, project_id)
|
||||
stage2 = next((s for s in state.get("stages", []) if int(s.get("stage_no", -1)) == 2), None)
|
||||
params = (stage2 or {}).get("params") or {}
|
||||
relative_old = str((old_long_row or {}).get("longitudinal_file_path") or "")
|
||||
old_long = _read_json(project_root / relative_old) if relative_old else {}
|
||||
old_xy = _xy(project_root, old_route.get("route_data_path"))
|
||||
old_pipes = read_pipe_points_file(pipe_points_path(stored_path))
|
||||
_, old_structures = load_structures(str(project_root))
|
||||
stage_dir = project_root / _STAGE_SUBDIR
|
||||
backup = _Backup(
|
||||
project_root,
|
||||
[
|
||||
project_root / str(old_route.get("route_data_path") or "B05_Profile/route/x"),
|
||||
pipe_points_path(stored_path),
|
||||
Path(structures_file_path(str(project_root))),
|
||||
],
|
||||
[stage_dir / "longitudinal", stage_dir / "cross_sections"],
|
||||
)
|
||||
mark_designing(project_root)
|
||||
try:
|
||||
# 구조물 = 땅 위 자리 따라감(옛 노선 파일이 아직 옛 것일 때 — 좌표 없는 옛 자료 채움)
|
||||
try:
|
||||
structures = await asyncio.to_thread(
|
||||
follow_structures_to_route, project_root, old_route.get("route_data_path"), points
|
||||
)
|
||||
except Exception: # noqa: BLE001 — 따라감 실패 = 옛 누가거리 그대로(비치명)
|
||||
logger.exception("구조물 따라감 실패(옛 누가거리 유지): project_id=%s", project_id)
|
||||
structures = {"moved": 0, "far": 0}
|
||||
request = redesign_solve_request(params, selection, surface_model_id)
|
||||
solved: Any = await solve_route_record(project_id, request, generate_sections=False)
|
||||
if isinstance(solved, JSONResponse):
|
||||
raise _Fail(f"노선 풀기 실패({solved.status_code}): {_reason(solved)}")
|
||||
new_route_id = int(solved.route_id)
|
||||
confirmed = await confirm_latest_route(project_id, None, mark_stage_complete=False)
|
||||
if isinstance(confirmed, JSONResponse):
|
||||
raise _Fail(f"노선 확정 실패({confirmed.status_code})")
|
||||
new_xy = _xy(project_root, solved.route_data_path)
|
||||
marks.append(("노선 풀기 · 구조물 따라감", time.perf_counter()))
|
||||
|
||||
pairs, pipes = await _place_pipes(
|
||||
project_id, project_root, stored_path, selection, old_pipes, old_xy, points
|
||||
)
|
||||
marks.append(("관 따라감 · 교차점 · 오목 보충", time.perf_counter()))
|
||||
|
||||
seed = asdict(_section_options(request, old_section_options))
|
||||
seed.pop("extra_stations", None)
|
||||
made = await regenerate_route_sections(
|
||||
project_id, new_route_id, float(seed["cross_half_width_m"]), stored_options=seed
|
||||
)
|
||||
if isinstance(made, JSONResponse):
|
||||
raise _Fail(f"종횡단 생성 실패({made.status_code}): {_reason(made)}")
|
||||
async with pool.acquire() as connection:
|
||||
new_long_row = await get_longitudinal_section(connection, project_id, new_route_id)
|
||||
relative = str((new_long_row or {}).get("longitudinal_file_path") or "")
|
||||
marks.append(("종횡단 한 번", time.perf_counter()))
|
||||
|
||||
if profile_target is not None:
|
||||
from B05_Profile.B05_Profile_Router_RouteInitial import apply_profile_target
|
||||
|
||||
await apply_profile_target(project_id, new_route_id, profile_target)
|
||||
else:
|
||||
await _carry_profile(
|
||||
project_root,
|
||||
relative,
|
||||
new_route_id,
|
||||
old_long.get("profile_alignment") or {},
|
||||
old_xy,
|
||||
new_xy,
|
||||
)
|
||||
marks.append(("계획고 옮김", time.perf_counter()))
|
||||
|
||||
standard = stored_standard_cross_section(old_long_row)
|
||||
dropped = await _carry_cross(
|
||||
project_id,
|
||||
project_root,
|
||||
new_route_id,
|
||||
relative,
|
||||
old_long,
|
||||
old_designs,
|
||||
old_pipes,
|
||||
old_structures,
|
||||
pairs,
|
||||
float(seed["station_interval_m"]),
|
||||
standard,
|
||||
)
|
||||
marks.append(("횡단 이월(측점 열쇠)", time.perf_counter()))
|
||||
|
||||
done = await confirm_sections(
|
||||
project_id,
|
||||
new_route_id,
|
||||
SectionConfirmRequest(standard_cross_section=standard) if standard else None,
|
||||
mark_stage_complete=False,
|
||||
)
|
||||
if isinstance(done, JSONResponse):
|
||||
raise _Fail(f"종횡단 확정 실패({done.status_code})")
|
||||
marks.append(("B06 확정 · 서버 재계산 · 3D", time.perf_counter()))
|
||||
except Exception as exc: # noqa: BLE001 — 어느 단계든 옛 노선으로 되살림
|
||||
if not isinstance(exc, _Fail):
|
||||
logger.exception("노선 저장 체인 실패: project_id=%s", project_id)
|
||||
await asyncio.to_thread(backup.restore)
|
||||
await _rollback(project_id, old_route_id, previous_params)
|
||||
return {"error": str(exc) if isinstance(exc, _Fail) else f"노선 저장 중 오류: {exc}"}
|
||||
finally:
|
||||
backup.drop()
|
||||
clear_designing(project_root)
|
||||
_log_steps("재확정 체인(노선 저장)", marks)
|
||||
return {
|
||||
"route_id": new_route_id,
|
||||
"total_length_m": float(solved.total_length_m or 0.0),
|
||||
"pipes": pipes,
|
||||
"structures": structures,
|
||||
"cross_dropped": dropped,
|
||||
"drainage_stale": True,
|
||||
"seconds": round(marks[-1][1] - marks[0][1], 2),
|
||||
}
|
||||
@@ -0,0 +1,224 @@
|
||||
"""노선 저장 길(346-3) 순수 부품 — 관 따라감 · 새 교차점 기본 관 · 오목 지점 보충 · 측점 열쇠 ·
|
||||
횡단 이월.
|
||||
|
||||
DB · 파일 없이 값만 받아 값을 돌려준다(시험이 그대로 부름). 차례 · 저장 · 되살림은
|
||||
`B05_Profile_Route_Save.py`.
|
||||
|
||||
- 관 = 좌표가 정본 — 새 노선에 투영해 누가거리만 다시(`project_pipe_points` · 정수 m). 지우지 않음
|
||||
(사용자 관 편집 지킴).
|
||||
- 새 세류선 교차점(옛 노선 위에 없던 자리)에 관이 없으면 기본 관 하나(`_base_pipes` 결과를 받음).
|
||||
- 관 최대 간격 보충 = 옛 노선에서 벗어난 자리의 **오목 지점**(지반 종단 낮은 곳)에만 — 격자 없음.
|
||||
- 측점 열쇠(계약 `design_pipeline_types.ts` 머리) = 일반 `grid:k`(누가거리 ÷ 측점 간격 반올림) ·
|
||||
관 `pipe:<id>` · 구조물 `structure:<id>`(구간 시작 · 끝은 `:start` · `:end`) · 시종점
|
||||
`end:BP` · `end:EP`. 관 id = 저장 파일 차례(노선 저장 한 번 안에서 옛 · 새가 같음).
|
||||
- 횡단 이월 = 같은 열쇠 측점에 `CARRY_INPUT_KEYS` 일곱 · 관 · 구조물 열쇠면
|
||||
`STRUCTURE_ADJUST_KEYS` 도(구조물 조정은 구조물 따라) · 열쇠가 사라진 측점 수 =
|
||||
「노선 밖 고침」.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import math
|
||||
from typing import Any, Sequence
|
||||
|
||||
import numpy as np
|
||||
from shapely.geometry import LineString, Point
|
||||
|
||||
from common_util.common_util_drainage_pipes import (
|
||||
PIPE_SOURCE_SPACING,
|
||||
PIPE_SOURCE_STREAM,
|
||||
PipePoint,
|
||||
project_pipe_points,
|
||||
)
|
||||
from common_util.common_util_route_geometry import RouteVertex
|
||||
from common_util.common_util_structure_station import plan_stations, snap_structure_station
|
||||
from config.config_system import DRAINAGE_PIPE_MAX_SPACING_M, DRAINAGE_PIPE_MIN_SPACING_M
|
||||
|
||||
# 옛 노선에서 이만큼(m) 넘게 떨어진 자리 = 노선이 바뀐 자리(교차점 · 보충 대상).
|
||||
MOVED_AWAY_M = 0.5
|
||||
# 오목 지점을 찾을 지반 표본 간격(m) · 고름 창(표본 수) — DTM 잔물결을 오목으로 안 읽게.
|
||||
# ponytail: 이동 평균 + 국소 최저 · 계곡이 얕아 놓치면 창을 줄이거나 흐름 강도와 섞을 것.
|
||||
CONCAVE_STEP_M = 2.0
|
||||
CONCAVE_WINDOW = 5
|
||||
_SAME_M = 0.01
|
||||
|
||||
|
||||
def vertices_of(xy: Sequence[Sequence[float]]) -> list[RouteVertex]:
|
||||
"""평면 점열 → 누가거리 붙은 정점(z 0)."""
|
||||
out: list[RouteVertex] = []
|
||||
walked = 0.0
|
||||
for index, point in enumerate(xy):
|
||||
x, y = float(point[0]), float(point[1])
|
||||
if index:
|
||||
walked += math.hypot(x - out[-1].x, y - out[-1].y)
|
||||
out.append(RouteVertex(x=x, y=y, z=0.0, chainage_m=walked))
|
||||
return out
|
||||
|
||||
|
||||
def follow_pipes(
|
||||
old_pipes: list[PipePoint], vertices: list[RouteVertex]
|
||||
) -> list[tuple[int, PipePoint]]:
|
||||
"""관이 새 노선 좌표를 따라감 → [(관 id = 옛 차례, 새 관)] 누가거리 순 · 옛 목록 그대로."""
|
||||
pairs = [(index, copy.deepcopy(pipe)) for index, pipe in enumerate(old_pipes)]
|
||||
for _, pipe in pairs:
|
||||
project_pipe_points([pipe], vertices)
|
||||
pairs.sort(key=lambda pair: pair[1].chainage_m)
|
||||
return pairs
|
||||
|
||||
|
||||
def _moved_away(old_line: LineString | None, x: float, y: float) -> bool:
|
||||
return old_line is None or old_line.distance(Point(x, y)) > MOVED_AWAY_M
|
||||
|
||||
|
||||
def add_crossing_pipes(
|
||||
pairs: list[tuple[int, PipePoint]],
|
||||
crossings: Sequence[Any],
|
||||
old_xy: Sequence[Sequence[float]] | None,
|
||||
total_m: float,
|
||||
) -> int:
|
||||
"""옛 노선에 없던 세류선 교차점에 관이 없으면 기본 관을 더함 · 더한 수."""
|
||||
old_line = LineString(old_xy) if old_xy and len(old_xy) >= 2 else None
|
||||
added = 0
|
||||
for crossing in crossings:
|
||||
if not _moved_away(old_line, crossing.x, crossing.y):
|
||||
continue # 옛 교차점 — 사용자가 지운 관을 되살리지 않음
|
||||
at = snap_structure_station(float(crossing.chainage_m), [total_m])
|
||||
if any(abs(pipe.chainage_m - at) < DRAINAGE_PIPE_MIN_SPACING_M for _, pipe in pairs):
|
||||
continue
|
||||
pairs.append(
|
||||
(-1, PipePoint(chainage_m=at, source=PIPE_SOURCE_STREAM, x=crossing.x, y=crossing.y))
|
||||
)
|
||||
added += 1
|
||||
pairs.sort(key=lambda pair: pair[1].chainage_m)
|
||||
return added
|
||||
|
||||
|
||||
def concave_chainages(chainage: np.ndarray, ground: np.ndarray) -> list[tuple[float, float]]:
|
||||
"""지반 종단의 오목 지점(이동 평균 국소 최저) → [(누가거리, 고른 지반고)]."""
|
||||
if ground.size < 3:
|
||||
return []
|
||||
kernel = np.ones(CONCAVE_WINDOW) / CONCAVE_WINDOW
|
||||
pad = CONCAVE_WINDOW // 2
|
||||
smooth = np.convolve(np.pad(ground, pad, mode="edge"), kernel, mode="valid")
|
||||
return [
|
||||
(float(chainage[i]), float(smooth[i]))
|
||||
for i in range(1, smooth.size - 1)
|
||||
if smooth[i] < smooth[i - 1] and smooth[i] <= smooth[i + 1]
|
||||
]
|
||||
|
||||
|
||||
def fill_concave_pipes(
|
||||
pairs: list[tuple[int, PipePoint]],
|
||||
concave: list[tuple[float, float]],
|
||||
vertices: list[RouteVertex],
|
||||
old_xy: Sequence[Sequence[float]] | None,
|
||||
) -> int:
|
||||
"""관 사이(시종점 포함)가 최대 간격을 넘으면 옛 노선 밖 오목 지점 중 가장 낮은 곳에 관."""
|
||||
line = LineString([(vertex.x, vertex.y) for vertex in vertices])
|
||||
old_line = LineString(old_xy) if old_xy and len(old_xy) >= 2 else None
|
||||
total = float(line.length)
|
||||
added = 0
|
||||
while True:
|
||||
edges = [0.0, *(pipe.chainage_m for _, pipe in pairs), total]
|
||||
best: tuple[float, float] | None = None
|
||||
for start, end in zip(edges, edges[1:]):
|
||||
if end - start <= DRAINAGE_PIPE_MAX_SPACING_M:
|
||||
continue
|
||||
for at, z in concave:
|
||||
if not (
|
||||
start + DRAINAGE_PIPE_MIN_SPACING_M < at < end - DRAINAGE_PIPE_MIN_SPACING_M
|
||||
):
|
||||
continue
|
||||
point = line.interpolate(at)
|
||||
if _moved_away(old_line, point.x, point.y) and (best is None or z < best[1]):
|
||||
best = (at, z)
|
||||
if best is None:
|
||||
return added
|
||||
at = snap_structure_station(best[0], [total])
|
||||
point = line.interpolate(at)
|
||||
pairs.append(
|
||||
(-1, PipePoint(chainage_m=at, source=PIPE_SOURCE_SPACING, x=point.x, y=point.y))
|
||||
)
|
||||
pairs.sort(key=lambda pair: pair[1].chainage_m)
|
||||
concave = [item for item in concave if abs(item[0] - best[0]) > _SAME_M]
|
||||
added += 1
|
||||
|
||||
|
||||
def _facility_keys(
|
||||
pipes: Sequence[tuple[int, PipePoint]], structures: Sequence[Any]
|
||||
) -> dict[float, str]:
|
||||
"""측점 자리(0.01 m) → 관 · 구조물 열쇠. 같은 자리면 관이 먼저(`plan_stations` 이름 규칙)."""
|
||||
keys: dict[float, str] = {}
|
||||
|
||||
def put(marks: list[dict[str, object]], head: str) -> None:
|
||||
for mark in marks:
|
||||
role = str(mark.get("role") or "anchor")
|
||||
keys.setdefault(
|
||||
round(float(mark["chainage_m"]), 2), head if role == "anchor" else f"{head}:{role}"
|
||||
)
|
||||
|
||||
for pipe_id, pipe in pipes:
|
||||
if pipe_id >= 0:
|
||||
put(plan_stations([pipe], []), f"pipe:{pipe_id}")
|
||||
for item in structures:
|
||||
if getattr(item, "structure_id", None):
|
||||
put(plan_stations([], [item]), f"structure:{item.structure_id}")
|
||||
return keys
|
||||
|
||||
|
||||
def station_keys(
|
||||
stations: Sequence[dict[str, Any]],
|
||||
pipes: Sequence[tuple[int, PipePoint]],
|
||||
structures: Sequence[Any],
|
||||
interval_m: float,
|
||||
) -> dict[str, float]:
|
||||
"""측점 열쇠 → 누가거리. 관 · 구조물 자리 > 시종점 > 일반 격자(간격 배수인 것만)."""
|
||||
facility = _facility_keys(pipes, structures)
|
||||
values = [float(row["chainage_m"]) for row in stations if row.get("chainage_m") is not None]
|
||||
out: dict[str, float] = {}
|
||||
for index, chainage in enumerate(values):
|
||||
key = facility.get(round(chainage, 2))
|
||||
if key is None and index == 0:
|
||||
key = "end:BP"
|
||||
elif key is None and index == len(values) - 1:
|
||||
key = "end:EP"
|
||||
elif key is None and interval_m > 0:
|
||||
k = round(chainage / interval_m)
|
||||
key = f"grid:{k}" if abs(chainage - k * interval_m) < _SAME_M else None
|
||||
if key is not None:
|
||||
out.setdefault(key, chainage)
|
||||
return out
|
||||
|
||||
|
||||
def carry_cross_inputs(
|
||||
old_keys: dict[str, float],
|
||||
new_keys: dict[str, float],
|
||||
old_designs: dict[float, dict[str, Any]],
|
||||
input_keys: Sequence[str],
|
||||
adjust_keys: Sequence[str],
|
||||
) -> tuple[dict[float, dict[str, Any]], dict[float, dict[str, Any]], int]:
|
||||
"""같은 열쇠 측점으로 횡단 입력을 넘김 → (입력{새 누가거리 0.001: 값}, 구조물 조정{…}, 버린 수).
|
||||
|
||||
`old_designs` = 옛 노선 행 {누가거리 0.001: design}. 버린 수 = 설계가 있던 옛 측점 중 새 노선에
|
||||
같은 열쇠가 없는 것(「노선 밖 고침」).
|
||||
"""
|
||||
inputs: dict[float, dict[str, Any]] = {}
|
||||
adjusts: dict[float, dict[str, Any]] = {}
|
||||
dropped = 0
|
||||
for key, old_at in old_keys.items():
|
||||
design = old_designs.get(round(old_at, 3))
|
||||
if not isinstance(design, dict):
|
||||
continue
|
||||
new_at = new_keys.get(key)
|
||||
if new_at is None:
|
||||
dropped += 1
|
||||
continue
|
||||
carried = {name: design[name] for name in input_keys if design.get(name) is not None}
|
||||
if carried:
|
||||
inputs[round(new_at, 3)] = carried
|
||||
if key.startswith(("pipe:", "structure:")):
|
||||
adjust = {name: design[name] for name in adjust_keys if design.get(name) is not None}
|
||||
if adjust:
|
||||
adjusts[round(new_at, 3)] = adjust
|
||||
return inputs, adjusts, dropped
|
||||
@@ -14,17 +14,13 @@
|
||||
노선 초기화는 수정본 파일을 지우는 것이다 — 그러면 초기 폴리라인을 읽으므로 「초기본을
|
||||
수정본으로 복사」와 결과가 같다.
|
||||
|
||||
[확인]을 눌렀을 때만 계산이 돈다. 재계산은 **초기 업로드 체인의 로직을 그대로 재사용**한다
|
||||
(2026-09-06 사용자 제안) — `run_redesign_chain`이 배수유역 다시 분석 → 기본 관 저장 → 관
|
||||
정착 계획선 재산출 → 종횡단 재생성 → 옛 측점 설계를 누가거리로 이월 → B05·B06 확정까지 한
|
||||
줄로 돈다. 여기서 하는 일은 그 앞에 **노선을 갈아 끼우는 것**뿐이다.
|
||||
|
||||
시설 처리(2026-09-06 사용자 확정) — 횡단배수 지점은 노선 자리에 따라 생기고 없어지므로
|
||||
저장분을 버리고 **새로 계산한 값**을 쓴다. 그 밖의 구조물은 **옛 측점값을 그대로** 이어받는다
|
||||
(`structures.json`은 프로젝트 정본이라 손대지 않으면 그대로 남는다).
|
||||
노선 저장(346-3) — 여기서 노선을 갈아 끼운 뒤 `B05_Profile_Route_Save.run_route_save_chain` 이
|
||||
풀기 → 관 · 구조물 좌표 따라감 → 종횡단 한 번 → 계획고 옮김 → 횡단 이월(측점 열쇠) → B06 확정까지
|
||||
돈다. 배수유역은 다시 안 셈(낡음) · 관 정본은 안 지움 · 초기값은 추천 [적용](`as_initial`)일 때만.
|
||||
요청 · 응답 · 오류 모양 = 그 모듈 머리.
|
||||
|
||||
GET /api/projects/{id}/route/plan → 예상노선·계획노선 정점(사업지 좌표계)
|
||||
POST /api/projects/{id}/route/replan → 고친 계획노선으로 갈아 끼우고 재계산
|
||||
POST /api/projects/{id}/route/replan → 노선 초안 저장(노선 저장 체인)
|
||||
POST /api/projects/{id}/route/replan/reset → 계획노선을 예상노선으로 되돌리고 재계산
|
||||
POST /api/projects/{id}/route/initial → 초기 계획노선 갈래 하나를 만들어 돌려줌(추천 저장본만 씀)
|
||||
"""
|
||||
@@ -45,6 +41,7 @@ from fastapi.responses import JSONResponse
|
||||
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 B05_Profile.B05_Profile_Engine_Sections import load_station_snap
|
||||
from B05_Profile.B05_Profile_Route_Save import RouteSaveRequest
|
||||
from B05_Profile.B05_Profile_Schema import ( # noqa: F401 — 부르는 곳이 `replan.` 으로 씀
|
||||
RouteInitialCriteria,
|
||||
RouteInitialRequest,
|
||||
@@ -340,22 +337,19 @@ async def _recompute(
|
||||
from B03_FileInput.B03_FileInput_Service_Chain import (
|
||||
_log_steps,
|
||||
_planned_route_points_in_project_crs,
|
||||
run_redesign_chain,
|
||||
)
|
||||
from B04_PreProcess.B04_PreProcess_Service import (
|
||||
find_surface_model_for_selection,
|
||||
switch_design_terrain,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Repository import get_latest_route
|
||||
from common_util.common_util_drainage_pipes import clear_pipe_points
|
||||
from B05_Profile.B05_Profile_Route_Save import run_route_save_chain
|
||||
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
|
||||
|
||||
marks = [("시작", time.perf_counter())]
|
||||
|
||||
# 횡단배수 지점은 새 노선에서 새로 계산한다 — 저장분을 남기면 옛 자리가 살아난다.
|
||||
await asyncio.to_thread(clear_pipe_points, stored_path)
|
||||
|
||||
# 관 정본은 지우지 않음(346-3) — 사용자 관 편집을 지키고 좌표로 새 노선을 따라감(체인 안).
|
||||
pool = get_db_pool()
|
||||
async with pool.acquire() as connection:
|
||||
await connection.begin() # 고른 지형 = 설계 지형 — 뒤 단계가 모두 이 확정값을 읽음
|
||||
@@ -408,7 +402,8 @@ async def _recompute(
|
||||
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 {})
|
||||
previous_params = dict((stage2 or {}).get("params") or {}) # 실패 때 되살림
|
||||
params = dict(previous_params)
|
||||
params["points"] = {
|
||||
"bp": points[0],
|
||||
"ep": points[-1],
|
||||
@@ -418,23 +413,28 @@ async def _recompute(
|
||||
await start_stage(cursor, str(project_id), 2, params)
|
||||
await connection.commit()
|
||||
|
||||
marks.append(("재계산 준비(관 삭제·노선 읽기·stage 기록)", time.perf_counter()))
|
||||
|
||||
# 구조물은 땅 위 자리(좌표)를 따라감(344-2) — 옛 노선 경로 파일은 체인 전이라 아직 옛 것
|
||||
from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route
|
||||
|
||||
try:
|
||||
old_path = (latest or {}).get("route_data_path")
|
||||
await asyncio.to_thread(follow_structures_to_route, project_root, old_path, points)
|
||||
except Exception: # 따라감 실패 = 옛 누가거리 그대로(지금까지와 같음)
|
||||
logger.exception("구조물 따라감 실패(옛 누가거리 유지): project_id=%s", project_id)
|
||||
marks.append(("재계산 준비(노선 읽기·stage 기록)", time.perf_counter()))
|
||||
|
||||
target = None
|
||||
if balanced is not None:
|
||||
from B05_Profile.B05_Profile_Router_RouteInitial import balanced_target
|
||||
|
||||
target = await balanced_target(project_id, project_root, points, balanced.criteria)
|
||||
failure = await run_redesign_chain(project_id, int(surface_model_id), selection, target)
|
||||
if not latest:
|
||||
return {"error": "저장된 노선이 없어 노선을 저장할 수 없습니다(자동설계 먼저)."}
|
||||
# 노선 저장 체인(346-3) — 배수유역 뗌 · 관 · 구조물 따라감 · 종횡단 한 번 · 계획고 옮김
|
||||
saved = await run_route_save_chain(
|
||||
project_id,
|
||||
project_root,
|
||||
stored_path,
|
||||
int(surface_model_id),
|
||||
selection,
|
||||
latest,
|
||||
points,
|
||||
previous_params,
|
||||
target,
|
||||
)
|
||||
failure = saved.get("error")
|
||||
marks.append(("재확정 체인", time.perf_counter()))
|
||||
_log_steps("노선 [확인] 재계산", marks)
|
||||
if failure and switched: # 노선과 함께 설계 지형도 되돌림
|
||||
@@ -448,14 +448,7 @@ async def _recompute(
|
||||
# 부르는 쪽이 계획노선을 되돌리도록 사유를 올려 보낸다.
|
||||
return {"error": failure}
|
||||
|
||||
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),
|
||||
}
|
||||
return {**saved, "vertex_count": len(points)}
|
||||
|
||||
|
||||
@router.get("/{project_id}/route/plan", response_model=None)
|
||||
@@ -530,10 +523,16 @@ async def read_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse:
|
||||
|
||||
|
||||
@router.post("/{project_id}/route/replan", response_model=None)
|
||||
async def replan_route(
|
||||
project_id: UUID, request: RouteReplanRequest
|
||||
) -> dict[str, Any] | JSONResponse:
|
||||
"""고친 계획노선으로 갈아 끼우고 배수유역부터 다시 계산한다(모달 [확인])."""
|
||||
async def replan_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, Any] | JSONResponse:
|
||||
"""노선 초안을 정본으로 저장한다(346-3 · 요청 · 응답 모양 = `B05_Profile_Route_Save` 머리).
|
||||
|
||||
배수유역은 다시 안 셈(낡음 표시) · 초기값은 추천 [적용](`as_initial`)일 때만 다시 뜬다.
|
||||
"""
|
||||
as_initial = bool(getattr(body, "as_initial", False)) # 옛 요청 꼴(파이썬 호출)도 받음
|
||||
try:
|
||||
request = body.resolved() if isinstance(body, RouteSaveRequest) else body
|
||||
except ValueError as error:
|
||||
return JSONResponse(status_code=400, content={"status": "error", "message": str(error)})
|
||||
if len(request.vertices) < 2:
|
||||
return JSONResponse(
|
||||
status_code=400,
|
||||
@@ -616,15 +615,18 @@ async def replan_route(
|
||||
if choice
|
||||
else await asyncio.to_thread(read_initial_choice, project_root)
|
||||
)
|
||||
# 확정 노선 = 새 초기값(PLAN 74-4) — [저장] 없이 이 결과를 [초기화]가 돌아갈 자리로 뜬다.
|
||||
from B05_Profile.B05_Profile_Router_RouteInitial import refresh_initial_snapshot
|
||||
# 초기값 = 자동설계 체인 직후 + 추천 노선 [적용] 때만(346-3 · 사용자 확정 10-16) — 손으로 고친
|
||||
# 노선 저장은 [초기화]가 돌아갈 자리를 안 바꿈.
|
||||
if as_initial:
|
||||
from B05_Profile.B05_Profile_Router_RouteInitial import refresh_initial_snapshot
|
||||
|
||||
await refresh_initial_snapshot(project_id, project_root, result.get("route_id"))
|
||||
await refresh_initial_snapshot(project_id, project_root, result.get("route_id"))
|
||||
return {
|
||||
"status": "success",
|
||||
"project_id": str(project_id),
|
||||
**result,
|
||||
"current_initial": current,
|
||||
"initial_refreshed": as_initial,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -337,7 +337,7 @@ async def balanced_profile_of(
|
||||
async def refresh_initial_snapshot(
|
||||
project_id: UUID, project_root: Path, route_id: int | None
|
||||
) -> None:
|
||||
"""노선 [확인] 결과로 초기값(initial_snapshot)을 다시 뜬다(PLAN 74-4) — 실패는 비치명적."""
|
||||
"""추천 노선 [적용](as_initial) 결과로 초기값을 다시 뜬다(346-3) — 실패는 비치명적."""
|
||||
from common_util.common_util_initial_snapshot import (
|
||||
discard_initial_snapshot,
|
||||
save_initial_snapshot,
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""346-3 노선 저장 실경로 시간 — 검증 프로젝트의 지금 노드 · 곡선 그대로 저장 한 번(수동 실행).
|
||||
|
||||
같은 노선이라 값은 그대로(새 노선 행 번호만 바뀜) · 초기값은 안 건드림(`as_initial` 끔).
|
||||
쓰는 법: `./venv/Scripts/python.exe resources/tester/run_346_3_route_save_timing.py <project_id>`
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from uuid import UUID
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from config.config_db import close_db_pool, init_db_pool # noqa: E402
|
||||
|
||||
|
||||
async def main(project_id: UUID) -> None:
|
||||
from B05_Profile import B05_Profile_Router_Replan as replan
|
||||
from B05_Profile.B05_Profile_Route_Save import RouteSaveRequest
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
await init_db_pool()
|
||||
try:
|
||||
plan = await replan.read_route_plan(project_id)
|
||||
nodes = plan["nodes"]
|
||||
radii: dict[int, float] = {}
|
||||
for curve in plan["curves"]:
|
||||
if curve.get("node_first") == curve.get("node_last"):
|
||||
radii[int(curve["node_first"])] = float(curve["radius_m"])
|
||||
body = RouteSaveRequest(
|
||||
vertices=[
|
||||
{
|
||||
"x": node["x"],
|
||||
"y": node["y"],
|
||||
"curve": index in radii,
|
||||
"radius_m": radii.get(index),
|
||||
}
|
||||
for index, node in enumerate(nodes)
|
||||
],
|
||||
station_snap=bool(plan.get("station_snap")),
|
||||
)
|
||||
started = time.perf_counter()
|
||||
result = await replan.replan_route(project_id, body)
|
||||
seconds = time.perf_counter() - started
|
||||
body_text = getattr(result, "body", None)
|
||||
print(json.dumps(json.loads(body_text) if body_text else result, ensure_ascii=False))
|
||||
print(f"[시간] 노선 저장 전체 {seconds:.1f}s")
|
||||
finally:
|
||||
await close_db_pool()
|
||||
|
||||
|
||||
asyncio.run(main(UUID(sys.argv[1])))
|
||||
@@ -18,7 +18,7 @@ import numpy as np
|
||||
import B03_FileInput.B03_FileInput_Service_Chain as chain
|
||||
import B04_PreProcess.B04_PreProcess_Service as service
|
||||
import B05_Profile.B05_Profile_Repository as route_repo
|
||||
import common_util.common_util_drainage_pipes as drainage_pipes
|
||||
import B05_Profile.B05_Profile_Route_Save as route_save
|
||||
import common_util.common_util_workflow_state as workflow_state
|
||||
from B05_Profile import B05_Profile_Router_Replan as replan
|
||||
from B06_Section import B06_Section_Router_Stations as stations
|
||||
@@ -212,16 +212,15 @@ def _patch_recompute(monkeypatch, tmp_path: Path, db: _Db, failure: str | None):
|
||||
async def start(*_args):
|
||||
return None
|
||||
|
||||
async def run_chain(_project_id, model_id, selection, _target=None):
|
||||
async def run_chain(_project_id, _root, _stored, model_id, selection, *_rest):
|
||||
seen["model_id"], seen["selection"] = model_id, dict(selection)
|
||||
return failure
|
||||
return {"error": failure} if failure else {"route_id": 238}
|
||||
|
||||
points = [{"x": 0.0, "y": 0.0}, {"x": 5.0, "y": 5.0}]
|
||||
monkeypatch.setattr(replan, "get_db_pool", lambda: _Pool())
|
||||
monkeypatch.setattr(drainage_pipes, "clear_pipe_points", lambda _s: None)
|
||||
monkeypatch.setattr(route_repo, "get_latest_route", latest)
|
||||
monkeypatch.setattr(chain, "_planned_route_points_in_project_crs", lambda *_a: points)
|
||||
monkeypatch.setattr(chain, "run_redesign_chain", run_chain)
|
||||
monkeypatch.setattr(route_save, "run_route_save_chain", run_chain)
|
||||
monkeypatch.setattr(workflow_state, "get_workflow_state", state)
|
||||
monkeypatch.setattr(workflow_state, "start_stage", start)
|
||||
return seen
|
||||
|
||||
@@ -0,0 +1,541 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""346-3 노선 저장 서버 길 — 차례 · 관 지킴 · 새 교차점 기본 관 · 오목 보충 · 이월 열쇠 ·
|
||||
초기값 규칙 ·
|
||||
배수유역 낡음 · 실패 때 옛 노선 되살림.
|
||||
|
||||
순수 부품(`B05_Profile_Route_Save_Parts`)은 값으로 · 체인(`run_route_save_chain`)과 API
|
||||
(`replan_route`)는 DB · 엔진을 가짜로 바꿔 차례 · 되살림 · 초기값만 잰다.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from uuid import uuid4
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from fastapi.responses import JSONResponse # noqa: E402
|
||||
|
||||
import B03_FileInput.B03_FileInput_Service_Chain as chain # noqa: E402
|
||||
import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed # noqa: E402
|
||||
import B05_Profile.B05_Profile_Route_Save as save # noqa: E402
|
||||
import B05_Profile.B05_Profile_Router as b05_router # noqa: E402
|
||||
import B05_Profile.B05_Profile_Router_Replan as replan # noqa: E402
|
||||
import B05_Profile.B05_Profile_Router_RouteInitial as route_initial # noqa: E402
|
||||
import B05_Profile.B05_Profile_Structures_Router as structures_router # noqa: E402
|
||||
import B06_Section.B06_Section_Repository as b06_repo # noqa: E402
|
||||
import B06_Section.B06_Section_Router as b06_router # noqa: E402
|
||||
import B06_Section.B06_Section_Router_Confirm as b06_confirm # noqa: E402
|
||||
import common_util.common_util_drainage_pipes as drainage_pipes # noqa: E402
|
||||
import common_util.common_util_initial_snapshot as snapshot # noqa: E402
|
||||
import common_util.common_util_workflow_state as workflow # noqa: E402
|
||||
from B05_Profile.B05_Profile_Route_Save_Parts import ( # noqa: E402
|
||||
add_crossing_pipes,
|
||||
carry_cross_inputs,
|
||||
concave_chainages,
|
||||
fill_concave_pipes,
|
||||
follow_pipes,
|
||||
station_keys,
|
||||
vertices_of,
|
||||
)
|
||||
from common_util.common_util_drainage_pipes import ( # noqa: E402
|
||||
PIPE_SOURCE_SPACING,
|
||||
PIPE_SOURCE_STREAM,
|
||||
PipePoint,
|
||||
analysis_route_signature,
|
||||
route_signature,
|
||||
)
|
||||
|
||||
OLD = [(0.0, 0.0), (100.0, 0.0), (200.0, 0.0), (300.0, 0.0), (400.0, 0.0)]
|
||||
# 새 노선 = 100 ~ 300 m 를 북으로 휘게(앞 · 뒤는 옛 노선 위)
|
||||
NEW = [(0.0, 0.0), (100.0, 0.0), (200.0, 60.0), (300.0, 0.0), (400.0, 0.0)]
|
||||
|
||||
|
||||
# ── 관 따라감 · 교차점 · 오목 보충 ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_관은_지우지_않고_좌표를_따라감():
|
||||
old = [
|
||||
PipePoint(chainage_m=50.0, x=50.0, y=0.0, options={"pipe_kind": "흄관"}),
|
||||
PipePoint(chainage_m=350.0, x=350.0, y=0.0, start_m=345.0, end_m=355.0),
|
||||
]
|
||||
pairs = follow_pipes(old, vertices_of(NEW))
|
||||
assert [pipe_id for pipe_id, _ in pairs] == [0, 1]
|
||||
assert pairs[0][1].options == {"pipe_kind": "흄관"} # 사용자 관 속성 그대로
|
||||
shift = pairs[1][1].chainage_m - 350.0 # 새 노선이 길어진 만큼 뒤 관이 밀림
|
||||
assert shift > 30 and pairs[1][1].start_m == pytest.approx(345.0 + shift)
|
||||
assert old[1].chainage_m == 350.0 # 옛 목록은 안 고침
|
||||
|
||||
|
||||
def test_새_교차점에만_기본_관():
|
||||
pairs = follow_pipes([PipePoint(chainage_m=50.0, x=50.0, y=0.0)], vertices_of(NEW))
|
||||
crossings = [
|
||||
SimpleNamespace(chainage_m=20.0, x=20.0, y=0.0), # 옛 노선 위 — 사용자가 지운 관 안 되살림
|
||||
SimpleNamespace(chainage_m=150.4, x=150.0, y=30.0), # 새 자리 · 관 없음 → 더함
|
||||
SimpleNamespace(chainage_m=52.0, x=52.0, y=0.0), # 옛 자리
|
||||
]
|
||||
added = add_crossing_pipes(pairs, crossings, OLD, 456.0)
|
||||
assert added == 1
|
||||
new = [pipe for pipe_id, pipe in pairs if pipe_id < 0]
|
||||
assert new[0].chainage_m == 150.0 and new[0].source == PIPE_SOURCE_STREAM
|
||||
# 같은 자리 다시 = 이미 관이 있음
|
||||
assert add_crossing_pipes(pairs, crossings, OLD, 456.0) == 0
|
||||
|
||||
|
||||
def test_오목_지점_보충은_새_자리의_오목만():
|
||||
vertices = vertices_of(NEW)
|
||||
total = vertices[-1].chainage_m
|
||||
chainage = np.arange(0.0, total, 2.0)
|
||||
ground = 100.0 + np.abs(chainage - 220.0) * 0.05 # 220 m 에 골(새 자리)
|
||||
ground[chainage < 60] = (
|
||||
100.0 + np.abs(chainage[chainage < 60] - 30.0) * 0.05
|
||||
) # 30 m 골(옛 자리)
|
||||
concave = concave_chainages(chainage, ground)
|
||||
assert any(abs(at - 220.0) < 3 for at, _ in concave)
|
||||
pairs: list = []
|
||||
added = fill_concave_pipes(pairs, concave, vertices, OLD)
|
||||
places = [pipe.chainage_m for _, pipe in pairs]
|
||||
assert added >= 1 and all(pipe.source == PIPE_SOURCE_SPACING for _, pipe in pairs)
|
||||
assert any(abs(at - 220.0) <= 3 for at in places)
|
||||
assert not any(at < 60 for at in places) # 옛 노선 위 오목엔 안 둠
|
||||
|
||||
|
||||
def test_간격이_좁으면_보충_없음():
|
||||
vertices = vertices_of(NEW)
|
||||
pairs = [(0, PipePoint(chainage_m=at)) for at in (0.0, 90.0, 180.0, 270.0, 360.0, 450.0)]
|
||||
concave = [(220.0, 99.0)]
|
||||
assert fill_concave_pipes(pairs, concave, vertices, OLD) == 0
|
||||
|
||||
|
||||
# ── 측점 열쇠 · 횡단 이월 ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _structure(sid, type_id, anchor, start=None, end=None):
|
||||
return SimpleNamespace(
|
||||
structure_id=sid,
|
||||
type_id=type_id,
|
||||
chainage_m=anchor,
|
||||
start_m=start,
|
||||
end_m=end,
|
||||
options={},
|
||||
)
|
||||
|
||||
|
||||
def test_측점_열쇠_규칙():
|
||||
stations = [{"chainage_m": s} for s in (0.0, 20.0, 40.0, 50.0, 60.0, 77.0, 80.0, 95.5)]
|
||||
pipes = [(3, PipePoint(chainage_m=50.0))]
|
||||
structures = [_structure("w1", "retaining_wall", 77.0, 60.0, 80.0)]
|
||||
keys = station_keys(stations, pipes, structures, 20.0)
|
||||
assert keys == {
|
||||
"end:BP": 0.0,
|
||||
"grid:1": 20.0,
|
||||
"grid:2": 40.0,
|
||||
"pipe:3": 50.0,
|
||||
"structure:w1:start": 60.0,
|
||||
"structure:w1": 77.0,
|
||||
"structure:w1:end": 80.0,
|
||||
"end:EP": 95.5,
|
||||
}
|
||||
|
||||
|
||||
def test_이월은_열쇠로_누가거리가_밀려도():
|
||||
old_keys = {
|
||||
"end:BP": 0.0,
|
||||
"grid:1": 20.0,
|
||||
"pipe:0": 50.0,
|
||||
"structure:w1": 77.0,
|
||||
"grid:5": 100.0,
|
||||
}
|
||||
new_keys = {"end:BP": 0.0, "grid:1": 20.0, "pipe:0": 57.0, "structure:w1": 84.0}
|
||||
designs = {
|
||||
0.0: {"ground_type": "soil"},
|
||||
20.0: {"ground_type": "rock", "berm": {"on": True}, "cut_area_m2": 9.0},
|
||||
50.0: {"ditch_type": "u", "inlet_structure": "basin", "basin_adjust": {"d": 1}},
|
||||
77.0: {"rock_boundary_offset_m": 0.8, "revet_adjust": {"h": 2}},
|
||||
100.0: {"ground_type": "soil"},
|
||||
}
|
||||
inputs, adjusts, dropped = carry_cross_inputs(
|
||||
old_keys,
|
||||
new_keys,
|
||||
designs,
|
||||
("ground_type", "ditch_type", "berm", "rock_boundary_offset_m"),
|
||||
("inlet_structure", "basin_adjust", "revet_adjust"),
|
||||
)
|
||||
assert inputs == {
|
||||
0.0: {"ground_type": "soil"},
|
||||
20.0: {"ground_type": "rock", "berm": {"on": True}}, # 결과 칸(면적)은 안 넘김
|
||||
57.0: {"ditch_type": "u"},
|
||||
84.0: {"rock_boundary_offset_m": 0.8},
|
||||
}
|
||||
# 구조물 조정은 관 · 구조물 열쇠만 따라감
|
||||
assert adjusts == {
|
||||
57.0: {"inlet_structure": "basin", "basin_adjust": {"d": 1}},
|
||||
84.0: {"revet_adjust": {"h": 2}},
|
||||
}
|
||||
assert dropped == 1 # grid:5 — 새 노선에 없음(노선 밖 고침)
|
||||
|
||||
|
||||
# ── 노선 초안 요청 ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_초안_모양을_옛_칸으로():
|
||||
body = save.RouteSaveRequest(
|
||||
draft={
|
||||
"rev": 3,
|
||||
"nodes": [[0, 0], [10, 5], [20, 0]],
|
||||
"curveOn": [True, False, True],
|
||||
"curveRadius": [None, 30.0, None],
|
||||
"curveLock": [None, "radius", None],
|
||||
"curveArc": [None, None, None],
|
||||
"stationSnap": True,
|
||||
"terrain": "sheet",
|
||||
}
|
||||
)
|
||||
request = body.resolved()
|
||||
assert [(v.x, v.y, v.curve, v.radius_m) for v in request.vertices] == [
|
||||
(0, 0, True, None),
|
||||
(10, 5, False, 30.0),
|
||||
(20, 0, True, None),
|
||||
]
|
||||
assert request.station_snap is True and request.terrain == "sheet"
|
||||
with pytest.raises(ValueError):
|
||||
save.RouteSaveRequest(draft={"nodes": [[0, 0], [1, 1]], "curveOn": [True]}).resolved()
|
||||
|
||||
|
||||
# ── 체인 차례 · 되살림 · 배수유역 낡음 ─────────────────────────────────────────
|
||||
|
||||
|
||||
class _Cursor:
|
||||
def __init__(self, log):
|
||||
self.log = log
|
||||
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc):
|
||||
return False
|
||||
|
||||
async def execute(self, sql, params=None):
|
||||
self.log.append(" ".join(sql.split()))
|
||||
|
||||
|
||||
class _Connection:
|
||||
def __init__(self, log):
|
||||
self.log = log
|
||||
|
||||
def cursor(self, *args):
|
||||
return _Cursor(self.log)
|
||||
|
||||
async def begin(self):
|
||||
return None
|
||||
|
||||
async def commit(self):
|
||||
return None
|
||||
|
||||
async def rollback(self):
|
||||
return None
|
||||
|
||||
|
||||
class _Pool:
|
||||
def __init__(self, log):
|
||||
self.log = log
|
||||
|
||||
def acquire(self):
|
||||
connection = _Connection(self.log)
|
||||
|
||||
class _Acquire:
|
||||
async def __aenter__(self):
|
||||
return connection
|
||||
|
||||
async def __aexit__(self, *exc):
|
||||
return False
|
||||
|
||||
return _Acquire()
|
||||
|
||||
|
||||
def _async(value, order=None, name=None):
|
||||
async def fake(*args, **kwargs):
|
||||
if order is not None:
|
||||
order.append(name)
|
||||
return value
|
||||
|
||||
return fake
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def chain_world(monkeypatch, tmp_path):
|
||||
order: list[str] = []
|
||||
sql: list[str] = []
|
||||
root = tmp_path
|
||||
route_file = root / "B05_Profile" / "route" / "route_main.geojson"
|
||||
route_file.parent.mkdir(parents=True)
|
||||
route_file.write_text(
|
||||
json.dumps({"geometry": {"coordinates": [[*p, 0] for p in OLD]}}), "utf-8"
|
||||
)
|
||||
long_dir = root / "B06_Section" / "longitudinal"
|
||||
long_dir.mkdir(parents=True)
|
||||
(long_dir / "longitudinal.json").write_text(json.dumps({"stations": [], "old": True}), "utf-8")
|
||||
drain = root / "drain"
|
||||
pipes_file = drain / "edits" / "pipe_points.json"
|
||||
pipes_file.parent.mkdir(parents=True)
|
||||
pipes_file.write_text(
|
||||
json.dumps(
|
||||
{"route_signature": "old", "points": [{"chainage_m": 50.0, "x": 50.0, "y": 0.0}]}
|
||||
),
|
||||
"utf-8",
|
||||
)
|
||||
(drain / "00_watershed_response.json").write_text(
|
||||
json.dumps({"route_signature": "old"}), "utf-8"
|
||||
)
|
||||
state = {"stages": [{"stage_no": 2, "params": {"points": {}, "options": {}}}]}
|
||||
|
||||
def solve(project_id, request, *, generate_sections=True):
|
||||
order.append("solve")
|
||||
assert generate_sections is False
|
||||
route_file.write_text(
|
||||
json.dumps({"geometry": {"coordinates": [[*p, 0] for p in NEW]}}), "utf-8"
|
||||
)
|
||||
return _async(
|
||||
SimpleNamespace(
|
||||
route_id=8,
|
||||
total_length_m=456.0,
|
||||
route_data_path="B05_Profile/route/route_main.geojson",
|
||||
)
|
||||
)()
|
||||
|
||||
async def regenerate(*args, **kwargs):
|
||||
order.append("sections")
|
||||
(long_dir / "longitudinal.json").write_text(
|
||||
json.dumps({"stations": [], "new": True}), "utf-8"
|
||||
)
|
||||
return SimpleNamespace()
|
||||
|
||||
async def watershed_must_not_run(*args, **kwargs):
|
||||
raise AssertionError("배수유역은 체인에서 뗌")
|
||||
|
||||
async def prepare(project_id):
|
||||
order.append("pipes")
|
||||
return {"vertices": vertices_of(NEW), "streams": []}
|
||||
|
||||
monkeypatch.setattr(save, "get_db_pool", lambda: _Pool(sql))
|
||||
monkeypatch.setattr(workflow, "get_workflow_state", _async(state))
|
||||
monkeypatch.setattr(workflow, "start_stage", _async(None, order, "restore_params"))
|
||||
monkeypatch.setattr(
|
||||
b06_repo,
|
||||
"get_longitudinal_section",
|
||||
_async(
|
||||
{"data": {}, "longitudinal_file_path": "B06_Section/longitudinal/longitudinal.json"}
|
||||
),
|
||||
)
|
||||
monkeypatch.setattr(b06_repo, "get_cross_section_designs", _async([]))
|
||||
monkeypatch.setattr(b06_repo, "get_latest_section_options", _async({}))
|
||||
monkeypatch.setattr(
|
||||
structures_router,
|
||||
"follow_structures_to_route",
|
||||
lambda *a: order.append("structures") or {"moved": 0, "far": 0},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
chain,
|
||||
"redesign_solve_request",
|
||||
lambda *a: b05_router.RouteSolveRequest(
|
||||
filter_key="csf",
|
||||
method="dtm",
|
||||
smooth=False,
|
||||
algorithm="as_planned",
|
||||
bp={"x": 0, "y": 0},
|
||||
ep={"x": 400, "y": 0},
|
||||
),
|
||||
)
|
||||
monkeypatch.setattr(b05_router, "solve_route_record", solve)
|
||||
monkeypatch.setattr(b05_router, "confirm_latest_route", _async({}))
|
||||
monkeypatch.setattr(watershed, "get_primary_region", watershed_must_not_run)
|
||||
monkeypatch.setattr(watershed, "_prepare", prepare)
|
||||
monkeypatch.setattr(drainage_pipes, "clear_pipe_points", watershed_must_not_run)
|
||||
monkeypatch.setattr(drainage_pipes, "drainage_dir", lambda stored: drain)
|
||||
monkeypatch.setattr(save, "_ground_along", lambda *a: (np.zeros(0), np.zeros(0)))
|
||||
monkeypatch.setattr(b06_router, "regenerate_route_sections", regenerate)
|
||||
monkeypatch.setattr(save, "_carry_profile", _async(None, order, "profile"))
|
||||
monkeypatch.setattr(save, "_carry_cross", _async(2, order, "cross"))
|
||||
monkeypatch.setattr(
|
||||
b06_confirm, "confirm_sections", _async(SimpleNamespace(), order, "confirm")
|
||||
)
|
||||
monkeypatch.setattr(snapshot, "mark_designing", lambda root: None)
|
||||
monkeypatch.setattr(snapshot, "clear_designing", lambda root: None)
|
||||
return SimpleNamespace(
|
||||
order=order,
|
||||
sql=sql,
|
||||
root=root,
|
||||
route_file=route_file,
|
||||
pipes_file=pipes_file,
|
||||
long_dir=long_dir,
|
||||
drain=drain,
|
||||
)
|
||||
|
||||
|
||||
def _run(world, **kwargs):
|
||||
old_route = {"id": 7, "route_data_path": "B05_Profile/route/route_main.geojson"}
|
||||
return asyncio.run(
|
||||
save.run_route_save_chain(
|
||||
uuid4(),
|
||||
world.root,
|
||||
"p",
|
||||
1,
|
||||
{},
|
||||
old_route,
|
||||
[{"x": x, "y": y} for x, y in NEW],
|
||||
{"old": 1},
|
||||
**kwargs,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def test_차례와_배수유역_뗌(chain_world):
|
||||
result = _run(chain_world)
|
||||
assert "error" not in result, result
|
||||
assert chain_world.order == [
|
||||
"structures",
|
||||
"solve",
|
||||
"pipes",
|
||||
"sections",
|
||||
"profile",
|
||||
"cross",
|
||||
"confirm",
|
||||
]
|
||||
assert result["cross_dropped"] == 2 and result["drainage_stale"] is True
|
||||
assert result["pipes"] == {"kept": 1, "crossing": 0, "concave": 0}
|
||||
# 관 정본 = 새 노선 지문 · 배수유역 분석 지문(옛) 과 달라 B04 가 「낡음」
|
||||
saved = json.loads(chain_world.pipes_file.read_text("utf-8"))
|
||||
assert saved["route_signature"] == route_signature(vertices_of(NEW))
|
||||
assert analysis_route_signature("p") != saved["route_signature"]
|
||||
assert len(saved["points"]) == 1 # 사용자 관 그대로(지우지 않음)
|
||||
|
||||
|
||||
def test_실패하면_옛_노선을_되살림(chain_world, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
b06_confirm, "confirm_sections", _async(JSONResponse(status_code=500, content={}))
|
||||
)
|
||||
before_pipes = chain_world.pipes_file.read_text("utf-8")
|
||||
result = _run(chain_world)
|
||||
assert result["error"].startswith("종횡단 확정 실패")
|
||||
assert json.loads(chain_world.route_file.read_text("utf-8"))["geometry"]["coordinates"] == [
|
||||
[*p, 0] for p in OLD
|
||||
]
|
||||
assert chain_world.pipes_file.read_text("utf-8") == before_pipes
|
||||
assert json.loads((chain_world.long_dir / "longitudinal.json").read_text("utf-8")) == {
|
||||
"stations": [],
|
||||
"old": True,
|
||||
}
|
||||
assert any(
|
||||
sql.startswith("DELETE FROM routes WHERE project_id = %s AND id > %s")
|
||||
for sql in chain_world.sql
|
||||
)
|
||||
assert chain_world.order[-1] == "restore_params"
|
||||
assert not list(chain_world.root.glob("route_save_*")) # 손에 쥔 사본 치움
|
||||
|
||||
|
||||
def test_추천_종단_반영이면_균형_종단(chain_world, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
route_initial, "apply_profile_target", _async(None, chain_world.order, "balanced")
|
||||
)
|
||||
_run(chain_world, profile_target=([0.0, 456.0], [100.0, 110.0]))
|
||||
assert "balanced" in chain_world.order and "profile" not in chain_world.order
|
||||
|
||||
|
||||
# ── 초기값 규칙 ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def api_world(monkeypatch, tmp_path):
|
||||
calls: list = []
|
||||
monkeypatch.setattr(replan, "_project_paths", _async((tmp_path, "p")))
|
||||
monkeypatch.setattr(replan, "_ensure_expected_route", lambda root: None)
|
||||
monkeypatch.setattr(replan, "_plan_criteria", _async((20.0, 0.0, 0.0)))
|
||||
monkeypatch.setattr(replan, "_ensure_planned_initial", lambda *a: None)
|
||||
monkeypatch.setattr(
|
||||
replan,
|
||||
"_write_planned_polyline",
|
||||
lambda *a, **k: {"vertices": 3, "curves": 0, "violations": 0},
|
||||
)
|
||||
monkeypatch.setattr(replan, "_recompute", _async({"route_id": 9, "total_length_m": 10.0}))
|
||||
monkeypatch.setattr(replan, "read_initial_choice", lambda root: None)
|
||||
monkeypatch.setattr(
|
||||
route_initial, "refresh_initial_snapshot", lambda *a: calls.append(a) or _async(None)()
|
||||
)
|
||||
return calls
|
||||
|
||||
|
||||
def _body(**kwargs):
|
||||
return save.RouteSaveRequest(vertices=[{"x": 0, "y": 0}, {"x": 5, "y": 5}], **kwargs)
|
||||
|
||||
|
||||
def test_손으로_고친_노선_저장은_초기값_그대로(api_world):
|
||||
result = asyncio.run(replan.replan_route(uuid4(), _body()))
|
||||
assert result["initial_refreshed"] is False and api_world == []
|
||||
|
||||
|
||||
def test_추천_적용은_초기값_다시(api_world):
|
||||
result = asyncio.run(replan.replan_route(uuid4(), _body(as_initial=True)))
|
||||
assert result["initial_refreshed"] is True and len(api_world) == 1
|
||||
|
||||
|
||||
def test_초안_칸이_어긋나면_400(api_world):
|
||||
body = save.RouteSaveRequest(draft={"nodes": [[0, 0], [1, 1]], "curveOn": [True]})
|
||||
response = asyncio.run(replan.replan_route(uuid4(), body))
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def _longitudinal(total: float) -> dict:
|
||||
from B05_Profile.B05_Profile_Engine_Grade import resolve_grade_options
|
||||
from B05_Profile.B05_Profile_Engine_Grade_Profile import design_ground_following_profile
|
||||
|
||||
samples = [
|
||||
{"chainage_m": float(s), "elevation_m": 100.0 + 0.05 * s + 2.0 * np.sin(s / 17.0)}
|
||||
for s in range(int(total) + 1)
|
||||
]
|
||||
stations = [{"chainage_m": float(s)} for s in range(0, int(total) + 1, 20)]
|
||||
longitudinal = {"samples": samples, "stations": stations}
|
||||
alignment, profile = design_ground_following_profile(
|
||||
longitudinal, resolve_grade_options("trunk")
|
||||
)
|
||||
return {**longitudinal, "profile_alignment": alignment, "design_profiles": [profile]}
|
||||
|
||||
|
||||
def test_계획고_옮김은_앞_그대로_뒤_밂(tmp_path, monkeypatch):
|
||||
new_xy = [list(p) for p in NEW]
|
||||
total = float(vertices_of(NEW)[-1].chainage_m)
|
||||
old = _longitudinal(400.0)
|
||||
old["profile_alignment"]["edits"] = {
|
||||
"station_offsets": {"40.000": 0.7, "200.000": -0.4, "380.000": 0.3},
|
||||
"curve_radii": {},
|
||||
}
|
||||
path = tmp_path / "B06_Section" / "longitudinal" / "longitudinal.json"
|
||||
path.parent.mkdir(parents=True)
|
||||
path.write_text(json.dumps(_longitudinal(total)), "utf-8")
|
||||
import B05_Profile.B05_Profile_Repository as b05_repo
|
||||
|
||||
monkeypatch.setattr(save, "get_db_pool", lambda: _Pool([]))
|
||||
monkeypatch.setattr(b05_repo, "update_longitudinal_grade_summary", _async(None))
|
||||
asyncio.run(
|
||||
save._carry_profile(
|
||||
tmp_path,
|
||||
"B06_Section/longitudinal/longitudinal.json",
|
||||
8,
|
||||
old["profile_alignment"],
|
||||
[list(p) for p in OLD],
|
||||
new_xy,
|
||||
)
|
||||
)
|
||||
edits = json.loads(path.read_text("utf-8"))["profile_alignment"]["edits"]["station_offsets"]
|
||||
delta = total - 400.0
|
||||
assert edits["40.000"] == 0.7 # 바뀐 구간 앞 — 그대로
|
||||
assert "200.000" not in edits # 바뀐 구간 안 — 새 지반
|
||||
assert edits[f"{380.0 + delta:.3f}"] == 0.3 # 뒤 — 길이 차만큼 밂
|
||||
Reference in New Issue
Block a user