Merge remote-tracking branch 'origin/sub_laptop_1' into sub_laptop_8
This commit is contained in:
@@ -0,0 +1,127 @@
|
||||
"""노선이 바뀔 때 계획고 옮김 — 바뀐 구간만 지반에 다시 맞추고 그 밖 편집은 길이 차만큼 민다(344-2).
|
||||
|
||||
사용자 확정(10-16 · 341 결정 1): 노선 끝쪽을 고쳤는데 앞쪽 계획고가 날아가지 않게 한다.
|
||||
|
||||
- 바뀐 구간 = 옛 · 새 노선 정점열의 같은 앞머리 · 같은 꼬리를 뺀 가운데(`changed_span`) — 바뀐
|
||||
노드 앞 · 뒤 접선점 사이. 원호 정점까지 같은지 보므로 곡선이 바뀐 이웃 노드 범위도 들어간다.
|
||||
- 구간 앞 = 같은 땅 · 같은 누가거리 → 그대로. 구간 뒤 = 같은 땅 · 누가거리만 길이 차만큼 밀림.
|
||||
- 구간 안 = 새 지반 첫 계획선(`design_ground_following_profile` 의 base_pvi) 꼭짓점으로
|
||||
갈아 끼움 · 그 안 편집(계획고 · 종단곡선 R)은 버림.
|
||||
- 옮기는 것 = base_pvi · station_offsets · curve_radii(누가거리 글자 키).
|
||||
|
||||
짝 = TS `B05_Profile/logic/B05_Profile_Grade_Carry.ts`(같은 규칙) · 거울 시험 =
|
||||
`resources/tester/test_344_2_route_pairs.py`. 서버 정본은 이쪽(346-3 [저장]이 부름).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from typing import Any
|
||||
|
||||
from B05_Profile.B05_Profile_Engine_Grade_Alignment import chainage_key
|
||||
|
||||
# 같은 정점으로 볼 거리(m) — 저장 좌표 0.0001 m 자리 · 원호 다시 뽑기 잡음보다 크게.
|
||||
SAME_VERTEX_M = 1e-3
|
||||
_EPS = 1e-6
|
||||
|
||||
|
||||
def _cumulative(xy: list[tuple[float, float]]) -> list[float]:
|
||||
total = [0.0]
|
||||
for (x0, y0), (x1, y1) in zip(xy, xy[1:]):
|
||||
total.append(total[-1] + math.hypot(x1 - x0, y1 - y0))
|
||||
return total
|
||||
|
||||
|
||||
def _same(a: tuple[float, float], b: tuple[float, float]) -> bool:
|
||||
return abs(a[0] - b[0]) <= SAME_VERTEX_M and abs(a[1] - b[1]) <= SAME_VERTEX_M
|
||||
|
||||
|
||||
def changed_span(
|
||||
old_xy: list[tuple[float, float]], new_xy: list[tuple[float, float]]
|
||||
) -> dict[str, Any] | None:
|
||||
"""옛 · 새 노선 정점열 → 바뀐 누가거리 구간. 같으면 None.
|
||||
|
||||
`start_m` = 같은 앞머리 끝(옛 = 새) · `old_end_m` · `new_end_m` = 같은 꼬리 시작 ·
|
||||
`delta_m` = 새 연장 − 옛 연장 · `start_fixed` · `end_fixed` = 앞머리 · 꼬리가 한 점이라도
|
||||
같은가.
|
||||
"""
|
||||
old_xy = [(float(x), float(y)) for x, y in old_xy]
|
||||
new_xy = [(float(x), float(y)) for x, y in new_xy]
|
||||
limit = min(len(old_xy), len(new_xy))
|
||||
prefix = 0
|
||||
while prefix < limit and _same(old_xy[prefix], new_xy[prefix]):
|
||||
prefix += 1
|
||||
if prefix == len(old_xy) == len(new_xy):
|
||||
return None
|
||||
suffix = 0
|
||||
while suffix < limit - prefix and _same(old_xy[-1 - suffix], new_xy[-1 - suffix]):
|
||||
suffix += 1
|
||||
old_cum, new_cum = _cumulative(old_xy), _cumulative(new_xy)
|
||||
return {
|
||||
"start_m": old_cum[prefix - 1] if prefix else 0.0,
|
||||
"old_end_m": old_cum[len(old_xy) - suffix] if suffix else old_cum[-1],
|
||||
"new_end_m": new_cum[len(new_xy) - suffix] if suffix else new_cum[-1],
|
||||
"delta_m": new_cum[-1] - old_cum[-1],
|
||||
"start_fixed": prefix > 0,
|
||||
"end_fixed": suffix > 0,
|
||||
}
|
||||
|
||||
|
||||
def _before(span: dict[str, Any], s: float) -> bool:
|
||||
"""구간 앞(그대로 둠) — 앞머리가 같으면 그 끝점까지."""
|
||||
start = float(span["start_m"])
|
||||
return s < start - _EPS or (bool(span["start_fixed"]) and s <= start + _EPS)
|
||||
|
||||
|
||||
def _after(span: dict[str, Any], s: float, end: float) -> bool:
|
||||
"""구간 뒤(밂) — 꼬리가 같으면 그 시작점부터."""
|
||||
return s > end + _EPS or (bool(span["end_fixed"]) and s >= end - _EPS)
|
||||
|
||||
|
||||
def carry_profile_edits(
|
||||
base_pvi: list[dict[str, float]],
|
||||
edits: dict[str, Any] | None,
|
||||
span: dict[str, Any],
|
||||
ground_pvi: list[dict[str, float]],
|
||||
) -> tuple[list[dict[str, float]], dict[str, dict[str, float]]]:
|
||||
"""옛 꼭짓점 · 편집을 새 노선으로 옮긴다 → (새 base_pvi, 새 edits).
|
||||
|
||||
`ground_pvi` = 새 노선 지반 첫 계획선 꼭짓점(전 구간 · `design_ground_following_profile`) —
|
||||
바뀐 구간 안 몫만 쓴다.
|
||||
"""
|
||||
delta = float(span["delta_m"])
|
||||
old_end, new_end = float(span["old_end_m"]), float(span["new_end_m"])
|
||||
|
||||
def moved(s: float) -> float | None:
|
||||
if _before(span, s):
|
||||
return s
|
||||
if _after(span, s, old_end):
|
||||
return s + delta
|
||||
return None # 바뀐 구간 안 — 버림
|
||||
|
||||
nodes: dict[float, dict[str, float]] = {}
|
||||
for node in base_pvi:
|
||||
at = moved(float(node["chainage_m"]))
|
||||
if at is not None:
|
||||
key = round(at, 3)
|
||||
nodes[key] = {"chainage_m": round(at, 6), "elevation_m": float(node["elevation_m"])}
|
||||
for node in ground_pvi:
|
||||
s = float(node["chainage_m"])
|
||||
if _before(span, s) or _after(span, s, new_end):
|
||||
continue
|
||||
nodes.setdefault(
|
||||
round(s, 3), {"chainage_m": round(s, 6), "elevation_m": float(node["elevation_m"])}
|
||||
)
|
||||
|
||||
carried: dict[str, dict[str, float]] = {}
|
||||
for name in ("station_offsets", "curve_radii"):
|
||||
values: dict[str, float] = {}
|
||||
for key, value in ((edits or {}).get(name) or {}).items():
|
||||
try:
|
||||
at = moved(float(key))
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
if at is not None and value is not None:
|
||||
values[chainage_key(at)] = float(value)
|
||||
carried[name] = values
|
||||
return [nodes[key] for key in sorted(nodes)], carried
|
||||
@@ -19,6 +19,11 @@
|
||||
- `rebuild_alignment_profile()`: 사용자 편집 확정 시. **저장된 자동 선형(base_pvi)과
|
||||
정책을 그대로 재사용**하고 편집 델타만 다시 얹는다. DP를 다시 돌리면 기준선이
|
||||
흔들려 "원복" 이 원래 위치로 돌아가지 않기 때문이다.
|
||||
|
||||
짝(344-2): `design_ground_following_profile` 의 base_pvi 몫(`ground_profile` · `pass_through`) ↔ TS
|
||||
`B05_Profile/logic/B05_Profile_Grade_Ground.ts` `groundFollowingBasePvi` · 노선 바뀜 때 계획고 옮김은
|
||||
`B05_Profile_Engine_Grade_Carry.py` ↔ `logic/B05_Profile_Grade_Carry.ts` · 거울 시험 =
|
||||
`resources/tester/test_344_2_route_pairs.py`.
|
||||
"""
|
||||
|
||||
from collections import Counter
|
||||
|
||||
@@ -420,6 +420,15 @@ async def _recompute(
|
||||
|
||||
marks.append(("재계산 준비(관 삭제·노선 읽기·stage 기록)", time.perf_counter()))
|
||||
|
||||
# 구조물은 땅 위 자리(좌표)를 따라감(344-2) — 옛 노선 경로 파일은 체인 전이라 아직 옛 것
|
||||
from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route
|
||||
|
||||
try:
|
||||
old_path = (latest or {}).get("route_data_path")
|
||||
await asyncio.to_thread(follow_structures_to_route, project_root, old_path, points)
|
||||
except Exception: # 따라감 실패 = 옛 누가거리 그대로(지금까지와 같음)
|
||||
logger.exception("구조물 따라감 실패(옛 누가거리 유지): project_id=%s", project_id)
|
||||
|
||||
target = None
|
||||
if balanced is not None:
|
||||
from B05_Profile.B05_Profile_Router_RouteInitial import balanced_target
|
||||
|
||||
@@ -42,6 +42,7 @@ from common_util.common_util_drainage_pipes import (
|
||||
pipe_points_path_in,
|
||||
read_pipe_points_file,
|
||||
)
|
||||
from common_util.common_util_route_follow import fill_structure_coordinates, follow_structures
|
||||
from common_util.common_util_storage import resolve_stored_project_path
|
||||
from common_util.common_util_structure_station import plan_stations
|
||||
from config.config_db import get_db_pool
|
||||
@@ -96,6 +97,61 @@ async def _route_length(project_id: UUID) -> float | None:
|
||||
return None
|
||||
|
||||
|
||||
async def _route_xy(project_id: UUID, root: str) -> list[list[float]] | None:
|
||||
"""최신 노선 정점(x · y) — 구조물 좌표를 채울 선. 없으면 None(좌표 없이 진행)."""
|
||||
from B05_Profile.B05_Profile_Engine_Sections import _load_route_polyline
|
||||
|
||||
try:
|
||||
pool = get_db_pool()
|
||||
async with pool.acquire() as connection:
|
||||
latest = await get_latest_route(connection, project_id)
|
||||
path = (latest or {}).get("route_data_path")
|
||||
return (
|
||||
await asyncio.to_thread(_load_route_polyline, Path(root), str(path)) if path else None
|
||||
)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"B05 구조물 좌표: 노선을 못 읽음(좌표 없이 진행): project_id=%s", project_id
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def follow_structures_to_route(
|
||||
project_root: Path, old_route_data_path: str | None, new_xy: list[dict[str, float]]
|
||||
) -> dict[str, int]:
|
||||
"""노선이 바뀔 때 구조물이 땅 위 자리를 따라감(344-2) → 정본 저장. {옮김, 멀어짐} 수.
|
||||
|
||||
좌표가 없는 옛 자료는 옛 노선(아직 안 바뀐 경로 파일)으로 먼저 채운다. 실패는 비치명 —
|
||||
구조물은 옛 누가거리 그대로 남는다(지금까지와 같음).
|
||||
"""
|
||||
from B05_Profile.B05_Profile_Engine_Sections import _load_route_polyline
|
||||
|
||||
root = str(project_root)
|
||||
revision, items = load_structures(root)
|
||||
if not items:
|
||||
return {"moved": 0, "far": 0}
|
||||
if old_route_data_path:
|
||||
try:
|
||||
fill_structure_coordinates(
|
||||
items, _load_route_polyline(project_root, old_route_data_path), only_missing=True
|
||||
)
|
||||
except (OSError, ValueError):
|
||||
logger.warning("B05 구조물 따라감: 옛 노선을 못 읽어 좌표 없는 구조물은 그대로")
|
||||
before = [(item.chainage_m, item.start_m, item.end_m) for item in items]
|
||||
result = follow_structures(items, [(point["x"], point["y"]) for point in new_xy])
|
||||
moved = sum(
|
||||
before[index] != (item.chainage_m, item.start_m, item.end_m)
|
||||
for index, item in enumerate(items)
|
||||
)
|
||||
far = sum(1 for _, flag in result if flag)
|
||||
if moved:
|
||||
# 범위는 따라감이 0 ~ 새 연장 안으로 이미 자름 — DB 연장(옛 노선)으로 재지 않음
|
||||
save_structures(root, items, base_revision=revision)
|
||||
if far:
|
||||
logger.warning("B05 구조물 %d건이 새 노선에서 멀어짐(버리지 않음)", far)
|
||||
return {"moved": moved, "far": far}
|
||||
|
||||
|
||||
def _migrate_pipe_facilities(
|
||||
root: str, facilities: list[PipePoint], migrated_before: set[str]
|
||||
) -> tuple[int, set[str]]:
|
||||
@@ -160,6 +216,11 @@ async def read_structures(project_id: UUID) -> StructureListResponse | JSONRespo
|
||||
if root is None:
|
||||
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
|
||||
revision, structures = load_structures(root)
|
||||
# 옛 자료(좌표 없음)는 지금 노선 위 기준 자리로 채워 보냄 — 파일은 다음 [저장] 때(344-2)
|
||||
if any(item.x is None for item in structures):
|
||||
xy = await _route_xy(project_id, root)
|
||||
if xy:
|
||||
fill_structure_coordinates(structures, xy, only_missing=True)
|
||||
return StructureListResponse(
|
||||
project_id=str(project_id),
|
||||
revision=revision,
|
||||
@@ -287,6 +348,14 @@ async def write_structures(
|
||||
if root is None:
|
||||
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
|
||||
_, previous = load_structures(root)
|
||||
# 땅 위 자리(344-2) — 기준 측점의 노선 위 좌표 · 자리가 그대로면 옛 좌표 지킴
|
||||
xy = await _route_xy(project_id, root)
|
||||
if xy:
|
||||
fill_structure_coordinates(
|
||||
payload.structures,
|
||||
xy,
|
||||
previous={item.structure_id: item for item in previous if item.structure_id},
|
||||
)
|
||||
revision = save_structures(
|
||||
root,
|
||||
payload.structures,
|
||||
|
||||
@@ -197,6 +197,10 @@ class StructureInstance(BaseModel):
|
||||
summary_item: dict[str, Any] | None = None
|
||||
# 옵션 그룹(M02 PLAN 46)마다 고른 집계 항목 `{그룹 id: summary_item}` — 본체는 `summary_item`
|
||||
group_items: dict[str, Any] | None = None
|
||||
# 땅 위 자리(344-2) — 기준 측점(`chainage_m`)의 노선 위 좌표 · 서버가 저장 때 채움 ·
|
||||
# 노선이 바뀌면 새 노선에 투영해 누가거리 다시(`common_util_route_follow`) · 옛 파일 = 읽을 때
|
||||
x: float | None = None
|
||||
y: float | None = None
|
||||
|
||||
@field_validator("summary_item")
|
||||
@classmethod
|
||||
|
||||
@@ -281,13 +281,13 @@ function resolvePvi(
|
||||
return [...nodes.values()].sort((a, b) => a.chainage_m - b.chainage_m);
|
||||
}
|
||||
|
||||
interface WorkingCurve extends AlignmentCurve {
|
||||
export interface WorkingCurve extends AlignmentCurve {
|
||||
grade_in: number;
|
||||
grade_out: number;
|
||||
}
|
||||
|
||||
/** 각 변화점에 대칭 종단곡선을 삽입한다 (좌우 직선이 남도록 반쪽 길이를 제한). */
|
||||
function buildCurves(
|
||||
export function buildCurves(
|
||||
pvi: AlignmentNode[],
|
||||
policy: AlignmentPolicy,
|
||||
curveRadii: Record<string, number>,
|
||||
@@ -352,7 +352,7 @@ function buildCurves(
|
||||
}
|
||||
|
||||
/** 직선 + 종단곡선으로 구성된 계획선을 임의 chainage에서 평가한다. */
|
||||
function evaluateAt(
|
||||
export function evaluateAt(
|
||||
pviS: number[],
|
||||
pviZ: number[],
|
||||
curves: WorkingCurve[],
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
/* =============================================================================
|
||||
* B05_Profile/logic/B05_Profile_Grade_Carry.ts
|
||||
* 노선이 바뀔 때 계획고 옮김 — 바뀐 구간만 지반에 다시 맞추고 그 밖 편집은 길이 차만큼 밂(344-2).
|
||||
*
|
||||
* 짝 = 파이썬 `B05_Profile_Engine_Grade_Carry.py`(서버 정본 · 같은 규칙) · 거울 시험 =
|
||||
* resources/tester/test_344_2_route_pairs.py.
|
||||
* - 바뀐 구간 = 옛 · 새 노선 정점열의 같은 앞머리 · 꼬리를 뺀 가운데(바뀐 노드 앞 · 뒤 접선점 사이).
|
||||
* - 앞 = 그대로 · 뒤 = 누가거리만 길이 차만큼 밂 · 안 = 새 지반 첫 계획선 꼭짓점(`groundFollowingBasePvi`)
|
||||
* · 안의 편집(계획고 · 종단곡선 R)은 버림.
|
||||
* - 옮기는 것 = base_pvi · station_offsets · curve_radii(누가거리 글자 키).
|
||||
* ========================================================================== */
|
||||
|
||||
import {
|
||||
chainageKey,
|
||||
type AlignmentEdits,
|
||||
type AlignmentNode,
|
||||
} from "../B05_Profile_UI_Profile_Alignment";
|
||||
|
||||
/** 같은 정점으로 볼 거리(m) — 파이썬 `SAME_VERTEX_M` 짝. */
|
||||
export const SAME_VERTEX_M = 1e-3;
|
||||
const EPS = 1e-6;
|
||||
|
||||
export interface XY {
|
||||
x: number;
|
||||
y: number;
|
||||
}
|
||||
|
||||
export interface ChangedSpan {
|
||||
start_m: number;
|
||||
old_end_m: number;
|
||||
new_end_m: number;
|
||||
delta_m: number;
|
||||
start_fixed: boolean;
|
||||
end_fixed: boolean;
|
||||
}
|
||||
|
||||
const roundTo = (value: number, digits: number): number => Number(value.toFixed(digits));
|
||||
|
||||
function cumulative(xy: ReadonlyArray<XY>): number[] {
|
||||
const total = [0];
|
||||
for (let index = 1; index < xy.length; index += 1) {
|
||||
const prev = xy[index - 1];
|
||||
total.push(total[index - 1] + Math.hypot(xy[index].x - prev.x, xy[index].y - prev.y));
|
||||
}
|
||||
return total;
|
||||
}
|
||||
|
||||
const same = (a: XY, b: XY): boolean =>
|
||||
Math.abs(a.x - b.x) <= SAME_VERTEX_M && Math.abs(a.y - b.y) <= SAME_VERTEX_M;
|
||||
|
||||
/** 옛 · 새 노선 정점열 → 바뀐 누가거리 구간 · 같으면 null(파이썬 `changed_span` 짝). */
|
||||
export function changedSpan(
|
||||
oldXy: ReadonlyArray<XY>,
|
||||
newXy: ReadonlyArray<XY>,
|
||||
): ChangedSpan | null {
|
||||
const limit = Math.min(oldXy.length, newXy.length);
|
||||
let prefix = 0;
|
||||
while (prefix < limit && same(oldXy[prefix], newXy[prefix])) prefix += 1;
|
||||
if (prefix === oldXy.length && prefix === newXy.length) return null;
|
||||
let suffix = 0;
|
||||
while (
|
||||
suffix < limit - prefix &&
|
||||
same(oldXy[oldXy.length - 1 - suffix], newXy[newXy.length - 1 - suffix])
|
||||
)
|
||||
suffix += 1;
|
||||
const oldCum = cumulative(oldXy);
|
||||
const newCum = cumulative(newXy);
|
||||
return {
|
||||
start_m: prefix ? oldCum[prefix - 1] : 0,
|
||||
old_end_m: suffix ? oldCum[oldXy.length - suffix] : oldCum[oldCum.length - 1],
|
||||
new_end_m: suffix ? newCum[newXy.length - suffix] : newCum[newCum.length - 1],
|
||||
delta_m: newCum[newCum.length - 1] - oldCum[oldCum.length - 1],
|
||||
start_fixed: prefix > 0,
|
||||
end_fixed: suffix > 0,
|
||||
};
|
||||
}
|
||||
|
||||
const before = (span: ChangedSpan, s: number): boolean =>
|
||||
s < span.start_m - EPS || (span.start_fixed && s <= span.start_m + EPS);
|
||||
|
||||
const after = (span: ChangedSpan, s: number, end: number): boolean =>
|
||||
s > end + EPS || (span.end_fixed && s >= end - EPS);
|
||||
|
||||
/** 옛 꼭짓점 · 편집을 새 노선으로 옮김 — `groundPvi` = 새 노선 지반 첫 계획선(전 구간). */
|
||||
export function carryProfileEdits(
|
||||
basePvi: ReadonlyArray<AlignmentNode>,
|
||||
edits: Partial<AlignmentEdits> | null | undefined,
|
||||
span: ChangedSpan,
|
||||
groundPvi: ReadonlyArray<AlignmentNode>,
|
||||
): { basePvi: AlignmentNode[]; edits: AlignmentEdits } {
|
||||
const moved = (s: number): number | null => {
|
||||
if (before(span, s)) return s;
|
||||
if (after(span, s, span.old_end_m)) return s + span.delta_m;
|
||||
return null; // 바뀐 구간 안 — 버림
|
||||
};
|
||||
|
||||
const nodes = new Map<number, AlignmentNode>();
|
||||
basePvi.forEach((node) => {
|
||||
const at = moved(node.chainage_m);
|
||||
if (at !== null)
|
||||
nodes.set(roundTo(at, 3), { chainage_m: roundTo(at, 6), elevation_m: node.elevation_m });
|
||||
});
|
||||
groundPvi.forEach((node) => {
|
||||
const s = node.chainage_m;
|
||||
if (before(span, s) || after(span, s, span.new_end_m)) return;
|
||||
const slot = roundTo(s, 3);
|
||||
if (!nodes.has(slot))
|
||||
nodes.set(slot, { chainage_m: roundTo(s, 6), elevation_m: node.elevation_m });
|
||||
});
|
||||
|
||||
const carry = (values: Record<string, number> | undefined): Record<string, number> => {
|
||||
const out: Record<string, number> = {};
|
||||
Object.entries(values ?? {}).forEach(([key, value]) => {
|
||||
const chainage = Number(key);
|
||||
if (!Number.isFinite(chainage) || value == null) return;
|
||||
const at = moved(chainage);
|
||||
if (at !== null) out[chainageKey(at)] = Number(value);
|
||||
});
|
||||
return out;
|
||||
};
|
||||
|
||||
return {
|
||||
basePvi: [...nodes.keys()].sort((a, b) => a - b).map((slot) => nodes.get(slot)!),
|
||||
edits: {
|
||||
station_offsets: carry(edits?.station_offsets),
|
||||
curve_radii: carry(edits?.curve_radii),
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
/* =============================================================================
|
||||
* B05_Profile/logic/B05_Profile_Grade_Ground.ts
|
||||
* 첫 계획선(계획고 = 지반고) 꼭짓점 — 종단 지반 표본 · 측점 → base_pvi (344-2).
|
||||
*
|
||||
* 짝 = 파이썬 `B05_Profile_Engine_Grade_Profile.design_ground_following_profile` 의 base_pvi 몫
|
||||
* (1순위만 · 관 정착 · 직선 분할 폴백은 짝 없음). 그 위 선형은 이미 짝인 `buildAlignment`
|
||||
* (`B05_Profile_UI_Profile_Alignment.ts`). 거울 시험 = resources/tester/test_344_2_route_pairs.py.
|
||||
* 1순위가 못 서는 땅(유효 지반 둘 미만 · 연장 0)이면 null — 부르는 쪽이 「서버로 다시 셈」 표시만.
|
||||
* ========================================================================== */
|
||||
|
||||
import {
|
||||
buildCurves,
|
||||
evaluateAt,
|
||||
type AlignmentNode,
|
||||
type AlignmentPolicy,
|
||||
} from "../B05_Profile_UI_Profile_Alignment";
|
||||
|
||||
/** 종단 지반 표본 한 점 — 저장 종단 `samples[]` 꼴. */
|
||||
export interface GroundSample {
|
||||
chainage_m: number;
|
||||
elevation_m?: number | null;
|
||||
valid?: boolean;
|
||||
}
|
||||
|
||||
export interface GroundFollowingInput {
|
||||
samples: ReadonlyArray<GroundSample>;
|
||||
/** 측점 누가거리 — 모든 측점이 꼭짓점. */
|
||||
stations: ReadonlyArray<number>;
|
||||
policy: AlignmentPolicy;
|
||||
startOffsetM?: number;
|
||||
endOffsetM?: number;
|
||||
}
|
||||
|
||||
/** 파이썬 `round(value, digits)` 와 같은 값(이진 값 그대로 십진 반올림). */
|
||||
const roundTo = (value: number, digits: number): number => Number(value.toFixed(digits));
|
||||
|
||||
/** `np.interp` — 오름차순 xs · 범위 밖은 양 끝값. */
|
||||
function interp(xs: number[], ys: number[], value: number): number {
|
||||
if (value <= xs[0]) return ys[0];
|
||||
if (value >= xs[xs.length - 1]) return ys[ys.length - 1];
|
||||
let low = 0;
|
||||
let high = xs.length - 1;
|
||||
while (high - low > 1) {
|
||||
const mid = (low + high) >> 1;
|
||||
if (xs[mid] <= value) low = mid;
|
||||
else high = mid;
|
||||
}
|
||||
const span = xs[high] - xs[low];
|
||||
return span <= 0 ? ys[high] : ys[low] + ((ys[high] - ys[low]) * (value - xs[low])) / span;
|
||||
}
|
||||
|
||||
/** 파이썬 `ground_profile` 짝 — 누가거리 순 · 빈 지반은 선형 보간. 유효 둘 미만이면 null. */
|
||||
export function groundProfile(
|
||||
samples: ReadonlyArray<GroundSample>,
|
||||
): { chainage: number[]; ground: number[] } | null {
|
||||
const rows = samples
|
||||
.map((sample) => ({
|
||||
s: Number(sample.chainage_m ?? 0),
|
||||
z: (sample.valid ?? true) && sample.elevation_m != null ? Number(sample.elevation_m) : NaN,
|
||||
}))
|
||||
.sort((a, b) => a.s - b.s);
|
||||
const finite = rows.filter((row) => Number.isFinite(row.z));
|
||||
if (finite.length < 2 || rows.length < 2) return null;
|
||||
const fs = finite.map((row) => row.s);
|
||||
const fz = finite.map((row) => row.z);
|
||||
return { chainage: rows.map((row) => row.s), ground: rows.map((row) => interp(fs, fz, row.s)) };
|
||||
}
|
||||
|
||||
/** 종단곡선을 넣어도 측점마다 `target` 을 지나게 하는 꼭짓점 표고 — 파이썬 `pass_through` 짝. */
|
||||
export function passThrough(baseS: number[], target: number[], policy: AlignmentPolicy): number[] {
|
||||
let baseZ = [...target];
|
||||
for (let round = 0; round < 12; round += 1) {
|
||||
const pvi = baseS.map((s, index) => ({ chainage_m: s, elevation_m: baseZ[index] }));
|
||||
const curves = buildCurves(pvi, policy, {}, []);
|
||||
const error = baseS.map((s, index) => target[index] - evaluateAt(baseS, baseZ, curves, s));
|
||||
error[0] = 0;
|
||||
error[error.length - 1] = 0;
|
||||
if (Math.max(...error.map(Math.abs)) < 1e-4) break;
|
||||
baseZ = baseZ.map((z, index) => z + error[index]);
|
||||
}
|
||||
return baseZ;
|
||||
}
|
||||
|
||||
/** 첫 계획선 꼭짓점(base_pvi · 0.000001 m) — 못 서는 땅이면 null(서버로 다시 셈). */
|
||||
export function groundFollowingBasePvi(input: GroundFollowingInput): AlignmentNode[] | null {
|
||||
const profile = groundProfile(input.samples);
|
||||
if (!profile) return null;
|
||||
const { chainage, ground } = profile;
|
||||
const total = chainage[chainage.length - 1];
|
||||
if (!(total > 0)) return null;
|
||||
|
||||
let nodeS = [
|
||||
...new Set(
|
||||
input.stations
|
||||
.filter((value) => Number.isFinite(value) && value >= 0 && value <= total)
|
||||
.map((value) => roundTo(value, 3)),
|
||||
),
|
||||
].sort((a, b) => a - b);
|
||||
if (nodeS.length < 2) nodeS = chainage.map((value) => roundTo(value, 3));
|
||||
if (nodeS[0] > 0) nodeS.unshift(0);
|
||||
if (nodeS[nodeS.length - 1] < total) nodeS.push(roundTo(total, 3));
|
||||
|
||||
const target = nodeS.map((s) => interp(chainage, ground, s));
|
||||
target[0] += input.startOffsetM ?? 0;
|
||||
target[target.length - 1] += input.endOffsetM ?? 0;
|
||||
const baseZ = passThrough(nodeS, target, input.policy);
|
||||
return nodeS.map((s, index) => ({
|
||||
chainage_m: roundTo(s, 6),
|
||||
elevation_m: roundTo(baseZ[index], 6),
|
||||
}));
|
||||
}
|
||||
@@ -19,6 +19,8 @@
|
||||
|
||||
짝: 시설 속성 붙이기 `carry_indices` ↔ 화면 `B05_Profile/B05_Profile_UI_Drainage_Facility.ts`
|
||||
`carryIndices` — 같은 규칙(허용 오차 안 · 한 원본 한 번) · 거울 시험 test_b05_facility_carry.
|
||||
짝: 관 자리 투영 `project_pipe_points` ↔ TS `common_util/common_util_route_follow.ts`
|
||||
`projectPipePoints`(344-2 · 노선 바뀜 때 관이 좌표를 따라감) · 거울 시험 test_344_2_route_pairs.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -0,0 +1,103 @@
|
||||
"""노선이 바뀔 때 구조물이 땅 위 자리(좌표)를 따라감 — 누가거리를 다시 매김(344-2).
|
||||
|
||||
사용자 확정(10-16 · 341 결정 2): 관 · 구조물의 자리 정본 = 좌표(x · y) · 누가거리는 그 자리를 지금
|
||||
노선에서 읽은 값. 노선이 바뀌면 좌표를 새 노선에 투영해 누가거리를 다시 매기고(정수 m · 234-1),
|
||||
새 노선에서 `STRUCTURE_ROUTE_DRIFT_MAX_M` 넘게 멀면 「노선에서 멀어짐」 표시만(버리지 않음).
|
||||
|
||||
- `fill_structure_coordinates` — 저장 때 노선 위 기준 자리로 x · y 채움(자리가 그대로면 옛 좌표
|
||||
지킴) · 옛 자료는 읽을 때 없는 것만 채움.
|
||||
- `follow_structures` — 좌표 → 새 노선 누가거리 · 구간 끝은 같은 만큼 밂 · 노선과의 거리.
|
||||
|
||||
짝 = TS `common_util/common_util_route_follow.ts` `followStructures`(구조물) · `projectPipePoints`
|
||||
(관 — 파이썬 본체 = `common_util_drainage_pipes.project_pipe_points`) · 거울 시험 =
|
||||
`resources/tester/test_344_2_route_pairs.py`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Iterable, Sequence
|
||||
|
||||
from shapely.geometry import LineString, Point
|
||||
|
||||
from common_util.common_util_structure_station import snap_structure_station
|
||||
from config.config_system import STRUCTURE_ROUTE_DRIFT_MAX_M
|
||||
|
||||
_POSITION_KEYS = ("chainage_m", "start_m", "end_m")
|
||||
|
||||
|
||||
def _line(xy: Sequence[Sequence[float]]) -> LineString | None:
|
||||
points = [(float(point[0]), float(point[1])) for point in xy]
|
||||
if len(points) < 2:
|
||||
return None
|
||||
line = LineString(points)
|
||||
return line if line.length > 0 else None
|
||||
|
||||
|
||||
def _positions(item: Any) -> tuple[Any, ...]:
|
||||
return tuple(getattr(item, key, None) for key in _POSITION_KEYS)
|
||||
|
||||
|
||||
def fill_structure_coordinates(
|
||||
items: Iterable[Any],
|
||||
xy: Sequence[Sequence[float]],
|
||||
*,
|
||||
previous: dict[str, Any] | None = None,
|
||||
only_missing: bool = False,
|
||||
) -> int:
|
||||
"""구조물 x · y 를 노선 위 기준 자리(`chainage_m`)로 채운다(제자리) · 채운 수.
|
||||
|
||||
`previous` = 저장 전 정본 {structure_id: 항목} — 자리(누가거리 셋)가 그대로면 옛 좌표를 지킨다
|
||||
(노선이 바뀌어 멀어진 땅 자리를 다음 저장이 노선 위로 덮지 않게). `only_missing` = 읽을 때.
|
||||
"""
|
||||
line = _line(xy)
|
||||
if line is None:
|
||||
return 0
|
||||
filled = 0
|
||||
for item in items:
|
||||
anchor = getattr(item, "chainage_m", None)
|
||||
if anchor is None:
|
||||
continue
|
||||
has_xy = getattr(item, "x", None) is not None and getattr(item, "y", None) is not None
|
||||
if only_missing and has_xy:
|
||||
continue
|
||||
old = (previous or {}).get(getattr(item, "structure_id", None) or "")
|
||||
if old is not None and _positions(old) == _positions(item) and old.x is not None:
|
||||
item.x, item.y = old.x, old.y
|
||||
continue
|
||||
point = line.interpolate(float(anchor))
|
||||
item.x, item.y = round(float(point.x), 4), round(float(point.y), 4)
|
||||
filled += 1
|
||||
return filled
|
||||
|
||||
|
||||
def follow_structures(
|
||||
items: Iterable[Any],
|
||||
xy: Sequence[Sequence[float]],
|
||||
max_drift_m: float = STRUCTURE_ROUTE_DRIFT_MAX_M,
|
||||
) -> list[tuple[float | None, bool]]:
|
||||
"""좌표를 새 노선에 투영해 누가거리를 다시 매긴다(제자리) → 항목마다 (노선과 거리, 멀어짐).
|
||||
|
||||
기준 자리 = 투영 → 정수 m(노선 끝 안) · 구간 끝(`start_m` · `end_m`)은 같은 만큼 밀고 0 ~ 연장
|
||||
안으로 자름. 좌표가 없는 옛 자료는 그대로(거리 None).
|
||||
"""
|
||||
items = list(items)
|
||||
line = _line(xy)
|
||||
if line is None:
|
||||
return [(None, False)] * len(items)
|
||||
length = float(line.length)
|
||||
result: list[tuple[float | None, bool]] = []
|
||||
for item in items:
|
||||
x, y = getattr(item, "x", None), getattr(item, "y", None)
|
||||
anchor = getattr(item, "chainage_m", None)
|
||||
if x is None or y is None or anchor is None:
|
||||
result.append((None, False))
|
||||
continue
|
||||
point = Point(float(x), float(y))
|
||||
shift = snap_structure_station(float(line.project(point)), [length]) - float(anchor)
|
||||
for key in _POSITION_KEYS:
|
||||
value = getattr(item, key, None)
|
||||
if value is not None:
|
||||
setattr(item, key, min(max(float(value) + shift, 0.0), length))
|
||||
drift = float(line.distance(point))
|
||||
result.append((drift, drift > max_drift_m))
|
||||
return result
|
||||
@@ -0,0 +1,116 @@
|
||||
/* =============================================================================
|
||||
* common_util/common_util_route_follow.ts
|
||||
* 노선이 바뀔 때 관 · 구조물이 땅 위 자리(좌표)를 따라감 — 누가거리 다시 매김(344-2).
|
||||
*
|
||||
* 자리 정본 = 좌표(x · y) · 누가거리 = 그 자리를 지금 노선에서 읽은 값(정수 m · 234-1).
|
||||
* 짝:
|
||||
* - `projectPipePoints` ↔ 파이썬 `common_util_drainage_pipes.project_pipe_points`(xy 투영 + 정수 m ·
|
||||
* 구간 끝 같은 만큼 밂 · 누가거리 순).
|
||||
* - `followStructures` ↔ 파이썬 `common_util_route_follow.follow_structures`(+ 0 ~ 연장 자름 ·
|
||||
* 노선과 거리 · `STRUCTURE_ROUTE_DRIFT_MAX_M` 넘으면 「노선에서 멀어짐」 · 버리지 않음).
|
||||
* - `projectOnPolyline` = shapely `LineString.project`(가장 가까운 첫 선분) · `distance`.
|
||||
* 거울 시험 = resources/tester/test_344_2_route_pairs.py.
|
||||
* ========================================================================== */
|
||||
|
||||
import { snapStructureStation } from "./common_util_structure_station";
|
||||
|
||||
/** 「노선에서 멀어짐」 거리(m) — 파이썬 `config_system.STRUCTURE_ROUTE_DRIFT_MAX_M` 짝(거울 시험이 대조). */
|
||||
export const STRUCTURE_ROUTE_DRIFT_MAX_M = 20;
|
||||
|
||||
export interface XY {
|
||||
x: number;
|
||||
y: number;
|
||||
}
|
||||
|
||||
/** 노선 위 자리 — 누가거리 · 노선과의 거리. 노선이 없으면(점 둘 미만 · 길이 0) null. */
|
||||
export function projectOnPolyline(
|
||||
line: ReadonlyArray<XY>,
|
||||
point: XY,
|
||||
): { chainage_m: number; distance_m: number; length_m: number } | null {
|
||||
let walked = 0;
|
||||
let best: { chainage_m: number; distance_m: number } | null = null;
|
||||
for (let index = 1; index < line.length; index += 1) {
|
||||
const a = line[index - 1];
|
||||
const b = line[index];
|
||||
const dx = b.x - a.x;
|
||||
const dy = b.y - a.y;
|
||||
const span = Math.hypot(dx, dy);
|
||||
if (span > 0) {
|
||||
const t = Math.min(1, Math.max(0, ((point.x - a.x) * dx + (point.y - a.y) * dy) / span ** 2));
|
||||
const distance = Math.hypot(a.x + dx * t - point.x, a.y + dy * t - point.y);
|
||||
if (best === null || distance < best.distance_m)
|
||||
best = { chainage_m: walked + span * t, distance_m: distance };
|
||||
}
|
||||
walked += span;
|
||||
}
|
||||
return best && walked > 0 ? { ...best, length_m: walked } : null;
|
||||
}
|
||||
|
||||
export interface FollowingPoint {
|
||||
chainage_m: number;
|
||||
x?: number | null;
|
||||
y?: number | null;
|
||||
start_m?: number | null;
|
||||
end_m?: number | null;
|
||||
}
|
||||
|
||||
/** 저장 좌표를 새 노선에 투영해 누가거리를 다시 매긴 사본(누가거리 순) — 관 지점. */
|
||||
export function projectPipePoints<P extends FollowingPoint>(
|
||||
points: ReadonlyArray<P>,
|
||||
line: ReadonlyArray<XY>,
|
||||
): P[] {
|
||||
const length = line.reduce(
|
||||
(sum, point, index) =>
|
||||
index ? sum + Math.hypot(point.x - line[index - 1].x, point.y - line[index - 1].y) : 0,
|
||||
0,
|
||||
);
|
||||
if (!(length > 0)) return [...points]; // 파이썬도 노선이 없으면 그대로(차례도)
|
||||
return points
|
||||
.map((point) => {
|
||||
if (point.x == null || point.y == null) return { ...point };
|
||||
const hit = projectOnPolyline(line, { x: point.x, y: point.y })!;
|
||||
const moved = snapStructureStation(hit.chainage_m, [hit.length_m]);
|
||||
const shift = moved - point.chainage_m;
|
||||
return {
|
||||
...point,
|
||||
chainage_m: moved,
|
||||
...(point.start_m != null ? { start_m: point.start_m + shift } : {}),
|
||||
...(point.end_m != null ? { end_m: point.end_m + shift } : {}),
|
||||
};
|
||||
})
|
||||
.sort((a, b) => a.chainage_m - b.chainage_m);
|
||||
}
|
||||
|
||||
export interface FollowingStructure {
|
||||
chainage_m?: number | null;
|
||||
start_m?: number | null;
|
||||
end_m?: number | null;
|
||||
x?: number | null;
|
||||
y?: number | null;
|
||||
}
|
||||
|
||||
/** 구조물이 좌표를 따라감 — 사본 · 노선과 거리(좌표 없으면 null) · 멀어짐(차례는 그대로). */
|
||||
export function followStructures<S extends FollowingStructure>(
|
||||
items: ReadonlyArray<S>,
|
||||
line: ReadonlyArray<XY>,
|
||||
maxDriftM: number = STRUCTURE_ROUTE_DRIFT_MAX_M,
|
||||
): Array<{ item: S; drift_m: number | null; far: boolean }> {
|
||||
return items.map((item) => {
|
||||
const hit =
|
||||
item.x != null && item.y != null ? projectOnPolyline(line, { x: item.x, y: item.y }) : null;
|
||||
if (hit === null || item.chainage_m == null)
|
||||
return { item: { ...item }, drift_m: null, far: false };
|
||||
const shift = snapStructureStation(hit.chainage_m, [hit.length_m]) - item.chainage_m;
|
||||
const clamp = (value: number): number => Math.min(Math.max(value + shift, 0), hit.length_m);
|
||||
return {
|
||||
item: {
|
||||
...item,
|
||||
chainage_m: clamp(item.chainage_m),
|
||||
...(item.start_m != null ? { start_m: clamp(item.start_m) } : {}),
|
||||
...(item.end_m != null ? { end_m: clamp(item.end_m) } : {}),
|
||||
},
|
||||
drift_m: hit.distance_m,
|
||||
far: hit.distance_m > maxDriftM,
|
||||
};
|
||||
});
|
||||
}
|
||||
@@ -241,6 +241,9 @@ PIPE_DEFAULT_DIAMETER_MM = int(os.getenv("PIPE_DEFAULT_DIAMETER_MM", "800"))
|
||||
ESCAPE_ROUTE_WIDTHS_M = (1.5, 2.0, 2.5, 3.0)
|
||||
ESCAPE_ROUTE_DEFAULT_WIDTH_M = 2.0
|
||||
STRUCTURE_ETC_DEFAULT_NAME = "기타 구조물"
|
||||
# 노선이 바뀌어 구조물 좌표가 새 노선에서 이만큼(m) 넘게 멀면 「노선에서 멀어짐」 표시(버리지 않음 · 344-2)
|
||||
# TS 짝 = common_util/common_util_route_follow.ts `STRUCTURE_ROUTE_DRIFT_MAX_M`(거울 시험이 대조).
|
||||
STRUCTURE_ROUTE_DRIFT_MAX_M = 20.0
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
/* 344-2 노선 짝 거울 — 진짜 TS(`logic/B05_Profile_Grade_Ground` · `logic/B05_Profile_Grade_Carry` ·
|
||||
* `common_util_route_follow`)를 Node 로 돌림. 입력 argv[2] JSON(파이썬과 같은 값) · 표준출력 JSON 한 벌.
|
||||
*/
|
||||
const fs = require("fs");
|
||||
const path = require("path");
|
||||
const ts = require(path.join(__dirname, "..", "..", "config", "node_modules", "typescript"));
|
||||
|
||||
const ROOT = path.join(__dirname, "..", "..");
|
||||
|
||||
function loadTs(file, stubs = {}) {
|
||||
const js = ts.transpileModule(fs.readFileSync(path.join(ROOT, file), "utf8"), {
|
||||
compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 },
|
||||
}).outputText;
|
||||
const box = { exports: {} };
|
||||
const localRequire = (request) => {
|
||||
if (stubs[request]) return stubs[request];
|
||||
throw new Error(`stub 없음: ${request} (${file})`);
|
||||
};
|
||||
new Function("exports", "module", "require", js)(box.exports, box, localRequire);
|
||||
return box.exports;
|
||||
}
|
||||
|
||||
const alignment = loadTs("B05_Profile/B05_Profile_UI_Profile_Alignment.ts");
|
||||
const ground = loadTs("B05_Profile/logic/B05_Profile_Grade_Ground.ts", {
|
||||
"../B05_Profile_UI_Profile_Alignment": alignment,
|
||||
});
|
||||
const carry = loadTs("B05_Profile/logic/B05_Profile_Grade_Carry.ts", {
|
||||
"../B05_Profile_UI_Profile_Alignment": alignment,
|
||||
});
|
||||
const catalog = require("./helper_322_catalog_module.cjs");
|
||||
const station = loadTs("common_util/common_util_structure_station.ts", {
|
||||
"./common_util_structure_catalog": catalog,
|
||||
});
|
||||
const follow = loadTs("common_util/common_util_route_follow.ts", {
|
||||
"./common_util_structure_station": station,
|
||||
});
|
||||
|
||||
const input = JSON.parse(fs.readFileSync(process.argv[2], "utf8"));
|
||||
const out = {
|
||||
ground: input.ground.map((item) => ground.groundFollowingBasePvi(item)),
|
||||
carry: input.carry.map((item) => {
|
||||
const span = carry.changedSpan(item.old_xy, item.new_xy);
|
||||
return {
|
||||
span,
|
||||
result: span
|
||||
? carry.carryProfileEdits(item.base_pvi, item.edits, span, item.ground_pvi)
|
||||
: null,
|
||||
};
|
||||
}),
|
||||
pipes: input.pipes.map((item) => follow.projectPipePoints(item.points, item.line)),
|
||||
structures: input.structures.map((item) => follow.followStructures(item.items, item.line)),
|
||||
driftMax: follow.STRUCTURE_ROUTE_DRIFT_MAX_M,
|
||||
};
|
||||
process.stdout.write(JSON.stringify(out));
|
||||
@@ -0,0 +1,402 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""344-2 실시간 바탕 짝 — 첫 계획선 · 관 자리 투영 · 구조물 좌표 따라감 · 노선 바뀜 때 계획고 옮김.
|
||||
|
||||
거울(파이썬 = 서버 정본 · TS = 브라우저 즉시 · 같은 값):
|
||||
① 첫 계획선 base_pvi — `design_ground_following_profile` ↔ `logic/B05_Profile_Grade_Ground.ts`
|
||||
(1순위만 · 못 서는 땅이면 TS 는 null = 「서버로 다시 셈」).
|
||||
② 관 자리 투영 — `common_util_drainage_pipes.project_pipe_points` ↔
|
||||
`common_util_route_follow.ts` `projectPipePoints`.
|
||||
③ 구조물 따라감 — `common_util_route_follow.follow_structures` ↔ `followStructures`(20 m 상수 같음).
|
||||
④ 계획고 옮김 — `B05_Profile_Engine_Grade_Carry` ↔ `logic/B05_Profile_Grade_Carry.ts`.
|
||||
서버 쪽: 저장 때 좌표 채움(자리가 그대로면 옛 좌표 지킴) · 노선 [확인] 때 구조물 정본이 따라감.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
ROOT = HERE.parents[1]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
import B05_Profile.B05_Profile_Structures_Router as structures_router # noqa: E402
|
||||
from B05_Profile.B05_Profile_Engine_Grade import resolve_grade_options # noqa: E402
|
||||
from B05_Profile.B05_Profile_Engine_Grade_Carry import ( # noqa: E402
|
||||
carry_profile_edits,
|
||||
changed_span,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Engine_Grade_Profile import ( # noqa: E402
|
||||
design_ground_following_profile,
|
||||
)
|
||||
from common_util.common_util_drainage_pipes import PipePoint, project_pipe_points # noqa: E402
|
||||
from common_util.common_util_route_follow import ( # noqa: E402
|
||||
fill_structure_coordinates,
|
||||
follow_structures,
|
||||
)
|
||||
from common_util.common_util_route_geometry import RouteVertex # noqa: E402
|
||||
from config.config_system import STRUCTURE_ROUTE_DRIFT_MAX_M # noqa: E402
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
not (ROOT / "config" / "node_modules" / "typescript").is_dir(),
|
||||
reason="프론트엔드 의존성 미설치",
|
||||
)
|
||||
|
||||
|
||||
# ── 입력 만들기 ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _longitudinal(total: float, *, wave: float, holes: tuple[int, ...] = ()) -> dict:
|
||||
samples = []
|
||||
for index in range(int(total) + 1):
|
||||
s = float(min(index, total))
|
||||
z = 100.0 + 0.06 * s + 4.0 * math.sin(s / wave) + 1.5 * math.cos(s / 7.3)
|
||||
samples.append({"chainage_m": s, "elevation_m": z, "valid": index not in holes})
|
||||
if samples[-1]["chainage_m"] < total:
|
||||
samples.append({"chainage_m": total, "elevation_m": 100.0 + 0.06 * total, "valid": True})
|
||||
stations = [float(s) for s in range(0, int(total) + 1, 20)] + [47.3, 133.0, total]
|
||||
return {"samples": samples, "stations": [{"chainage_m": s} for s in sorted(set(stations))]}
|
||||
|
||||
|
||||
def _ground_case(total: float, wave: float, stored: dict | None = None, paved=False) -> dict:
|
||||
longitudinal = _longitudinal(total, wave=wave, holes=(3, 4, 90))
|
||||
options = resolve_grade_options("trunk", terrain_type="normal", paved=paved, stored=stored)
|
||||
alignment, _ = design_ground_following_profile(longitudinal, options)
|
||||
return {
|
||||
"py": alignment["base_pvi"],
|
||||
"ts_input": {
|
||||
"samples": longitudinal["samples"],
|
||||
"stations": [row["chainage_m"] for row in longitudinal["stations"]],
|
||||
"policy": alignment["policy"],
|
||||
"startOffsetM": options.start_elevation_offset_m,
|
||||
"endOffsetM": options.end_elevation_offset_m,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
GROUND_CASES = [
|
||||
_ground_case(200.0, 11.0),
|
||||
_ground_case(279.8, 23.0, {"start_elevation_offset_m": 0.4, "end_elevation_offset_m": -0.3}),
|
||||
_ground_case(160.4, 5.0, paved=True),
|
||||
]
|
||||
GROUND_FAIL = {
|
||||
"samples": [{"chainage_m": 0.0, "elevation_m": 100.0}, {"chainage_m": 50.0, "valid": False}],
|
||||
"stations": [0.0, 50.0],
|
||||
"policy": GROUND_CASES[0]["ts_input"]["policy"],
|
||||
}
|
||||
|
||||
OLD_ROUTE = [(0.0, 0.0), (40.0, 5.0), (80.0, 0.0), (120.0, 10.0), (160.0, 5.0), (200.0, 0.0)]
|
||||
L_OLD = round(sum(math.dist(a, b) for a, b in zip(OLD_ROUTE, OLD_ROUTE[1:])), 3)
|
||||
|
||||
|
||||
def _carry_case(new_route: list[tuple[float, float]]) -> dict:
|
||||
total = sum(math.dist(a, b) for a, b in zip(new_route, new_route[1:]))
|
||||
longitudinal = _longitudinal(total, wave=13.0)
|
||||
options = resolve_grade_options("trunk", terrain_type="normal")
|
||||
alignment, _ = design_ground_following_profile(longitudinal, options)
|
||||
base_pvi = [
|
||||
{"chainage_m": s, "elevation_m": 200.0 + s / 10.0}
|
||||
for s in (0.0, 20.0, 40.0, 60.0, 90.5, 120.0, 150.0, 180.0, L_OLD)
|
||||
]
|
||||
edits = {
|
||||
"station_offsets": {"20.000": 0.5, "90.500": -0.3, "180.000": 0.2, f"{L_OLD:.3f}": 0.1},
|
||||
"curve_radii": {"40.000": 300.0, "150.000": 250.0},
|
||||
}
|
||||
return {
|
||||
"old_xy": [{"x": x, "y": y} for x, y in OLD_ROUTE],
|
||||
"new_xy": [{"x": x, "y": y} for x, y in new_route],
|
||||
"base_pvi": base_pvi,
|
||||
"edits": edits,
|
||||
"ground_pvi": alignment["base_pvi"],
|
||||
}
|
||||
|
||||
|
||||
CARRY_CASES = [
|
||||
# 가운데 노드(80, 0) 옮김 — 앞(0 ~ 40) 그대로 · 뒤(120 ~) 밂
|
||||
_carry_case(
|
||||
[(0.0, 0.0), (40.0, 5.0), (85.0, -12.0), (120.0, 10.0), (160.0, 5.0), (200.0, 0.0)]
|
||||
),
|
||||
# 끝 노드 옮김(꼬리 없음)
|
||||
_carry_case([(0.0, 0.0), (40.0, 5.0), (80.0, 0.0), (120.0, 10.0), (160.0, 5.0), (215.0, 9.0)]),
|
||||
# 앞 노드 옮김(앞머리 없음) · 점 하나 더함
|
||||
_carry_case(
|
||||
[
|
||||
(-6.0, 4.0),
|
||||
(20.0, 6.0),
|
||||
(40.0, 5.0),
|
||||
(80.0, 0.0),
|
||||
(120.0, 10.0),
|
||||
(160.0, 5.0),
|
||||
(200.0, 0.0),
|
||||
]
|
||||
),
|
||||
# 같음
|
||||
_carry_case(OLD_ROUTE),
|
||||
]
|
||||
|
||||
NEW_LINE = [(0.0, 0.0), (50.0, 0.0), (50.0, 50.0), (120.0, 60.0)]
|
||||
PIPES = [
|
||||
{"chainage_m": 10.0, "x": 12.3, "y": 1.1, "start_m": 8.0, "end_m": 14.0},
|
||||
{"chainage_m": 70.0, "x": 48.0, "y": 25.0},
|
||||
{"chainage_m": 5.0}, # 좌표 없음 — 누가거리 그대로
|
||||
{"chainage_m": 90.0, "x": 55.0, "y": -5.0, "start_m": 88.0}, # 모서리 둘 사이
|
||||
{"chainage_m": 150.0, "x": 119.5, "y": 61.0, "end_m": 152.0}, # 끝 밖
|
||||
]
|
||||
STRUCTURES = [
|
||||
{"chainage_m": 20.0, "x": 30.0, "y": 2.0},
|
||||
{
|
||||
"chainage_m": 3.0,
|
||||
"start_m": 0.0,
|
||||
"end_m": 12.0,
|
||||
"x": 1.0,
|
||||
"y": 0.5,
|
||||
}, # 앞으로 밀면 0 에서 자름
|
||||
{"chainage_m": 100.0, "x": 80.0, "y": 90.0}, # 노선에서 멀어짐(>20 m)
|
||||
{"chainage_m": 40.0}, # 좌표 없음
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def ts(tmp_path_factory) -> dict:
|
||||
source = tmp_path_factory.mktemp("s344") / "in.json"
|
||||
source.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"ground": [case["ts_input"] for case in GROUND_CASES] + [GROUND_FAIL],
|
||||
"carry": CARRY_CASES,
|
||||
"pipes": [{"points": PIPES, "line": [{"x": x, "y": y} for x, y in NEW_LINE]}],
|
||||
"structures": [
|
||||
{"items": STRUCTURES, "line": [{"x": x, "y": y} for x, y in NEW_LINE]}
|
||||
],
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
proc = subprocess.run( # noqa: S603 — 고정 실행 파일
|
||||
["node", str(HERE / "helper_344_2_route_pairs.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 _close(a: float, b: float, tol: float = 1e-6) -> bool:
|
||||
return abs(float(a) - float(b)) <= tol
|
||||
|
||||
|
||||
# ── ① 첫 계획선 ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.parametrize("index", range(len(GROUND_CASES)))
|
||||
def test_첫_계획선_거울(ts, index):
|
||||
py, got = GROUND_CASES[index]["py"], ts["ground"][index]
|
||||
assert len(got) == len(py)
|
||||
for a, b in zip(py, got):
|
||||
assert _close(a["chainage_m"], b["chainage_m"]) and _close(
|
||||
a["elevation_m"], b["elevation_m"]
|
||||
)
|
||||
|
||||
|
||||
def test_첫_계획선은_측점에서_지반고(ts):
|
||||
case = GROUND_CASES[0]
|
||||
assert len(case["py"]) >= 13 # 모든 측점이 꼭짓점(호 보정으로 꼭짓점 표고는 지반에서 뜸)
|
||||
|
||||
|
||||
def test_못_서는_땅은_서버로(ts):
|
||||
assert ts["ground"][-1] is None
|
||||
longitudinal = {"samples": GROUND_FAIL["samples"], "stations": []}
|
||||
with pytest.raises(ValueError):
|
||||
design_ground_following_profile(longitudinal, resolve_grade_options("trunk"))
|
||||
|
||||
|
||||
# ── ② 관 자리 투영 ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _vertices(line):
|
||||
out, walked = [], 0.0
|
||||
for index, (x, y) in enumerate(line):
|
||||
if index:
|
||||
walked += math.dist(line[index - 1], (x, y))
|
||||
out.append(RouteVertex(x=x, y=y, z=0.0, chainage_m=walked))
|
||||
return out
|
||||
|
||||
|
||||
def test_관_자리_투영_거울(ts):
|
||||
points = [
|
||||
PipePoint(
|
||||
chainage_m=p["chainage_m"],
|
||||
x=p.get("x"),
|
||||
y=p.get("y"),
|
||||
start_m=p.get("start_m"),
|
||||
end_m=p.get("end_m"),
|
||||
)
|
||||
for p in PIPES
|
||||
]
|
||||
py = project_pipe_points(points, _vertices(NEW_LINE))
|
||||
got = ts["pipes"][0]
|
||||
assert [p.chainage_m for p in py] == [g["chainage_m"] for g in got]
|
||||
for p, g in zip(py, got):
|
||||
assert p.start_m == g.get("start_m") and p.end_m == g.get("end_m")
|
||||
assert (p.x, p.y) == (g.get("x"), g.get("y"))
|
||||
# (55, -5) 은 두 선분에서 같은 거리 — 앞 선분 끝(50) · 끝 밖은 연장 안 정수(170)
|
||||
assert [g["chainage_m"] for g in got] == [5.0, 12.0, 50.0, 75.0, 170.0]
|
||||
|
||||
|
||||
# ── ③ 구조물 따라감 ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_구조물_따라감_거울(ts):
|
||||
items = [SimpleNamespace(**{"start_m": None, "end_m": None, **s}) for s in STRUCTURES]
|
||||
for item in items:
|
||||
item.x, item.y = getattr(item, "x", None), getattr(item, "y", None)
|
||||
py = follow_structures(items, NEW_LINE)
|
||||
got = ts["structures"][0]
|
||||
for item, (drift, far), entry in zip(items, py, got):
|
||||
moved = entry["item"]
|
||||
assert item.chainage_m == moved["chainage_m"]
|
||||
assert item.start_m == moved.get("start_m") and item.end_m == moved.get("end_m")
|
||||
assert (drift is None) == (entry["drift_m"] is None)
|
||||
if drift is not None:
|
||||
assert _close(drift, entry["drift_m"], 1e-9)
|
||||
assert far == entry["far"]
|
||||
assert [entry["far"] for entry in got] == [False, False, True, False]
|
||||
assert got[1]["item"]["start_m"] == 0.0 # 앞으로 밀려도 0 에서 자름
|
||||
assert got[3]["item"]["chainage_m"] == 40.0 # 좌표 없는 옛 자료 그대로
|
||||
|
||||
|
||||
def test_멀어짐_거리_상수_한_벌(ts):
|
||||
assert ts["driftMax"] == STRUCTURE_ROUTE_DRIFT_MAX_M == 20.0
|
||||
|
||||
|
||||
def test_저장_때_좌표_채움_자리가_그대로면_옛_좌표():
|
||||
line = [(0.0, 0.0), (100.0, 0.0)]
|
||||
kept = SimpleNamespace(structure_id="a", chainage_m=30.0, start_m=None, end_m=None, x=31, y=9)
|
||||
moved = SimpleNamespace(structure_id="b", chainage_m=50.0, start_m=None, end_m=None, x=1, y=1)
|
||||
fresh = SimpleNamespace(
|
||||
structure_id=None, chainage_m=70.0, start_m=None, end_m=None, x=None, y=None
|
||||
)
|
||||
previous = {
|
||||
"a": SimpleNamespace(chainage_m=30.0, start_m=None, end_m=None, x=31, y=9),
|
||||
"b": SimpleNamespace(chainage_m=40.0, start_m=None, end_m=None, x=1, y=1),
|
||||
}
|
||||
fill_structure_coordinates([kept, moved, fresh], line, previous=previous)
|
||||
assert (kept.x, kept.y) == (31, 9) # 따라간 땅 자리(노선 밖) 그대로
|
||||
assert (moved.x, moved.y) == (50.0, 0.0) # 사용자가 옮김 → 노선 위 새 자리
|
||||
assert (fresh.x, fresh.y) == (70.0, 0.0)
|
||||
|
||||
|
||||
def test_읽을_때는_없는_것만():
|
||||
line = [(0.0, 0.0), (100.0, 0.0)]
|
||||
old = SimpleNamespace(chainage_m=30.0, x=None, y=None)
|
||||
has = SimpleNamespace(chainage_m=60.0, x=5.0, y=5.0)
|
||||
assert fill_structure_coordinates([old, has], line, only_missing=True) == 1
|
||||
assert (old.x, old.y) == (30.0, 0.0) and (has.x, has.y) == (5.0, 5.0)
|
||||
|
||||
|
||||
def test_노선_확인_때_구조물_정본이_따라감(tmp_path, monkeypatch):
|
||||
route = tmp_path / "route.geojson"
|
||||
route.write_text(
|
||||
json.dumps({"geometry": {"coordinates": [[0, 0, 0], [100, 0, 0]]}}), encoding="utf-8"
|
||||
)
|
||||
items = [
|
||||
SimpleNamespace(chainage_m=30.0, start_m=None, end_m=None, x=None, y=None), # 옛 자료
|
||||
SimpleNamespace(chainage_m=60.0, start_m=55.0, end_m=65.0, x=60.0, y=0.0),
|
||||
]
|
||||
saved: dict = {}
|
||||
monkeypatch.setattr(structures_router, "load_structures", lambda root: (4, items))
|
||||
monkeypatch.setattr(
|
||||
structures_router,
|
||||
"save_structures",
|
||||
lambda root, values, *, base_revision: saved.update(values=values, revision=base_revision),
|
||||
)
|
||||
# 새 노선 = 앞쪽 10 m 늘림(같은 땅 자리는 누가거리 +10)
|
||||
new_xy = [{"x": -10.0, "y": 0.0}, {"x": 100.0, "y": 0.0}]
|
||||
result = structures_router.follow_structures_to_route(tmp_path, "route.geojson", new_xy)
|
||||
assert result == {"moved": 2, "far": 0}
|
||||
assert saved["revision"] == 4
|
||||
assert [(i.chainage_m, i.start_m, i.end_m) for i in items] == [
|
||||
(40.0, None, None),
|
||||
(70.0, 65.0, 75.0),
|
||||
]
|
||||
assert (items[0].x, items[0].y) == (30.0, 0.0) # 옛 노선으로 채운 땅 자리
|
||||
|
||||
|
||||
# ── ④ 계획고 옮김 ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _py_carry(case):
|
||||
span = changed_span(
|
||||
[(p["x"], p["y"]) for p in case["old_xy"]], [(p["x"], p["y"]) for p in case["new_xy"]]
|
||||
)
|
||||
result = (
|
||||
carry_profile_edits(case["base_pvi"], case["edits"], span, case["ground_pvi"])
|
||||
if span
|
||||
else None
|
||||
)
|
||||
return span, result
|
||||
|
||||
|
||||
@pytest.mark.parametrize("index", range(len(CARRY_CASES)))
|
||||
def test_계획고_옮김_거울(ts, index):
|
||||
span, result = _py_carry(CARRY_CASES[index])
|
||||
got = ts["carry"][index]
|
||||
if span is None:
|
||||
assert got["span"] is None and got["result"] is None
|
||||
return
|
||||
for key, value in span.items():
|
||||
assert (
|
||||
(value == got["span"][key])
|
||||
if isinstance(value, bool)
|
||||
else _close(value, got["span"][key], 1e-9)
|
||||
)
|
||||
base_pvi, edits = result
|
||||
ts_result = got["result"]
|
||||
assert len(base_pvi) == len(ts_result["basePvi"])
|
||||
for a, b in zip(base_pvi, ts_result["basePvi"]):
|
||||
assert _close(a["chainage_m"], b["chainage_m"]) and _close(
|
||||
a["elevation_m"], b["elevation_m"]
|
||||
)
|
||||
assert edits == ts_result["edits"]
|
||||
|
||||
|
||||
def test_가운데를_고치면_앞은_그대로_뒤는_밂():
|
||||
span, (base_pvi, edits) = _py_carry(CARRY_CASES[0])
|
||||
assert span["start_fixed"] and span["end_fixed"]
|
||||
delta = span["delta_m"]
|
||||
assert span["start_m"] == pytest.approx(math.dist(OLD_ROUTE[0], OLD_ROUTE[1]))
|
||||
# 앞 편집 그대로 · 안(90.5) 버림 · 뒤(180 · 207.6) 길이 차만큼
|
||||
assert edits["station_offsets"]["20.000"] == 0.5
|
||||
assert "90.500" not in edits["station_offsets"]
|
||||
assert edits["station_offsets"][f"{180.0 + delta:.3f}"] == 0.2
|
||||
assert edits["curve_radii"] == {"40.000": 300.0, f"{150.0 + delta:.3f}": 250.0}
|
||||
old = {round(p["chainage_m"], 3): p["elevation_m"] for p in CARRY_CASES[0]["base_pvi"]}
|
||||
got = {round(p["chainage_m"], 3): p["elevation_m"] for p in base_pvi}
|
||||
assert got[20.0] == old[20.0] and got[round(180.0 + delta, 3)] == old[180.0]
|
||||
# 바뀐 구간 안은 새 지반 첫 계획선 꼭짓점
|
||||
ground = {round(p["chainage_m"], 3): p["elevation_m"] for p in CARRY_CASES[0]["ground_pvi"]}
|
||||
inside = [s for s in got if span["start_m"] + 1e-6 < s < span["new_end_m"] - 1e-6]
|
||||
assert inside and all(got[s] == ground[s] for s in inside)
|
||||
assert min(got) == 0.0 and max(got) == pytest.approx(L_OLD + delta, abs=1e-3)
|
||||
|
||||
|
||||
def test_끝을_고치면_끝_꼭짓점은_새_지반():
|
||||
span, (base_pvi, edits) = _py_carry(CARRY_CASES[1])
|
||||
assert not span["end_fixed"]
|
||||
assert f"{L_OLD:.3f}" not in edits["station_offsets"] # 바뀐 끝 편집은 버림
|
||||
assert edits["station_offsets"]["90.500"] == -0.3 # 앞(바뀐 구간 앞) 편집 그대로
|
||||
ground_end = CARRY_CASES[1]["ground_pvi"][-1]
|
||||
assert base_pvi[-1] == {
|
||||
"chainage_m": round(ground_end["chainage_m"], 6),
|
||||
"elevation_m": ground_end["elevation_m"],
|
||||
}
|
||||
Reference in New Issue
Block a user