Merge remote-tracking branch 'origin/sub_laptop_7' into main_laptop_1

This commit is contained in:
2026-10-10 13:36:56 +09:00
4 changed files with 233 additions and 26 deletions
+80 -5
View File
@@ -13,6 +13,8 @@ from uuid import UUID
import aiomysql
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B05_Profile.B05_Profile_Engine_Grade_Alignment import chainage_key
from B05_Profile.B05_Profile_Engine_Grade_Profile import rebuild_alignment_profile
from B05_Profile.B05_Profile_Engine_Sections import generate_irregular_sections
from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions
from B05_Profile.B05_Profile_Repository import get_surface_crs_epsg
@@ -181,28 +183,101 @@ async def sync_uphill_overrides_into_designs(
)
#: 반올림만 다른 같은 측점(m) — 서버가 관 측점을 0.01m 로 반올림.
#: 화면 `applyIrregularStations` 0.006 과 짝.
_SAME_STATION_M = 0.006
def _follow_moved_stations(data: dict[str, Any], before: list[float]) -> dict[str, Any] | None:
"""사라진 비정규 측점의 계획선 꼭짓점(base_pvi) · 편집을 새 측점으로 옮기거나 지운다.
남기면 측점선 없는 변화점이 계획선을 꺾는다(192-1 — 관 769.28 → 769.38 뒤 두 자리에
편집이 남아 0.1m 사이 −465% 계단). 반올림 차이(0.006m 안)만 옮기고 나머지는 지운다.
바뀌었으면 새 등급 요약(DB 몫)을, 아니면 `None`.
"""
stored = data.get("profile_alignment")
if not isinstance(stored, dict) or not stored.get("base_pvi"):
return None
after = [float(station.get("chainage_m", 0.0)) for station in data["stations"]]
after_keys = {chainage_key(value) for value in after}
before_keys = {chainage_key(value) for value in before}
moves: dict[str, str | None] = {}
for old in before:
if chainage_key(old) in after_keys:
continue
near = next(
(
value
for value in after
if abs(value - old) < _SAME_STATION_M and chainage_key(value) not in before_keys
),
None,
)
moves[chainage_key(old)] = None if near is None else chainage_key(near)
edits = stored.get("edits") or {}
next_edits: dict[str, dict[str, Any]] = {}
changed = False
for name in ("station_offsets", "curve_radii"):
values = dict(edits.get(name) or {})
for old_key, new_key in moves.items():
if old_key in values:
value = values.pop(old_key)
if new_key is not None:
values.setdefault(new_key, value)
changed = True
next_edits[name] = values
base: list[dict[str, Any]] = []
for node in stored["base_pvi"]:
key = chainage_key(node["chainage_m"])
if key not in moves:
base.append(node)
continue
changed = True
if moves[key] is not None:
base.append({**node, "chainage_m": float(moves[key])})
if not changed:
return None
stored["base_pvi"] = base
alignment, profile = rebuild_alignment_profile(data, next_edits)
data["profile_alignment"] = alignment
profiles = data.get("design_profiles")
if isinstance(profiles, list) and profiles:
profiles[0] = profile
else:
data["design_profiles"] = [profile]
return {"id": profile["id"], **profile["summary"]}
def _merge_irregular_into_longitudinal(
project_root: Path, longitudinal_file_path: str, irregular_stations: list[dict[str, Any]]
) -> None:
) -> dict[str, Any] | None:
"""종단 정본 파일의 stations에 비정규 측점을 병합한다(기존 비정규는 교체·정렬).
B06 상세는 종단 파일의 stations를 읽고 그에 맞는 cross 파일만 노출하므로, 이 병합과
(엔진이 이미 쓴) cross 파일만으로 다음 페이지에 횡단이 나타난다. 계획선·규칙 측점·표고
샘플은 손대지 않는다. 파일 기반이라 재확정해도 중복이 생기지 않는다.
(엔진이 이미 쓴) cross 파일만으로 다음 페이지에 횡단이 나타난다. 규칙 측점·표고 샘플은
손대지 않는다. 계획선은 사라진 비정규 측점 자리만 고친다(`_follow_moved_stations` —
바뀌면 새 등급 요약을 돌려줌). 파일 기반이라 재확정해도 중복이 생기지 않는다.
"""
root = project_root.resolve()
path = (root / longitudinal_file_path).resolve()
if root not in path.parents or not path.is_file():
return
return None
data = json.loads(path.read_text(encoding="utf-8"))
stations = data.get("stations")
if not isinstance(stations, list):
return
return None
before = [
float(station.get("chainage_m", 0.0))
for station in stations
if station.get("kind") == "irregular"
]
regular = [station for station in stations if station.get("kind") != "irregular"]
merged = regular + list(irregular_stations)
merged.sort(key=lambda station: float(station.get("chainage_m", 0.0)))
data["stations"] = merged
summary = _follow_moved_stations(data, before)
atomic_write_json(path, data)
return summary
#: 지표 샘플링 조건을 적어 두는 파일 — **나중에 측점을 더 만들 때 같은 조건으로** 뜨기 위해.
+22 -13
View File
@@ -28,7 +28,11 @@ from fastapi import APIRouter
from fastapi.responses import JSONResponse
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B05_Profile.B05_Profile_Repository import get_latest_route, get_surface_crs_epsg
from B05_Profile.B05_Profile_Repository import (
get_latest_route,
get_surface_crs_epsg,
update_longitudinal_grade_summary,
)
from B05_Profile.B05_Profile_Router_Confirm import load_sampling_snapshot
from common_util.common_util_storage import resolve_stored_project_path
from config.config_db import run_with_connection
@@ -37,14 +41,10 @@ logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/projects", tags=["B06 Section"])
#: 「그 자리에 측점이 있다」고 볼 거리(m).
#: ⚠⚠ **0.5m 다 — 0.05m 가 아니다**(2026-09-09 실측으로 뒤집힌 자리).
#: 측점을 만들 때 **정수 미터가 같은 격자 측점이 있으면 그리로 스냅**한다
#: (`B05_Profile_Engine_Sections_Core` 의 파일명 가드). 그래서 관 440.241 은 측점 440.0
#: 위에 서고, 그 측점은 **구조물 이름표까지 달고 있다**(`structure`).
#: 0.05m 로 보면 그런 자리를 「측점 없음」으로 잘못 세어 **있는 측점을 또 만들라고 한다.**
#: 스냅 폭이 「정수 미터 반올림」이므로 최대 어긋남은 0.5m 다.
STATION_MATCH_TOLERANCE_M = 0.5
#: 「그 자리에 측점이 있다」고 볼 거리(m) — 측점 생성의 스냅 폭과 같음
#: (`B05_Profile_Engine_Sections_Core` · 같은 횡단 이름이 되는 0.005m 안만 한 측점).
#: 넓으면 조금 옮긴 관(769.28 → 769.38)을 「측점 있음」으로 세어 측점이 옛 자리에 남음(192-1).
STATION_MATCH_TOLERANCE_M = 0.005
SNAPSHOT_MISSING = (
"지표 샘플링 조건을 찾을 수 없어 측점을 만들 수 없음 — 1단계(지표 확정)를 마친 뒤"
@@ -167,7 +167,7 @@ async def _regenerate_structure_stations(
stored_options, crs_epsg, longitudinal = await run_with_connection(_load)
def _regenerate() -> int:
def _regenerate() -> tuple[int, dict[str, Any] | None]:
polyline = _load_route_polyline(project_root, str(route["route_data_path"]))
pipes = _load_pipe_points(project_root, polyline)
extras = resolve_extra_stations(project_root, pipes)
@@ -181,13 +181,22 @@ async def _regenerate_structure_stations(
options=_section_options_from_stored(stored_options),
crs=f"EPSG:{crs_epsg}" if crs_epsg is not None else None,
)
summary = None
if stations and longitudinal:
_merge_irregular_into_longitudinal(
summary = _merge_irregular_into_longitudinal(
project_root, str(longitudinal["longitudinal_file_path"]), stations
)
return len(stations)
return len(stations), summary
return await asyncio.to_thread(_regenerate)
made, summary = await asyncio.to_thread(_regenerate)
if summary is not None:
# 옮긴 관 자리의 계획선이 다시 셈됨 — DB 등급 요약도 같이(계획선 저장과 같은 길).
await run_with_connection(
lambda connection: update_longitudinal_grade_summary(
connection, route_id=route["id"], grade_summary=summary
)
)
return made
async def ensure_structure_stations(project_id: UUID) -> int:
@@ -0,0 +1,128 @@
"""192-1 — 종단 계획선 숨은 꺾임(검증_랩탑메인 226fc805 · No.38+9).
까닭: 관을 769.28 → 769.38(0.1m) 옮김 → 「측점 있음」 판정 폭이 0.5m 라 측점이 옛 자리
769.28 에 남음 → B06(측점 769.28)과 B05(관 769.38) 두 자리에 계획고 편집이 남아 0.1m 사이
−465% 계단 · 769.38 은 측점선 없는 꼭짓점.
고침: 판정 폭 0.005m(측점 스냅 폭) · 측점을 다시 뜨면 사라진 측점의 꼭짓점 · 편집을 지움
(반올림 차이 0.006m 안은 옮김). 사용자 storage 는 읽기만 — 사본(tmp_path)으로 재현.
"""
from __future__ import annotations
import copy
import json
import shutil
import sys
from pathlib import Path
import pytest
PROJECT_ROOT = Path(__file__).resolve().parents[2]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
import B05_Profile.B05_Profile_Engine_Sections as sections # noqa: E402
from B05_Profile.B05_Profile_Router_Confirm import _merge_irregular_into_longitudinal # noqa: E402
from B06_Section.B06_Section_Router_Stations import _missing_marks # noqa: E402
SOURCE = (
PROJECT_ROOT
/ "storage/1/3/226fc805-26c8-4f5a-9956-bacb200e10a5/B06_Section/longitudinal/longitudinal.json"
)
REL = "B06_Section/longitudinal/longitudinal.json"
needs_copy = pytest.mark.skipif(not SOURCE.is_file(), reason="검증_랩탑메인 storage 없음")
def _copy(tmp_path: Path) -> Path:
target = tmp_path / REL
target.parent.mkdir(parents=True)
shutil.copyfile(SOURCE, target)
return target
def _load(path: Path) -> dict:
return json.loads(path.read_text(encoding="utf-8"))
def _irregular(data: dict, old: float, new: float) -> list[dict]:
"""다시 뜬 비정규 측점 목록 흉내 — `old` 측점만 `new` 로 옮김."""
result = []
for station in data["stations"]:
if station.get("kind") != "irregular":
continue
item = copy.deepcopy(station)
if abs(item["chainage_m"] - old) < 1e-6:
item["chainage_m"] = new
result.append(item)
return result
@needs_copy
def test_원본_자료가_숨은_꼭짓점을_가짐() -> None:
"""까닭 근거 — 769.38 꼭짓점은 측점 목록에 없고 두 편집 사이 −465%."""
data = _load(SOURCE)
alignment = data["profile_alignment"]
stations = [s["chainage_m"] for s in data["stations"]]
pvis = [p["chainage_m"] for p in alignment["pvi"]]
assert 769.38 in pvis and 769.38 not in stations and 769.28 in stations
assert set(alignment["edits"]["station_offsets"]) == {"769.280", "769.380"}
step = next(s for s in alignment["segments"] if s["from_m"] == 769.28)
assert step["to_m"] == 769.38 and step["grade_percent"] < -400
@needs_copy
def test_관_옮긴_측점_다시_뜨면_숨은_꼭짓점_사라짐(tmp_path: Path) -> None:
path = _copy(tmp_path)
data = _load(path)
summary = _merge_irregular_into_longitudinal(tmp_path, REL, _irregular(data, 769.28, 769.38))
after = _load(path)
alignment = after["profile_alignment"]
stations = {round(s["chainage_m"], 3) for s in after["stations"]}
near = [p for p in alignment["pvi"] if 760 < p["chainage_m"] < 780]
assert [p["chainage_m"] for p in near] == [769.38] # 꼭짓점 하나 · 측점선 있음
assert 769.38 in stations and 769.28 not in stations
assert alignment["edits"]["station_offsets"] == {"769.380": pytest.approx(0.051428)}
assert all(abs(b["chainage_m"] - 769.28) > 1e-6 for b in alignment["base_pvi"])
grades = [s["grade_percent"] for s in alignment["segments"] if 760 <= s["from_m"] < 780]
assert max(abs(g) for g in grades) < 20
assert summary is not None and summary["max_grade_pct"] < 20
assert after["design_profiles"][0]["summary"]["max_grade_pct"] == summary["max_grade_pct"]
@needs_copy
def test_반올림_차이만_다르면_편집을_옮김(tmp_path: Path) -> None:
path = _copy(tmp_path)
data = _load(path)
data["profile_alignment"]["edits"]["station_offsets"]["484.210"] = 0.3
path.write_text(json.dumps(data), encoding="utf-8")
_merge_irregular_into_longitudinal(tmp_path, REL, _irregular(data, 484.21, 484.213))
offsets = _load(path)["profile_alignment"]["edits"]["station_offsets"]
assert "484.210" not in offsets and offsets["484.213"] == pytest.approx(0.3)
@needs_copy
def test_측점이_그대로면_계획선을_건드리지_않음(tmp_path: Path) -> None:
path = _copy(tmp_path)
data = _load(path)
before = data["profile_alignment"]
assert _merge_irregular_into_longitudinal(tmp_path, REL, _irregular(data, -1, -1)) is None
assert _load(path)["profile_alignment"] == before
def test_조금_옮긴_관도_측점_없음으로_셈(tmp_path: Path, monkeypatch) -> None:
"""0.1m 옮긴 관 — 옛 판정(0.5m)은 「측점 있음」으로 놓쳤음."""
target = tmp_path / REL
target.parent.mkdir(parents=True)
target.write_text(
json.dumps({"stations": [{"chainage_m": 760.0}, {"chainage_m": 769.28}]}),
encoding="utf-8",
)
monkeypatch.setattr(sections, "_load_route_polyline", lambda *_: [])
monkeypatch.setattr(sections, "_load_pipe_points", lambda *_: [])
monkeypatch.setattr(
sections, "resolve_extra_stations", lambda *_: [(769.38, "파형강관 D800"), (760.0, "옹벽")]
)
assert _missing_marks(tmp_path, "route.csv") == [
{"chainage_m": 769.38, "label": "파형강관 D800"}
]
@@ -47,14 +47,9 @@ def test_측점이_없으면_빈_목록(tmp_path: Path) -> None:
assert _station_chainages(tmp_path) == []
def test_같은_자리_판정은_50cm() -> None:
"""⚠ 스냅 때문이다 — 관 440.241 은 **측점 440.0** 위에 선다(2026-09-09 실측).
측점을 만들 때 정수 미터가 같은 격자 측점이 있으면 그리로 스냅하므로(횡단 파일명이
정수 미터라 두 측점이 한 파일을 덮어쓰는 것을 막는 가드) 최대 어긋남이 0.5m 다.
0.05m 로 보면 **있는 측점을 없다고 세어** 또 만들라고 한다.
"""
assert STATION_MATCH_TOLERANCE_M == 0.5
def test_같은_자리_판정은_측점_스냅_폭() -> None:
"""측점 생성 스냅 폭(0.005m · 같은 횡단 이름)과 같음 — 0.5m 면 0.1m 옮긴 관을 놓침(192-1)."""
assert STATION_MATCH_TOLERANCE_M == 0.005
def test_샘플링_조건이_없으면_None(tmp_path: Path) -> None: