eomsangdonandClaude Opus 5
0979927dfc
[초기화] 후 관 지점 횡단(비정규 측점)이 통째로 사라지던 문제. 측점 정본은 longitudinal.json stations + cross_*.json 파일인데, 그 파일을 쓰는 run_section_generation 에는 공급자가 없었다. 유일한 공급자가 프론트 [임시저장]이라 초기화·반폭 재생성이 파일을 다시 쓸 때마다 측점이 날아갔다 (실측: 초기화 직후 cross 23 -> 19, irregular 4 -> 0). - resolve_extra_stations 신설 — 관 정본(pipe_points.json) + 구조물 정본 A군에서 매번 파생한다. 라벨 형식은 프론트 structureLabel()과 같다. 물넘이포장은 비정규 측점으로 올리지 않는다(2026-08-28 사용자 확정). - run_section_generation 이 generate_sections 앞에서 이 목록을 심는다. 호출자가 넘긴 extra_stations 는 무시한다 — DB 스냅샷으로 되쓰이는 값이라 되먹이면 B04에서 지운 관이 파일 정본을 이기고 부활한다. - _load_pipe_anchors -> _load_pipe_points. 계획선 정착과 측점 생성이 같은 한 번의 읽기를 쓴다. 지문 불일치 경고에 버린 건수를 남겨 침묵 실패를 없앤다. - B06_Section_Router._extra_stations 삭제(자기참조라 늘 빈 값이었다). 706 -> 689줄로 700줄 규칙 위반도 함께 해소. - 격자와 같은 정수 미터인 비정규 측점은 격자로 스냅한다 — cross_*.json 파일명이 정수 미터라 측점 수와 파일 수가 어긋났다. - 구조물 이관 필터를 A군 6종 라벨로 넓힌다 — 서버가 심은 세월교·BOX암거 라벨을 "기타"로 재이관하던 고스트를 막는다. 검증: pytest 229 passed / 7 skipped(신규 14건). 공용 브라우저 실측 — 초기화 직후 임시저장 없이 cross 23 / irregular 4(라벨 일치), B06 카드 23장, 반폭 재생성 20->25->20 에도 23 유지, DB cross_sections 23행. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
456 lines
20 KiB
Python
456 lines
20 KiB
Python
"""B05 경로 계산 후 종횡단을 생성하는 엔진 오케스트레이터.
|
|
|
|
확정 경로 GeoJSON과 확정 지표면 모델 sampler로 종단·횡단을 생성하고, 종단은
|
|
longitudinal/, 각 횡단은 cross_sections/ 아래 파일로 저장한다. DB 기록용
|
|
데이터(경로·요약·측점별 파일)를 준비해 반환한다. 라우터에서 asyncio.to_thread로
|
|
호출한다.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from B05_Profile.B05_Profile_Engine_Grade import GradeDesignOptions, design_grade_line
|
|
from B05_Profile.B05_Profile_Engine_Grade_Profile import (
|
|
design_alignment_profile,
|
|
design_pipe_anchored_profile,
|
|
)
|
|
from B05_Profile.B05_Profile_Engine_Sections_Core import (
|
|
SectionGenerationOptions,
|
|
generate_sections,
|
|
)
|
|
from B05_Profile.B05_Profile_Structures_Repository import load_structures
|
|
from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
|
|
from common_util.common_util_drainage_pipes import (
|
|
PIPE_FACILITY_FORD_PAVEMENT,
|
|
PIPE_FACILITY_PIPE,
|
|
PipePoint,
|
|
parse_pipe_points,
|
|
pipe_anchor_clearances,
|
|
route_signature,
|
|
)
|
|
from common_util.common_util_json import atomic_write_json
|
|
from common_util.common_util_route_geometry import RouteVertex
|
|
from common_util.common_util_surface_sampler import build_surface_sampler
|
|
from config.config_system import (
|
|
DRAINAGE_CACHE_DIRNAME,
|
|
DRAINAGE_EDITS_DIRNAME,
|
|
DRAINAGE_PIPE_POINTS_FILENAME,
|
|
FOREST_ROAD_PROFILE_CRITERIA,
|
|
PIPE_DEFAULT_DIAMETER_MM,
|
|
PIPE_DEFAULT_TYPE,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_STAGE_SUBDIR = Path("B06_Section")
|
|
_MODELS_SUBDIR = Path("B04_PreProcess") / "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 _load_pipe_points(project_root: Path, polyline: list[list[float]]) -> list[PipePoint]:
|
|
"""배수유역도가 확정한 계곡 통과 시설(관 지점 정본)을 읽는다.
|
|
|
|
계획선 정착과 구조물 측점 생성이 **같은 한 번의 읽기**를 쓴다 — 따로 읽으면 계획선이
|
|
물린 자리와 측점 자리가 어긋난다.
|
|
|
|
관 지점 파일에는 저장 당시 노선 지문이 함께 있다 — 노선이 바뀌었으면 버린다
|
|
(옛 노선의 배관 자리로 계획선을 앉히면 전부 어긋난다). 파일이 없거나 못 읽으면 빈 목록.
|
|
"""
|
|
path = (
|
|
project_root
|
|
/ "B04_PreProcess"
|
|
/ DRAINAGE_CACHE_DIRNAME
|
|
/ DRAINAGE_EDITS_DIRNAME
|
|
/ DRAINAGE_PIPE_POINTS_FILENAME
|
|
)
|
|
if not path.is_file():
|
|
return []
|
|
try:
|
|
document = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
logger.warning("B05 계획선: 관 지점 파일을 읽지 못했습니다 (%s)", path)
|
|
return []
|
|
vertices = [
|
|
RouteVertex(
|
|
x=float(p[0]), y=float(p[1]), z=float(p[2]) if len(p) > 2 else 0.0, chainage_m=0.0
|
|
)
|
|
for p in polyline
|
|
]
|
|
if str(document.get("route_signature") or "") != route_signature(vertices):
|
|
# 계획선 정착과 구조물 측점이 함께 빠지므로 버린 건수를 남긴다(침묵 실패 금지).
|
|
logger.warning(
|
|
"B05 계획선: 노선 지문이 달라 저장된 관 지점 %d건을 쓰지 않습니다.",
|
|
len(document.get("points") or []),
|
|
)
|
|
return []
|
|
return parse_pipe_points(document.get("points"))
|
|
|
|
|
|
def resolve_extra_stations(
|
|
project_root: Path, pipes: list[PipePoint]
|
|
) -> tuple[tuple[float, str], ...]:
|
|
"""구조물(비정규) 측점의 **단일 공급자** — 관 정본 + 구조물 정본 A군에서 파생한다.
|
|
|
|
종전에는 이 목록을 서버에 알려 주는 곳이 프론트 [임시저장] 한 줄뿐이었다. 그래서
|
|
초기화·재생성이 종횡단 파일을 다시 쓸 때마다 구조물 측점이 통째로 사라졌다
|
|
(2026-08-28 사용자 보고). 정본 파일은 초기화에도 살아남으므로 여기서 매번 다시 만든다
|
|
— B05·B06은 같은 응답 하나를 보는 한 페이지라 공급자도 하나여야 한다.
|
|
|
|
물넘이포장은 비정규 측점으로 올리지 않는다 — 종단 그래프 하단 구조물 표시로만 다룬다
|
|
(2026-08-28 사용자 확정).
|
|
|
|
라벨 형식은 프론트 structureLabel()과 같아야 한다 — 재진입 이관이 라벨 문자열로 종류를
|
|
되짚기 때문이다(B05_Profile_Structures_Migration).
|
|
"""
|
|
types = structure_type_map()
|
|
derived: list[tuple[float, str]] = []
|
|
for pipe in pipes:
|
|
if pipe.facility == PIPE_FACILITY_FORD_PAVEMENT:
|
|
continue
|
|
if pipe.facility == PIPE_FACILITY_PIPE:
|
|
values = pipe.options or {}
|
|
kind = str(values.get("pipe_kind") or PIPE_DEFAULT_TYPE)
|
|
diameter = int(float(values.get("pipe_diameter_mm") or PIPE_DEFAULT_DIAMETER_MM))
|
|
label = f"{kind} D{diameter}"
|
|
else:
|
|
definition = types.get(pipe.facility)
|
|
if definition is None:
|
|
continue
|
|
label = definition.name
|
|
derived.append((float(pipe.chainage_m), label))
|
|
# A군 수동 구조물(노출형 횡단수로·개거)도 같은 자리에 측점이 필요하다. 관 정본이
|
|
# 관리하는 타입(managed_by)은 structures.json에 저장되지 않아 중복되지 않는다.
|
|
try:
|
|
for structure in load_structures(str(project_root))[1]:
|
|
definition = types.get(structure.type_id)
|
|
if definition and definition.group == "A" and not definition.managed_by:
|
|
derived.append((float(structure.anchor_m()), definition.name))
|
|
except Exception:
|
|
logger.exception("B05: 구조물 정본을 읽지 못했습니다 (관 유래 측점만 씁니다)")
|
|
return tuple(sorted(derived))
|
|
|
|
|
|
def _append_design_profiles(
|
|
longitudinal: dict[str, Any],
|
|
grade_options: GradeDesignOptions | None,
|
|
station_interval_m: float | None = None,
|
|
pipe_chainages: list[float] | None = None,
|
|
pipe_clearances: dict[float, float] | None = None,
|
|
) -> dict[str, Any] | None:
|
|
"""종단 계획선을 산출해 longitudinal에 붙이고 요약을 반환한다.
|
|
|
|
1순위는 **배관 정착 선형** — 배수유역도의 배관 배치 측점마다 계획선이 지면선과
|
|
만나도록 직선으로 잇고 기본 R을 얹는다(2026-08-03 사용자 확정). 배관이 없거나
|
|
산출이 불가하면 2순위로 기존 지반 추종 직선 분할 선형(`design_alignment_profile`),
|
|
그마저 실패하면 구 균형 최적화 계획선으로 폴백하고(편집 불가), 모두 실패해도
|
|
종횡단 생성 자체는 유지한다.
|
|
"""
|
|
longitudinal.setdefault("design_profiles", [])
|
|
if grade_options is None:
|
|
return None
|
|
alignment = None
|
|
profile = None
|
|
if pipe_chainages:
|
|
try:
|
|
alignment, profile = design_pipe_anchored_profile(
|
|
longitudinal,
|
|
grade_options,
|
|
pipe_chainages,
|
|
station_interval_m=station_interval_m,
|
|
pipe_clearances=pipe_clearances,
|
|
)
|
|
except (ValueError, KeyError, ArithmeticError):
|
|
logger.exception("B05 배관 정착 계획선 산출 실패 — 직선 분할 선형으로 대체")
|
|
if profile is None:
|
|
try:
|
|
alignment, profile = design_alignment_profile(
|
|
longitudinal, grade_options, station_interval_m=station_interval_m
|
|
)
|
|
except (ValueError, KeyError, ArithmeticError):
|
|
logger.exception("B05 계획선 선형 산출 실패 — 균형 최적화 계획선으로 대체")
|
|
try:
|
|
profile = design_grade_line(longitudinal, grade_options)
|
|
except (ValueError, KeyError, ArithmeticError):
|
|
logger.exception("B05 종단 계획선 산출 실패 (종횡단은 유지)")
|
|
return None
|
|
if alignment is not None:
|
|
longitudinal["profile_alignment"] = alignment
|
|
longitudinal["design_profiles"].append(profile)
|
|
return {"id": profile["id"], **profile["summary"]}
|
|
|
|
|
|
def _annotate_pavement_suggestions(
|
|
longitudinal: dict[str, Any], grade_options: GradeDesignOptions | None
|
|
) -> None:
|
|
"""계획선 국소 종단경사가 비포장 법정 상한을 넘는 측점에 포장 제안을 표기한다.
|
|
|
|
근거: 「임도설치 및 관리 등에 관한 규정」[별표 1-2] — 기준(설계속도×지형) 초과
|
|
구간은 노면포장 시에 한해 상한(`paved_exception_grade_pct`)까지 허용된다.
|
|
stations에 `pavement_suggested`(bool)와 판정 근거(경사·상한)를 남기며, B06이
|
|
포장 기본값과 법정 근거 문구 표기에 그대로 사용한다. 실패해도 종횡단은 유지한다.
|
|
"""
|
|
if grade_options is None:
|
|
return
|
|
profiles = longitudinal.get("design_profiles") or []
|
|
samples = profiles[0].get("samples", []) if profiles else []
|
|
points = [
|
|
(float(s["chainage_m"]), float(s["elevation_m"]))
|
|
for s in samples
|
|
if isinstance(s.get("chainage_m"), (int, float))
|
|
and isinstance(s.get("elevation_m"), (int, float))
|
|
]
|
|
stations = longitudinal.get("stations")
|
|
if len(points) < 2 or not isinstance(stations, list):
|
|
return
|
|
criteria = FOREST_ROAD_PROFILE_CRITERIA["design_speed"].get(grade_options.design_speed_kph)
|
|
if not criteria:
|
|
return
|
|
terrain = grade_options.terrain_type if grade_options.terrain_type else "normal"
|
|
unpaved_limit = float(
|
|
criteria["max_grade_pct"].get(terrain, criteria["max_grade_pct"]["normal"])
|
|
)
|
|
|
|
def local_grade_pct(chainage: float) -> float:
|
|
"""측점을 감싸는 인접 계획선 구간들의 경사 중 최댓값(절댓값 %)을 돌려준다."""
|
|
worst = 0.0
|
|
for index in range(1, len(points)):
|
|
c0, z0 = points[index - 1]
|
|
c1, z1 = points[index]
|
|
if c1 < chainage - 1e-6 or c0 > chainage + 1e-6:
|
|
continue
|
|
span = c1 - c0
|
|
if span <= 1e-9:
|
|
continue
|
|
worst = max(worst, abs((z1 - z0) / span) * 100.0)
|
|
return worst
|
|
|
|
for station in stations:
|
|
chainage = station.get("chainage_m")
|
|
if not isinstance(chainage, (int, float)):
|
|
continue
|
|
grade_pct = local_grade_pct(float(chainage))
|
|
station["pavement_suggested"] = grade_pct > unpaved_limit + 1e-6
|
|
station["pavement_grade_pct"] = round(grade_pct, 2)
|
|
station["pavement_grade_limit_pct"] = round(unpaved_limit, 2)
|
|
|
|
|
|
def run_section_generation(
|
|
project_root: Path,
|
|
route_data_path: str,
|
|
filter_key: str,
|
|
method: str,
|
|
smooth: bool,
|
|
*,
|
|
options: SectionGenerationOptions | None = None,
|
|
grade_options: GradeDesignOptions | None = None,
|
|
grade_overrides: dict[str, Any] | None = None,
|
|
crs: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""종횡단을 생성·저장하고 DB 기록용 데이터를 반환한다.
|
|
|
|
비정규(구조물) 측점은 정본 파일에서 해석하며 **호출자 옵션의 extra_stations는 무시한다**
|
|
(단일 공급자 — resolve_extra_stations).
|
|
|
|
반환 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)
|
|
|
|
# 구조물(비정규) 측점의 단일 공급자 — 정본 파일에서 매번 다시 만든다. 호출자가 넘긴
|
|
# extra_stations는 쓰지 않는다: 이 옵션은 아래 long_summary["options"]로 DB에 되쓰이는
|
|
# 값이라, 되먹이면 B04에서 지운 관이 파일 정본을 이기고 되살아난다.
|
|
pipes = _load_pipe_points(project_root, polyline)
|
|
try:
|
|
extras = resolve_extra_stations(project_root, pipes)
|
|
except Exception:
|
|
logger.exception("B05 구조물 측점 해석 실패 (규칙 격자만 생성)")
|
|
extras = ()
|
|
logger.info("B05 종횡단: 구조물 측점 %d건 파생(관 %d건)", len(extras), len(pipes))
|
|
|
|
result = generate_sections(
|
|
polyline,
|
|
sampler,
|
|
replace(options or SectionGenerationOptions(), extra_stations=extras),
|
|
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)
|
|
|
|
# 종단면 저장 (계획선은 저장 직전에 종단 데이터에 붙인다)
|
|
grade_summary = _append_design_profiles(
|
|
result["longitudinal"],
|
|
grade_options,
|
|
(result.get("options") or {}).get("station_interval_m"),
|
|
# 1차 계획선의 변화점 = 배수유역도가 확정한 배관 배치 측점(측점 생성과 같은 목록).
|
|
pipe_chainages=[pipe.chainage_m for pipe in pipes],
|
|
# 시설 제원이 요구하는 최소 여유(관경+토피 등) — 계획선이 그만큼 들려야 관·구체가
|
|
# 들어갈 자리가 생긴다(2026-08-23 사용자 지시).
|
|
pipe_clearances=dict(pipe_anchor_clearances(pipes)),
|
|
)
|
|
# 계획선 경사 기반 포장 제안(법정 상한 초과 측점) — 비치명적.
|
|
try:
|
|
_annotate_pavement_suggestions(result["longitudinal"], grade_options)
|
|
except Exception:
|
|
logger.exception("B05 포장 제안 판정 실패 (종횡단은 유지)")
|
|
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"],
|
|
# 해석이 끝난 기준값은 기록·표시용으로만 보관한다. 이 값을 재계산 시
|
|
# 폴백으로 되쓰면 등급·지형을 바꿔도 옛 기본값이 법정값을 이겨 갱신되지 않는다.
|
|
"grade_options": grade_options.as_dict() if grade_options else None,
|
|
# 재계산 폴백에 쓰는 단일 소스: 사용자가 명시적으로 입력한 값만 담는다.
|
|
"grade_overrides": {
|
|
key: value for key, value in (grade_overrides or {}).items() if value is not None
|
|
},
|
|
"grade_summary": grade_summary,
|
|
}
|
|
|
|
# 측점별 횡단면 저장 (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,
|
|
}
|
|
|
|
|
|
def generate_irregular_sections(
|
|
project_root: Path,
|
|
route_data_path: str,
|
|
filter_key: str,
|
|
method: str,
|
|
smooth: bool,
|
|
*,
|
|
extra_stations: tuple[tuple[float, str], ...],
|
|
options: SectionGenerationOptions | None = None,
|
|
crs: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""비정규 측점만 샘플링해 **횡단 파일을 쓰고** 측점 dict 목록을 돌려준다.
|
|
|
|
기존 종단 계획선·규칙 측점·횡단은 건드리지 않는다. 격자 측점과 똑같은 접선·오프셋
|
|
샘플링을 거치되(`generate_sections` 재사용) kind="irregular"인 것만 골라내 각각
|
|
`cross_*.json`으로 저장한다. B06 상세 조회가 이 폴더를 glob하고 종단 파일의 stations로
|
|
필터링하므로, 이 파일들 + 종단 stations 병합만으로 다음 페이지에 횡단이 나타난다(DB 불필요).
|
|
"""
|
|
if not extra_stations:
|
|
return []
|
|
base = options or SectionGenerationOptions()
|
|
merged_options = SectionGenerationOptions(
|
|
station_interval_m=base.station_interval_m,
|
|
cross_half_width_m=base.cross_half_width_m,
|
|
cross_sample_interval_m=base.cross_sample_interval_m,
|
|
long_sample_interval_m=base.long_sample_interval_m,
|
|
include_endpoint=base.include_endpoint,
|
|
extra_stations=tuple(extra_stations),
|
|
)
|
|
polyline = _load_route_polyline(project_root, route_data_path)
|
|
sampler = build_surface_sampler(project_root / _MODELS_SUBDIR, filter_key, method, smooth)
|
|
result = generate_sections(
|
|
polyline,
|
|
sampler,
|
|
merged_options,
|
|
source_snapshot={"filter": filter_key, "method": method, "smooth": smooth},
|
|
crs=crs,
|
|
)
|
|
irregular_stations = [
|
|
station
|
|
for station in result["longitudinal"]["stations"]
|
|
if station.get("kind") == "irregular"
|
|
]
|
|
if not irregular_stations:
|
|
return []
|
|
cross_dir = project_root / _STAGE_SUBDIR / "cross_sections"
|
|
cross_dir.mkdir(parents=True, exist_ok=True)
|
|
for cross_section in result["cross_sections"]:
|
|
if cross_section.get("kind") != "irregular":
|
|
continue
|
|
cross_file = cross_dir / cross_filename(float(cross_section["chainage_m"]))
|
|
atomic_write_json(cross_file, cross_section)
|
|
return irregular_stations
|