472 lines
20 KiB
Python
472 lines
20 KiB
Python
"""B05 종단 계획선 선형(직선 + 측점 위 종단곡선) 파생 모듈.
|
||
|
||
계획선을 "폴리라인 샘플"이 아니라 **변화점(PVI) 구조**로 1차 표현하고, 도면
|
||
테이블(구배·절토고·성토고·계획고·지반고·누가거리·거리·측점·곡선)에 필요한 값을
|
||
전부 여기서 파생시킨다. 폴리라인 샘플은 이 구조에서 만들어지므로 B06/B07은
|
||
기존 `design_profiles[].samples` 계약을 그대로 쓴다.
|
||
|
||
기하 규칙 (사용자 확정 사항):
|
||
- 변화점은 **기준 측점 위에만** 놓인다 → 곡선 중심이 측점 수직선상에 있다.
|
||
- 종단곡선은 변화점 대칭 배치이며 좌우에 직선 구간이 반드시 남는다.
|
||
- 곡선 기본 **반경** R = 측점간격 × `curve_radius_ratio`, 길이 L = R × |A| 로 파생.
|
||
|
||
수치 최적화(직선 분할 DP·표고 결정)는 [[B05_wf2_Route_Engine_Grade_Solver]]가,
|
||
기준 해석과 오케스트레이션은 [[B05_wf2_Route_Engine_Grade]]가 담당한다. 이 모듈은
|
||
scipy에 의존하지 않는 순수 기하 계산만 두어 프론트엔드 구현과 1:1로 대응시킨다.
|
||
"""
|
||
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
|
||
import numpy as np
|
||
|
||
from config.config_system import FOREST_ROAD_PROFILE_ALIGNMENT
|
||
|
||
ALIGNMENT_SCHEMA_VERSION = 1
|
||
# chainage를 dict 키로 쓸 때의 표기. 프론트엔드(`toFixed(3)`)와 반드시 같아야 한다.
|
||
_KEY_DECIMALS = 3
|
||
|
||
|
||
def chainage_key(value: float) -> str:
|
||
"""chainage를 편집 델타 dict의 키 문자열로 바꾼다(프론트와 동일 규칙)."""
|
||
return f"{float(value):.{_KEY_DECIMALS}f}"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class AlignmentPolicy:
|
||
"""계획선 선형·편집 정책. config 기본값 위에 노선별 해석값을 얹는다."""
|
||
|
||
station_interval_m: float
|
||
curve_radius_ratio: float
|
||
curve_tangent_max_ratio: float
|
||
curve_skip_legal_exception: bool
|
||
balance_tolerance_percent: float
|
||
pvi_penalty_m2: float
|
||
min_segment_stations: int
|
||
edit_step_m: float
|
||
grade_violation_policy: str
|
||
max_grade_pct: float
|
||
curve_skip_delta_pct: float
|
||
paved: bool
|
||
|
||
@classmethod
|
||
def from_config(
|
||
cls,
|
||
*,
|
||
station_interval_m: float,
|
||
max_grade_pct: float,
|
||
curve_skip_delta_pct: float,
|
||
paved: bool,
|
||
) -> "AlignmentPolicy":
|
||
config = FOREST_ROAD_PROFILE_ALIGNMENT
|
||
return cls(
|
||
station_interval_m=float(station_interval_m),
|
||
curve_radius_ratio=float(config["curve_radius_ratio"]),
|
||
curve_tangent_max_ratio=float(config["curve_tangent_max_ratio"]),
|
||
curve_skip_legal_exception=bool(config["curve_skip_legal_exception"]),
|
||
balance_tolerance_percent=float(config["balance_tolerance_percent"]),
|
||
pvi_penalty_m2=float(config["pvi_penalty_m2"]),
|
||
min_segment_stations=int(config["min_segment_stations"]),
|
||
edit_step_m=float(config["edit_step_m"]),
|
||
grade_violation_policy=str(config["grade_violation_policy"]),
|
||
max_grade_pct=float(max_grade_pct),
|
||
curve_skip_delta_pct=float(curve_skip_delta_pct),
|
||
paved=bool(paved),
|
||
)
|
||
|
||
@classmethod
|
||
def from_dict(cls, payload: dict[str, Any]) -> "AlignmentPolicy":
|
||
"""저장된 정책 스냅샷을 복원한다(편집 저장 시 기준을 흔들지 않기 위함)."""
|
||
config = FOREST_ROAD_PROFILE_ALIGNMENT
|
||
return cls(
|
||
station_interval_m=float(payload.get("station_interval_m") or 20.0),
|
||
curve_radius_ratio=float(
|
||
payload.get("curve_radius_ratio", config["curve_radius_ratio"])
|
||
),
|
||
curve_tangent_max_ratio=float(
|
||
payload.get("curve_tangent_max_ratio", config["curve_tangent_max_ratio"])
|
||
),
|
||
curve_skip_legal_exception=bool(
|
||
payload.get("curve_skip_legal_exception", config["curve_skip_legal_exception"])
|
||
),
|
||
balance_tolerance_percent=float(
|
||
payload.get("balance_tolerance_percent", config["balance_tolerance_percent"])
|
||
),
|
||
pvi_penalty_m2=float(config["pvi_penalty_m2"]),
|
||
min_segment_stations=int(config["min_segment_stations"]),
|
||
edit_step_m=float(payload.get("edit_step_m", config["edit_step_m"])),
|
||
grade_violation_policy=str(
|
||
payload.get("grade_violation_policy", config["grade_violation_policy"])
|
||
),
|
||
max_grade_pct=float(payload.get("max_grade_pct") or 9.0),
|
||
curve_skip_delta_pct=float(payload.get("curve_skip_delta_pct") or 5.0),
|
||
paved=bool(payload.get("paved", False)),
|
||
)
|
||
|
||
@property
|
||
def default_curve_radius_m(self) -> float:
|
||
"""측점간격 기준 기본 종단곡선 반경.
|
||
|
||
R을 1차 값으로 고정해야 계획고를 편집해도 곡선 반경이 흔들리지 않는다.
|
||
곡선 길이는 L = R × |대수차| 로 따라 움직인다.
|
||
"""
|
||
return max(1e-6, self.station_interval_m * self.curve_radius_ratio)
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
"""프론트엔드가 동일 기하를 재현하는 데 필요한 상수 묶음."""
|
||
return {
|
||
"station_interval_m": self.station_interval_m,
|
||
"curve_radius_ratio": self.curve_radius_ratio,
|
||
"curve_tangent_max_ratio": self.curve_tangent_max_ratio,
|
||
"curve_skip_legal_exception": self.curve_skip_legal_exception,
|
||
"default_curve_radius_m": self.default_curve_radius_m,
|
||
"balance_tolerance_percent": self.balance_tolerance_percent,
|
||
"edit_step_m": self.edit_step_m,
|
||
"grade_violation_policy": self.grade_violation_policy,
|
||
"max_grade_pct": self.max_grade_pct,
|
||
"curve_skip_delta_pct": self.curve_skip_delta_pct,
|
||
"paved": self.paved,
|
||
}
|
||
|
||
|
||
def tangent_elevation(pvi_s: np.ndarray, pvi_z: np.ndarray, targets: np.ndarray) -> np.ndarray:
|
||
"""종단곡선을 빼고 변화점 직선만으로 본 표고(탄젠트 표고)."""
|
||
return np.interp(targets, pvi_s, pvi_z)
|
||
|
||
|
||
def resolve_pvi(
|
||
base_s: np.ndarray,
|
||
base_z: np.ndarray,
|
||
station_offsets: dict[str, float],
|
||
) -> tuple[np.ndarray, np.ndarray, list[str]]:
|
||
"""자동 변화점에 사용자 편집 측점을 합쳐 최종 변화점 집합을 만든다.
|
||
|
||
편집 델타는 **자동 선형의 탄젠트 표고 기준**으로 해석한다. 이렇게 해야 다른
|
||
측점을 나중에 건드려도 이미 편집한 측점의 표고가 흔들리지 않고, 키를 지우면
|
||
자동 선형으로 정확히 돌아온다(원복).
|
||
"""
|
||
nodes: dict[float, float] = {
|
||
round(float(s), _KEY_DECIMALS): float(z) for s, z in zip(base_s, base_z)
|
||
}
|
||
sources: dict[float, str] = {key: "auto" for key in nodes}
|
||
for key, offset in (station_offsets or {}).items():
|
||
try:
|
||
chainage = round(float(key), _KEY_DECIMALS)
|
||
delta = float(offset)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if not np.isfinite(chainage) or not np.isfinite(delta):
|
||
continue
|
||
if chainage < float(base_s[0]) - 1e-6 or chainage > float(base_s[-1]) + 1e-6:
|
||
continue
|
||
base_value = float(np.interp(chainage, base_s, base_z))
|
||
nodes[chainage] = base_value + delta
|
||
sources[chainage] = "user"
|
||
ordered = sorted(nodes.items())
|
||
pvi_s = np.array([item[0] for item in ordered], dtype=np.float64)
|
||
pvi_z = np.array([item[1] for item in ordered], dtype=np.float64)
|
||
return pvi_s, pvi_z, [sources[value] for value in pvi_s.tolist()]
|
||
|
||
|
||
def build_curves(
|
||
pvi_s: np.ndarray,
|
||
pvi_z: np.ndarray,
|
||
policy: AlignmentPolicy,
|
||
curve_radii: dict[str, float] | None = None,
|
||
) -> tuple[list[dict[str, Any]], list[str]]:
|
||
"""각 변화점에 대칭 종단곡선을 삽입하고 곡선 제원을 만든다.
|
||
|
||
**반경 R이 1차 값**이고 곡선길이는 L = R × |대수차| 로 따라온다. 계획고를 편집하면
|
||
대수차가 바뀌므로 L은 변하지만 R은 지정한 값 그대로 유지된다.
|
||
|
||
곡선 반쪽 길이는 짧은 쪽 인접 직선의 `curve_tangent_max_ratio` 이내로 제한해
|
||
좌우에 직선이 반드시 남게 한다(R을 크게 넣어도 곡선끼리 겹치지 않는다).
|
||
"""
|
||
overrides = curve_radii or {}
|
||
spans = pvi_s[1:] - pvi_s[:-1]
|
||
grades = (pvi_z[1:] - pvi_z[:-1]) / np.where(spans > 0, spans, 1.0)
|
||
curves: list[dict[str, Any]] = []
|
||
warnings: list[str] = []
|
||
skip_delta = policy.curve_skip_delta_pct / 100.0
|
||
for index in range(1, len(pvi_s) - 1):
|
||
grade_in, grade_out = float(grades[index - 1]), float(grades[index])
|
||
delta = grade_out - grade_in
|
||
chainage = float(pvi_s[index])
|
||
key = chainage_key(chainage)
|
||
if abs(delta) < 1e-9:
|
||
continue
|
||
# 법정 다-(3)-(다)는 "종단곡선을 두지 않을 수 있다"는 허용 조항이다.
|
||
# 실무 도면은 대수차가 작아도 변화점을 원곡선으로 처리하므로, 기본은 곡선을
|
||
# 삽입하고 생략 가능 구간이라는 표시만 남긴다(config로 실제 생략 전환 가능).
|
||
skip_allowed = not policy.paved and abs(delta) <= skip_delta + 1e-12
|
||
omitted = skip_allowed and policy.curve_skip_legal_exception
|
||
requested = overrides.get(key)
|
||
try:
|
||
radius = float(requested) if requested is not None else policy.default_curve_radius_m
|
||
except (TypeError, ValueError):
|
||
radius = policy.default_curve_radius_m
|
||
if not np.isfinite(radius) or radius <= 0:
|
||
radius = policy.default_curve_radius_m
|
||
length = radius * abs(delta)
|
||
half_limit = (
|
||
min(float(spans[index - 1]), float(spans[index])) * policy.curve_tangent_max_ratio
|
||
)
|
||
half = min(length / 2.0, half_limit)
|
||
if half <= 1e-9:
|
||
continue
|
||
if length / 2.0 - half > 1e-6:
|
||
warnings.append(
|
||
f"chainage {chainage:.1f}m: 인접 직선이 짧아 종단곡선 반경을 "
|
||
f"{half * 2.0 / abs(delta):.0f}m(길이 {half * 2.0:.1f}m)로 줄였습니다."
|
||
)
|
||
length = half * 2.0
|
||
curves.append(
|
||
{
|
||
"pvi_index": index,
|
||
"chainage_m": chainage,
|
||
"elevation_m": float(pvi_z[index]),
|
||
"grade_in": grade_in,
|
||
"grade_out": grade_out,
|
||
"delta_pct": delta * 100.0,
|
||
"bvc_m": chainage - half,
|
||
"evc_m": chainage + half,
|
||
"length_m": length,
|
||
"radius_m": length / abs(delta),
|
||
# K = 곡선길이 / 기울기 대수차(%) — 도면 주기 표기와 동일 정의
|
||
"k": length / (abs(delta) * 100.0),
|
||
# 중앙종거: 변화점에서 탄젠트와 곡선의 수직 거리 (|A|·L/8)
|
||
"middle_ordinate_m": abs(delta) * length / 8.0,
|
||
# 포물선 삽입으로 직선 폴리라인 대비 발생하는 면적 차 (A·L²/24)
|
||
"area_offset_m2": delta * length * length / 24.0,
|
||
"omitted": bool(omitted),
|
||
"skip_allowed": bool(skip_allowed),
|
||
"omit_reason": (
|
||
f"비포장 대수차 {policy.curve_skip_delta_pct:.0f}% 이하 (법정 생략 가능)"
|
||
if skip_allowed
|
||
else None
|
||
),
|
||
}
|
||
)
|
||
return curves, warnings
|
||
|
||
|
||
def evaluate(
|
||
pvi_s: np.ndarray,
|
||
pvi_z: np.ndarray,
|
||
curves: list[dict[str, Any]],
|
||
targets: np.ndarray,
|
||
) -> np.ndarray:
|
||
"""직선 + 종단곡선으로 구성된 계획선을 임의 chainage에서 평가한다."""
|
||
values = np.interp(targets, pvi_s, pvi_z)
|
||
for curve in curves:
|
||
if curve["omitted"]:
|
||
continue
|
||
start, end = curve["bvc_m"], curve["evc_m"]
|
||
mask = (targets >= start) & (targets <= end)
|
||
if not np.any(mask):
|
||
continue
|
||
local = targets[mask] - start
|
||
half = curve["length_m"] / 2.0
|
||
start_z = curve["elevation_m"] - curve["grade_in"] * half
|
||
delta = curve["grade_out"] - curve["grade_in"]
|
||
values[mask] = (
|
||
start_z + curve["grade_in"] * local + delta / (2.0 * curve["length_m"]) * local * local
|
||
)
|
||
return values
|
||
|
||
|
||
def _trapezoid_weights(chainage: np.ndarray) -> np.ndarray:
|
||
"""사다리꼴 적분용 샘플 가중치(ds)."""
|
||
weights = np.zeros_like(chainage)
|
||
if len(chainage) < 2:
|
||
return weights
|
||
weights[1:-1] = (chainage[2:] - chainage[:-2]) / 2.0
|
||
weights[0] = (chainage[1] - chainage[0]) / 2.0
|
||
weights[-1] = (chainage[-1] - chainage[-2]) / 2.0
|
||
return weights
|
||
|
||
|
||
def _segments(pvi_s: np.ndarray, pvi_z: np.ndarray) -> list[dict[str, Any]]:
|
||
"""변화점 사이 직선 구간의 연장·고저차·구배(%)."""
|
||
rows: list[dict[str, Any]] = []
|
||
for index in range(len(pvi_s) - 1):
|
||
length = float(pvi_s[index + 1] - pvi_s[index])
|
||
height = float(pvi_z[index + 1] - pvi_z[index])
|
||
rows.append(
|
||
{
|
||
"index": index,
|
||
"from_m": round(float(pvi_s[index]), 6),
|
||
"to_m": round(float(pvi_s[index + 1]), 6),
|
||
"length_m": round(length, 6),
|
||
"height_m": round(height, 6),
|
||
"grade_percent": round(height / length * 100.0 if length > 0 else 0.0, 6),
|
||
}
|
||
)
|
||
return rows
|
||
|
||
|
||
def _station_rows(
|
||
stations: list[dict[str, Any]],
|
||
pvi_s: np.ndarray,
|
||
pvi_z: np.ndarray,
|
||
curves: list[dict[str, Any]],
|
||
chainage: np.ndarray,
|
||
ground: np.ndarray,
|
||
) -> list[dict[str, Any]]:
|
||
"""도면 테이블 한 열에 해당하는 측점별 값 (거리·누가거리·지반고·계획고·절성토고)."""
|
||
if not stations:
|
||
return []
|
||
targets = np.array([float(item["chainage_m"]) for item in stations], dtype=np.float64)
|
||
plan = evaluate(pvi_s, pvi_z, curves, targets)
|
||
natural = np.interp(targets, chainage, ground)
|
||
rows: list[dict[str, Any]] = []
|
||
for index, station in enumerate(stations):
|
||
difference = float(natural[index] - plan[index])
|
||
rows.append(
|
||
{
|
||
"station_id": station.get("station_id"),
|
||
"chainage_m": round(float(targets[index]), 6),
|
||
"distance_m": round(
|
||
float(targets[index] - targets[index - 1]) if index else 0.0, 6
|
||
),
|
||
"ground_elevation_m": round(float(natural[index]), 6),
|
||
"plan_elevation_m": round(float(plan[index]), 6),
|
||
"cut_m": round(max(difference, 0.0), 6),
|
||
"fill_m": round(max(-difference, 0.0), 6),
|
||
}
|
||
)
|
||
return rows
|
||
|
||
|
||
def build_alignment(
|
||
*,
|
||
base_s: np.ndarray,
|
||
base_z: np.ndarray,
|
||
chainage: np.ndarray,
|
||
ground: np.ndarray,
|
||
stations: list[dict[str, Any]],
|
||
policy: AlignmentPolicy,
|
||
edits: dict[str, Any] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""자동 변화점 + 사용자 편집으로 계획선 선형 전체를 파생한다."""
|
||
edits = edits or {}
|
||
station_offsets = {
|
||
str(key): float(value)
|
||
for key, value in (edits.get("station_offsets") or {}).items()
|
||
if value is not None
|
||
}
|
||
curve_radii = {
|
||
str(key): float(value)
|
||
for key, value in (edits.get("curve_radii") or {}).items()
|
||
if value is not None
|
||
}
|
||
|
||
pvi_s, pvi_z, sources = resolve_pvi(base_s, base_z, station_offsets)
|
||
curves, warnings = build_curves(pvi_s, pvi_z, policy, curve_radii)
|
||
segments = _segments(pvi_s, pvi_z)
|
||
plan = evaluate(pvi_s, pvi_z, curves, chainage)
|
||
difference = plan - ground
|
||
weights = _trapezoid_weights(chainage)
|
||
|
||
cut_area = float(weights[difference < 0] @ -difference[difference < 0])
|
||
fill_area = float(weights[difference > 0] @ difference[difference > 0])
|
||
reference = max(cut_area, fill_area)
|
||
imbalance = abs(cut_area - fill_area) / reference * 100.0 if reference > 1e-9 else 0.0
|
||
|
||
curve_by_pvi = {curve["pvi_index"]: curve for curve in curves}
|
||
pvi_rows: list[dict[str, Any]] = []
|
||
for index in range(len(pvi_s)):
|
||
curve = curve_by_pvi.get(index)
|
||
pvi_rows.append(
|
||
{
|
||
"chainage_m": round(float(pvi_s[index]), 6),
|
||
"elevation_m": round(float(pvi_z[index]), 6),
|
||
"source": sources[index],
|
||
"kind": "bp" if index == 0 else "ep" if index == len(pvi_s) - 1 else "pvi",
|
||
"grade_in_pct": round(segments[index - 1]["grade_percent"], 6) if index else None,
|
||
"grade_out_pct": (
|
||
round(segments[index]["grade_percent"], 6) if index < len(segments) else None
|
||
),
|
||
"curve_l_m": round(float(curve["length_m"]), 6) if curve else None,
|
||
"curve_r_m": round(float(curve["radius_m"]), 6) if curve else None,
|
||
}
|
||
)
|
||
|
||
violations = [
|
||
{
|
||
"segment_index": segment["index"],
|
||
"type": "grade_over",
|
||
"value": segment["grade_percent"],
|
||
"limit": policy.max_grade_pct,
|
||
}
|
||
for segment in segments
|
||
if abs(segment["grade_percent"]) > policy.max_grade_pct + 1e-6
|
||
]
|
||
if violations:
|
||
worst = max(abs(item["value"]) for item in violations)
|
||
warnings.append(
|
||
f"종단기울기 {worst:.2f}%가 기준 {policy.max_grade_pct:.2f}%를 초과하는 구간이 "
|
||
f"{len(violations)}개 있습니다."
|
||
)
|
||
within_tolerance = imbalance <= policy.balance_tolerance_percent + 1e-9
|
||
if not within_tolerance:
|
||
warnings.append(
|
||
f"절·성토 불균형 {imbalance:.1f}%가 허용치 "
|
||
f"{policy.balance_tolerance_percent:.1f}%를 초과합니다."
|
||
)
|
||
|
||
return {
|
||
"schema_version": ALIGNMENT_SCHEMA_VERSION,
|
||
"policy": policy.as_dict(),
|
||
"base_pvi": [
|
||
{"chainage_m": round(float(s), 6), "elevation_m": round(float(z), 6)}
|
||
for s, z in zip(base_s, base_z)
|
||
],
|
||
"edits": {"station_offsets": station_offsets, "curve_radii": curve_radii},
|
||
"pvi": pvi_rows,
|
||
"segments": segments,
|
||
"curves": [
|
||
{
|
||
"pvi_index": curve["pvi_index"],
|
||
"chainage_m": round(float(curve["chainage_m"]), 6),
|
||
"bvc_m": round(float(curve["bvc_m"]), 6),
|
||
"evc_m": round(float(curve["evc_m"]), 6),
|
||
"bvc_elevation_m": round(
|
||
float(evaluate(pvi_s, pvi_z, curves, np.array([curve["bvc_m"]]))[0]), 6
|
||
),
|
||
"evc_elevation_m": round(
|
||
float(evaluate(pvi_s, pvi_z, curves, np.array([curve["evc_m"]]))[0]), 6
|
||
),
|
||
"l_m": round(float(curve["length_m"]), 6),
|
||
"r_m": round(float(curve["radius_m"]), 6),
|
||
"k": round(float(curve["k"]), 6),
|
||
"delta_pct": round(float(curve["delta_pct"]), 6),
|
||
"middle_ordinate_m": round(float(curve["middle_ordinate_m"]), 6),
|
||
"omitted": curve["omitted"],
|
||
"skip_allowed": curve["skip_allowed"],
|
||
"omit_reason": curve["omit_reason"],
|
||
}
|
||
for curve in curves
|
||
],
|
||
"stations": _station_rows(stations, pvi_s, pvi_z, curves, chainage, ground),
|
||
"samples": [
|
||
{
|
||
"chainage_m": round(float(chainage[index]), 6),
|
||
"elevation_m": round(float(plan[index]), 6),
|
||
"ground_elevation_m": round(float(ground[index]), 6),
|
||
"difference_m": round(float(difference[index]), 6),
|
||
}
|
||
for index in range(len(chainage))
|
||
],
|
||
"balance": {
|
||
"cut_area_m2": round(cut_area, 6),
|
||
"fill_area_m2": round(fill_area, 6),
|
||
"net_area_m2": round(fill_area - cut_area, 6),
|
||
"imbalance_percent": round(imbalance, 6),
|
||
"tolerance_percent": policy.balance_tolerance_percent,
|
||
"within_tolerance": bool(within_tolerance),
|
||
},
|
||
"violations": violations,
|
||
"warnings": warnings,
|
||
}
|