147 lines
5.7 KiB
Python
147 lines
5.7 KiB
Python
"""B05 경로 계산 후 종횡단을 생성하는 엔진 오케스트레이터.
|
|
|
|
확정 경로 GeoJSON과 확정 지표면 모델 sampler로 종단·횡단을 생성하고, 종단은
|
|
longitudinal/, 각 횡단은 cross_sections/ 아래 파일로 저장한다. DB 기록용
|
|
데이터(경로·요약·측점별 파일)를 준비해 반환한다. 라우터에서 asyncio.to_thread로
|
|
호출한다.
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from B05_wf2_Route.B05_wf2_Route_Engine_Sections_Core import (
|
|
SectionGenerationOptions,
|
|
generate_sections,
|
|
)
|
|
from B05_wf2_Route.B05_wf2_Route_Engine_Sections_Sampler import build_surface_sampler
|
|
from common_util.common_util_json import atomic_write_json
|
|
|
|
_STAGE_SUBDIR = Path("B06_wf3_ProfileCross")
|
|
_MODELS_SUBDIR = Path("B04_wf1_Surface") / "models"
|
|
|
|
|
|
def _load_route_polyline(project_root: Path, route_data_path: str) -> list[list[float]]:
|
|
"""B05가 저장한 경로 GeoJSON에서 3D 폴리라인 좌표열을 로드한다."""
|
|
geojson_path = project_root / Path(route_data_path)
|
|
if not geojson_path.is_file():
|
|
raise FileNotFoundError(f"경로 GeoJSON을 찾을 수 없습니다: {route_data_path}")
|
|
geo = json.loads(geojson_path.read_text(encoding="utf-8"))
|
|
coords = geo.get("geometry", {}).get("coordinates")
|
|
if not coords or len(coords) < 2:
|
|
raise ValueError("경로 GeoJSON에 유효한 LineString 좌표가 없습니다.")
|
|
return [[float(c[0]), float(c[1]), float(c[2]) if len(c) > 2 else 0.0] for c in coords]
|
|
|
|
|
|
def cross_filename(chainage_m: float) -> str:
|
|
"""측점 chainage에 대응하는 횡단면 파일명(단일 규칙)."""
|
|
return f"cross_{int(round(float(chainage_m))):05d}m.json"
|
|
|
|
|
|
def prune_stale_cross_files(cross_dir: Path, stations: list[Any]) -> set[str]:
|
|
"""stations에 없는 잔재 cross_*.json을 삭제하고 유효 파일명 집합을 반환한다.
|
|
|
|
측점 정보가 비어 있으면 오삭제를 피하기 위해 아무것도 지우지 않고
|
|
빈 집합을 반환한다(호출부는 빈 집합이면 필터를 생략한다).
|
|
"""
|
|
valid = {
|
|
cross_filename(station["chainage_m"])
|
|
for station in stations
|
|
if isinstance(station, dict) and isinstance(station.get("chainage_m"), (int, float))
|
|
}
|
|
if not valid:
|
|
return valid
|
|
for path in cross_dir.glob("cross_*.json"):
|
|
if path.name not in valid:
|
|
path.unlink(missing_ok=True)
|
|
return valid
|
|
|
|
|
|
def _cross_summary(cross_section: dict[str, Any]) -> dict[str, Any]:
|
|
"""횡단면 상세에서 DB data 컬럼에 저장할 요약을 만든다."""
|
|
samples = cross_section.get("samples", [])
|
|
valid_z = [
|
|
s["elevation_m"] for s in samples if s.get("valid") and s.get("elevation_m") is not None
|
|
]
|
|
return {
|
|
"chainage_m": cross_section.get("chainage_m"),
|
|
"center_z": cross_section.get("center_z"),
|
|
"azimuth_deg": cross_section.get("azimuth_deg"),
|
|
"sample_count": len(samples),
|
|
"min_elevation_m": min(valid_z) if valid_z else None,
|
|
"max_elevation_m": max(valid_z) if valid_z else None,
|
|
}
|
|
|
|
|
|
def run_section_generation(
|
|
project_root: Path,
|
|
route_data_path: str,
|
|
filter_key: str,
|
|
method: str,
|
|
smooth: bool,
|
|
*,
|
|
options: SectionGenerationOptions | None = None,
|
|
crs: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""종횡단을 생성·저장하고 DB 기록용 데이터를 반환한다.
|
|
|
|
반환 dict:
|
|
- longitudinal: {file_path, data(요약)}
|
|
- cross_sections: [{chainage_m, sequence_num, data(요약), cross_section_file_path}]
|
|
- result: generate_sections 원본 결과
|
|
"""
|
|
polyline = _load_route_polyline(project_root, route_data_path)
|
|
models_dir = project_root / _MODELS_SUBDIR
|
|
sampler = build_surface_sampler(models_dir, filter_key, method, smooth)
|
|
|
|
result = generate_sections(
|
|
polyline,
|
|
sampler,
|
|
options,
|
|
source_snapshot={"filter": filter_key, "method": method, "smooth": smooth},
|
|
crs=crs,
|
|
)
|
|
|
|
stage_root = project_root / _STAGE_SUBDIR
|
|
long_dir = stage_root / "longitudinal"
|
|
cross_dir = stage_root / "cross_sections"
|
|
long_dir.mkdir(parents=True, exist_ok=True)
|
|
cross_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 종단면 저장
|
|
long_file = long_dir / "longitudinal.json"
|
|
atomic_write_json(long_file, result["longitudinal"])
|
|
long_summary = {
|
|
"length_m": result["longitudinal"]["length_m"],
|
|
"station_count": result["summary"]["station_count"],
|
|
"invalid_samples": result["summary"]["invalid_longitudinal_samples"],
|
|
# 사용자 선택값의 단일 소스(DB): 재생성·재탐색 시 이 값을 우선 사용한다.
|
|
"options": result["options"],
|
|
}
|
|
|
|
# 측점별 횡단면 저장 (detail 조회가 폴더 전체를 glob하므로 이전 실행 잔재를 먼저 비운다)
|
|
for stale_file in cross_dir.glob("cross_*.json"):
|
|
stale_file.unlink()
|
|
cross_records: list[dict[str, Any]] = []
|
|
for seq, cross_section in enumerate(result["cross_sections"]):
|
|
chainage = float(cross_section["chainage_m"])
|
|
cross_file = cross_dir / cross_filename(chainage)
|
|
atomic_write_json(cross_file, cross_section)
|
|
cross_records.append(
|
|
{
|
|
"chainage_m": chainage,
|
|
"sequence_num": seq,
|
|
"data": _cross_summary(cross_section),
|
|
"cross_section_file_path": cross_file.relative_to(project_root).as_posix(),
|
|
}
|
|
)
|
|
|
|
return {
|
|
"longitudinal": {
|
|
"file_path": long_file.relative_to(project_root).as_posix(),
|
|
"data": long_summary,
|
|
},
|
|
"cross_sections": cross_records,
|
|
"result": result,
|
|
}
|