eomsangdonandClaude Opus 5.5
aa87dcfd68
- 새 `B05_Profile_Engine_InitialProfile` — 후보 노드 → 화면 펴기 짝(`draft_nodes` ↔ TS flattenServerPlan) → 노선 짜기(csv 4자리) → 관 · 구조물 · 추가 측점 따라감(새 세류선 교차 관 포함) → build_sections(438 1차 계획선) → 켬 = 1 m 균형 종단 얹기(seat_base_pvi)
- 새 `B05_Profile_Router_RouteInitial_Initial.initial_profile_of` — [적용] 체인과 같은 입력 모으기 · 지문(후보 · 지형 · 페이지 설정 · 시설 파일 · 측점 옵션 · 균형 인자 · 코드) · 저장본 `route_recommend/initial/<키>.json` · 실패 = {error}
- compute_initial 응답에 `initial_profile`(stations_m · ground_m · plan_m · line · pvi_count · curve_count · straightened · max_grade_pct · curve_length_m · warnings · with_profile · seconds)
- 시험 test_476_1_recommend_initial — 미리 셈 ↔ 체인 파일 길 꼭짓점 차 0 mm(끔 · 노선 바뀜 + 옹벽 + 추가 측점 · 켬) · 지문 · 응답 칸
- 위키 calculations N2 · rules 줄
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
410 lines
16 KiB
Python
410 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""476-1 추천 후보 초기 종단 — 미리 셈 = [적용] 체인 종단 · 화면 펴기 짝 · 응답 칸 · 저장본 지문.
|
|
|
|
고정 자료 = `fixtures/golden_472`(사본) · 체인 쪽 = `run_route_save_chain` 파일 길 차례(노선 csv →
|
|
구조물 · 추가 측점 따라감 → 노선 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 follow_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_Router import follow_structures_to_route
|
|
from common_util.common_util_drainage_pipes import (
|
|
ensure_pipe_ids,
|
|
fill_pipe_coordinates,
|
|
pipe_points_path_in,
|
|
read_pipe_points_file,
|
|
route_signature,
|
|
)
|
|
from common_util.common_util_json import atomic_write_json
|
|
from common_util.common_util_route_geometry import planned_route_working_path
|
|
|
|
old_xy = _old_xy(root)
|
|
old_pipes = read_pipe_points_file(pipe_points_path_in(root))
|
|
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)]
|
|
follow_structures_to_route(root, P.ROUTE, points)
|
|
follow_plus_stations(root, [(p["x"], p["y"]) for p in points])
|
|
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(old_pipes, old_xy, 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_drainage_pipes import pipe_points_path_in, read_pipe_points_file
|
|
from common_util.common_util_surface_sampler import build_surface_sampler
|
|
|
|
line, curves = ip.candidate_line(candidate["nodes"], candidate["curves"], RADIUS)
|
|
facilities = ip.followed_facilities(
|
|
root,
|
|
line,
|
|
P.ROUTE,
|
|
_old_xy(root),
|
|
read_pipe_points_file(pipe_points_path_in(root)),
|
|
[],
|
|
)
|
|
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 "다시 셈" in gather["run"]()["error"]
|
|
|
|
|
|
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 # 옛 코드 = 칸 없음
|