Files
Aislo/B05_Profile/B05_Profile_Engine_Grade_Node.py
T

225 lines
8.7 KiB
Python

"""종단 계획선 셈 = 서버 Node(PLAN 511-B B-3) — 파이썬은 입력을 모으고 정본 꼴만 적음.
셈 = TS 한 벌(`design_calc_node_profile.ts` → `B05_Profile/logic/` 의
`B05_Profile_UI_Profile_Alignment.ts` `buildAlignment` · `B05_Profile_Grade_Ground.ts`
`groundFollowingNodes` · `B05_Profile_Grade_Carry.ts` `carryEdits` · `B05_Profile_Pipeline_Grade.ts`
`seatRouteMoved`) · 늘 떠 있는 Node(`run_calc_served`).
Node 출력은 반올림 전 값 — 정본 꼴(0.000001 m · 칸 차례 · 실수형)은 `alignment_record` 가 적음
(옛 파이썬 `build_alignment` 출력과 같은 글). 파이썬 `build_alignment` 은 1순위 못 선 땅의
2~4 순위 폴백(관 정착 · 직선 분할)만 씀. 시험 = `resources/tester/test_511_b3_profile_node.py`.
"""
from __future__ import annotations
from typing import Any
from B05_Profile.B05_Profile_Engine_Grade_Alignment import ALIGNMENT_SCHEMA_VERSION
from common_util.common_util_calc_runner import run_calc_served
def run_profile(ids: list[str], inputs: dict[str, Any]) -> dict[str, Any]:
"""종단 id 를 한 요청으로 · Node 못 돌림 · 셈 실패는 RuntimeError(파이썬 짝 없음)."""
answer = run_calc_served(ids, inputs)
if not answer or "results" not in answer:
reason = (answer or {}).get("error") or "Node 응답 없음"
raise RuntimeError(f"종단 계산 Node 실패({', '.join(ids)}) — {reason}")
return answer["results"]
def _r6(value: Any) -> float:
return round(float(value), 6)
def alignment_record(
raw: dict[str, Any], policy: dict[str, Any], *, only_explicit_curves: bool = False
) -> dict[str, Any]:
"""Node `buildAlignment` 출력 → 저장 정본 꼴(옛 `build_alignment` 반환 · 칸 · 차례 · 반올림)."""
segments = [
{
"index": int(row["index"]),
"from_m": _r6(row["from_m"]),
"to_m": _r6(row["to_m"]),
"length_m": _r6(row["length_m"]),
"height_m": _r6(row["height_m"]),
"grade_percent": _r6(row["grade_percent"]),
}
for row in raw["segments"]
]
pvi = []
for index, row in enumerate(raw["pvi"]):
pvi.append(
{
"chainage_m": _r6(row["chainage_m"]),
"elevation_m": _r6(row["elevation_m"]),
"source": row["source"],
"kind": row["kind"],
"grade_in_pct": _r6(segments[index - 1]["grade_percent"]) if index else None,
"grade_out_pct": (
_r6(segments[index]["grade_percent"]) if index < len(segments) else None
),
"curve_l_m": None if row["curve_l_m"] is None else _r6(row["curve_l_m"]),
"curve_r_m": None if row["curve_r_m"] is None else _r6(row["curve_r_m"]),
}
)
max_grade = float(policy["max_grade_pct"])
stations = []
for row in raw["stations"]:
difference = float(row["ground_elevation_m"]) - float(row["plan_elevation_m"])
stations.append(
{
"station_id": row["station_id"],
"chainage_m": _r6(row["chainage_m"]),
"distance_m": _r6(row["distance_m"]),
"ground_elevation_m": _r6(row["ground_elevation_m"]),
"plan_elevation_m": _r6(row["plan_elevation_m"]),
"cut_m": _r6(max(difference, 0.0)),
"fill_m": _r6(max(-difference, 0.0)),
}
)
balance = raw["balance"]
cut, fill = float(balance["cut_area_m2"]), float(balance["fill_area_m2"])
return {
"schema_version": ALIGNMENT_SCHEMA_VERSION,
"policy": policy,
"only_explicit_curves": bool(only_explicit_curves),
"base_pvi": [
{"chainage_m": _r6(node["chainage_m"]), "elevation_m": _r6(node["elevation_m"])}
for node in raw["base_pvi"]
],
"edits": {
name: {str(key): float(value) for key, value in raw["edits"][name].items()}
for name in ("station_offsets", "curve_radii")
},
"pvi": pvi,
"segments": segments,
"curves": [
{
"pvi_index": int(curve["pvi_index"]),
**{
name: _r6(curve[name])
for name in (
"chainage_m",
"bvc_m",
"evc_m",
"bvc_elevation_m",
"evc_elevation_m",
"l_m",
"r_m",
"k",
"delta_pct",
"middle_ordinate_m",
)
},
"omitted": bool(curve["omitted"]),
"skip_allowed": bool(curve["skip_allowed"]),
"omit_reason": curve["omit_reason"],
}
for curve in raw["curves"]
],
"stations": stations,
"samples": [
{
"chainage_m": _r6(sample["chainage_m"]),
"elevation_m": _r6(sample["elevation_m"]),
"ground_elevation_m": _r6(sample["ground_elevation_m"]),
"difference_m": _r6(
float(sample["elevation_m"]) - float(sample["ground_elevation_m"])
),
}
for sample in raw["samples"]
],
"balance": {
"cut_area_m2": _r6(cut),
"fill_area_m2": _r6(fill),
"net_area_m2": _r6(fill - cut),
"imbalance_percent": _r6(balance["imbalance_percent"]),
"tolerance_percent": policy["balance_tolerance_percent"],
"within_tolerance": bool(balance["within_tolerance"]),
},
"violations": [
{
"segment_index": row["index"],
"type": "grade_over",
"value": row["grade_percent"],
"limit": policy["max_grade_pct"],
}
for row in segments
if abs(row["grade_percent"]) > max_grade + 1e-6
],
"warnings": list(raw["warnings"]),
}
def node_alignment(
chainage: Any,
ground: Any,
stations: list[dict[str, Any]],
policy: dict[str, Any],
edits: dict[str, Any] | None,
*,
base_pvi: list[dict[str, Any]] | None = None,
ground_following: dict[str, Any] | None = None,
) -> dict[str, Any] | None:
"""계획선 정본 — 꼭짓점(`base_pvi`) 또는 첫 계획선(`ground_following`) 위에 편집을 얹음.
첫 계획선이 못 서는 땅이면 None(부르는 쪽 2~4 순위). 편집 None 값 · 글 값은 옛 파이썬처럼 거름.
"""
edits = edits or {}
clean = {
name: {
str(key): float(value)
for key, value in (edits.get(name) or {}).items()
if value is not None
}
for name in ("station_offsets", "curve_radii")
}
profile: dict[str, Any] = {
"chainage": [float(value) for value in chainage],
"ground": [float(value) for value in ground],
"policy": policy,
"stations": [
{"station_id": row.get("station_id"), "chainage_m": float(row["chainage_m"])}
for row in stations
],
"edits": clean,
}
if base_pvi is not None:
profile["base_pvi"] = [
{"chainage_m": float(node["chainage_m"]), "elevation_m": float(node["elevation_m"])}
for node in base_pvi
]
if ground_following is not None:
profile["ground_following"] = ground_following
raw = run_profile(["profile.alignment"], {"profile": profile})["profile.alignment"]
return None if raw is None else alignment_record(raw, policy)
def carry_seat(
old_xy: list[Any],
new_xy: list[Any],
old_base: list[dict[str, Any]],
edits: dict[str, Any] | None,
stations: list[float],
half_m: float,
seat: dict[str, Any] | None = None,
) -> tuple[dict[str, dict[str, float]], list[dict[str, Any]] | None]:
"""노선 바뀜 계획고 옮김 + 옮긴 시설 꼭짓점 세움 = Node 한 부름(`profile.carry`).
`seat` = { ground_s · ground_z · policy · base_pvi · born } · 없으면 꼭짓점 None.
"""
carry = {
"old_xy": [[float(x), float(y)] for x, y in old_xy],
"new_xy": [[float(x), float(y)] for x, y in new_xy],
"old_base": old_base,
"edits": edits or {},
"stations": [float(value) for value in stations],
"half_m": float(half_m),
"seat": seat,
}
out = run_profile(["profile.carry"], {"carry_profile": carry})["profile.carry"]
moved = {
name: {str(key): float(value) for key, value in out["edits"][name].items()}
for name in ("station_offsets", "curve_radii")
}
return moved, out["base_pvi"]