eomsangdonandClaude Opus 5.5
7a857d626a
- 판정 = 저장 · 새 계획 폴리라인 1 mm 견줌(`route_shape_changed` · TS 짝 `routeShapeChanged` = sub1) · 스냅 · 간격만 = 따라감 그대로
- 체인 초기화 = 사용자 구조물 · 추가 측점 · 관(옵션 포함) 버림 · 관 = 새 노선 세류선 교차만 · 1차 계획선 · 횡단 기본 설계 · 응답 route_reset {reset, dropped}
- 한 본문 초기화 = 보낸 관 · 구조물 · 추가 측점 · 계획고 · 관 옵션 · 넘김 칸 안 씀(페이지 설정만) · 되돌림 범위에 initial_snapshot
- 추천 초기 종단 미리 셈 = 같은 초기화 시설 · 지문에서 시설 파일 뺌
- 시험: 새 test_283_route_change_reset(옛 코드 8 실패) · 검증_sub8 자료 고정 · 실DB 시험 둘 = storage 임시 사본(helper_temp_storage) · 따라감 기대 시험 문구 고침
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
402 lines
16 KiB
Python
402 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""476-1 추천 후보 초기 종단 — 미리 셈 = [적용] 체인 종단 · 화면 펴기 짝 · 응답 칸 · 저장본 지문.
|
|
|
|
고정 자료 = `fixtures/golden_472`(사본) · 체인 쪽 = `run_route_save_chain` 파일 길 차례(노선 csv →
|
|
초기화 = 구조물 · 추가 측점 비움(492) → 노선 geojson → 관 = 새 교차만 · 정본 →
|
|
`run_section_generation` →
|
|
켬이면 `_set_base_pvi` + `_apply_alignment_edits`) · 미리 셈 쪽 = 메모리 조각(파일 안 씀).
|
|
화면 펴기 = TS `flattenServerPlan`(Node `helper_476_1_flatten.cjs`) ↔ 파이썬 `draft_nodes`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import copy
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
TESTER = Path(__file__).resolve().parent
|
|
for path in (ROOT, TESTER):
|
|
if str(path) not in sys.path:
|
|
sys.path.insert(0, str(path))
|
|
|
|
import helper_472_1_paths as P # noqa: E402
|
|
from test_224_3_recommend_store import _ask, wired # noqa: E402, F401 — 가짜 세계 fixture
|
|
|
|
from B05_Profile import B05_Profile_Engine_InitialProfile as ip # noqa: E402
|
|
|
|
RADIUS = 12.0
|
|
HELPER = TESTER / "helper_476_1_flatten.cjs"
|
|
NODES_CSV = "B05_Profile/route/planned_route_initial_nodes.csv"
|
|
WALL = {
|
|
"structure_id": "wall600",
|
|
"type_id": "retaining_wall",
|
|
"placement": "interval",
|
|
"chainage_m": 600.0,
|
|
"start_m": 595.0,
|
|
"end_m": 605.0,
|
|
"options": {"wall_form": "콘크리트", "length_m": 10, "height_m": 2},
|
|
"memo": "",
|
|
"placement_source": "manual",
|
|
"status": "draft",
|
|
"revision": 0,
|
|
"geometry": None,
|
|
"summary_item": None,
|
|
"group_items": None,
|
|
"x": 208785.614,
|
|
"y": 367845.1028,
|
|
}
|
|
|
|
|
|
def _candidate(root: Path, shift: float = 0.0) -> dict:
|
|
"""엔진 결과 꼴 후보(`nodes` · `curves` as_dict) — 교점 9 를 동으로 `shift` m."""
|
|
from B05_Profile.B05_Profile_Router_Replan import _vertices_of
|
|
from common_util.common_util_route_polyline import build_planned_polyline
|
|
|
|
nodes = [tuple(p) for p in _vertices_of(root / NODES_CSV)]
|
|
nodes[9] = (nodes[9][0] + shift, nodes[9][1])
|
|
outline = build_planned_polyline(nodes, min_radius_m=RADIUS, simplify=False)
|
|
return {
|
|
"nodes": [n.as_dict() for n in outline.nodes],
|
|
"curves": [c.as_dict() for c in outline.curves],
|
|
}
|
|
|
|
|
|
def _browser_draft(candidate: dict, work: Path) -> dict:
|
|
if not (ROOT / "config" / "node_modules" / "typescript").is_dir():
|
|
pytest.skip("프론트엔드 의존성 미설치")
|
|
body = work / "candidate.json"
|
|
body.write_text(json.dumps(candidate), encoding="utf-8")
|
|
out = subprocess.run(
|
|
["node", str(HELPER), str(body)], capture_output=True, text=True, check=True
|
|
)
|
|
return json.loads(out.stdout)
|
|
|
|
|
|
def _settings() -> dict:
|
|
from B05_Profile.B05_Profile_Engine_Grade import GradeDesignOptions
|
|
from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions
|
|
|
|
summary = P.read("summary.json")["longitudinal"]["data"]
|
|
options = {k: v for k, v in summary["options"].items() if k != "extra_stations"}
|
|
return {
|
|
"options": SectionGenerationOptions(**options),
|
|
"grade_options": GradeDesignOptions(**summary["grade_options"]),
|
|
"crs": f"EPSG:{P.read('context.json')['crs_epsg']}",
|
|
}
|
|
|
|
|
|
def _old_xy(root: Path) -> list[list[float]]:
|
|
coords = json.loads((root / P.ROUTE).read_text("utf-8"))["geometry"]["coordinates"]
|
|
return [[c[0], c[1]] for c in coords]
|
|
|
|
|
|
def _seed(root: Path, extras: bool) -> None:
|
|
from B05_Profile.B05_Profile_Plus_Stations import save_plus_stations
|
|
from common_util.common_util_json import atomic_write_json
|
|
|
|
if extras:
|
|
atomic_write_json(
|
|
root / "B05_Profile" / "route" / "structures.json",
|
|
{"revision": 1, "structures": [copy.deepcopy(WALL)], "definition_version": None},
|
|
)
|
|
save_plus_stations(root, [510], _old_xy(root))
|
|
|
|
|
|
def _chain(root: Path, draft: dict, target) -> dict:
|
|
"""`run_route_save_chain` 파일 길 차례(DB 없이) — [적용] 뒤 종단 정본."""
|
|
from B05_Profile.B05_Profile_Engine import _route_geojson
|
|
from B05_Profile.B05_Profile_Engine_AsPlanned import _vertex_z
|
|
from B05_Profile.B05_Profile_Engine_Sections import run_section_generation
|
|
from B05_Profile.B05_Profile_Plus_Stations import save_plus_stations
|
|
from B05_Profile.B05_Profile_Route_Save import followed_pipes
|
|
from B05_Profile.B05_Profile_Router import _apply_alignment_edits
|
|
from B05_Profile.B05_Profile_Router_Replan import _vertices_of, _write_planned_polyline
|
|
from B05_Profile.B05_Profile_Router_RouteInitial import _set_base_pvi
|
|
from B05_Profile.B05_Profile_Structures_Repository import load_structures, save_structures
|
|
from common_util.common_util_drainage_pipes import (
|
|
ensure_pipe_ids,
|
|
fill_pipe_coordinates,
|
|
pipe_points_path_in,
|
|
route_signature,
|
|
)
|
|
from common_util.common_util_json import atomic_write_json
|
|
from common_util.common_util_route_geometry import planned_route_working_path
|
|
|
|
working = planned_route_working_path(root)
|
|
count = len(draft["nodes"])
|
|
_write_planned_polyline(
|
|
working,
|
|
[tuple(n) for n in draft["nodes"]],
|
|
RADIUS,
|
|
simplify=False,
|
|
curve_flags=draft["curveOn"],
|
|
radii=draft["curveRadius"],
|
|
station_snap=True,
|
|
station_snap_step=None,
|
|
locks=[None] * count,
|
|
arcs=[None] * count,
|
|
)
|
|
points = [{"x": x, "y": y} for x, y in _vertices_of(working)]
|
|
revision, items = load_structures(str(root)) # [적용] = 초기화(492) — 사용자 시설 버림
|
|
if items:
|
|
save_structures(str(root), [], base_revision=revision)
|
|
save_plus_stations(root, [], None)
|
|
xy = [(p["x"], p["y"]) for p in points]
|
|
z = _vertex_z(root, *P.SURFACE, xy)
|
|
atomic_write_json(root / P.ROUTE, _route_geojson([[x, y, h] for (x, y), h in zip(xy, z)]))
|
|
pairs, _crossing, vertices = followed_pipes([], [], points, [])
|
|
pipes = [pipe for _, pipe in pairs]
|
|
fill_pipe_coordinates(pipes, vertices)
|
|
ensure_pipe_ids(pipes)
|
|
atomic_write_json(
|
|
pipe_points_path_in(root),
|
|
{"route_signature": route_signature(vertices), "points": [p.as_dict() for p in pipes]},
|
|
)
|
|
settings = _settings()
|
|
run_section_generation(root, P.ROUTE, *P.SURFACE, **settings)
|
|
if target is not None:
|
|
_set_base_pvi(root / P.LONG, *target)
|
|
_apply_alignment_edits(root, P.LONG, {})
|
|
return json.loads((root / P.LONG).read_text("utf-8"))
|
|
|
|
|
|
def _preview(root: Path, candidate: dict, target) -> tuple[dict, list, list]:
|
|
from common_util.common_util_surface_sampler import build_surface_sampler
|
|
|
|
line, curves = ip.candidate_line(candidate["nodes"], candidate["curves"], RADIUS)
|
|
facilities = ip.reset_facilities(line, [])
|
|
sampler = build_surface_sampler(root / "B04_PreProcess" / "models", *P.SURFACE)
|
|
longitudinal = ip.initial_longitudinal(
|
|
root,
|
|
line,
|
|
curves,
|
|
sampler,
|
|
facilities,
|
|
source_snapshot={"filter": "classification", "method": "dtm", "smooth": True},
|
|
target=target,
|
|
**_settings(),
|
|
)
|
|
return longitudinal, line, curves
|
|
|
|
|
|
def _target(root: Path, line, curves):
|
|
from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import route_profile
|
|
from common_util.common_util_surface_sampler import build_surface_sampler
|
|
|
|
sampler = build_surface_sampler(root / "B04_PreProcess" / "models", *P.SURFACE)
|
|
profile = route_profile(line, sampler, max_grade=0.14, design_speed=20, design_curves=curves)
|
|
return profile["stations_m"], profile["plan_m"]
|
|
|
|
|
|
CASES = {
|
|
"same_off": {"shift": 0.0, "extras": False, "balanced": False},
|
|
"moved_off": {"shift": 5.0, "extras": True, "balanced": False},
|
|
"moved_on": {"shift": 5.0, "extras": True, "balanced": True},
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="module", params=list(CASES))
|
|
def pair(request, tmp_path_factory):
|
|
case = CASES[request.param]
|
|
work = tmp_path_factory.mktemp(request.param)
|
|
chain_root = P.stage_project(work / "chain")
|
|
preview_root = P.stage_project(work / "preview")
|
|
for root in (chain_root, preview_root):
|
|
_seed(root, case["extras"])
|
|
candidate = _candidate(preview_root, case["shift"])
|
|
draft = _browser_draft(candidate, work)
|
|
target = None
|
|
if case["balanced"]:
|
|
line, curves = ip.candidate_line(candidate["nodes"], candidate["curves"], RADIUS)
|
|
target = _target(preview_root, line, curves)
|
|
preview, line, curves = _preview(preview_root, candidate, target)
|
|
chain = _chain(chain_root, draft, target)
|
|
return {"preview": preview, "chain": chain, "line": line, "curves": curves, "root": chain_root}
|
|
|
|
|
|
def test_화면_펴기_짝(tmp_path):
|
|
root = P.stage_project(tmp_path / "p")
|
|
candidate = _candidate(root, 5.0)
|
|
draft = _browser_draft(candidate, tmp_path)
|
|
points, flags, radii = ip.draft_nodes(candidate["nodes"], candidate["curves"])
|
|
assert [list(p) for p in points] == draft["nodes"]
|
|
assert flags == draft["curveOn"] and radii == draft["curveRadius"]
|
|
|
|
|
|
def test_노선_곡선표_같음(pair):
|
|
from B05_Profile.B05_Profile_Engine_Sections import _load_design_curves
|
|
from B05_Profile.B05_Profile_Router_Replan import _vertices_of
|
|
from common_util.common_util_route_geometry import planned_route_working_path
|
|
|
|
root = pair["root"]
|
|
assert [tuple(p) for p in _vertices_of(planned_route_working_path(root))] == pair["line"]
|
|
assert _load_design_curves(root) == pair["curves"]
|
|
|
|
|
|
def test_미리_셈_종단_체인과_1mm(pair):
|
|
preview, chain = pair["preview"], pair["chain"]
|
|
got = [float(s["chainage_m"]) for s in preview["stations"]]
|
|
want = [float(s["chainage_m"]) for s in chain["stations"]]
|
|
assert got == pytest.approx(want, abs=1e-6)
|
|
a = preview["profile_alignment"]["base_pvi"]
|
|
b = chain["profile_alignment"]["base_pvi"]
|
|
assert len(a) == len(b)
|
|
for p, q in zip(a, b):
|
|
assert abs(p["chainage_m"] - q["chainage_m"]) <= 1e-3
|
|
assert abs(p["elevation_m"] - q["elevation_m"]) <= 1e-3
|
|
assert ip.profile_summary(preview)["plan_m"] == pytest.approx(
|
|
ip.profile_summary(chain)["plan_m"], abs=1e-3
|
|
)
|
|
|
|
|
|
def test_초기_종단_438_규칙(pair):
|
|
summary = ip.profile_summary(pair["preview"])
|
|
assert summary["curve_length_m"] == 20.0 # 432 L 20
|
|
assert summary["straightened"] > 0 # R > 500 꼭짓점 묶음 = 직선 위(438)
|
|
curves = pair["preview"]["profile_alignment"]["curves"]
|
|
assert not [c for c in curves if not c["omitted"] and c["r_m"] > 500.0 + 1e-6]
|
|
assert len(summary["line"]) > len(summary["stations_m"]) # 종단곡선 포함 선
|
|
|
|
|
|
# ── 응답 칸 · 저장본 지문 ──────────────────────────────────────────────────────
|
|
|
|
|
|
class _Pool:
|
|
def acquire(self):
|
|
pool = self
|
|
|
|
class _Acquire:
|
|
async def __aenter__(self):
|
|
return pool
|
|
|
|
async def __aexit__(self, *exc):
|
|
return False
|
|
|
|
return _Acquire()
|
|
|
|
def cursor(self, *args):
|
|
return self.acquire()
|
|
|
|
|
|
def _async(value):
|
|
async def fake(*args, **kwargs):
|
|
return value() if callable(value) else value
|
|
|
|
return fake
|
|
|
|
|
|
@pytest.fixture
|
|
def gather(tmp_path, monkeypatch):
|
|
import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed
|
|
import B05_Profile.B05_Profile_Repository as b05_repo
|
|
import B05_Profile.B05_Profile_Router_Replan as replan
|
|
import B05_Profile.B05_Profile_Router_RouteInitial as route_initial
|
|
import B05_Profile.B05_Profile_Router_RouteInitial_Initial as gatherer
|
|
import B06_Section.B06_Section_Repository as b06_repo
|
|
import common_util.common_util_drainage_pipes as drainage
|
|
import common_util.common_util_workflow_state as workflow
|
|
|
|
root = P.stage_project(tmp_path / "project")
|
|
summary = P.read("summary.json")["longitudinal"]["data"]
|
|
grade = summary["grade_options"]
|
|
params = {
|
|
"options": {
|
|
"grade_class": grade.get("grade_class") or "trunk",
|
|
"terrain_type": grade.get("terrain_type") or "special",
|
|
"design_speed_kph": grade.get("design_speed_kph"),
|
|
},
|
|
"max_grade_pct": grade.get("max_grade_pct"),
|
|
}
|
|
state = {"stages": [{"stage_no": 2, "params": params}]}
|
|
pipes_file = root / "B04_PreProcess" / "drainage" / "edits" / "pipe_points.json"
|
|
monkeypatch.setattr(gatherer, "get_db_pool", lambda: _Pool())
|
|
monkeypatch.setattr(replan, "_plan_criteria", _async((RADIUS, 0.0, 0.0)))
|
|
monkeypatch.setattr(replan, "_road_settings", _async(("trunk", 20, "special")))
|
|
monkeypatch.setattr(b05_repo, "get_latest_route", _async({"id": 1, "route_data_path": P.ROUTE}))
|
|
monkeypatch.setattr(b06_repo, "get_route_generation_source", _async({"crs_epsg": 5186}))
|
|
monkeypatch.setattr(b06_repo, "get_latest_section_options", _async(summary["options"]))
|
|
monkeypatch.setattr(workflow, "get_workflow_state", _async(lambda: copy.deepcopy(state)))
|
|
monkeypatch.setattr(route_initial, "project_standard", _async(None))
|
|
monkeypatch.setattr(watershed, "_prepare", _async({"streams": []}))
|
|
monkeypatch.setattr(watershed, "_sheet_dir", lambda stored: root / "none")
|
|
monkeypatch.setattr(drainage, "pipe_points_path", lambda stored: pipes_file)
|
|
request = replan.RouteInitialRequest(mode="follow", with_profile=False)
|
|
results = root / "B05_Profile" / "route_recommend"
|
|
results.mkdir(parents=True, exist_ok=True)
|
|
(results / "k1.json").write_text("{}", "utf-8") # 결과 저장본 짝(없으면 초기 종단 파일 지움)
|
|
candidate = _candidate(root)
|
|
selection = {"source_filter": "classification", "method": "dtm", "smooth": True}
|
|
|
|
def run(key: str = "k1") -> dict:
|
|
return asyncio.run(
|
|
gatherer.initial_profile_of(uuid4(), root, "p", candidate, request, selection, key)
|
|
)
|
|
|
|
return {"run": run, "root": root, "state": state, "pipes": pipes_file}
|
|
|
|
|
|
def test_응답_칸(gather):
|
|
got = gather["run"]()
|
|
assert "error" not in got, got
|
|
for name in (
|
|
"stations_m",
|
|
"ground_m",
|
|
"plan_m",
|
|
"line",
|
|
"pvi_count",
|
|
"curve_count",
|
|
"straightened",
|
|
"max_grade_pct",
|
|
"curve_length_m",
|
|
"warnings",
|
|
):
|
|
assert name in got, name
|
|
assert got["with_profile"] is False and got["curve_length_m"] == 20.0
|
|
assert len(got["stations_m"]) == len(got["plan_m"]) == len(got["ground_m"])
|
|
|
|
|
|
def test_지문_같으면_꺼냄_다르면_다시(gather, monkeypatch):
|
|
first = gather["run"]()
|
|
assert "error" not in first, first
|
|
|
|
def broken(*args, **kwargs):
|
|
raise RuntimeError("다시 셈")
|
|
|
|
monkeypatch.setattr(ip, "initial_longitudinal", broken)
|
|
assert gather["run"]() == first # 같은 입력 = 저장본
|
|
gather["state"]["stages"][0]["params"]["max_grade_pct"] = 9.0 # 페이지 설정 최대 기울기
|
|
assert "다시 셈" in gather["run"]()["error"]
|
|
gather["state"]["stages"][0]["params"].pop("max_grade_pct")
|
|
doc = json.loads(gather["pipes"].read_text("utf-8"))
|
|
doc["points"][0]["x"] = float(doc["points"][0]["x"]) + 1.0 # 관 하나 옮김
|
|
gather["pipes"].write_text(json.dumps(doc), "utf-8")
|
|
assert gather["run"]() == first # 초기화(492) = 저장 관을 안 씀 → 지문 그대로
|
|
|
|
|
|
def test_코드_바뀌면_다시(gather, monkeypatch):
|
|
assert "error" not in gather["run"]()
|
|
monkeypatch.setattr(ip, "source_fingerprint", lambda: "other")
|
|
monkeypatch.setattr(ip, "initial_longitudinal", lambda *a, **k: {})
|
|
assert "error" in gather["run"]() # 438 등 소스 바뀜 = 다시 셈(빈 종단 = 실패 글)
|
|
|
|
|
|
def test_추천_응답에_initial_profile(wired, monkeypatch):
|
|
import B05_Profile.B05_Profile_Router_RouteInitial_Initial as gatherer
|
|
|
|
seen: list = []
|
|
|
|
async def fake(project_id, root, stored, result, request, selection, key):
|
|
seen.append(key)
|
|
return {"pvi_count": 3}
|
|
|
|
monkeypatch.setattr(gatherer, "initial_profile_of", fake)
|
|
body = _ask(mode="earthwork") # 224-3 가짜 세계
|
|
assert body["initial_profile"] == {"pvi_count": 3} and seen # 옛 코드 = 칸 없음
|