Merge remote-tracking branch 'origin/sub_laptop_1' into main_laptop_1
This commit is contained in:
@@ -375,11 +375,13 @@ async def _ensure_initial_polyline(
|
||||
"""계획노선 초기 폴리라인을 세운다 — 이미 있으면 그대로 둔다. 실패는 비치명적."""
|
||||
import asyncio
|
||||
|
||||
from B05_Profile.B05_Profile_Router_FollowLand import initial_land
|
||||
from B05_Profile.B05_Profile_Router_Replan import _ensure_planned_initial, _min_plan_radius_m
|
||||
|
||||
try:
|
||||
radius_m = await _min_plan_radius_m(project_id)
|
||||
summary = await asyncio.to_thread(_ensure_planned_initial, project_root, radius_m)
|
||||
land = await initial_land(project_id, project_root)
|
||||
summary = await asyncio.to_thread(_ensure_planned_initial, project_root, radius_m, land)
|
||||
if summary:
|
||||
logger.info(
|
||||
"초기 계획노선 폴리라인: project_id=%s 노드 %d · 곡선 %d · 위반 %d (R %.1fm)",
|
||||
|
||||
@@ -55,15 +55,18 @@ from __future__ import annotations
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
from dataclasses import replace
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
from shapely.geometry import LineString
|
||||
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_CrossEarth import (
|
||||
cross_earthwork,
|
||||
cross_earthwork_delta,
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_CrossEarth import cross_earthwork
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Follow import (
|
||||
FollowLand,
|
||||
cross_delta,
|
||||
cross_total,
|
||||
merged_follow,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Merge import RouteEarth
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Metrics import GRADE_SMOOTH_M, route_metrics
|
||||
@@ -72,12 +75,10 @@ from B05_Profile.B05_Profile_Engine_RouteInitial_Pool import (
|
||||
build_pool,
|
||||
choose,
|
||||
hard_limits,
|
||||
settle,
|
||||
similar,
|
||||
weighted,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import (
|
||||
STATION_M,
|
||||
_reference,
|
||||
attach_profile,
|
||||
cover_shortfall,
|
||||
@@ -94,8 +95,8 @@ from B05_Profile.B05_Profile_Engine_RouteInitial_Search import (
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Station import (
|
||||
off_station_count,
|
||||
snap_curves_to_stations,
|
||||
with_station_misses,
|
||||
)
|
||||
from common_util.common_util_route_polyline import build_planned_polyline
|
||||
|
||||
__all__ = ["NoRoute", "Terrain", "generate_initial_route"]
|
||||
|
||||
@@ -360,12 +361,13 @@ def generate_initial_route(
|
||||
moved = snap_curves_to_stations(outline, min_radius_m, guard=guard)
|
||||
return outline if moved is None else moved
|
||||
|
||||
def follow_line() -> tuple[Any, dict[str, Any]]:
|
||||
# 예상노선 추적 = 전처리 초기값(`planned_route_initial.csv` · 383-1 측점 맞춤)과 같은
|
||||
# 규칙 · 같은 입력 → 같은 선. 제약은 따지지 않고 넘은 곳만 목록에 뜸.
|
||||
outline = build_planned_polyline(expected, min_radius_m=min_radius_m)
|
||||
route = (outline, measure(outline, bound=None, follow=True))
|
||||
return settle(route, fit, measure, bound=None, follow=True)
|
||||
def follow_line(grid: Any = None) -> tuple[Any, dict[str, Any]]:
|
||||
# 예상노선 추적 = 전처리 초기값(`planned_route_initial.csv`)과 같은 함수(곡선 최적 · 측점
|
||||
# 맞춤 · 758-1 `..._Follow.merged_follow`) · 같은 입력 → 같은 선. 넘은 곳만 목록에 뜸.
|
||||
land = FollowLand(terrain, streams, max_grade, stream_offset, grade_smooth, goal_grade)
|
||||
land = replace(land, standard=standard, corridor_m=corridor_m)
|
||||
outline = merged_follow(expected, min_radius_m, land, grid)
|
||||
return outline, with_station_misses(measure(outline, bound=None, follow=True), outline)
|
||||
|
||||
def profiled(route: Any, exact: bool = True) -> None:
|
||||
if route is not None:
|
||||
@@ -395,29 +397,10 @@ def generate_initial_route(
|
||||
|
||||
def route_earth(line: Any) -> float:
|
||||
# 합치기 멈춤 잣대 = 계획고 지반고 횡단(1차 종단 규칙 · 종단 끔과 같음)
|
||||
try:
|
||||
found = cross(line)
|
||||
except ValueError:
|
||||
return math.inf
|
||||
return found["cut"] + found["fill"]
|
||||
return cross_total(line, terrain, standard)
|
||||
|
||||
def merge_delta(before: Any, after: Any, window: tuple[float, float]) -> float:
|
||||
# 바뀐 자리 측점만 — 뒤 측점은 같은 선을 따라 밀린 것뿐(옛 선 끝 s1 → 새 선 s1 + 연장 차)
|
||||
shift = LineString(after.vertices).length - LineString(before.vertices).length
|
||||
first = math.ceil(window[0] / STATION_M) * STATION_M
|
||||
try:
|
||||
return cross_earthwork_delta(
|
||||
before.vertices,
|
||||
after.vertices,
|
||||
np.arange(first, window[1] + shift + 1e-9, STATION_M),
|
||||
terrain,
|
||||
standard,
|
||||
before_chainages=np.arange(first, window[1] + 1e-9, STATION_M),
|
||||
design_curves_before=[curve.as_dict() for curve in before.curves],
|
||||
design_curves_after=[curve.as_dict() for curve in after.curves],
|
||||
)["earth"]
|
||||
except ValueError:
|
||||
return math.inf
|
||||
return cross_delta(before, after, window, terrain, standard)
|
||||
|
||||
def cross_score(route: Any, capped: bool) -> float:
|
||||
# 후보 줄 세움(419) — 켬 = 상한 누른 계획선(`_reference`) 횡단 · 끔 = 계획고 지반고 횡단
|
||||
@@ -469,7 +452,7 @@ def generate_initial_route(
|
||||
# 합치기 멈춤 토공 = 노선 횡단 + 바뀐 측점 횡단 차이(420 `..._CrossEarth` · 419)
|
||||
_POOLS[key] = build_pool(
|
||||
graph,
|
||||
follow_line(),
|
||||
follow_line(grid),
|
||||
max_grade,
|
||||
min_radius_m,
|
||||
measure,
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
"""예정노선 따라가기 곡선 최적(758-1 · 요구 598-1) — 따라간 선에 「토공 최소」와 같은 둘째 단계.
|
||||
|
||||
`B05_Profile_Route_Polyline_Write.followed_table`(체인 처음 · `/route/follow` · 받은 노선으로) ·
|
||||
추천 「예상노선 추적」(`..._RouteInitial.generate_initial_route` `follow_line`)이 같이 부름.
|
||||
|
||||
차례(`merged_follow`) = ① 예정노선 폴리라인(꺾임점 · 묶기 · R 맞춤 · 기본 규칙) → 큰 R
|
||||
(`..._Radius.widen`) → 꺾임만 곡선으로(`curve_every_kink`) → 같은 쪽 이웃 곡선 합치기
|
||||
(`..._Merge.merge_same_side`) → ② 곡선 MC 일반측점 맞춤. 큰 R · 합치기 · 측점 맞춤 모두 직접 잼
|
||||
(`Guard` · 지형 · 계류 격자 · 토공 최소 갈래와 같은 판 — 이격도 판 `keeping_offset` 은 합치기 0 ·
|
||||
측점 맞춤이 더 벗어나 안 씀) · 합치기 멈춤 = 못 지킨 곳 수 · 노선 횡단 토공 +10 %.
|
||||
지형이 없으면(`FollowLand` None) 부르는 쪽이 옛 길(순수 기하 `snapped_initial`)을 씀.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from collections import OrderedDict
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
from shapely.geometry import LineString
|
||||
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_CrossEarth import (
|
||||
cross_earthwork,
|
||||
cross_earthwork_delta,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Merge import Guard, RouteEarth, merge_same_side
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Metrics import STATION_M, route_metrics
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Pool import hard_limits
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Radius import curve_every_kink, widen
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Search import (
|
||||
CORRIDOR_M,
|
||||
ROAD_WIDTH_M,
|
||||
SearchGrid,
|
||||
Terrain,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Station import snap_curves_to_stations
|
||||
from common_util.common_util_route_polyline import PlannedPolyline, build_planned_polyline
|
||||
|
||||
Xy = tuple[float, float]
|
||||
# 같은 입력 다시 셈 막기(끌 때 · 체인 · 추천이 같은 선을 거듭 부름) — 최근 몇 벌만
|
||||
MEMO_SIZE = 8
|
||||
_MEMO: OrderedDict[tuple, PlannedPolyline] = OrderedDict()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FollowLand:
|
||||
"""따라가기 곡선 최적의 지형 · 제약 — 추천 엔진과 같은 풀이 값(`resolve_inputs` 뒤)."""
|
||||
|
||||
terrain: Terrain
|
||||
streams: list[list[Xy]]
|
||||
max_grade: float
|
||||
stream_offset_m: float
|
||||
grade_smooth_m: float
|
||||
target_grade: float
|
||||
standard: dict[str, Any] | None = None
|
||||
corridor_m: float = CORRIDOR_M
|
||||
|
||||
|
||||
def cross_total(line: PlannedPolyline, terrain: Terrain, standard: dict[str, Any] | None) -> float:
|
||||
"""합치기 멈춤 잣대 — 계획고 지반고 횡단 토공(1차 종단 규칙 · 종단 끔과 같음) · 못 재면 inf."""
|
||||
try:
|
||||
found = cross_earthwork(
|
||||
line.vertices,
|
||||
None,
|
||||
None,
|
||||
terrain,
|
||||
standard,
|
||||
design_curves=[curve.as_dict() for curve in line.curves],
|
||||
)
|
||||
except ValueError:
|
||||
return math.inf
|
||||
return found["cut"] + found["fill"]
|
||||
|
||||
|
||||
def cross_delta(
|
||||
before: PlannedPolyline,
|
||||
after: PlannedPolyline,
|
||||
window: tuple[float, float],
|
||||
terrain: Terrain,
|
||||
standard: dict[str, Any] | None,
|
||||
) -> float:
|
||||
"""바뀐 자리 측점만 횡단 토공 차 — 뒤 측점은 같은 선 따라 밀림(옛 끝 s1 → 새 s1 + 연장 차)."""
|
||||
shift = LineString(after.vertices).length - LineString(before.vertices).length
|
||||
first = math.ceil(window[0] / STATION_M) * STATION_M
|
||||
try:
|
||||
return cross_earthwork_delta(
|
||||
before.vertices,
|
||||
after.vertices,
|
||||
np.arange(first, window[1] + shift + 1e-9, STATION_M),
|
||||
terrain,
|
||||
standard,
|
||||
before_chainages=np.arange(first, window[1] + 1e-9, STATION_M),
|
||||
design_curves_before=[curve.as_dict() for curve in before.curves],
|
||||
design_curves_after=[curve.as_dict() for curve in after.curves],
|
||||
)["earth"]
|
||||
except ValueError:
|
||||
return math.inf
|
||||
|
||||
|
||||
def merged_follow(
|
||||
points: list[Xy], min_radius_m: float, land: FollowLand, grid: SearchGrid | None = None
|
||||
) -> PlannedPolyline:
|
||||
"""예정노선 점 묶음 → 곡선 최적 · 측점 맞춘 폴리라인(못 맞추면 맞추기 전 선). `grid` = 이미 짠
|
||||
탐색 격자(같은 예정노선 · 추천 묶음) · 없으면 짬."""
|
||||
key = (
|
||||
tuple((float(x), float(y)) for x, y in points),
|
||||
round(min_radius_m, 6),
|
||||
land.terrain.signature or id(land.terrain),
|
||||
land.max_grade,
|
||||
land.stream_offset_m,
|
||||
land.grade_smooth_m,
|
||||
land.target_grade,
|
||||
land.corridor_m,
|
||||
repr(land.standard),
|
||||
len(land.streams),
|
||||
)
|
||||
if key in _MEMO:
|
||||
_MEMO.move_to_end(key)
|
||||
return _MEMO[key]
|
||||
|
||||
def measure(line: PlannedPolyline) -> dict[str, Any]:
|
||||
# 추천 「예상노선 추적」 잣대(follow · 못 지킨 곳 = R · 계류 · 측점 · 원지반 기울기 참고)
|
||||
return hard_limits(
|
||||
route_metrics(
|
||||
line.vertices,
|
||||
points,
|
||||
land.terrain,
|
||||
land.streams,
|
||||
line.curves,
|
||||
max_grade=land.max_grade,
|
||||
min_radius_m=min_radius_m,
|
||||
stream_offset_m=land.stream_offset_m,
|
||||
road_width_m=ROAD_WIDTH_M,
|
||||
bound_pct=None,
|
||||
grade_smooth_m=land.grade_smooth_m,
|
||||
follow=True,
|
||||
target_grade=land.target_grade,
|
||||
)
|
||||
)
|
||||
|
||||
if grid is None:
|
||||
grid = SearchGrid(
|
||||
points,
|
||||
land.terrain,
|
||||
land.streams,
|
||||
stream_offset_m=land.stream_offset_m,
|
||||
grade_smooth_m=land.grade_smooth_m,
|
||||
target_grade=land.target_grade,
|
||||
corridor_m=land.corridor_m,
|
||||
)
|
||||
guard = Guard(grid, land.max_grade)
|
||||
outline = build_planned_polyline(points, min_radius_m=min_radius_m)
|
||||
route = (outline, measure(outline))
|
||||
larger = widen(outline, points, min_radius_m, guard=guard)
|
||||
if larger is not outline:
|
||||
trial = (larger, measure(larger))
|
||||
if trial[1]["over_limit"]["total"] <= route[1]["over_limit"]["total"]:
|
||||
route = trial
|
||||
kinked = curve_every_kink(route[0], min_radius_m)
|
||||
if kinked is not route[0]:
|
||||
route = (kinked, measure(kinked))
|
||||
route = merge_same_side(
|
||||
route,
|
||||
guard,
|
||||
min_radius_m,
|
||||
measure,
|
||||
RouteEarth(
|
||||
lambda line: cross_total(line, land.terrain, land.standard),
|
||||
lambda before, after, window: cross_delta(
|
||||
before, after, window, land.terrain, land.standard
|
||||
),
|
||||
),
|
||||
)
|
||||
snapped = snap_curves_to_stations(route[0], min_radius_m, guard=guard)
|
||||
found = route[0] if snapped is None else snapped
|
||||
_MEMO[key] = found
|
||||
while len(_MEMO) > MEMO_SIZE:
|
||||
_MEMO.popitem(last=False)
|
||||
return found
|
||||
@@ -204,16 +204,23 @@ def polyline_table(
|
||||
|
||||
|
||||
def followed_table(
|
||||
points: list[tuple[float, float]], radius_m: float
|
||||
points: list[tuple[float, float]], radius_m: float, land: Any = None
|
||||
) -> tuple[PlannedPolyline, dict[str, Any]]:
|
||||
"""예정노선 추적(383-1) — 점 묶음 → 곡선 MC 를 일반측점에 맞춘 폴리라인 + 곡선표(스냅 켬).
|
||||
|
||||
체인 초기 폴리라인(`_ensure_planned_initial`) · `/route/follow`(626-2)가 같이 씀 · 못 맞추면
|
||||
지금 선.
|
||||
지금 선. `land`(`..._RouteInitial_Follow.FollowLand`) 주면 곡선 최적(큰 R · 같은 쪽 합치기 ·
|
||||
758-1 · 추천 「예상노선 추적」과 같은 함수) · 없으면 순수 기하.
|
||||
"""
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Station import snapped_initial
|
||||
|
||||
fit = snapped_initial(points, radius_m)
|
||||
if land is not None:
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Follow import merged_follow
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Radius import _flat
|
||||
|
||||
fit = _flat(merged_follow(points, radius_m, land))
|
||||
else:
|
||||
fit = snapped_initial(points, radius_m)
|
||||
shape = {"simplify": False, "curve_flags": fit[1], "radii": fit[2]} if fit else {}
|
||||
return polyline_table(fit[0] if fit else points, radius_m, station_snap=True, **shape)
|
||||
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
"""따라가기 곡선 최적의 지형 · 제약 읽기(758-1) — 체인 처음 · `/route/plan` · `/route/replan` ·
|
||||
`/route/follow` 가 `followed_table` 에 넘김.
|
||||
|
||||
값 = 추천 엔진 기본 풀이와 같은 자리(`..._Router_RouteInitial.compute_initial`): 종단기울기 상한 =
|
||||
법정(임도 종류 · 설계속도 · 지형 · 비포장) · 지형 = 설계 지형 비용면 · 계류 · 표준 횡단면 = 프로젝트
|
||||
저장값 · 나머지는 엔진 기본값. 못 읽으면 None(옛 길 — 순수 기하 따라가기).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from uuid import UUID
|
||||
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Follow import FollowLand
|
||||
from common_util.common_util_route_geometry import (
|
||||
expected_route_csv_path,
|
||||
planned_route_initial_path,
|
||||
)
|
||||
from config.config_db import get_db_pool
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def follow_land(project_id: UUID, project_root: Path) -> FollowLand | None:
|
||||
"""이 프로젝트의 따라가기 곡선 최적 입력 · 못 읽으면 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,
|
||||
resolve_inputs,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Router_Replan import _road_settings
|
||||
from B05_Profile.B05_Profile_Router_RouteInitial import project_standard
|
||||
from common_util.common_util_surface_confirmation import (
|
||||
analysis_terrain_params,
|
||||
get_surface_confirmation_params,
|
||||
)
|
||||
|
||||
try:
|
||||
grade_class, design_speed, terrain_type = await _road_settings(project_id)
|
||||
max_grade = legal_grade_limit_pct(grade_class, terrain_type, False, design_speed) / 100.0
|
||||
async with get_db_pool().acquire() as connection:
|
||||
selection = await get_surface_confirmation_params(connection, str(project_id))
|
||||
terrain, streams = await asyncio.to_thread(
|
||||
load_terrain_and_streams, project_root, analysis_terrain_params(selection, None)
|
||||
)
|
||||
stream_offset, grade_smooth, goal_grade = resolve_inputs(max_grade, None, None, None)
|
||||
return FollowLand(
|
||||
terrain,
|
||||
streams,
|
||||
max_grade,
|
||||
stream_offset,
|
||||
grade_smooth,
|
||||
goal_grade,
|
||||
await project_standard(project_id),
|
||||
)
|
||||
except Exception: # noqa: BLE001 — 지형이 없어도 따라가기는 이어 감(순수 기하)
|
||||
logger.warning("따라가기 곡선 최적 지형을 못 읽어 순수 기하로 따라갑니다: %s", project_id)
|
||||
return None
|
||||
|
||||
|
||||
async def initial_land(project_id: UUID, project_root: Path) -> FollowLand | None:
|
||||
"""초기 폴리라인을 새로 세울 때만 지형을 읽음(`_ensure_planned_initial` 과 같은 낡음 판정)."""
|
||||
target = planned_route_initial_path(project_root)
|
||||
source = expected_route_csv_path(project_root)
|
||||
if target.is_file():
|
||||
if not source.is_file() or source.stat().st_mtime <= target.stat().st_mtime:
|
||||
return None
|
||||
return await follow_land(project_id, project_root)
|
||||
@@ -46,6 +46,7 @@ from B05_Profile.B05_Profile_Route_Polyline_Write import ( # noqa: F401 — 부
|
||||
followed_table,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Route_Save import RouteSaveRequest
|
||||
from B05_Profile.B05_Profile_Router_FollowLand import initial_land
|
||||
from B05_Profile.B05_Profile_Schema import ( # noqa: F401 — 부르는 곳이 `replan.` 으로 씀
|
||||
RouteInitialCriteria,
|
||||
RouteInitialRequest,
|
||||
@@ -207,9 +208,11 @@ def write_expected_edit(project_root: Path, points: list[tuple[float, float]]) -
|
||||
write_route_csv(target, [{"x": x, "y": y} for x, y in points])
|
||||
|
||||
|
||||
def _ensure_planned_initial(project_root: Path, radius_m: float) -> dict | None:
|
||||
def _ensure_planned_initial(project_root: Path, radius_m: float, land: Any = None) -> dict | None:
|
||||
"""계획노선 **초기 폴리라인**이 없거나 낡았으면 예상노선을 폴리라인화해 세운다.
|
||||
|
||||
`land` = 곡선 최적 지형(`..._Router_FollowLand.initial_land` · 758-1) · 없으면 순수 기하.
|
||||
|
||||
「낡았다」 = 예상노선 파일이 더 나중에 쓰였다. 파일을 다시 올리면 예상노선이 새로
|
||||
깔리는데 초기본이 옛 노선인 채로 남으면 노선 초기화가 옛 자리로 돌아간다.
|
||||
"""
|
||||
@@ -223,7 +226,7 @@ def _ensure_planned_initial(project_root: Path, radius_m: float) -> dict | None:
|
||||
if len(points) < 2:
|
||||
return None
|
||||
# 383-1 — 곡선 MC = 일반측점(기본 제약) + [측점 스냅] 켬 · 못 맞추면 지금 선 + 켬
|
||||
summary = _write_table(target, *followed_table(points, radius_m))
|
||||
summary = _write_table(target, *followed_table(points, radius_m, land))
|
||||
logger.info(
|
||||
"계획노선 초기 폴리라인 생성: %s (노드 %d · 곡선 %d · 위반 %d · 정점 %d)",
|
||||
target,
|
||||
@@ -393,7 +396,8 @@ async def read_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse:
|
||||
expected = await asyncio.to_thread(_vertices_of, design_route_csv_path(project_root))
|
||||
|
||||
radius_m, radius_limit_m, arc_limit_m = await _plan_criteria(project_id)
|
||||
await asyncio.to_thread(_ensure_planned_initial, project_root, radius_m)
|
||||
land = await initial_land(project_id, project_root)
|
||||
await asyncio.to_thread(_ensure_planned_initial, project_root, radius_m, land)
|
||||
working = await asyncio.to_thread(_vertices_of, planned_route_working_path(project_root))
|
||||
initial = await asyncio.to_thread(_vertices_of, planned_route_initial_path(project_root))
|
||||
planned = working or initial or expected
|
||||
@@ -522,7 +526,8 @@ async def _save_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, Any
|
||||
# 고치기 전에 예상노선(원본)·초기 폴리라인이 서 있는지 본다 — 초기화가 돌아갈 자리다.
|
||||
await asyncio.to_thread(_ensure_expected_route, project_root)
|
||||
radius_m, radius_limit_m, arc_limit_m = await _plan_criteria(project_id)
|
||||
await asyncio.to_thread(_ensure_planned_initial, project_root, radius_m)
|
||||
land = await initial_land(project_id, project_root)
|
||||
await asyncio.to_thread(_ensure_planned_initial, project_root, radius_m, land)
|
||||
# 화면이 보낸 것은 **노드(꺾임점)** 다 — 같은 R 규칙으로 다시 폴리라인을 만든다.
|
||||
# 노드만 옮기면 선이 저절로 규칙을 지키는 것이 이 구조의 목적이다(2026-09-06 사용자).
|
||||
# 받은 것은 이미 꺾임점이므로 **다시 뽑지 않는다** — 뽑으면 사용자가 둔 노드가 지워진다.
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
"""예정노선 다시 추적(626-2 · 요구 530) — 고친 예정노선 → 평면선(노드 · 곡선 · 편집표) · 저장 0.
|
||||
|
||||
계산 자리 ③ 서버 단독 — 체인 초기 폴리라인(`_ensure_planned_initial`)과 같은 함수
|
||||
`followed_table`(지형 안 씀 · 순수 기하) · 추적 전 체인 예정노선과 같은 간격 조밀화(635-2 —
|
||||
`followed_table`(곡선 최적 · 지형 · 계류 격자 잼 · 758-1 · 못 읽으면 순수 기하) · 추적 전 체인
|
||||
예정노선과 같은 간격 조밀화(635-2 —
|
||||
성긴 원본 꼭짓점 = 체인 초기선과 같은 선 · 촘촘한 입력은 그대로).
|
||||
화면은 받은 노드를 편집기 한 걸음으로 싣고 페이지 [저장] 길로 보냄(고친 예정노선 = 초안 `expected`
|
||||
→ `expected_route_edit.csv`). `/route/replan` · `as_initial` · `expected_route.csv` · 초기값 ·
|
||||
@@ -25,6 +26,7 @@ from pydantic import BaseModel
|
||||
|
||||
from B05_Profile.B05_Profile_Repository import get_surface_crs_epsg
|
||||
from B05_Profile.B05_Profile_Route_Polyline_Write import followed_table
|
||||
from B05_Profile.B05_Profile_Router_FollowLand import follow_land
|
||||
from B05_Profile.B05_Profile_Router_Replan import (
|
||||
_PROJECT_PATH_MISSING,
|
||||
_plan_criteria,
|
||||
@@ -46,10 +48,12 @@ class RouteFollowRequest(BaseModel):
|
||||
crs: Literal["tm", "lonlat"] = "tm"
|
||||
|
||||
|
||||
def follow_route_line(points: list[tuple[float, float]], radius_m: float) -> dict[str, Any]:
|
||||
"""점 묶음(TM) → 예정노선 추적 평면선 — 파일 꼴(csv 4자리)과 같은 값."""
|
||||
def follow_route_line(
|
||||
points: list[tuple[float, float]], radius_m: float, land: Any = None
|
||||
) -> dict[str, Any]:
|
||||
"""점 묶음(TM) → 예정노선 추적 평면선 — 파일 꼴(csv 4자리)과 같은 값 · `land` = 곡선 최적."""
|
||||
spacing = ROUTE_DIRECT_LINK_CELL_FACTOR * ROUTE_GRID_RES_M * ROUTE_PLANNED_DENSIFY_SAFETY
|
||||
result, table = followed_table(densify_route(points, spacing), radius_m)
|
||||
result, table = followed_table(densify_route(points, spacing), radius_m, land)
|
||||
return {
|
||||
"planned": [[round(x, 4), round(y, 4)] for x, y in result.vertices],
|
||||
"nodes": [node.as_dict() for node in result.nodes],
|
||||
@@ -70,7 +74,8 @@ async def follow_route(project_id: UUID, body: RouteFollowRequest) -> dict[str,
|
||||
status_code=400,
|
||||
content={"status": "error", "message": "예정노선은 점이 2개 이상이어야 합니다."},
|
||||
)
|
||||
if await _project_paths(project_id) is None:
|
||||
paths = await _project_paths(project_id)
|
||||
if paths is None:
|
||||
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
|
||||
points = [(float(x), float(y)) for x, y in body.points]
|
||||
if body.crs == "lonlat":
|
||||
@@ -86,5 +91,6 @@ async def follow_route(project_id: UUID, body: RouteFollowRequest) -> dict[str,
|
||||
)
|
||||
points = [lonlat_to_tm(lon, lat, epsg) for lon, lat in points]
|
||||
radius_m = (await _plan_criteria(project_id))[0]
|
||||
line = await asyncio.to_thread(follow_route_line, points, radius_m)
|
||||
land = await follow_land(project_id, paths[0])
|
||||
line = await asyncio.to_thread(follow_route_line, points, radius_m, land)
|
||||
return {"status": "success", "project_id": str(project_id), **line}
|
||||
|
||||
@@ -133,12 +133,11 @@ export async function mountRouteEditor(
|
||||
onOpen: () => ((picked = -1), syncCurveBar()),
|
||||
measuring: () => measureControls.mode(), // 켠 동안 재기 = D03 재기 겹(755-2)
|
||||
onHistory: () => historyControls.sync(), // 켠 동안 되돌리기 = 예정노선 판(755-1)
|
||||
onFollow: (flat, line) =>
|
||||
restoreSnapshot(
|
||||
{ ...flat, apexLock: [], picked: -1, rec: initial.tag(), expected: line },
|
||||
"예정노선을 따라 평면을 다시 추적했습니다.",
|
||||
true,
|
||||
),
|
||||
onFollow: (flat, line) => {
|
||||
const step = { ...flat, apexLock: [], picked: -1, expected: line };
|
||||
initial.restoreTag(initial.followed(step)); // 지금 추천 = 예상노선 추적(758-1) · 걸음 사진에 실림
|
||||
restoreSnapshot(step, "예정노선을 따라 평면을 다시 추적했습니다.", true);
|
||||
},
|
||||
});
|
||||
|
||||
let expected: Vertex[] = [];
|
||||
|
||||
@@ -44,6 +44,8 @@ export interface InitialPanel {
|
||||
/** 지금 추천 — 편집 걸음(사진)에 실어 둔다. 되돌리기 · 다시하기가 `restoreTag` 로 되살린다. */
|
||||
tag: () => InitialChoice | null;
|
||||
restoreTag: (tag: unknown) => void;
|
||||
/** 예정노선 따라가기(758-1) 걸음의 꼬리표 — 지금 추천 = 「예상노선 추적」(그 선 · 종단 반영 값 그대로). */
|
||||
followed: (snapshot: RouteEditSnapshot) => InitialChoice;
|
||||
/** [취소] — 이 창에서 바꾼 추천을 창을 열 때 것으로 되돌린다. */
|
||||
cancel: () => void;
|
||||
/** [확인] 뒤 결과 캐시를 비웠다 — 추천은 새 노선의 출발점이라 다시 적는다. */
|
||||
@@ -87,6 +89,15 @@ export function createInitialPanel(params: InitialPanelParams): InitialPanel {
|
||||
choice = tag as InitialChoice | null;
|
||||
writeInitialChoice(projectId, choice);
|
||||
},
|
||||
followed: (snapshot) => ({
|
||||
logic: "follow",
|
||||
weights: null,
|
||||
snapshot,
|
||||
metrics: null,
|
||||
criteria: null,
|
||||
criteriaOverrides: null,
|
||||
withProfile: choiceWithProfile(choice ?? base),
|
||||
}),
|
||||
cancel: () => writeInitialChoice(projectId, openingChoice),
|
||||
keep: () => writeInitialChoice(projectId, choice),
|
||||
};
|
||||
|
||||
@@ -67,12 +67,17 @@ def test_new_copy_initial_on_stations_and_flag_on(tmp_path: Path) -> None:
|
||||
|
||||
|
||||
def test_follow_nodes_equal_initial(tmp_path: Path) -> None:
|
||||
"""추천 「예상노선 추적」(측점 맞춤 켬) = 새 초기 노선 · 측점 밖 곡선 수 = station 목록 수."""
|
||||
"""추천 「예상노선 추적」(측점 맞춤 켬) = 새 초기 노선(같은 지형 곡선 최적 · 758-1) · 측점 밖
|
||||
곡선 수 = station 목록 수."""
|
||||
import test_351_2_curve_stations as t351
|
||||
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial import resolve_inputs
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Follow import FollowLand
|
||||
|
||||
root = _copy_expected(tmp_path)
|
||||
_ensure_planned_initial(root, RADIUS)
|
||||
terrain, streams = t351._terrain(), []
|
||||
land = FollowLand(terrain, streams, 0.14, *resolve_inputs(0.14, None, None, None))
|
||||
_ensure_planned_initial(root, RADIUS, land)
|
||||
expected = [(v.x, v.y) for v in read_planned_route_csv(expected_route_csv_path(root)).vertices]
|
||||
result = generate_initial_route(
|
||||
expected,
|
||||
|
||||
@@ -43,10 +43,11 @@ def test_다시_추적_follow_만_저장_0_초기값_추천_안_건드림():
|
||||
def test_결과는_편집기_한_걸음_저장은_손_편집_같은_길():
|
||||
main = _src(B05 / "B05_Profile_UI_RouteEdit.ts")
|
||||
assert "createExpectedDraw({" in main and "restoreSnapshot(" in main
|
||||
assert re.search(r"onFollow: \(flat, line\) =>\s+restoreSnapshot\(", main)
|
||||
assert (
|
||||
'"예정노선을 따라 평면을 다시 추적했습니다.",\n true,' in main
|
||||
) # 새 걸음(되돌리기 됨)
|
||||
assert re.search(r"onFollow: \(flat, line\) => \{[\s\S]{0,300}?restoreSnapshot\(", main)
|
||||
# 지금 추천 = 예상노선 추적(758-1 · 「토공 최소」 안 남음) — 걸음 사진(`snapshotNow`)에 실림
|
||||
assert "initial.restoreTag(initial.followed(step));" in main
|
||||
# 새 걸음(되돌리기 됨)
|
||||
assert 'restoreSnapshot(step, "예정노선을 따라 평면을 다시 추적했습니다.", true);' in main
|
||||
assert "expected: expectedDraw.edit()," in main # 걸음 사진 칸
|
||||
assert "expectedDraw.setEdit(snapshot.expected ?? null);" in main # 되돌리기 = 예정노선도
|
||||
assert "expected: () => expectedDraw.edit()," in main # [저장] 초안 칸
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
"""758-1 예정노선 따라가기 곡선 최적(요구 598-1) — 「토공 최소」와 같은 큰 R · 합치기.
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_758_1_follow_merge.py -q
|
||||
|
||||
- 합성 언덕 지형 · 굽은 예정노선 → 곡선 수 줆 · MC 모두 일반측점 · 최소 R 지킴.
|
||||
- 한 함수 — 체인 처음(`followed_table` + 지형) = `/route/follow`(`follow_route_line`) =
|
||||
추천 「예상노선 추적」(`generate_initial_route` follow) 노드 같음.
|
||||
- 지형 못 읽으면(None) 옛 길(순수 기하 `snapped_initial`) 그대로.
|
||||
- 글 대조 — 체인 · `/route/plan` · `/route/replan` 은 `initial_land` · `/route/follow` 는
|
||||
`follow_land` 를 넘김.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
from shapely.geometry import LineString
|
||||
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial import generate_initial_route, resolve_inputs
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Follow import FollowLand, merged_follow
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Search import Terrain
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Station import (
|
||||
off_station_count,
|
||||
snapped_initial,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Route_Polyline_Write import followed_table
|
||||
from B05_Profile.B05_Profile_Router_RouteFollow import follow_route_line
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
RADIUS = 12.0
|
||||
# 3 m 간격(조밀화 3.6 m 안 — `/route/follow` 가 더 안 쪼갬 · 체인 예정노선과 같은 입력)
|
||||
EXPECTED = [(float(x), 200.0 + 40.0 * math.sin(x / 60.0)) for x in np.arange(50.0, 752.0, 3.0)]
|
||||
|
||||
|
||||
def _land() -> FollowLand:
|
||||
x = np.arange(0.0, 802.0, 2.0)
|
||||
y = np.arange(0.0, 402.0, 2.0)
|
||||
xx, yy = np.meshgrid(x, y)
|
||||
z = 40.0 * np.exp(-((xx - 400.0) ** 2 + (yy - 200.0) ** 2) / 90.0**2)
|
||||
terrain = Terrain(x, y, z, np.ones(xx.shape, dtype=bool))
|
||||
return FollowLand(terrain, [], 0.14, *resolve_inputs(0.14, None, None, None))
|
||||
|
||||
|
||||
def _nodes(line) -> list[tuple[float, float]]:
|
||||
return [(round(node.x, 3), round(node.y, 3)) for node in line.nodes]
|
||||
|
||||
|
||||
def test_곡선_줆_측점_위_최소_R():
|
||||
plain = followed_table(EXPECTED, RADIUS)[0]
|
||||
merged = merged_follow(EXPECTED, RADIUS, _land())
|
||||
assert len(merged.curves) < len(plain.curves)
|
||||
assert off_station_count(merged) == 0
|
||||
assert min(curve.radius_m for curve in merged.curves) >= RADIUS - 1e-6
|
||||
gap = LineString(EXPECTED).hausdorff_distance(LineString(merged.vertices))
|
||||
assert gap < 30.0 # 예정노선 따름(토공 최소 후보처럼 멀리 안 감)
|
||||
|
||||
|
||||
def test_체인_follow_추천_한_함수():
|
||||
land = _land()
|
||||
chain = followed_table(EXPECTED, RADIUS, land)[0]
|
||||
follow = follow_route_line(EXPECTED, RADIUS, land)
|
||||
shown = generate_initial_route(
|
||||
EXPECTED,
|
||||
land.terrain,
|
||||
[],
|
||||
mode="follow",
|
||||
weights=None,
|
||||
max_grade=0.14,
|
||||
min_radius_m=RADIUS,
|
||||
with_profile=False,
|
||||
)
|
||||
want = _nodes(chain)
|
||||
assert [(round(n["x"], 3), round(n["y"], 3)) for n in follow["nodes"]] == want
|
||||
assert [(round(n["x"], 3), round(n["y"], 3)) for n in shown["nodes"]] == want
|
||||
assert follow["curve_count"] == len(chain.curves) == len(shown["curves"])
|
||||
|
||||
|
||||
def test_지형_없으면_옛_길():
|
||||
fit = snapped_initial(EXPECTED, RADIUS)
|
||||
line = followed_table(EXPECTED, RADIUS)[0]
|
||||
assert fit is not None
|
||||
assert [(round(x, 3), round(y, 3)) for x, y in fit[0]] == _nodes(line)
|
||||
|
||||
|
||||
def test_부르는_자리_지형_넘김():
|
||||
def src(name: str) -> str:
|
||||
return (ROOT / name).read_text(encoding="utf-8")
|
||||
|
||||
follow = src("B05_Profile/B05_Profile_Router_RouteFollow.py")
|
||||
assert "land = await follow_land(project_id, paths[0])" in follow
|
||||
assert "follow_route_line, points, radius_m, land" in follow
|
||||
replan = src("B05_Profile/B05_Profile_Router_Replan.py")
|
||||
assert replan.count("land = await initial_land(project_id, project_root)") == 2
|
||||
assert "followed_table(points, radius_m, land)" in replan
|
||||
chain = src("B03_FileInput/B03_FileInput_Service_Chain.py")
|
||||
assert "_ensure_planned_initial, project_root, radius_m, land" in chain
|
||||
engine = src("B05_Profile/B05_Profile_Engine_RouteInitial.py")
|
||||
assert "merged_follow(expected, min_radius_m, land, grid)" in engine
|
||||
Reference in New Issue
Block a user