121 lines
4.2 KiB
Python
121 lines
4.2 KiB
Python
"""B05 경로 설계 엔진 오케스트레이터.
|
|
|
|
경로점(BP/CP/EP/AP/FP)과 옵션을 받아 최적 경로를 계산하고, 폴리라인을
|
|
GeoJSON으로 저장하며 DB 기록용 데이터(메타·렌더링 샘플·통계)를 준비한다.
|
|
라우터에서 asyncio.to_thread로 호출한다.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from B05_wf2_Route.B05_wf2_Route_Engine_RidgeValley import solve_ridge_valley_route
|
|
from B05_wf2_Route.B05_wf2_Route_Engine_Solver import solve_optimal_route
|
|
from common_util.common_util_json import atomic_write_json
|
|
|
|
_ROUTE_SUBDIR = Path("B05_wf2_Route") / "route"
|
|
# route_points 테이블에 저장할 렌더링 샘플 최대 개수
|
|
_MAX_RENDER_POINTS = 500
|
|
|
|
|
|
def _route_geojson(polyline: list[list[float]]) -> dict[str, Any]:
|
|
"""폴리라인을 3D LineString GeoJSON Feature로 변환한다."""
|
|
return {
|
|
"type": "Feature",
|
|
"geometry": {
|
|
"type": "LineString",
|
|
"coordinates": [[round(x, 3), round(y, 3), round(z, 3)] for x, y, z in polyline],
|
|
},
|
|
"properties": {},
|
|
}
|
|
|
|
|
|
def _sample_render_points(
|
|
polyline: list[list[float]], chainage_m: list[float], maximum: int
|
|
) -> list[dict[str, Any]]:
|
|
"""폴리라인을 최대 maximum개로 균등 샘플링해 렌더링용 포인트를 만든다."""
|
|
n = len(polyline)
|
|
if n == 0:
|
|
return []
|
|
if n <= maximum:
|
|
indices = range(n)
|
|
else:
|
|
step = n / maximum
|
|
indices = (int(i * step) for i in range(maximum))
|
|
|
|
points: list[dict[str, Any]] = []
|
|
for seq, idx in enumerate(indices):
|
|
idx = min(idx, n - 1)
|
|
_, _, z = polyline[idx]
|
|
# 국소 경사(%) — 직전 정점과의 차이
|
|
slope_pct = 0.0
|
|
if idx > 0:
|
|
x0, y0, z0 = polyline[idx - 1]
|
|
x1, y1, z1 = polyline[idx]
|
|
h = ((x1 - x0) ** 2 + (y1 - y0) ** 2) ** 0.5
|
|
if h > 1e-6:
|
|
slope_pct = abs(z1 - z0) / h * 100.0
|
|
points.append(
|
|
{
|
|
"chainage_m": round(chainage_m[idx], 3) if idx < len(chainage_m) else None,
|
|
"elevation_m": round(z, 3),
|
|
"slope_percent": round(slope_pct, 3),
|
|
"sequence_num": seq,
|
|
}
|
|
)
|
|
return points
|
|
|
|
|
|
def run_route_design(
|
|
project_root: Path,
|
|
filter_key: str,
|
|
method: str,
|
|
smooth: bool,
|
|
points_data: dict[str, Any],
|
|
options: dict[str, Any],
|
|
algorithm: str = "dijkstra",
|
|
) -> dict[str, Any]:
|
|
"""경로 탐색을 실행하고 GeoJSON 저장 + DB 기록용 데이터를 반환한다.
|
|
|
|
algorithm: "dijkstra"(격자 Dijkstra) 또는 "ridge_valley"(능선-계곡 정속경사).
|
|
|
|
반환 dict:
|
|
- route_data_path: 저장한 GeoJSON의 프로젝트 상대 경로
|
|
- solver_result: solver 원본 결과 (polyline, metrics, segments 등)
|
|
- render_points: route_points 테이블 저장용 샘플
|
|
- statistics: route_statistics 저장용 요약
|
|
"""
|
|
if algorithm == "ridge_valley":
|
|
result = solve_ridge_valley_route(
|
|
project_root, filter_key, smooth, points_data, options, method=method
|
|
)
|
|
else:
|
|
result = solve_optimal_route(
|
|
project_root, filter_key, smooth, points_data, options, method=method
|
|
)
|
|
polyline = result["polyline"]
|
|
chainage_m = result["chainage_m"]
|
|
|
|
route_dir = project_root / _ROUTE_SUBDIR
|
|
route_dir.mkdir(parents=True, exist_ok=True)
|
|
geojson_path = route_dir / "route_main.geojson"
|
|
atomic_write_json(geojson_path, _route_geojson(polyline))
|
|
|
|
# 통계 요약 (solver 메트릭에서 파생)
|
|
metrics = result["metrics"]
|
|
statistics = {
|
|
"min_slope": 0.0,
|
|
"max_slope": metrics.get("max_grade_pct"),
|
|
"mean_slope": metrics.get("avg_grade_pct"),
|
|
"cost_score": None,
|
|
}
|
|
|
|
return {
|
|
"route_data_path": geojson_path.relative_to(project_root).as_posix(),
|
|
"solver_result": result,
|
|
"render_points": _sample_render_points(polyline, chainage_m, _MAX_RENDER_POINTS),
|
|
"statistics": statistics,
|
|
"grade_percent": [seg.get("max_grade_pct") for seg in result.get("segments", [])],
|
|
"constraints": result.get("conditions_snapshot", {}),
|
|
"algorithm_params": options.get("weights") or {},
|
|
}
|