fix(B05·B06 측점): 구조물 놓음 · 뺌에도 서버 종단 곡선 측점 다시 셈 (407-3)
- [저장] 횡단 차례 ensure_structure_stations 가 곡선 측점 어긋남(_curve_marks)도 셈 · 노선 저장 안 기다림 - 다시 뜰 때 곡선 측점도 뜸(sample_structure_sections with_curves) · 종단 병합이 옛 곡선 측점도 갈음 - 측점 자리 한 벌 _station_layout 으로 뺌 · planned_curve_stations = generate_sections 와 같은 목록(표본 없음) - 시험: test_407_3_curve_station_follow(놓음 = 구조물 측점 하나 · 뺌 = 곡선 측점 되살아남) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013QyNXY2TUKXCFMwAca2APG
This commit is contained in:
@@ -553,14 +553,18 @@ def sample_structure_sections(
|
||||
extra_stations: tuple[tuple[float, str], ...],
|
||||
options: SectionGenerationOptions | None = None,
|
||||
crs: str | None = None,
|
||||
with_curves: bool = False,
|
||||
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||||
"""구조물 측점만 샘플링해 (측점 dict, 횡단 dict) 목록을 돌려준다 — 파일은 안 씀.
|
||||
|
||||
`with_curves` — 곡선 측점(`curve` · 252-2)도 같이(정본 쓰기 · 구조물이 MC 를 비우면
|
||||
되살림 · 407-3).
|
||||
|
||||
격자 측점과 똑같은 접선·오프셋 샘플링을 거치되(`generate_sections` 재사용) 구조물 측점
|
||||
(`irregular` · 격자 · 시종점에 얹힌 것 `grid` 포함)만 골라낸다. [저장] 때 정본 쓰기
|
||||
(`generate_irregular_sections`)와 B06 미리 뜨기(`preview_structure_sections`)가 같은 길(273-2).
|
||||
"""
|
||||
if not extra_stations:
|
||||
if not extra_stations and not with_curves:
|
||||
return [], []
|
||||
base = options or SectionGenerationOptions()
|
||||
merged_options = SectionGenerationOptions(
|
||||
@@ -587,6 +591,8 @@ def sample_structure_sections(
|
||||
)
|
||||
|
||||
def structural(entry: dict[str, Any]) -> bool:
|
||||
if with_curves and entry.get("kind") == "curve":
|
||||
return True
|
||||
return entry.get("kind") == "irregular" or bool(entry.get("structure"))
|
||||
|
||||
return (
|
||||
@@ -606,7 +612,8 @@ def generate_irregular_sections(
|
||||
options: SectionGenerationOptions | None = None,
|
||||
crs: str | None = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""구조물 측점만 샘플링해 **횡단 파일을 쓰고** 측점 dict 목록을 돌려준다(빈 목록 = 없음).
|
||||
"""구조물 측점(곡선 측점 포함 · 407-3)만 샘플링해 **횡단 파일을 쓰고** 측점 dict 목록을
|
||||
돌려준다(빈 목록 = 없음).
|
||||
|
||||
기존 종단 계획선·규칙 측점·횡단은 건드리지 않는다 — 받는 쪽이 목록 전체로 다시 맞춤(272).
|
||||
각각 `cross_*.json`으로 저장한다. B06 상세 조회가 이 폴더를 glob하고 종단 파일의 stations로
|
||||
@@ -621,6 +628,7 @@ def generate_irregular_sections(
|
||||
extra_stations=extra_stations,
|
||||
options=options,
|
||||
crs=crs,
|
||||
with_curves=True,
|
||||
)
|
||||
if not irregular_stations:
|
||||
return []
|
||||
|
||||
@@ -381,32 +381,17 @@ def _design_curve_widenings(
|
||||
return radii, sides, widenings
|
||||
|
||||
|
||||
def generate_sections(
|
||||
polyline: np.ndarray | list[list[float]],
|
||||
sampler: SurfaceElevationSampler,
|
||||
options: SectionGenerationOptions | None = None,
|
||||
*,
|
||||
source_snapshot: dict[str, Any] | None = None,
|
||||
crs: str | None = None,
|
||||
design_curves: list[dict[str, Any]] | None = None,
|
||||
station_snap: bool = False,
|
||||
station_snap_step: float | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""확정 경로로 CAD 인계 가능한 종단·횡단 원시 데이터를 생성한다.
|
||||
|
||||
`station_snap` — 호 안 측점 하나를 곡선 중점으로 옮김(`snap_curve_stations` · 요구 224-2) ·
|
||||
`station_snap_step` — 그 MC 측점 값 자릿수를 정하는 스냅 간격(390-5 · None = 측점 No.).
|
||||
MC 에 측점이 없으면 곡선 측점(kind `curve` · `curve_stations` · 요구 252-2) 하나를 더함.
|
||||
"""
|
||||
options = options or SectionGenerationOptions()
|
||||
options.validate()
|
||||
points, route_chainage = _clean_polyline(np.asarray(polyline, dtype=np.float64))
|
||||
def _station_layout(
|
||||
points: np.ndarray,
|
||||
route_chainage: np.ndarray,
|
||||
options: SectionGenerationOptions,
|
||||
design_curves: list[dict[str, Any]] | None,
|
||||
station_snap: bool,
|
||||
station_snap_step: float | None,
|
||||
) -> tuple[np.ndarray, list, set[float], dict[float, str], set[float]]:
|
||||
"""측점 자리 한 벌(지반 표본 없음) — (측점 누가거리 · 곡선 구간 · 격자 자리 · 구조물 측점 ·
|
||||
곡선 측점). `generate_sections` · `planned_curve_stations`(B06 측점 맞춤 판정 · 407-3)가 같은 길."""
|
||||
total = float(route_chainage[-1])
|
||||
|
||||
long_chainage = _chainages(total, options.long_sample_interval_m, True)
|
||||
long_xy = _interpolate_xy(points, route_chainage, long_chainage)
|
||||
long_z, long_valid = sampler.sample_xy(long_xy)
|
||||
|
||||
station_chainage = _chainages(total, options.station_interval_m, options.include_endpoint)
|
||||
spans = _design_curve_spans(points, route_chainage, design_curves or [])
|
||||
if station_snap:
|
||||
@@ -440,6 +425,56 @@ def generate_sections(
|
||||
curves = {round(value, 6) for value in curve_stations(station_chainage, spans, total, step)}
|
||||
if curves:
|
||||
station_chainage = np.unique(np.r_[station_chainage, np.array(sorted(curves))])
|
||||
return station_chainage, spans, regular_rounded, extras, curves
|
||||
|
||||
|
||||
def planned_curve_stations(
|
||||
polyline: np.ndarray | list[list[float]],
|
||||
options: SectionGenerationOptions | None = None,
|
||||
*,
|
||||
design_curves: list[dict[str, Any]] | None = None,
|
||||
station_snap: bool = False,
|
||||
station_snap_step: float | None = None,
|
||||
) -> list[float]:
|
||||
"""지금 노선 · 관 · 구조물(`options.extra_stations`)이 세울 곡선 측점 누가거리(오름차순) —
|
||||
`generate_sections` 와 같은 값 · 표본은 안 뜸."""
|
||||
options = options or SectionGenerationOptions()
|
||||
points, route_chainage = _clean_polyline(np.asarray(polyline, dtype=np.float64))
|
||||
layout = _station_layout(
|
||||
points, route_chainage, options, design_curves, station_snap, station_snap_step
|
||||
)
|
||||
return sorted(layout[4])
|
||||
|
||||
|
||||
def generate_sections(
|
||||
polyline: np.ndarray | list[list[float]],
|
||||
sampler: SurfaceElevationSampler,
|
||||
options: SectionGenerationOptions | None = None,
|
||||
*,
|
||||
source_snapshot: dict[str, Any] | None = None,
|
||||
crs: str | None = None,
|
||||
design_curves: list[dict[str, Any]] | None = None,
|
||||
station_snap: bool = False,
|
||||
station_snap_step: float | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""확정 경로로 CAD 인계 가능한 종단·횡단 원시 데이터를 생성한다.
|
||||
|
||||
`station_snap` — 호 안 측점 하나를 곡선 중점으로 옮김(`snap_curve_stations` · 요구 224-2) ·
|
||||
`station_snap_step` — 그 MC 측점 값 자릿수를 정하는 스냅 간격(390-5 · None = 측점 No.).
|
||||
MC 에 측점이 없으면 곡선 측점(kind `curve` · `curve_stations` · 요구 252-2) 하나를 더함.
|
||||
"""
|
||||
options = options or SectionGenerationOptions()
|
||||
options.validate()
|
||||
points, route_chainage = _clean_polyline(np.asarray(polyline, dtype=np.float64))
|
||||
total = float(route_chainage[-1])
|
||||
|
||||
long_chainage = _chainages(total, options.long_sample_interval_m, True)
|
||||
long_xy = _interpolate_xy(points, route_chainage, long_chainage)
|
||||
long_z, long_valid = sampler.sample_xy(long_xy)
|
||||
|
||||
station_chainage, spans, regular_rounded, extras, curves = _station_layout(
|
||||
points, route_chainage, options, design_curves, station_snap, station_snap_step
|
||||
)
|
||||
station_xy = _interpolate_xy(points, route_chainage, station_chainage)
|
||||
tangents = np.vstack(
|
||||
[
|
||||
|
||||
@@ -300,7 +300,8 @@ def _merge_irregular_into_longitudinal(
|
||||
`irregular_stations` = 지금 관 · 구조물로 새로 뜬 구조물 측점 전부(격자와 같은 m 인 것은
|
||||
`grid` 표시 · `generate_irregular_sections`). 옛 구조물 측점은 다 빼고(빈 목록이어도),
|
||||
구조물이 떠난 격자 자리(`grid`)는 일반 측점으로 되돌리고, 새 목록의 격자 자리 측점은 그
|
||||
격자 측점을 갈음한다(같은 m 측점 하나 · 구조물이 위). 표고 샘플은 손대지 않는다.
|
||||
격자 측점을 갈음한다(같은 m 측점 하나 · 구조물이 위). 곡선 측점(`curve`)도 새 목록이 통째로
|
||||
갈음(구조물이 MC 를 비우면 되살아남 · 407-3). 표고 샘플은 손대지 않는다.
|
||||
B06 상세는 종단 파일의 stations 에 맞는 cross 파일만 노출하고 남은 파일을 지운다
|
||||
(`prune_stale_cross_files`). 계획선은 사라진 구조물 측점 자리만 고친다(`_follow_moved_stations`
|
||||
— 바뀌면 새 등급 요약을 돌려줌). 파일 기반이라 다시 맞춰도 중복이 생기지 않는다.
|
||||
@@ -318,14 +319,16 @@ def _merge_irregular_into_longitudinal(
|
||||
before = [
|
||||
float(station.get("chainage_m", 0.0))
|
||||
for station in stations
|
||||
if station.get("kind") == "irregular"
|
||||
if station.get("kind") in ("irregular", "curve")
|
||||
]
|
||||
fresh = {round(float(s.get("chainage_m", 0.0)), 6): s for s in irregular_stations}
|
||||
kept: dict[float, dict[str, Any]] = {}
|
||||
left: list[dict[str, Any]] = []
|
||||
for station in stations:
|
||||
if station.get("kind") == "irregular" and not station.get("grid"):
|
||||
continue # 옛 구조물 측점 — 새 목록이 갈음
|
||||
if station.get("kind") == "curve" or (
|
||||
station.get("kind") == "irregular" and not station.get("grid")
|
||||
):
|
||||
continue # 옛 구조물 측점 · 곡선 측점(407-3) — 새 목록이 갈음
|
||||
key = round(float(station.get("chainage_m", 0.0)), 6)
|
||||
if key not in fresh and ("structure" in station or station.get("grid")):
|
||||
# 구조물이 떠난 격자 · 시종점 자리 — 일반 측점으로(옛 판 「regular + 글」 포함)
|
||||
|
||||
@@ -156,6 +156,47 @@ def _stale_marks(project_root: Path, route_data_path: str) -> int:
|
||||
return len(standing ^ planted)
|
||||
|
||||
|
||||
def _curve_marks(project_root: Path, route_data_path: str, options: Any) -> int:
|
||||
"""종단 정본의 곡선 측점(`curve`)과 **지금 노선 · 관 · 구조물**이 세울 곡선 측점이 어긋난 수
|
||||
(407-3 — 구조물을 MC 위에 놓거나 빼면 바뀜 · 화면 초안 `buildStations` 와 같은 목록).
|
||||
종단 정본이 없으면 0."""
|
||||
import json
|
||||
from dataclasses import replace
|
||||
|
||||
from B05_Profile.B05_Profile_Engine_Sections import (
|
||||
_load_design_curves,
|
||||
_load_pipe_points,
|
||||
_load_route_polyline,
|
||||
load_station_snap,
|
||||
load_station_snap_step,
|
||||
resolve_extra_stations,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Engine_Sections_Core import planned_curve_stations
|
||||
|
||||
files = sorted((project_root / "B06_Section" / "longitudinal").glob("*.json"))
|
||||
if not files:
|
||||
return 0
|
||||
standing: set[float] = set()
|
||||
for path in files:
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
for station in data.get("stations") or []:
|
||||
if station.get("kind") == "curve":
|
||||
standing.add(round(float(station.get("chainage_m") or 0.0), 2))
|
||||
polyline = _load_route_polyline(project_root, route_data_path)
|
||||
extras = resolve_extra_stations(project_root, _load_pipe_points(project_root, polyline))
|
||||
planned = planned_curve_stations(
|
||||
polyline,
|
||||
replace(options, extra_stations=extras),
|
||||
design_curves=_load_design_curves(project_root),
|
||||
station_snap=load_station_snap(project_root),
|
||||
station_snap_step=load_station_snap_step(project_root),
|
||||
)
|
||||
return len({round(value, 2) for value in planned} ^ standing)
|
||||
|
||||
|
||||
async def _project_paths(project_id: UUID) -> tuple[Path, dict[str, Any]] | None:
|
||||
async def _load(connection):
|
||||
stored = await get_project_storage_relative_path(connection, project_id)
|
||||
@@ -250,6 +291,7 @@ async def _regenerate_structure_stations(
|
||||
|
||||
async def ensure_structure_stations(project_id: UUID) -> int:
|
||||
"""빠진 구조물 측점을 세우고 낡은 것을 지운다 — 횡단 저장마다 부름(맞출 것이 있을 때만 돎).
|
||||
곡선 측점이 지금 관 · 구조물과 어긋나도 맞춤(407-3 · [저장] 의 관 · 구조물 차례 뒤).
|
||||
|
||||
옮긴 관 · 구조물 측점은 B05 확정 전이면 종단 측점 · 횡단 파일 · DB 행이 없어, 그 자리
|
||||
횡단 수정이 받을 곳 없이 버려진다. [측점 만들기] 단추와 같은 조건으로 세운다.
|
||||
@@ -262,6 +304,17 @@ async def ensure_structure_stations(project_id: UUID) -> int:
|
||||
missing = await asyncio.to_thread(_missing_marks, project_root, str(route["route_data_path"]))
|
||||
# 빠진 것만이 아니라 낡은 것(옮긴 관 옛 자리 · 떠난 격자 자리)도 맞춤 — 두 벌로 남지 않게(272)
|
||||
stale = await asyncio.to_thread(_stale_marks, project_root, str(route["route_data_path"]))
|
||||
# 곡선 측점도 지금 관 · 구조물로 — MC 위 구조물을 빼면 되살아남(노선 저장 안 기다림 · 407-3)
|
||||
from B05_Profile.B05_Profile_Router_Confirm import _section_options_from_stored
|
||||
from B06_Section.B06_Section_Repository import get_latest_section_options
|
||||
|
||||
stored = await run_with_connection(get_latest_section_options, project_id)
|
||||
stale += await asyncio.to_thread(
|
||||
_curve_marks,
|
||||
project_root,
|
||||
str(route["route_data_path"]),
|
||||
_section_options_from_stored(stored),
|
||||
)
|
||||
if not missing and not stale:
|
||||
return 0
|
||||
snapshot = await _sampling_conditions(project_id, project_root)
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
"""PLAN 407-3 — 구조물을 MC 위에 놓았다 빼면 서버 종단 정본의 곡선 측점이 되살아남.
|
||||
|
||||
[저장] 차례(관 · 구조물 → 횡단) 의 `ensure_structure_stations` 길 — `_curve_marks`(어긋남 판정) ·
|
||||
`sample_structure_sections(with_curves=True)`(곡선 측점도 뜸) · `_merge_irregular_into_longitudinal`
|
||||
(옛 곡선 측점도 새 목록이 갈음). 노선 저장을 안 기다림 · 놓은 동안엔 그 자리 구조물 측점 하나.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from B05_Profile import B05_Profile_Engine_Sections as sections # noqa: E402
|
||||
from B05_Profile import B05_Profile_Engine_Sections_Core as core # noqa: E402
|
||||
from B05_Profile.B05_Profile_Router_Confirm import ( # noqa: E402
|
||||
_merge_irregular_into_longitudinal,
|
||||
)
|
||||
from B06_Section.B06_Section_Router_Stations import _curve_marks # noqa: E402
|
||||
|
||||
LINE = [[0.0, 0.0, 100.0], [1000.3, 0.0, 100.0]]
|
||||
SHORT = {"start": [101.0, 0.0], "apex": [107.3, 2.0], "end": [113.6, 0.0], "radius_m": 30.0}
|
||||
LONG = {"start": [616.1, 0.0], "apex": [631.3, 3.0], "end": [646.5, 0.0], "radius_m": 40.0}
|
||||
|
||||
|
||||
class _Flat:
|
||||
def sample_xy(self, xy):
|
||||
count = len(np.asarray(xy, dtype=np.float64))
|
||||
return np.full(count, 100.0), np.ones(count, dtype=bool)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict:
|
||||
"""노선 · 곡선표 · 관 · 구조물 정본 대신 — 바꿀 수 있는 구조물 측점 하나(`extras`)."""
|
||||
state: dict = {"extras": ()}
|
||||
monkeypatch.setattr(sections, "_load_route_polyline", lambda *_: LINE)
|
||||
monkeypatch.setattr(sections, "_load_design_curves", lambda *_: [SHORT, LONG])
|
||||
monkeypatch.setattr(sections, "load_station_snap", lambda *_: False)
|
||||
monkeypatch.setattr(sections, "load_station_snap_step", lambda *_: None)
|
||||
monkeypatch.setattr(sections, "build_surface_sampler", lambda *_, **__: _Flat())
|
||||
monkeypatch.setattr(sections, "_load_pipe_points", lambda *_: [])
|
||||
monkeypatch.setattr(sections, "resolve_extra_stations", lambda *_: state["extras"])
|
||||
# 노선 저장 때 정본 = 전체 생성(곡선 측점 107 · 631)
|
||||
full = core.generate_sections(LINE, _Flat(), design_curves=[SHORT, LONG])["longitudinal"]
|
||||
folder = tmp_path / "B06_Section" / "longitudinal"
|
||||
folder.mkdir(parents=True)
|
||||
(tmp_path / "B06_Section" / "cross_sections").mkdir()
|
||||
(folder / "longitudinal.json").write_text(json.dumps(full), encoding="utf-8")
|
||||
state["root"] = tmp_path
|
||||
return state
|
||||
|
||||
|
||||
def _save(state: dict) -> list[dict]:
|
||||
"""[저장] 의 횡단 차례 — 어긋나면 다시 떠 병합(`ensure_structure_stations` 와 같은 길)."""
|
||||
root = state["root"]
|
||||
options = core.SectionGenerationOptions()
|
||||
if _curve_marks(root, "route.json", options) == 0 and not state["extras"]:
|
||||
return _stations(root)
|
||||
fresh = sections.generate_irregular_sections(
|
||||
root, "route.json", "f", "m", False, extra_stations=state["extras"], options=options
|
||||
)
|
||||
_merge_irregular_into_longitudinal(root, "B06_Section/longitudinal/longitudinal.json", fresh)
|
||||
return _stations(root)
|
||||
|
||||
|
||||
def _stations(root: Path) -> list[dict]:
|
||||
path = root / "B06_Section" / "longitudinal" / "longitudinal.json"
|
||||
return json.loads(path.read_text(encoding="utf-8"))["stations"]
|
||||
|
||||
|
||||
def _at(stations: list[dict], chainage: float) -> list[str]:
|
||||
return [s["kind"] for s in stations if abs(s["chainage_m"] - chainage) < 0.01]
|
||||
|
||||
|
||||
def test_처음은_맞음(project) -> None:
|
||||
assert _curve_marks(project["root"], "route.json", core.SectionGenerationOptions()) == 0
|
||||
assert _at(_stations(project["root"]), 107.0) == ["curve"]
|
||||
|
||||
|
||||
def test_MC_위_구조물_놓음_뺌(project) -> None:
|
||||
root = project["root"]
|
||||
project["extras"] = ((107.0, "관"),)
|
||||
assert _curve_marks(root, "route.json", core.SectionGenerationOptions()) == 1
|
||||
placed = _save(project)
|
||||
assert _at(placed, 107.0) == ["irregular"] # 놓은 동안 = 구조물 측점 하나
|
||||
assert _at(placed, 631.0) == ["curve"] # 다른 곡선 측점은 그대로
|
||||
assert list((root / "B06_Section" / "cross_sections").glob("cross_*.json"))
|
||||
|
||||
project["extras"] = ()
|
||||
assert _curve_marks(root, "route.json", core.SectionGenerationOptions()) == 1
|
||||
removed = _save(project)
|
||||
assert _at(removed, 107.0) == ["curve"] # 노선 저장 없이 되살아남
|
||||
assert _at(removed, 631.0) == ["curve"]
|
||||
assert _curve_marks(root, "route.json", core.SectionGenerationOptions()) == 0
|
||||
assert len(removed) == len(_stations_full())
|
||||
|
||||
|
||||
def _stations_full() -> list[dict]:
|
||||
return core.generate_sections(LINE, _Flat(), design_curves=[SHORT, LONG])["longitudinal"][
|
||||
"stations"
|
||||
]
|
||||
|
||||
|
||||
def test_planned_curve_stations_는_생성과_같음() -> None:
|
||||
extras = ((107.0, "관"), (300.0, "옹벽"))
|
||||
for snap, step in [(False, None), (True, None), (True, 0.1)]:
|
||||
options = core.SectionGenerationOptions(extra_stations=extras)
|
||||
made = core.generate_sections(
|
||||
LINE,
|
||||
_Flat(),
|
||||
options,
|
||||
design_curves=[SHORT, LONG],
|
||||
station_snap=snap,
|
||||
station_snap_step=step,
|
||||
)["longitudinal"]["stations"]
|
||||
planned = core.planned_curve_stations(
|
||||
LINE,
|
||||
options,
|
||||
design_curves=[SHORT, LONG],
|
||||
station_snap=snap,
|
||||
station_snap_step=step,
|
||||
)
|
||||
assert planned == [s["chainage_m"] for s in made if s["kind"] == "curve"]
|
||||
Reference in New Issue
Block a user