- follow 가 노선 초안 전 관 · 구조물이 따라가기만 한 자리를 남김(좌표 있으면 좌표 · 없으면 저장 노선 위 누가거리) - planline: ① 따라간 자리 측점으로 carryBase · 바뀐 구간 시설 세움(_carry_profile 짝) ② 사용자가 옮긴 · 놓은 것을 followMovedNodes + seatBornNodes(_follow_moved_stations 짝) · 격자 자리는 측점 남김 - 검증_sub4 화면 실측에서 찾음: 초안 355 m 계획고 = 지반 · [저장] 뒤 서버 = 옮기기 전 계획선 높이(균형 종단) → 1.4 m 갈림 - 시험 test_459 9 개(균형 종단 · 지반 종단 · 마커로 옮김 = 서버 두 차례와 0.00001 m 안) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
357 lines
15 KiB
Python
357 lines
15 KiB
Python
"""459 노선 바뀜(요구 270) — [저장] 구조물 맞춤 · 추가 측점 따라감 · 관 꼭짓점 세움 · 관 따라감 선.
|
|
|
|
- 옮긴 시설 측점(바뀐 구간 · 사용자가 옮김)에 427 꼭짓점 다시 세움 — 초안 planline(TS) ↔
|
|
서버 `_carry_profile`(`route_moved_stations` · `seat_route_moved`) 같은 값 · 같은 자리로
|
|
따라간 시설은 옛 높이.
|
|
- [저장] 구조물 초안 = 서버 따라감 위에(`rebaseStructureDraft` · 관 `rebasePipeDraft` 짝).
|
|
- 추가 측점 좌표 따라감 TS `followPlusStations` ↔ 파이썬 `follow_plus_stations`.
|
|
- 서버 관 따라감 선 = 구조물과 같은 `points`(사용 범위 자름) · 좌표 없는 옛 관은 옛 노선
|
|
자리 먼저.
|
|
helper_459_route_seat.cjs 가 진짜 TS 를 Node 로 돌림.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import math
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
ROOT = HERE.parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from B05_Profile.B05_Profile_Engine_Grade import GradeDesignOptions # noqa: E402
|
|
from B05_Profile.B05_Profile_Engine_Grade_Alignment import ( # noqa: E402
|
|
AlignmentPolicy,
|
|
build_curves,
|
|
evaluate,
|
|
)
|
|
from B05_Profile.B05_Profile_Engine_Grade_Profile import ( # noqa: E402
|
|
design_ground_following_profile,
|
|
rebuild_alignment_profile,
|
|
)
|
|
from B05_Profile.B05_Profile_Plus_Stations import ( # noqa: E402
|
|
follow_plus_stations,
|
|
plus_chainages,
|
|
save_plus_stations,
|
|
)
|
|
from B05_Profile.B05_Profile_Route_Save_Parts import ( # noqa: E402
|
|
route_moved_stations,
|
|
seat_route_moved,
|
|
)
|
|
from B05_Profile.B05_Profile_Router_Confirm import _seat_born_nodes # noqa: E402
|
|
from common_util.common_util_drainage_pipes import PipePoint # noqa: E402
|
|
|
|
OLD_XY = [[0.0, 0.0], [200.0, 0.0], [400.0, 0.0]]
|
|
NEW_XY = [[0.0, 0.0], [200.0, 0.0], [300.0, 60.0], [420.0, 60.0]]
|
|
NEW_LEN = 200.0 + math.hypot(100.0, 60.0) + 120.0
|
|
|
|
|
|
def _old_ground(s: float) -> float:
|
|
return 100.0 + 0.08 * s + 1.5 * math.sin(s / 7.0)
|
|
|
|
|
|
def _new_ground(s: float) -> float:
|
|
"""200 m 까지 옛 선과 같은 땅 · 뒤는 다른 땅(노선이 옮겨 감)."""
|
|
if s <= 200.0:
|
|
return _old_ground(s)
|
|
return _old_ground(200.0) + 0.05 * (s - 200.0) + 0.8 * math.cos((s - 200.0) / 9.0)
|
|
|
|
|
|
PIPE = 127.0 # 같은 자리 관(격자 아님)
|
|
MOVED = 133.0 # 노선 초안 중 마커로 옮긴 자리
|
|
|
|
|
|
def _old_alignment(balanced: bool = False) -> dict:
|
|
samples = [
|
|
{"chainage_m": float(s), "elevation_m": round(_old_ground(s), 6), "valid": True}
|
|
for s in range(0, 401)
|
|
]
|
|
stations = [{"chainage_m": float(s), "kind": "regular"} for s in range(0, 401, 20)]
|
|
stations.append({"chainage_m": PIPE, "kind": "irregular", "structure": "파형강관 D800"})
|
|
stations.sort(key=lambda row: row["chainage_m"])
|
|
options = GradeDesignOptions(
|
|
max_grade_pct=14.0,
|
|
max_reverse_grade_pct=5.0,
|
|
min_vertical_radius_m=100.0,
|
|
min_curve_length_m=20.0,
|
|
min_tangent_length_m=20.0,
|
|
vertical_curve_skip_delta_pct=0.5,
|
|
design_speed_kph=20,
|
|
)
|
|
data = {"samples": samples, "stations": stations}
|
|
alignment, profile = design_ground_following_profile(data, options, station_interval_m=20.0)
|
|
if balanced: # 계획선이 지반을 안 지남(균형 종단 꼴 · 427 세움 = 옮기기 전 계획선 높이)
|
|
for node in alignment["base_pvi"]:
|
|
node["elevation_m"] -= 1.0
|
|
data["profile_alignment"], data["design_profiles"] = alignment, [profile]
|
|
alignment, _ = rebuild_alignment_profile(data, {})
|
|
return alignment
|
|
|
|
|
|
def _stations(pipe_at: float = PIPE) -> list[dict]:
|
|
"""새 노선 측점(TS 꼴) — 격자 · 끝 · 관 둘 · 구조물 하나 · 추가 측점 하나."""
|
|
rows = [{"chainageM": float(s), "kind": "grid", "id": None} for s in range(0, 421, 20)]
|
|
rows.append({"chainageM": round(NEW_LEN, 3), "kind": "end", "id": None})
|
|
rows += [
|
|
{"chainageM": pipe_at, "kind": "pipe", "id": "p127"},
|
|
{"chainageM": 251.0, "kind": "structure", "id": "s251"},
|
|
{"chainageM": 333.0, "kind": "pipe", "id": "p333"},
|
|
{"chainageM": 77.0, "kind": "structure", "id": "plus-77"},
|
|
]
|
|
return sorted(rows, key=lambda row: row["chainageM"])
|
|
|
|
|
|
def _origin() -> dict:
|
|
"""노선 초안 전 관 · 구조물(저장 노선 위) — 새 노선에 따라가면 127 · 251 · 333."""
|
|
|
|
def on_new(at: float) -> tuple[float, float]:
|
|
walked = 0.0
|
|
for (ax, ay), (bx, by) in zip(NEW_XY, NEW_XY[1:]):
|
|
span = math.hypot(bx - ax, by - ay)
|
|
if walked + span >= at:
|
|
t = (at - walked) / span
|
|
return ax + (bx - ax) * t, ay + (by - ay) * t
|
|
walked += span
|
|
raise ValueError(at)
|
|
|
|
def item(at: float, **rest) -> dict:
|
|
x, y = on_new(at)
|
|
return {**rest, "chainage_m": float(round(min(max(x, 0.0), 400.0))), "x": x, "y": y}
|
|
|
|
return {
|
|
"pipes": [item(PIPE, id="p127"), item(333.0, id="p333")],
|
|
"structures": [item(251.0, structure_id="s251", placement="point")],
|
|
}
|
|
|
|
|
|
def _server_stations(stations: list[dict]) -> list[dict]:
|
|
"""같은 측점(서버 종단 꼴) — 관 · 구조물 = `structure` 이름 · 추가 측점 = 빈 이름."""
|
|
names = {"pipe": "파형강관 D800", "structure": "옹벽"}
|
|
return [
|
|
{
|
|
"chainage_m": row["chainageM"],
|
|
"kind": "irregular" if row["kind"] in names else "regular",
|
|
"structure": "" if (row["id"] or "").startswith("plus-") else names.get(row["kind"]),
|
|
}
|
|
for row in stations
|
|
]
|
|
|
|
|
|
def _ts(tmp_path: Path, cases: list[dict]) -> list:
|
|
source = tmp_path / "in459.json"
|
|
source.write_text(json.dumps(cases, ensure_ascii=False), encoding="utf-8")
|
|
proc = subprocess.run( # noqa: S603 — 고정 실행 파일
|
|
["node", str(HERE / "helper_459_route_seat.cjs"), str(source)],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
cwd=str(ROOT),
|
|
check=False,
|
|
)
|
|
assert proc.returncode == 0, proc.stderr
|
|
return json.loads(proc.stdout)
|
|
|
|
|
|
def _base(alignment: dict) -> dict:
|
|
return {
|
|
"basePvi": alignment["base_pvi"],
|
|
"chainage": [s["chainage_m"] for s in alignment["samples"]],
|
|
"ground": [s["ground_elevation_m"] for s in alignment["samples"]],
|
|
"policy": alignment["policy"],
|
|
"onlyExplicitCurves": False,
|
|
"stations": [],
|
|
}
|
|
|
|
|
|
def _planline_case(alignment: dict, stations: list[dict], **extra) -> dict:
|
|
chainage = [float(s) for s in range(0, int(NEW_LEN) + 1)]
|
|
return {
|
|
"op": "planline",
|
|
"base": _base(alignment),
|
|
"oldXy": OLD_XY,
|
|
"newXy": NEW_XY,
|
|
"stations": stations,
|
|
"chainage": chainage,
|
|
"ground": [round(_new_ground(s), 6) for s in chainage],
|
|
**extra,
|
|
}
|
|
|
|
|
|
def _new_alignment(alignment: dict, base: list[dict]) -> dict:
|
|
samples = [
|
|
{"chainage_m": float(s), "ground_elevation_m": round(_new_ground(s), 6)}
|
|
for s in range(0, int(NEW_LEN) + 1)
|
|
]
|
|
return {"samples": samples, "policy": alignment["policy"], "base_pvi": base}
|
|
|
|
|
|
def _seat_py(alignment: dict, carried: list[dict], born: list[float]) -> list[dict]:
|
|
return seat_route_moved(_new_alignment(alignment, carried), born)
|
|
|
|
|
|
def _plan_at(nodes: list[dict], policy: dict, at: float) -> float:
|
|
s = np.array([n["chainage_m"] for n in nodes], dtype=np.float64)
|
|
z = np.array([n["elevation_m"] for n in nodes], dtype=np.float64)
|
|
curves, _ = build_curves(s, z, AlignmentPolicy.from_dict(policy))
|
|
return float(evaluate(s, z, curves, np.array([at]))[0])
|
|
|
|
|
|
def _same_nodes(mine: list[dict], theirs: list[dict], tol: float = 1e-6) -> None:
|
|
assert [n["chainage_m"] for n in mine] == pytest.approx([n["chainage_m"] for n in theirs])
|
|
for a, b in zip(mine, theirs, strict=True):
|
|
assert a["elevation_m"] == pytest.approx(b["elevation_m"], abs=tol), a["chainage_m"]
|
|
|
|
|
|
def test_옮긴_시설_측점_TS_파이썬_같음(tmp_path: Path) -> None:
|
|
stations = _stations(pipe_at=MOVED) + [{"chainageM": 401.0, "kind": "pipe", "id": "new1"}]
|
|
server = route_moved_stations(_server_stations(_stations()), OLD_XY, NEW_XY)
|
|
plain, moves = _ts(
|
|
tmp_path,
|
|
[
|
|
{"op": "moved", "stations": _stations(), "oldXy": OLD_XY, "newXy": NEW_XY},
|
|
{
|
|
"op": "userMoves",
|
|
"stations": stations,
|
|
"spots": {"p127": [PIPE], "s251": [251.0], "p333": [333.0], "plus-77": [70.0]},
|
|
},
|
|
],
|
|
)
|
|
assert server == plain == [251.0, 333.0] # 같은 자리 관 127 · 추가 측점 77 은 뺌
|
|
# 사용자가 옮긴 관 = 따라간 자리 → 지금 자리 · 새로 놓은 관 = 옛 자리 없음
|
|
assert sorted(moves, key=lambda m: m["to"]) == [
|
|
{"from": PIPE, "to": MOVED},
|
|
{"from": None, "to": 401.0},
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("balanced", [False, True])
|
|
def test_초안_planline_은_서버와_같은_꼭짓점(tmp_path: Path, balanced: bool) -> None:
|
|
alignment = _old_alignment(balanced)
|
|
[out] = _ts(tmp_path, [_planline_case(alignment, _stations())])
|
|
server = _seat_py(alignment, out["carried"], [251.0, 333.0])
|
|
_same_nodes(out["basePvi"], server)
|
|
old = {n["chainage_m"]: n["elevation_m"] for n in out["carried"]}
|
|
new = {n["chainage_m"]: n["elevation_m"] for n in out["basePvi"]}
|
|
assert new[PIPE] == pytest.approx(old[PIPE], abs=1e-9) # 같은 자리 관 = 옛 높이
|
|
if not balanced: # 옮긴 시설 측점 계획선 = 새 원지반(옛 계획선이 지반 종단)
|
|
for at in (251.0, 333.0):
|
|
assert _plan_at(out["basePvi"], alignment["policy"], at) == pytest.approx(
|
|
_new_ground(at), abs=1e-4
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("balanced", [False, True])
|
|
def test_노선_초안_중_마커로_옮긴_관도_서버_차례와_같은_값(tmp_path: Path, balanced: bool) -> None:
|
|
"""서버 = 노선 저장(따라간 자리 127 로 새 계획선 · 바뀐 구간 세움) → 관 저장(133 · 427)."""
|
|
alignment = _old_alignment(balanced)
|
|
case = _planline_case(
|
|
alignment,
|
|
_stations(pipe_at=MOVED),
|
|
homeStations=_stations(),
|
|
homeBorn=[251.0, 333.0],
|
|
origin=_origin(),
|
|
pipes=[{"id": "p127", "chainage_m": MOVED}, {"id": "p333", "chainage_m": 333.0}],
|
|
)
|
|
[out] = _ts(tmp_path, [case])
|
|
home = _seat_py(alignment, out["carried"], [251.0, 333.0]) # 노선 저장 뒤
|
|
_same_nodes(out["home"], home)
|
|
kept = [n for n in out["home"] if abs(n["chainage_m"] - PIPE) >= 0.006]
|
|
server = _seat_born_nodes(_new_alignment(alignment, out["home"]), kept, [MOVED]) # 관 저장 뒤
|
|
_same_nodes(out["basePvi"], server, tol=1e-5) # 맞춤 풀이 수렴 폭(0.000001 m) 안팎
|
|
if not balanced:
|
|
assert _plan_at(out["basePvi"], alignment["policy"], MOVED) == pytest.approx(
|
|
_old_ground(MOVED), abs=1e-4
|
|
)
|
|
|
|
|
|
def test_구조물_초안은_서버_따라감_위에(tmp_path: Path) -> None:
|
|
before = [
|
|
{"structure_id": "a", "chainage_m": 100.0, "x": 100.0, "y": 0.0}, # 따라가기만(브라우저)
|
|
{"structure_id": "b", "chainage_m": 150.0, "x": 150.0, "y": 0.0}, # 안 따라감(옛 판)
|
|
{"structure_id": "c", "chainage_m": 180.0, "x": 180.0, "y": 0.0}, # 마커로 옮김(좌표 낡음)
|
|
{
|
|
"structure_id": "d",
|
|
"chainage_m": 300.0,
|
|
"start_m": 295.0,
|
|
"end_m": 305.0,
|
|
"x": 300.0,
|
|
"y": 0.0,
|
|
},
|
|
]
|
|
after = [
|
|
{"structure_id": "a", "chainage_m": 110.0, "x": 100.0, "y": 0.0},
|
|
{"structure_id": "b", "chainage_m": 160.0, "x": 150.0, "y": 0.0},
|
|
{"structure_id": "c", "chainage_m": 190.0, "x": 180.0, "y": 0.0},
|
|
{
|
|
"structure_id": "d",
|
|
"chainage_m": 310.0,
|
|
"start_m": 305.0,
|
|
"end_m": 315.0,
|
|
"x": 300.0,
|
|
"y": 0.0,
|
|
},
|
|
]
|
|
draft = [
|
|
{"structure_id": "a", "chainage_m": 110.0, "x": 100.0, "y": 0.0, "memo": "옵션"},
|
|
{"structure_id": "b", "chainage_m": 150.0, "x": 150.0, "y": 0.0},
|
|
{"structure_id": "c", "chainage_m": 197.0, "x": 180.0, "y": 0.0},
|
|
# 구간 끝만 늘림(따라간 뒤) — 사용자 고침 = 초안 그대로
|
|
{
|
|
"structure_id": "d",
|
|
"chainage_m": 310.0,
|
|
"start_m": 305.0,
|
|
"end_m": 320.0,
|
|
"x": 300.0,
|
|
"y": 0.0,
|
|
},
|
|
{"structure_id": "e", "chainage_m": 50.0}, # 새로 놓음
|
|
]
|
|
[out] = _ts(tmp_path, [{"op": "rebase", "draft": draft, "before": before, "after": after}])
|
|
by = {row["structure_id"]: row for row in out}
|
|
assert by["a"]["chainage_m"] == 110.0 and by["a"]["memo"] == "옵션"
|
|
assert by["b"]["chainage_m"] == 160.0 # 옛 판 = 서버 따라감
|
|
assert by["c"]["chainage_m"] == 197.0 # 사용자가 옮긴 자리 지킴
|
|
assert (by["d"]["start_m"], by["d"]["end_m"]) == (305.0, 320.0)
|
|
assert by["e"] == {"structure_id": "e", "chainage_m": 50.0}
|
|
|
|
|
|
@pytest.mark.parametrize("chainages", [[30.0, 120.0, 250.0, 390.0], [199.0, 200.0, 201.0]])
|
|
def test_추가_측점_따라감_TS_파이썬_같음(tmp_path: Path, chainages: list[float]) -> None:
|
|
save_plus_stations(tmp_path, chainages, OLD_XY)
|
|
follow_plus_stations(tmp_path, NEW_XY)
|
|
[mine] = _ts(tmp_path, [{"op": "plus", "list": chainages, "from": OLD_XY, "to": NEW_XY}])
|
|
assert mine == plus_chainages(tmp_path)
|
|
|
|
|
|
def test_관_따라감_선은_사용_범위_자른_points_좌표_없는_옛_관도(monkeypatch) -> None:
|
|
"""B04 노선(범위 안 자름 · 0 m 부터) 대신 `points`(50 m 부터) · 좌표 없는 옛 관 = 옛
|
|
노선 자리로 채운 뒤."""
|
|
import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed
|
|
import common_util.common_util_drainage_pipes as drainage
|
|
from B05_Profile import B05_Profile_Route_Save as route_save
|
|
from B05_Profile.B05_Profile_Route_Save_Parts import vertices_of
|
|
|
|
async def prepared(_project_id):
|
|
return {"vertices": vertices_of([(0.0, 0.0), (400.0, 0.0)]), "streams": []}
|
|
|
|
saved: list = []
|
|
monkeypatch.setattr(watershed, "_prepare", prepared)
|
|
monkeypatch.setattr(drainage, "save_pipe_points", lambda *args: saved.append(args[2]))
|
|
old = [
|
|
PipePoint(chainage_m=100.0, x=150.0, y=0.0),
|
|
PipePoint(chainage_m=200.0), # 좌표 없는 옛 관
|
|
]
|
|
points = [{"x": 40.0, "y": 0.0}, {"x": 350.0, "y": 0.0}] # 시점 10 m 당김
|
|
old_xy = [[50.0, 0.0], [350.0, 0.0]]
|
|
pairs, _ = asyncio.run(route_save._place_pipes("p", "s", old, old_xy, points))
|
|
assert [pipe.chainage_m for _, pipe in pairs] == [110.0, 210.0]
|
|
assert [pipe.chainage_m for pipe in saved[0]] == [110.0, 210.0]
|