"""종단 계획선 셈 = 서버 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"]