- 구조물: 노선 저장 뒤 rebaseStructureDraft(관 rebasePipeDraft 짝) — 따라가기만 한 것 = 서버 자리 · 사용자가 옮긴 것 = 초안 - 추가 측점: follow 단계가 좌표로 옮긴 정수 m 를 patch.plusStations 로(FOLLOW_KEYS · stations 단계 입력값) · followPlusDraft(가게 쓰기 · 초안 버림 되돌림 · page 부름은 sub7) · follow 만 쓴 초안은 [저장]이 PUT 안 함 - 427: 초안 planline · 서버 _carry_profile 이 바뀐 구간 관 · 구조물 측점(+ 초안 중 사용자가 옮긴 것)에 seatBornNodes 로 다시 세움(route_moved_stations · seat_route_moved 짝) - 서버 관 따라감 = 구조물과 같은 points(사용 범위 자름) · 좌표 없는 옛 관은 옛 노선 자리로 먼저 · 따라감 실패 = follow_failed 알림 - 시험 test_459_route_seat_mirror(거울 · 임시 storage · _place_pipes) · 헬퍼 400 · 449 · design_store_save 가짜 창구 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
311 lines
12 KiB
Python
311 lines
12 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,
|
|
)
|
|
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 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)
|
|
|
|
|
|
def _old_alignment() -> 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": 120.0, "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,
|
|
)
|
|
alignment, _ = design_ground_following_profile(
|
|
{"samples": samples, "stations": stations}, options, station_interval_m=20.0
|
|
)
|
|
return alignment
|
|
|
|
|
|
def _stations(pipe_at: float = 120.0) -> 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": "p120"},
|
|
{"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 _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 _seat_py(alignment: dict, carried: list[dict], born: list[float]) -> list[dict]:
|
|
samples = [
|
|
{"chainage_m": float(s), "ground_elevation_m": round(_new_ground(s), 6)}
|
|
for s in range(0, int(NEW_LEN) + 1)
|
|
]
|
|
return seat_route_moved(
|
|
{"samples": samples, "policy": alignment["policy"], "base_pvi": 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]) -> 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=1e-6), a["chainage_m"]
|
|
|
|
|
|
def test_옮긴_시설_측점_TS_파이썬_같음(tmp_path: Path) -> None:
|
|
stations = _stations()
|
|
server = route_moved_stations(_server_stations(stations), OLD_XY, NEW_XY)
|
|
plain, moved = _ts(
|
|
tmp_path,
|
|
[
|
|
{"op": "moved", "stations": stations, "oldXy": OLD_XY, "newXy": NEW_XY},
|
|
{
|
|
"op": "moved",
|
|
"stations": stations,
|
|
"oldXy": OLD_XY,
|
|
"newXy": NEW_XY,
|
|
"moved": ["p120"],
|
|
},
|
|
],
|
|
)
|
|
assert server == plain == [251.0, 333.0] # 같은 자리 관 120 · 추가 측점 77 은 뺌
|
|
assert moved == [120.0, 251.0, 333.0] # 사용자가 옮긴 관은 같은 자리여도 세움
|
|
|
|
|
|
def test_초안_planline_은_서버와_같은_꼭짓점(tmp_path: Path) -> None:
|
|
alignment = _old_alignment()
|
|
[out] = _ts(tmp_path, [_planline_case(alignment, _stations())])
|
|
server = _seat_py(alignment, out["carried"], [251.0, 333.0])
|
|
_same_nodes(out["basePvi"], server)
|
|
# 옮긴 시설 측점 계획선 = 새 원지반(옛 계획선이 지반 종단) · 같은 자리 관 120 = 옛 높이
|
|
for at in (251.0, 333.0):
|
|
assert _plan_at(out["basePvi"], alignment["policy"], at) == pytest.approx(
|
|
_new_ground(at), abs=1e-4
|
|
)
|
|
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[120.0] == pytest.approx(old[120.0], abs=1e-9)
|
|
|
|
|
|
def test_노선_초안_중_마커로_옮긴_관도_즉시_세움(tmp_path: Path) -> None:
|
|
alignment = _old_alignment()
|
|
case = _planline_case(
|
|
alignment,
|
|
_stations(pipe_at=125.0),
|
|
pipes=[{"id": "p120", "chainage_m": 125.0}],
|
|
followed={"pipes": [{"id": "p120", "chainage_m": 120.0}]},
|
|
)
|
|
[out] = _ts(tmp_path, [case])
|
|
server = _seat_py(alignment, out["carried"], [125.0, 251.0, 333.0])
|
|
_same_nodes(out["basePvi"], server)
|
|
assert _plan_at(out["basePvi"], alignment["policy"], 125.0) == pytest.approx(
|
|
_old_ground(125.0), 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]
|