feat(562-2): D03 노선 정본 GET · PUT network — 손 그은 선 · 갈래 · 마크 · 조각 → 조각 행 · 연도 행 · 사업연도 끝 · 옛 추천 엔진 · 꺾임점 R 걷음
- 새 D03_PlanAnalysis_Network.py — 서버가 marks 로 조각 경계 · 길이 · 측점 다시 셈(높이 저장 안 함) · 옛 연도 노선은 읽을 때 한 선으로 · 업체 지정 조각 지움 409 · 산림청만 - 새 db_management/021_plan_network.sql — plan_networks · plan_route_pieces · plan_route_assignees.piece_id(DB 반영함) - main.py 라우터 한 벌 · Router.py 의 POST analyze · GET/PUT analysis 덩이 지움(지형 · 진행 길 그대로) - 지움: D03_PlanAnalysis_Analyze.py · D03_PlanAnalysis_Edit.py · config PLAN_ANALYZE_WORKERS - 시험: test_530_2_plan_analyze = 노선 정본 시험으로 다시 씀(옛 엔진 시험은 엔진과 함께 걷음) · test_530_2_plan_router = 분석 API 시험 지우고 끈 지점 규칙만 · test_530_3_plan_logic = Analyze 짝 시험 지움 · test_543_plan_edit 지움(꺾임점 R 고침 걷음 — 요구 389 11 직선만) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -1,400 +0,0 @@
|
||||
"""D 예정노선 2단계 분석 — 연도 구간마다 B05 노선 추천 엔진으로 지점을 이음 (요구 345 · PLAN 530-2).
|
||||
|
||||
- 구간 = [P(i-1), P(i)] 한 번씩 `generate_initial_route` — 시종점을 옮기지 않아 연도 경계가 맞물림
|
||||
- 갈래 넷(`PLAN_MODES` · follow 뺌 — 두 점이면 직선) · 연도마다 고름
|
||||
- 추천 기준 = 프로젝트 임도 종류 · 설계속도 → B 법정 함수 · 지형 구분은 B 폴백(`TERRAIN_TYPE`)
|
||||
- 계류 이격 = 도엽 하천선(B 로더 그대로) · 복도 반폭 = `PLAN_CORRIDOR_M`(D 만)
|
||||
- 연도 경계는 꺾인 채(경계점 = 측점) · 누가거리 = 등록 시작 누가부터 고른 갈래 길이로 이어 셈
|
||||
- 파생 저장본 = `D03_PlanAnalysis/_derived/route_<연도>_<지점 열쇠>.json`(입력 지문 같으면 다시
|
||||
안 셈) · 끈 지점 분석이 저장 지점 분석본을 덮지 않음 · 연도마다 최근 `KEEP_PER_YEAR` 개 + 저장
|
||||
지점 것만 남김 · 옛 꼴 `route_<연도>.json` 은 읽기만(공유 storage — 옛 판 서버가 아직 읽음)
|
||||
|
||||
계산 자리 = ③ 서버 단독(B 추천과 같음).
|
||||
|
||||
⚠ 파이썬·TS 짝 — `pick_mode` · `chain` · `joined_line` · `with_edits` ↔
|
||||
`logic/D03_PlanAnalysis_Plan.ts` `pickMode` · `chainRows` · `joinedLine` · `withEdits`(화면이 고르는
|
||||
즉시) · 거울 시험 `test_530_3_plan_logic.py` · `test_543_plan_edit.py`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import math
|
||||
import multiprocessing
|
||||
from concurrent.futures import ProcessPoolExecutor, as_completed
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
import numpy as np
|
||||
import shapely
|
||||
from shapely.geometry import LineString, MultiLineString
|
||||
|
||||
from B05_Profile.B05_Profile_Engine_Grade import (
|
||||
legal_grade_limit_pct,
|
||||
legal_plan_radius_min_m,
|
||||
resolve_design_speed,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial import (
|
||||
_POOLS,
|
||||
_TERRAINS,
|
||||
generate_initial_route,
|
||||
load_terrain_and_streams,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import request_fingerprint
|
||||
from common_util.common_util_json import atomic_write_json
|
||||
from common_util.common_util_surface_confirmation import SHEET_ANALYSIS_FILTER
|
||||
from config.config_system_terrain import PLAN_ANALYZE_WORKERS
|
||||
from D03_PlanAnalysis.D03_PlanAnalysis_Terrain import PLAN_CORRIDOR_M, models_dir_of
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
PLAN_MODES = ("gradeline", "earthwork", "gentle", "weighted")
|
||||
# 도엽 DEM(스무딩) — B 「수치지형도」 분석 지형과 같은 고르기
|
||||
SELECTION = {"source_filter": SHEET_ANALYSIS_FILTER, "method": "dtm", "smooth": True}
|
||||
# D 등록에 지형 구분 칸 없음 — B `_road_settings` 못 읽을 때 폴백 그대로(가장 완화)
|
||||
TERRAIN_TYPE = "special"
|
||||
ALGO = 1
|
||||
# 계류 거르기 칸 · 지형 상자 여유(m) — `near_streams`
|
||||
STREAM_TILE_M = 200.0
|
||||
STREAM_PAD_M = 100.0
|
||||
# 연도마다 남길 분석본 수 — 저장 지점 분석본은 따로 늘 남김
|
||||
KEEP_PER_YEAR = 3
|
||||
|
||||
Progress = Callable[[int, str], None]
|
||||
Loader = Callable[..., tuple[Any, list]]
|
||||
|
||||
|
||||
def derived_dir(project_root: Path) -> Path:
|
||||
return project_root / "D03_PlanAnalysis" / "_derived"
|
||||
|
||||
|
||||
def plan_criteria(road_type: str | None, design_speed_kph: int | None) -> dict[str, Any]:
|
||||
"""추천 기준 — 종단 상한(비율) · 최소 R · 설계속도(B 법정 함수)."""
|
||||
grade_class = str(road_type or "work")
|
||||
speed = resolve_design_speed(grade_class, design_speed_kph)
|
||||
return {
|
||||
"max_grade": legal_grade_limit_pct(grade_class, TERRAIN_TYPE, False, speed) / 100.0,
|
||||
"min_radius_m": legal_plan_radius_min_m(speed, TERRAIN_TYPE),
|
||||
"design_speed": speed,
|
||||
}
|
||||
|
||||
|
||||
def _point(value: Any) -> tuple[float, float]:
|
||||
"""엔진 출력 자리수(0.1 mm)로 — 노선 끝점 = 지점 그대로 · 연도 경계가 정확히 맞물림."""
|
||||
return (round(float(value[0]), 4), round(float(value[1]), 4))
|
||||
|
||||
|
||||
def _same_point(a: Any, b: Any) -> bool:
|
||||
return abs(float(a[0]) - float(b[0])) < 1e-3 and abs(float(a[1]) - float(b[1])) < 1e-3
|
||||
|
||||
|
||||
def segment_path(project_root: Path, seg: dict[str, Any]) -> Path:
|
||||
"""연도 · 지점 열쇠 파일 — 지점이 다르면 다른 파일(덮지 않음)."""
|
||||
key = hashlib.blake2b(
|
||||
repr((_point(seg["start"]), _point(seg["end"]))).encode(), digest_size=4
|
||||
).hexdigest()
|
||||
return derived_dir(project_root) / f"route_{int(seg['year'])}_{key}.json"
|
||||
|
||||
|
||||
def _find(project_root: Path, seg: dict[str, Any]) -> tuple[Path, dict[str, Any]] | None:
|
||||
"""지점이 지금과 같은 저장본(1 mm) — 열쇠 파일 · 같은 연도 다른 열쇠(새것부터) · 옛 꼴 차례."""
|
||||
year = int(seg["year"])
|
||||
folder = derived_dir(project_root)
|
||||
exact = segment_path(project_root, seg)
|
||||
others = sorted(
|
||||
(p for p in folder.glob(f"route_{year}_*.json") if p != exact),
|
||||
key=lambda p: p.stat().st_mtime,
|
||||
reverse=True,
|
||||
)
|
||||
for path in (exact, *others, folder / f"route_{year}.json"):
|
||||
try:
|
||||
stored = json.loads(path.read_text("utf-8"))
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
if _same_point(stored["start"], seg["start"]) and _same_point(stored["end"], seg["end"]):
|
||||
return path, stored
|
||||
return None
|
||||
|
||||
|
||||
def read_segment(project_root: Path, seg: dict[str, Any]) -> dict[str, Any] | None:
|
||||
"""저장본 — 지점이 지금과 같을 때만(입력 지문은 분석 때 다시 봄)."""
|
||||
found = _find(project_root, seg)
|
||||
return found[1] if found else None
|
||||
|
||||
|
||||
def _prune(project_root: Path, seg: dict[str, Any]) -> None:
|
||||
"""연도 분석본 — 최근 `KEEP_PER_YEAR` 개 + 저장 지점(`saved` · 없으면 seg) 것만 남김 · 옛 꼴은
|
||||
안 지움."""
|
||||
start, end = seg.get("saved") or (seg["start"], seg["end"])
|
||||
canon = _find(project_root, {"year": seg["year"], "start": start, "end": end})
|
||||
files = sorted(
|
||||
derived_dir(project_root).glob(f"route_{int(seg['year'])}_*.json"),
|
||||
key=lambda p: p.stat().st_mtime,
|
||||
reverse=True,
|
||||
)
|
||||
for path in files[KEEP_PER_YEAR:]:
|
||||
if canon is None or path != canon[0]:
|
||||
path.unlink(missing_ok=True)
|
||||
|
||||
|
||||
def _route_view(result: dict[str, Any]) -> dict[str, Any]:
|
||||
keep = ("planned", "nodes", "curves", "metrics", "relaxed", "terrain_bound_pct")
|
||||
return {key: result.get(key) for key in keep}
|
||||
|
||||
|
||||
def _engine_inputs(criteria: dict[str, Any]) -> dict[str, Any]:
|
||||
return {
|
||||
"weights": None,
|
||||
"max_grade": criteria["max_grade"],
|
||||
"min_radius_m": criteria["min_radius_m"],
|
||||
"stream_offset_m": None,
|
||||
"grade_smooth_m": None,
|
||||
"target_grade": None,
|
||||
"with_profile": False,
|
||||
"design_speed": criteria["design_speed"],
|
||||
}
|
||||
|
||||
|
||||
def segment_fingerprint(
|
||||
seg: dict[str, Any],
|
||||
terrain: Any,
|
||||
streams: list,
|
||||
criteria: dict[str, Any],
|
||||
corridor_m: float = PLAN_CORRIDOR_M,
|
||||
) -> str:
|
||||
"""구간 입력 지문 — 갈래 넷 B 저장 키 · 복도 · 판."""
|
||||
expected = [_point(seg["start"]), _point(seg["end"])]
|
||||
inputs = _engine_inputs(criteria)
|
||||
keys = [
|
||||
request_fingerprint(expected, terrain, streams, mode=mode, **inputs)[0]
|
||||
for mode in PLAN_MODES
|
||||
]
|
||||
return hashlib.blake2b(
|
||||
repr((keys, float(corridor_m), ALGO)).encode(), digest_size=16
|
||||
).hexdigest()
|
||||
|
||||
|
||||
def analyze_segment(
|
||||
project_root: Path,
|
||||
seg: dict[str, Any],
|
||||
terrain: Any,
|
||||
streams: list,
|
||||
criteria: dict[str, Any],
|
||||
corridor_m: float = PLAN_CORRIDOR_M,
|
||||
) -> dict[str, Any]:
|
||||
"""연도 구간 하나 · 갈래 넷 — 입력 지문이 저장본과 같으면 그대로(`stored` True)."""
|
||||
expected = [_point(seg["start"]), _point(seg["end"])]
|
||||
inputs = _engine_inputs(criteria)
|
||||
fingerprint = segment_fingerprint(seg, terrain, streams, criteria, corridor_m)
|
||||
stored = read_segment(project_root, seg)
|
||||
if stored is not None and stored.get("fingerprint") == fingerprint:
|
||||
return {**stored, "stored": True}
|
||||
routes: dict[str, Any] = {}
|
||||
for mode in PLAN_MODES:
|
||||
try:
|
||||
routes[mode] = _route_view(
|
||||
generate_initial_route(
|
||||
expected, terrain, streams, mode=mode, corridor_m=corridor_m, **inputs
|
||||
)
|
||||
)
|
||||
except ValueError as error: # NoRoute 포함 — 그 갈래만 빔
|
||||
routes[mode] = {"error": str(error)}
|
||||
result = {
|
||||
"year": int(seg["year"]),
|
||||
"start": list(expected[0]),
|
||||
"end": list(expected[1]),
|
||||
"fingerprint": fingerprint,
|
||||
"corridor_m": float(corridor_m),
|
||||
"criteria": {
|
||||
"max_grade_pct": round(criteria["max_grade"] * 100.0, 2),
|
||||
"min_radius_m": criteria["min_radius_m"],
|
||||
"design_speed_kph": criteria["design_speed"],
|
||||
"terrain_type": TERRAIN_TYPE,
|
||||
},
|
||||
"routes": routes,
|
||||
}
|
||||
atomic_write_json(segment_path(project_root, seg), result)
|
||||
_prune(project_root, seg)
|
||||
return {**result, "stored": False}
|
||||
|
||||
|
||||
def near_streams(terrain: Any, streams: list) -> list:
|
||||
"""지형 상자 안 어느 점에서든 가장 가까운 계류일 수 있는 선만 — 나머지는 엔진 값에 안 닿음.
|
||||
|
||||
칸마다 닿는 거리 = 칸 가운데 계류 거리 + 칸 반대각(거리는 1-립시츠) — 그 안에 드는 선만 남기면
|
||||
상자 안 점의 계류 거리 · 건너기 · 격자 거리가 통째와 같음(통째 거리 = 추천 시간의 큰 몫).
|
||||
"""
|
||||
if not streams:
|
||||
return streams
|
||||
x0, x1 = float(np.min(terrain.x)) - STREAM_PAD_M, float(np.max(terrain.x)) + STREAM_PAD_M
|
||||
y0, y1 = float(np.min(terrain.y)) - STREAM_PAD_M, float(np.max(terrain.y)) + STREAM_PAD_M
|
||||
xs = np.linspace(x0, x1, max(1, math.ceil((x1 - x0) / STREAM_TILE_M)) + 1)
|
||||
ys = np.linspace(y0, y1, max(1, math.ceil((y1 - y0) / STREAM_TILE_M)) + 1)
|
||||
lo_x, lo_y = np.meshgrid(xs[:-1], ys[:-1])
|
||||
hi_x, hi_y = np.meshgrid(xs[1:], ys[1:])
|
||||
tiles = shapely.box(lo_x.ravel(), lo_y.ravel(), hi_x.ravel(), hi_y.ravel())
|
||||
reach = shapely.distance(shapely.centroid(tiles), MultiLineString(streams)) + 0.5 * math.hypot(
|
||||
xs[1] - xs[0], ys[1] - ys[0]
|
||||
)
|
||||
lines = np.array([LineString(line) for line in streams], dtype=object)
|
||||
keep = (shapely.distance(lines[:, None], tiles[None, :]) <= reach[None, :]).any(axis=1)
|
||||
return [line for line, kept in zip(streams, keep) if kept]
|
||||
|
||||
|
||||
def _load(project_root: Path, year: int, loader: Loader) -> tuple[Any, list]:
|
||||
terrain, streams = loader(
|
||||
project_root,
|
||||
SELECTION,
|
||||
models_dir_of(project_root, year),
|
||||
derived_dir(project_root) / f"cost_{year}",
|
||||
)
|
||||
return terrain, near_streams(terrain, streams)
|
||||
|
||||
|
||||
def _segment_job(
|
||||
project_root: Path,
|
||||
seg: dict[str, Any],
|
||||
criteria: dict[str, Any],
|
||||
corridor_m: float,
|
||||
loader: Loader,
|
||||
) -> dict[str, Any]:
|
||||
"""연도 구간 하나 — 지형 읽기부터(제 프로세스면 끝날 때 비용면 · 묶음 메모리를 놓음)."""
|
||||
terrain, streams = _load(project_root, int(seg["year"]), loader)
|
||||
return analyze_segment(project_root, seg, terrain, streams, criteria, corridor_m)
|
||||
|
||||
|
||||
def _worker_job(*args: Any) -> dict[str, Any]:
|
||||
"""나란히 셈 프로세스 — 구간 하나 뒤 B 기억(비용면 · 묶음)을 비워 다음 구간 메모리를 놓음."""
|
||||
try:
|
||||
return _segment_job(*args)
|
||||
finally:
|
||||
_POOLS.clear()
|
||||
_TERRAINS.clear()
|
||||
|
||||
|
||||
def analyze_plan(
|
||||
project_root: Path,
|
||||
segments: list[dict[str, Any]],
|
||||
criteria: dict[str, Any],
|
||||
*,
|
||||
loader: Loader = load_terrain_and_streams,
|
||||
corridor_m: float = PLAN_CORRIDOR_M,
|
||||
on_progress: Progress | None = None,
|
||||
workers: int = PLAN_ANALYZE_WORKERS,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""연도 구간마다(연도 DEM · 연도 비용면 캐시) `analyze_segment`.
|
||||
|
||||
다시 셀 구간이 둘 이상이면 프로세스 `workers` 개로 나란히 — 구간끼리 입력 · 저장본이 따로라
|
||||
결과 같음. `workers` 1 = 이 프로세스(가짜 지형 시험)."""
|
||||
|
||||
def report(percent: int, message: str) -> None:
|
||||
if on_progress is not None:
|
||||
on_progress(percent, message)
|
||||
|
||||
results: list[dict[str, Any] | None] = [None] * len(segments)
|
||||
todo = []
|
||||
for i, seg in enumerate(segments):
|
||||
stored = read_segment(project_root, seg)
|
||||
if stored is not None: # 저장본이 있을 때만 이 프로세스에서 지형을 읽어 지문을 봄
|
||||
terrain, streams = _load(project_root, int(seg["year"]), loader)
|
||||
if stored.get("fingerprint") == segment_fingerprint(
|
||||
seg, terrain, streams, criteria, corridor_m
|
||||
):
|
||||
results[i] = {**stored, "stored": True}
|
||||
continue
|
||||
todo.append(i)
|
||||
|
||||
count = min(workers, len(todo))
|
||||
if count <= 1:
|
||||
for n, i in enumerate(todo):
|
||||
report(100 * n // len(segments), f"{int(segments[i]['year'])}년 구간 노선 추천 중")
|
||||
results[i] = _segment_job(project_root, segments[i], criteria, corridor_m, loader)
|
||||
else:
|
||||
years = "·".join(str(int(segments[i]["year"])) for i in todo)
|
||||
report(0, f"{years}년 구간 노선 추천 중(나란히 {count})")
|
||||
context = multiprocessing.get_context("spawn")
|
||||
with ProcessPoolExecutor(count, mp_context=context) as pool:
|
||||
futures = {
|
||||
pool.submit(_worker_job, project_root, segments[i], criteria, corridor_m, loader): i
|
||||
for i in todo
|
||||
}
|
||||
for n, future in enumerate(as_completed(futures), 1):
|
||||
i = futures[future]
|
||||
results[i] = future.result()
|
||||
report(
|
||||
100 * n // len(todo), f"{int(segments[i]['year'])}년 구간 끝 ({n}/{len(todo)})"
|
||||
)
|
||||
report(100, "분석 끝")
|
||||
return results # type: ignore[return-value]
|
||||
|
||||
|
||||
def pick_mode(result: dict[str, Any], wanted: str | None) -> str | None:
|
||||
"""고른 갈래 — 없거나 빈 갈래면 차례대로 첫 성공 갈래."""
|
||||
routes = result.get("routes") or {}
|
||||
if wanted in routes and "error" not in routes[wanted]:
|
||||
return wanted
|
||||
return next((m for m in PLAN_MODES if m in routes and "error" not in routes[m]), None)
|
||||
|
||||
|
||||
def chain(
|
||||
results: list[dict[str, Any]], choices: dict[int, str | None], start_chainage_m: float
|
||||
) -> list[dict[str, Any]]:
|
||||
"""연도별 고른 갈래 · 시작 · 끝 누가(이어 셈) — 앞 연도 끝 누가 = 다음 연도 시작 누가."""
|
||||
at = float(start_chainage_m or 0.0)
|
||||
rows = []
|
||||
for result in results:
|
||||
mode = pick_mode(result, choices.get(int(result["year"])))
|
||||
length = float(result["routes"][mode]["metrics"]["length_m"]) if mode else 0.0
|
||||
rows.append(
|
||||
{
|
||||
"year": int(result["year"]),
|
||||
"mode": mode,
|
||||
"length_m": round(length, 3),
|
||||
"start_chainage_m": round(at, 3),
|
||||
"end_chainage_m": round(at + length, 3),
|
||||
}
|
||||
)
|
||||
at += length
|
||||
return rows
|
||||
|
||||
|
||||
def line_length(line: list) -> float:
|
||||
"""선 길이(m) — 짝: `lineLength`(같은 차례로 더함)."""
|
||||
total = 0.0
|
||||
for a, b in zip(line, line[1:]):
|
||||
total += math.hypot(float(b[0]) - float(a[0]), float(b[1]) - float(a[1]))
|
||||
return total
|
||||
|
||||
|
||||
def with_edits(
|
||||
results: list[dict[str, Any]], edits: dict[int, dict[str, Any]]
|
||||
) -> list[dict[str, Any]]:
|
||||
"""고친 노선(`{mode, planned, metrics?}`)을 그 갈래 자리에 갈아 끼운 분석 사본 — 길이 = 고친 선
|
||||
길이. 그 갈래가 없거나 빈 갈래면 그대로."""
|
||||
out = []
|
||||
for result in results:
|
||||
edit = edits.get(int(result["year"]))
|
||||
route = (result.get("routes") or {}).get(edit["mode"]) if edit else None
|
||||
if not edit or route is None or "error" in route:
|
||||
out.append(result)
|
||||
continue
|
||||
metrics = {**(edit.get("metrics") or {}), "length_m": line_length(edit["planned"])}
|
||||
routes = {
|
||||
**result["routes"],
|
||||
edit["mode"]: {**route, "planned": edit["planned"], "metrics": metrics, "edited": True},
|
||||
}
|
||||
out.append({**result, "routes": routes})
|
||||
return out
|
||||
|
||||
|
||||
def joined_line(results: list[dict[str, Any]], rows: list[dict[str, Any]]) -> list[list[float]]:
|
||||
"""고른 갈래를 이은 중심선 — 경계점 한 번만(꺾인 채)."""
|
||||
line: list[list[float]] = []
|
||||
for result, row in zip(results, rows):
|
||||
if row["mode"] is None:
|
||||
continue
|
||||
planned = result["routes"][row["mode"]]["planned"]
|
||||
line.extend(planned[1:] if line and _same_point(line[-1], planned[0]) else planned)
|
||||
return line
|
||||
@@ -1,169 +0,0 @@
|
||||
"""D 예정노선 꺾임점 고침 [저장] 정본(요구 363-1 · PLAN 543-3).
|
||||
|
||||
- 받는 것 = 연도별 마디 · 곡선 켬끔 · R(화면이 그은 선은 안 받음)
|
||||
- 선 = B05 편집 갈래 `node_polyline`(Node `route.polyline` = 화면 `buildEditedPolyline` 같은 TS) ·
|
||||
못 돌림 = 파이썬 짝 `build_planned_polyline(simplify=False)`
|
||||
- 지표 = 엔진과 같은 `route_metrics` → `hard_limits` → `with_station_misses`(연도 지형 · 계류)
|
||||
- 처음 · 끝 마디 = 연도 경계 지점(1 mm) · 고친 갈래 · 분석 지문이 지금과 같아야 함
|
||||
|
||||
계산 자리 = ① 파이썬 · TS 짝 — 화면 `logic/D03_PlanAnalysis_Edit.ts`(조작 중 즉시) ↔
|
||||
여기([저장] 정본) · 거울 시험 `resources/tester/test_543_plan_edit.py`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial import (
|
||||
ROAD_WIDTH_M,
|
||||
STREAM_OFFSET_M,
|
||||
load_terrain_and_streams,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Metrics import route_metrics
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Pool import hard_limits
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Station import with_station_misses
|
||||
from B05_Profile.B05_Profile_Route_Polyline_Write import node_polyline
|
||||
from common_util.common_util_route_polyline import PlannedPolyline, build_planned_polyline
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze
|
||||
|
||||
# 끝 마디 = 지점 판정(m) — 짝: `D02_PlanBase_Router.SAME_M`
|
||||
SAME_M = 1e-3
|
||||
INPUT_KEYS = ("base_mode", "base_fingerprint", "nodes", "curve_flags", "radii")
|
||||
|
||||
|
||||
class EditError(ValueError):
|
||||
"""고침을 받을 수 없음 — 까닭 그대로 화면에."""
|
||||
|
||||
|
||||
def edited_line(edit: dict[str, Any], min_radius_m: float) -> PlannedPolyline:
|
||||
"""마디 · 켬끔 · R → 선(화면과 같은 편집 갈래)."""
|
||||
points = [(float(x), float(y)) for x, y in edit["nodes"]]
|
||||
flags = [bool(on) for on in edit["curve_flags"]]
|
||||
radii = [None if r is None else float(r) for r in edit["radii"]]
|
||||
built = node_polyline(points, min_radius_m, flags, radii, None, None)
|
||||
if built is not None:
|
||||
return built[0]
|
||||
return build_planned_polyline(
|
||||
points, min_radius_m=min_radius_m, simplify=False, curve_flags=flags, radii=radii
|
||||
)
|
||||
|
||||
|
||||
def check_edit(edit: dict[str, Any], result: dict[str, Any], mode: str) -> None:
|
||||
"""고친 갈래 · 분석 지문 · 칸 수 · 끝 마디 = 지점."""
|
||||
year = int(result["year"])
|
||||
if edit["base_mode"] != mode:
|
||||
raise EditError(f"{year}년 고침이 고른 갈래와 다릅니다.")
|
||||
if edit["base_fingerprint"] != result.get("fingerprint"):
|
||||
raise EditError(f"{year}년 분석이 바뀌어 고침이 낡았습니다 — 다시 고치세요.")
|
||||
nodes = edit["nodes"]
|
||||
if len(nodes) < 2 or not (len(nodes) == len(edit["curve_flags"]) == len(edit["radii"])):
|
||||
raise EditError(f"{year}년 고침 마디 꼴이 다릅니다.")
|
||||
if any(r is not None and not float(r) > 0 for r in edit["radii"]):
|
||||
raise EditError(f"{year}년 R 은 0 보다 커야 합니다.")
|
||||
for got, want in ((nodes[0], result["start"]), (nodes[-1], result["end"])):
|
||||
if max(abs(float(a) - float(b)) for a, b in zip(got, want)) > SAME_M:
|
||||
raise EditError(f"{year}년 처음 · 끝 마디가 연도 경계 지점과 다릅니다.")
|
||||
|
||||
|
||||
def edit_metrics(
|
||||
line: PlannedPolyline, result: dict[str, Any], terrain: Any, streams: list
|
||||
) -> dict[str, Any]:
|
||||
"""엔진과 같은 잼 — 원지반 기울기는 참고(못 지킨 수에서 뺌) · 측점 밖 곡선 더함."""
|
||||
criteria = result["criteria"]
|
||||
metrics = route_metrics(
|
||||
line.vertices,
|
||||
[tuple(result["start"]), tuple(result["end"])],
|
||||
terrain,
|
||||
streams,
|
||||
line.curves,
|
||||
max_grade=float(criteria["max_grade_pct"]) / 100.0,
|
||||
min_radius_m=float(criteria["min_radius_m"]),
|
||||
stream_offset_m=STREAM_OFFSET_M,
|
||||
road_width_m=ROAD_WIDTH_M,
|
||||
bound_pct=None,
|
||||
)
|
||||
return with_station_misses(hard_limits(metrics), line)
|
||||
|
||||
|
||||
def saved_edit(route_json: Any) -> dict[str, Any] | None:
|
||||
"""저장된 고침(route_json `edit` + 그때 선 · 지표) — 없으면 None."""
|
||||
try:
|
||||
value = json.loads(route_json) if isinstance(route_json, str) else route_json
|
||||
except ValueError:
|
||||
return None
|
||||
edit = value.get("edit") if isinstance(value, dict) else None
|
||||
if not isinstance(edit, dict) or not all(key in edit for key in INPUT_KEYS):
|
||||
return None
|
||||
return {
|
||||
**{key: edit[key] for key in INPUT_KEYS},
|
||||
"planned": value.get("planned"),
|
||||
"nodes_info": value.get("nodes"),
|
||||
"curves": value.get("curves"),
|
||||
"metrics": value.get("metrics"),
|
||||
}
|
||||
|
||||
|
||||
def same_inputs(a: dict[str, Any] | None, b: dict[str, Any] | None) -> bool:
|
||||
"""같은 고침 입력인가(1 mm) — 같으면 저장된 선 · 지표를 다시 씀."""
|
||||
if not a or not b:
|
||||
return False
|
||||
|
||||
def near(x: Any, y: Any) -> bool:
|
||||
return (
|
||||
x is None
|
||||
and y is None
|
||||
or (x is not None and y is not None and abs(float(x) - float(y)) < 1e-3)
|
||||
)
|
||||
|
||||
return (
|
||||
a["base_mode"] == b["base_mode"]
|
||||
and a["base_fingerprint"] == b["base_fingerprint"]
|
||||
and len(a["nodes"]) == len(b["nodes"])
|
||||
and all(near(p[0], q[0]) and near(p[1], q[1]) for p, q in zip(a["nodes"], b["nodes"]))
|
||||
and [bool(v) for v in a["curve_flags"]] == [bool(v) for v in b["curve_flags"]]
|
||||
and all(near(x, y) for x, y in zip(a["radii"], b["radii"]))
|
||||
)
|
||||
|
||||
|
||||
def build_edits(
|
||||
project_root: Path,
|
||||
results: list[dict[str, Any]],
|
||||
rows: list[dict[str, Any]],
|
||||
edits: dict[int, dict[str, Any]],
|
||||
saved: dict[int, dict[str, Any] | None],
|
||||
loader: analyze.Loader = load_terrain_and_streams,
|
||||
) -> dict[int, dict[str, Any]]:
|
||||
"""연도 → 고친 노선 `{mode, planned, nodes, curves, metrics, edit}` — 저장된 고침과 입력이
|
||||
같으면 그 선 · 지표 그대로(지형 안 읽음)."""
|
||||
out: dict[int, dict[str, Any]] = {}
|
||||
for result, row in zip(results, rows):
|
||||
year = int(result["year"])
|
||||
edit = edits.get(year)
|
||||
if edit is None:
|
||||
continue
|
||||
check_edit(edit, result, row["mode"])
|
||||
inputs = {key: edit[key] for key in INPUT_KEYS}
|
||||
old = saved.get(year)
|
||||
if old is not None and old.get("metrics") and same_inputs(inputs, old):
|
||||
out[year] = {
|
||||
"mode": row["mode"],
|
||||
"planned": old["planned"],
|
||||
"nodes": old["nodes_info"],
|
||||
"curves": old["curves"],
|
||||
"metrics": old["metrics"],
|
||||
"edit": inputs,
|
||||
}
|
||||
continue
|
||||
line = edited_line(inputs, float(result["criteria"]["min_radius_m"]))
|
||||
terrain, streams = analyze._load(project_root, year, loader)
|
||||
out[year] = {
|
||||
"mode": row["mode"],
|
||||
"planned": [[round(x, 4), round(y, 4)] for x, y in line.vertices],
|
||||
"nodes": [node.as_dict() for node in line.nodes],
|
||||
"curves": [curve.as_dict() for curve in line.curves],
|
||||
"metrics": edit_metrics(line, result, terrain, streams),
|
||||
"edit": inputs,
|
||||
}
|
||||
return out
|
||||
@@ -0,0 +1,406 @@
|
||||
"""D 예정노선 2단계 노선 정본 — 손으로 그은 직선 · 갈래 · 구역
|
||||
(요구 389 6 · 11 · 13 · 14 · 391 · PLAN 562-2).
|
||||
|
||||
- `GET /api/projects/{id}/plan/network` — 정본(`plan_networks`) + 서버가 다시 센 길이 ·
|
||||
측점 · 조각 · 저장 없으면 옛 연도 노선(`plan_route_years.route_json.planned`)을 한 선으로 바꿔
|
||||
보임(legacy)
|
||||
- `PUT …/plan/network` — [저장] · [초기화](빈 선) — 선 · 마크 · 조각 연도만 받아 서버가 정본으로
|
||||
다시 셈 → `plan_networks` · 조각마다 한 줄(`plan_route_pieces`) · 연도 행(`plan_route_years`) ·
|
||||
사업연도 끝
|
||||
|
||||
길이 · 측점 · 높이는 저장하지 않음(늘 다시 셈 · 높이는 지형에서). 측점 = 선마다 No.0 부터.
|
||||
조각 = 선의 마크 사이 구간(구역 × 갈래) · 업체 지정은 조각 단위(`plan_route_assignees.piece_id`).
|
||||
계산 자리 = ① 파이썬·TS 짝(`D03_PlanAnalysis_Draw`) — 화면은 그리는 즉시 · [저장]은 여기서 다시 셈.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
import aiomysql
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from common_util.common_util_auth import is_forest_service, require_company
|
||||
from config.config_db import get_db_pool
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Draw as draw
|
||||
|
||||
router = APIRouter(prefix="/api/projects", tags=["D03 Plan Network"])
|
||||
|
||||
ID = r"^[A-Za-z0-9_-]{1,64}$"
|
||||
SAME_M = 0.001 # 1 mm 안 = 같은 점 · 같은 마크
|
||||
PIECE_TOL_M = 0.01 # 조각 끝 = 마크 경계(1 cm)
|
||||
STATION_M = 20.0 # 측점 간격(지식DB 설계제원_총괄)
|
||||
|
||||
|
||||
class Branch(BaseModel):
|
||||
line: str = Field(pattern=ID)
|
||||
vertex: int = Field(ge=0)
|
||||
|
||||
|
||||
class Line(BaseModel):
|
||||
id: str = Field(pattern=ID)
|
||||
vertices: list[tuple[float, float]] = Field(min_length=1)
|
||||
branch: Branch | None = None
|
||||
|
||||
|
||||
class Mark(BaseModel):
|
||||
line: str = Field(pattern=ID)
|
||||
at_m: float
|
||||
|
||||
|
||||
class Piece(BaseModel):
|
||||
id: str = Field(pattern=ID)
|
||||
line: str = Field(pattern=ID)
|
||||
from_m: float
|
||||
to_m: float
|
||||
year: int
|
||||
|
||||
|
||||
class Network(BaseModel):
|
||||
epsg: int
|
||||
lines: list[Line] = []
|
||||
marks: list[Mark] = []
|
||||
pieces: list[Piece] = []
|
||||
|
||||
|
||||
class NetworkError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def _check(ok: bool, message: str) -> None:
|
||||
if not ok:
|
||||
raise NetworkError(message)
|
||||
|
||||
|
||||
def derive(network: dict[str, Any], year_from: int | None = None) -> dict[str, Any]:
|
||||
"""선 · 마크 · 조각 → 선 길이 · 측점 · 조각(경계 = 마크 · 꺾은선 · 측점) ·
|
||||
틀리면 `NetworkError`.
|
||||
|
||||
`year_from` = 사업연도 시작(그 앞 연도 조각 거절 · None = 안 봄)."""
|
||||
lines: dict[str, list] = {}
|
||||
for line in network["lines"]:
|
||||
vertices = line["vertices"]
|
||||
_check(line["id"] not in lines, f"선 id 가 겹칩니다: {line['id']}")
|
||||
_check(
|
||||
all(math.isfinite(v) for point in vertices for v in point),
|
||||
f"선 {line['id']} 점이 수가 아닙니다.",
|
||||
)
|
||||
branch = line.get("branch")
|
||||
if branch:
|
||||
parent = lines.get(branch["line"])
|
||||
_check(parent is not None, f"갈래 {line['id']} 의 부모 선이 앞에 없습니다.")
|
||||
_check(branch["vertex"] < len(parent), f"갈래 {line['id']} 의 부모 점이 없습니다.")
|
||||
_check(
|
||||
draw.segment_length(vertices[0], parent[branch["vertex"]]) <= SAME_M,
|
||||
f"갈래 {line['id']} 첫 점이 부모 선 점과 다릅니다.",
|
||||
)
|
||||
lines[line["id"]] = vertices
|
||||
lengths = {key: draw.line_length(vertices) for key, vertices in lines.items()}
|
||||
cuts: dict[str, list[float]] = {key: [] for key in lines}
|
||||
for mark in network["marks"]:
|
||||
_check(mark["line"] in lines, f"마크의 선이 없습니다: {mark['line']}")
|
||||
at = mark["at_m"]
|
||||
_check(
|
||||
math.isfinite(at) and SAME_M < at < lengths[mark["line"]] - SAME_M,
|
||||
f"마크가 선 {mark['line']} 밖입니다: {at}",
|
||||
)
|
||||
cuts[mark["line"]].append(at)
|
||||
tiles: dict[str, list[tuple[float, float]]] = {}
|
||||
for key, marks in cuts.items():
|
||||
edges = [0.0, *sorted(marks), lengths[key]]
|
||||
_check(
|
||||
all(b - a > SAME_M for a, b in zip(edges[1:-1], edges[2:-1])),
|
||||
f"선 {key} 마크가 겹칩니다.",
|
||||
)
|
||||
tiles[key] = list(zip(edges, edges[1:])) if lengths[key] > SAME_M else []
|
||||
order = {key: i for i, key in enumerate(lines)}
|
||||
stations = {key: draw.station_chainages(lengths[key], STATION_M) for key in lines}
|
||||
used: set[tuple[str, int]] = set()
|
||||
pieces = []
|
||||
for piece in network["pieces"]:
|
||||
key = piece["line"]
|
||||
_check(key in lines, f"조각 {piece['id']} 의 선이 없습니다.")
|
||||
_check(year_from is None or piece["year"] >= year_from, f"조각 연도가 사업연도 앞: {piece}")
|
||||
hit = next(
|
||||
(
|
||||
i
|
||||
for i, (a, b) in enumerate(tiles[key])
|
||||
if abs(a - piece["from_m"]) <= PIECE_TOL_M and abs(b - piece["to_m"]) <= PIECE_TOL_M
|
||||
),
|
||||
None,
|
||||
)
|
||||
_check(hit is not None, f"조각 {piece['id']} 끝이 마크 경계와 다릅니다.")
|
||||
_check((key, hit) not in used, f"조각이 겹칩니다: {piece['id']}")
|
||||
used.add((key, hit))
|
||||
a, b = tiles[key][hit]
|
||||
pieces.append(
|
||||
{
|
||||
"id": piece["id"],
|
||||
"line": key,
|
||||
"year": int(piece["year"]),
|
||||
"from_m": a,
|
||||
"to_m": b,
|
||||
"length_m": b - a,
|
||||
"planned": [list(p) for p in draw.slice_line(lines[key], a, b)],
|
||||
"stations": sorted(
|
||||
{a, b, *(c for c in stations[key] if a - 1e-6 <= c <= b + 1e-6)}
|
||||
),
|
||||
}
|
||||
)
|
||||
_check(len({p["id"] for p in pieces}) == len(pieces), "조각 id 가 겹칩니다.")
|
||||
pieces.sort(key=lambda p: (order[p["line"]], p["from_m"]))
|
||||
return {
|
||||
"lines": [
|
||||
{"id": key, "length_m": lengths[key], "stations": stations[key]} for key in lines
|
||||
],
|
||||
"pieces": pieces,
|
||||
}
|
||||
|
||||
|
||||
def year_rows(pieces: list[dict[str, Any]]) -> dict[int, dict[str, Any]]:
|
||||
"""조각 → 연도 행(`plan_route_years`) — `planned` = 첫 조각(옛 읽는 곳 그대로) ·
|
||||
`pieces` = 그 연도 전부."""
|
||||
by_year: dict[int, list[dict[str, Any]]] = {}
|
||||
for piece in pieces:
|
||||
by_year.setdefault(piece["year"], []).append(piece)
|
||||
out = {}
|
||||
for year, items in sorted(by_year.items()):
|
||||
first = items[0]
|
||||
parts = [
|
||||
{
|
||||
"id": p["id"],
|
||||
"line": p["line"],
|
||||
"planned": p["planned"],
|
||||
"start_chainage_m": p["from_m"],
|
||||
"end_chainage_m": p["to_m"],
|
||||
"length_m": p["length_m"],
|
||||
}
|
||||
for p in items
|
||||
]
|
||||
route = {
|
||||
"mode": "manual",
|
||||
"planned": first["planned"],
|
||||
"start_chainage_m": first["from_m"],
|
||||
"end_chainage_m": first["to_m"],
|
||||
"length_m": sum(p["length_m"] for p in items),
|
||||
"pieces": parts,
|
||||
}
|
||||
out[year] = {
|
||||
"route": route,
|
||||
"fingerprint": _sha1(parts),
|
||||
"start": first["planned"][0],
|
||||
"end": first["planned"][-1],
|
||||
}
|
||||
return out
|
||||
|
||||
|
||||
def _sha1(value: Any) -> str:
|
||||
return hashlib.sha1(json.dumps(value, sort_keys=True).encode()).hexdigest() # noqa: S324
|
||||
|
||||
|
||||
def _json(value: Any) -> Any:
|
||||
if isinstance(value, (str, bytes)):
|
||||
try:
|
||||
return json.loads(value)
|
||||
except ValueError:
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def legacy_network(rows: list[dict[str, Any]]) -> dict[str, Any] | None:
|
||||
"""옛 연도 노선(연도 차례 · 직전 종점 = 다음 시점) → 본선 하나 + 연도 경계 마크 + 연도 조각."""
|
||||
vertices: list[list[float]] = []
|
||||
pieces: list[dict[str, Any]] = []
|
||||
epsg = None
|
||||
for row in rows:
|
||||
route = _json(row.get("route_json"))
|
||||
planned = route.get("planned") if isinstance(route, dict) else None
|
||||
if not planned or len(planned) < 2:
|
||||
continue
|
||||
epsg = epsg or row.get("epsg")
|
||||
points = [[float(p[0]), float(p[1])] for p in planned]
|
||||
if vertices and draw.segment_length(vertices[-1], points[0]) <= SAME_M:
|
||||
points = points[1:]
|
||||
start = draw.line_length(vertices)
|
||||
vertices.extend(points)
|
||||
end = draw.line_length(vertices)
|
||||
pieces.append(
|
||||
{
|
||||
"id": f"y{row['year']}",
|
||||
"line": "main",
|
||||
"from_m": start,
|
||||
"to_m": end,
|
||||
"year": row["year"],
|
||||
}
|
||||
)
|
||||
if not pieces:
|
||||
return None
|
||||
return {
|
||||
"epsg": int(epsg) if epsg is not None else None,
|
||||
"lines": [{"id": "main", "vertices": vertices, "branch": None}],
|
||||
"marks": [{"line": "main", "at_m": p["to_m"]} for p in pieces[:-1]],
|
||||
"pieces": pieces,
|
||||
}
|
||||
|
||||
|
||||
async def _load(connection: Any, project_id: UUID) -> tuple[dict[str, Any] | None, bool]:
|
||||
"""(정본 · 옛 연도 노선을 바꾼 것인가) — 둘 다 없으면 (None, False)."""
|
||||
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
||||
await cursor.execute(
|
||||
"SELECT network_json FROM plan_networks WHERE project_id = %s", (str(project_id),)
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if row:
|
||||
return _json(row["network_json"]), False
|
||||
await cursor.execute(
|
||||
"SELECT year, epsg, route_json FROM plan_route_years"
|
||||
" WHERE project_id = %s ORDER BY year",
|
||||
(str(project_id),),
|
||||
)
|
||||
legacy = legacy_network(list(await cursor.fetchall()))
|
||||
return legacy, legacy is not None
|
||||
|
||||
|
||||
def _view(network: dict[str, Any] | None, legacy: bool) -> dict[str, Any]:
|
||||
return {
|
||||
"status": "success",
|
||||
"network": network,
|
||||
"derived": derive(network) if network else None,
|
||||
"legacy": legacy,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/{project_id}/plan/network", response_model=None)
|
||||
async def get_plan_network(project_id: UUID) -> dict[str, Any]:
|
||||
async with get_db_pool().acquire() as connection:
|
||||
network, legacy = await _load(connection, project_id)
|
||||
try:
|
||||
return _view(network, legacy)
|
||||
except NetworkError as error:
|
||||
raise HTTPException(status_code=409, detail=f"저장된 노선이 틀립니다 — {error}") from error
|
||||
|
||||
|
||||
async def _store(
|
||||
connection: Any, project_id: UUID, network: dict[str, Any], derived: dict[str, Any]
|
||||
) -> None:
|
||||
pid = str(project_id)
|
||||
epsg = network["epsg"]
|
||||
years = year_rows(derived["pieces"])
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"SELECT DISTINCT piece_id FROM plan_route_assignees"
|
||||
" WHERE project_id = %s AND piece_id IS NOT NULL",
|
||||
(pid,),
|
||||
)
|
||||
assigned = {r[0] for r in await cursor.fetchall()}
|
||||
await cursor.execute(
|
||||
"SELECT DISTINCT year FROM plan_route_assignees"
|
||||
" WHERE project_id = %s AND piece_id IS NULL",
|
||||
(pid,),
|
||||
)
|
||||
assigned_years = {int(r[0]) for r in await cursor.fetchall()}
|
||||
lost = sorted(assigned - {p["id"] for p in derived["pieces"]})
|
||||
lost_years = sorted(assigned_years - set(years))
|
||||
if lost or lost_years:
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail=f"업체가 지정된 조각 · 연도를 지울 수 없습니다: {lost + lost_years}",
|
||||
)
|
||||
await cursor.execute(
|
||||
"INSERT INTO plan_networks (project_id, network_json) VALUES (%s, %s)"
|
||||
" ON DUPLICATE KEY UPDATE network_json = VALUES(network_json)",
|
||||
(pid, json.dumps(network, ensure_ascii=False)),
|
||||
)
|
||||
await cursor.execute("DELETE FROM plan_route_pieces WHERE project_id = %s", (pid,))
|
||||
for p in derived["pieces"]:
|
||||
route = {"planned": p["planned"], "stations": p["stations"]}
|
||||
await cursor.execute(
|
||||
"INSERT INTO plan_route_pieces (project_id, piece_id, line_id, year, from_m, to_m,"
|
||||
" length_m, epsg, route_json, fingerprint)"
|
||||
" VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
|
||||
(
|
||||
pid,
|
||||
p["id"],
|
||||
p["line"],
|
||||
p["year"],
|
||||
p["from_m"],
|
||||
p["to_m"],
|
||||
p["length_m"],
|
||||
epsg,
|
||||
json.dumps(route),
|
||||
_sha1(route),
|
||||
),
|
||||
)
|
||||
for year, row in years.items():
|
||||
# 내용이 바뀐 연도만 확정 풂(confirmed_at 을 fingerprint 보다 먼저 셈)
|
||||
await cursor.execute(
|
||||
"INSERT INTO plan_route_years (project_id, year, start_x, start_y, end_x, end_y,"
|
||||
" epsg, route_json, fingerprint) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)"
|
||||
" ON DUPLICATE KEY UPDATE start_x = VALUES(start_x), start_y = VALUES(start_y),"
|
||||
" end_x = VALUES(end_x), end_y = VALUES(end_y), epsg = VALUES(epsg),"
|
||||
" route_json = VALUES(route_json),"
|
||||
" confirmed_at = IF(fingerprint <=> VALUES(fingerprint), confirmed_at, NULL),"
|
||||
" fingerprint = VALUES(fingerprint)",
|
||||
(
|
||||
pid,
|
||||
year,
|
||||
*row["start"],
|
||||
*row["end"],
|
||||
epsg,
|
||||
json.dumps(row["route"], ensure_ascii=False),
|
||||
row["fingerprint"],
|
||||
),
|
||||
)
|
||||
if years:
|
||||
marks = ", ".join(["%s"] * len(years))
|
||||
await cursor.execute(
|
||||
"UPDATE plan_route_years SET route_json = NULL, fingerprint = NULL,"
|
||||
f" confirmed_at = NULL WHERE project_id = %s AND year NOT IN ({marks})",
|
||||
(pid, *years),
|
||||
)
|
||||
await cursor.execute(
|
||||
"UPDATE projects SET project_year_end = %s WHERE id = %s", (max(years), pid)
|
||||
)
|
||||
else:
|
||||
await cursor.execute(
|
||||
"UPDATE plan_route_years SET route_json = NULL, fingerprint = NULL,"
|
||||
" confirmed_at = NULL WHERE project_id = %s",
|
||||
(pid,),
|
||||
)
|
||||
|
||||
|
||||
@router.put("/{project_id}/plan/network", response_model=None)
|
||||
async def put_plan_network(
|
||||
project_id: UUID,
|
||||
payload: Network,
|
||||
session: dict[str, Any] = Depends(require_company),
|
||||
) -> dict[str, Any]:
|
||||
"""[저장] · [초기화](lines 빈 것) — 서버가 정본으로 다시 셈 → 정본 · 조각 행 · 연도 행."""
|
||||
if not is_forest_service(session):
|
||||
raise HTTPException(status_code=403, detail="예정노선은 산림청만 고칠 수 있습니다.")
|
||||
network = payload.model_dump()
|
||||
async with get_db_pool().acquire() as connection:
|
||||
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
||||
await cursor.execute(
|
||||
"SELECT project_year FROM projects WHERE id = %s", (str(project_id),)
|
||||
)
|
||||
project = await cursor.fetchone()
|
||||
if not project:
|
||||
raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.")
|
||||
try:
|
||||
derived = derive(network, project["project_year"])
|
||||
except NetworkError as error:
|
||||
raise HTTPException(status_code=400, detail=str(error)) from error
|
||||
await connection.begin()
|
||||
try:
|
||||
await _store(connection, project_id, network, derived)
|
||||
await connection.commit()
|
||||
except Exception:
|
||||
await connection.rollback()
|
||||
raise
|
||||
return {"status": "success", "network": network, "derived": derived, "legacy": False}
|
||||
@@ -1,40 +1,29 @@
|
||||
"""D 예정노선 2단계 분석 API (요구 345 · PLAN 530-1 · 530-2).
|
||||
"""D 예정노선 2단계 지형 API (요구 345 · PLAN 530-1).
|
||||
|
||||
- `POST /api/projects/{id}/plan/terrain` — 지형 받기(뒤에서) · 끝나면 연도별 `surface_models` 행
|
||||
- `POST …/plan/analyze` — 연도 구간마다 노선 추천 갈래 넷(뒤에서 · 파생 저장본)
|
||||
- `GET …/plan/progress` — 두 일의 진행률
|
||||
- `GET …/plan/analysis` — 저장본 + 고른 갈래 · 이은 누가 · 이은 중심선
|
||||
- `PUT …/plan/analysis` — [저장] 연도별 고른 갈래(+ 꺾임점 고침) → `plan_route_years.route_json` ·
|
||||
`fingerprint`
|
||||
- `GET …/plan/progress` — 일의 진행률
|
||||
|
||||
지점 정본은 `plan_route_years` 시점 · 종점 칸(529-3 몫). D03 화면에서 끈 지점(캐시 · PLAN 530-3)은
|
||||
`points`(작업 좌표계 m)로 받아 그 지점으로 받기 · 분석 · 보기 · [저장] 때만 D02 규칙으로 정본에 씀.
|
||||
계산 자리 = ③ 서버 단독 — [저장]은 노선을 받지 않고 갈래 이름만 받아 서버 저장본으로 다시 셈.
|
||||
꺾임점 고침(543)은 마디 · 켬끔 · R 만 받아 서버가 선 · 지표를 다시 셈(`D03_PlanAnalysis_Edit`) ·
|
||||
GET 은 분석 지문이 같은 저장 고침만 돌려줌.
|
||||
`points`(작업 좌표계 m)로 받아 그 지점으로 받음. 노선 정본(손 그은 선)은 `D03_PlanAnalysis_Network`
|
||||
(PLAN 562-2 · 옛 추천 엔진 · 꺾임점 R 걷음).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
import aiomysql
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from pydantic import BaseModel
|
||||
|
||||
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
||||
from common_util.common_util_auth import is_forest_service, require_company
|
||||
from common_util.common_util_crs import tm_to_lonlat
|
||||
from common_util.common_util_storage import resolve_stored_project_path
|
||||
from config.config_db import get_db_pool
|
||||
from D02_PlanBase import D02_PlanBase_Router as base
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Edit as plan_edit
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -58,21 +47,6 @@ class PointsBody(BaseModel):
|
||||
points: list[PlanPoint] | None = None
|
||||
|
||||
|
||||
class PlanEditBody(BaseModel):
|
||||
"""연도 하나의 꺾임점 고침 — 고친 갈래 · 그 분석 지문 · 마디(작업 좌표계 m) · 곡선 켬끔 · R."""
|
||||
|
||||
base_mode: str
|
||||
base_fingerprint: str
|
||||
nodes: list[tuple[float, float]]
|
||||
curve_flags: list[bool]
|
||||
radii: list[float | None]
|
||||
|
||||
|
||||
class ChoiceRequest(PointsBody):
|
||||
choices: dict[int, str] # 연도 → 갈래
|
||||
edits: dict[int, PlanEditBody] | None = None # 연도 → 꺾임점 고침(없는 연도 = 엔진 노선)
|
||||
|
||||
|
||||
def _override(segments: list[dict[str, Any]], points: list[PlanPoint]) -> list[dict[str, Any]]:
|
||||
"""연도 구간 지점을 끈 지점으로 — 연도 같음 · 직전 종점 = 다음 시점(1 mm) 아니면 거절."""
|
||||
by_year = {p.year: p for p in points}
|
||||
@@ -205,23 +179,6 @@ async def plan_terrain(project_id: UUID, body: PointsBody | None = None) -> dict
|
||||
return _start_job(project_id, "terrain", work)
|
||||
|
||||
|
||||
@router.post("/{project_id}/plan/analyze", response_model=None)
|
||||
async def plan_analyze(project_id: UUID, body: PointsBody | None = None) -> dict[str, Any]:
|
||||
"""연도 구간마다 노선 추천 갈래 넷(뒤에서 · 같은 입력이면 저장본)."""
|
||||
root, project, segments = await _plan_inputs(project_id, body.points if body else None)
|
||||
criteria = analyze.plan_criteria(project["road_type"], project["design_speed_kph"])
|
||||
missing = [s["year"] for s in segments if not terrain.models_dir_of(root, s["year"]).is_dir()]
|
||||
if missing:
|
||||
raise HTTPException(status_code=409, detail=f"지형을 먼저 받으세요: {missing}년")
|
||||
|
||||
async def work(report: Any) -> None:
|
||||
await asyncio.to_thread(
|
||||
lambda: analyze.analyze_plan(root, segments, criteria, on_progress=report)
|
||||
)
|
||||
|
||||
return _start_job(project_id, "analyze", work)
|
||||
|
||||
|
||||
@router.get("/{project_id}/plan/progress", response_model=None)
|
||||
async def plan_progress(project_id: UUID) -> dict[str, Any]:
|
||||
idle = {"running": False, "percent": 0, "message": None, "error": None}
|
||||
@@ -229,187 +186,3 @@ async def plan_progress(project_id: UUID) -> dict[str, Any]:
|
||||
"status": "success",
|
||||
**{kind: dict(_JOBS.get((str(project_id), kind)) or idle) for kind in KINDS},
|
||||
}
|
||||
|
||||
|
||||
def _saved_mode(route_json: Any) -> str | None:
|
||||
try:
|
||||
value = json.loads(route_json) if isinstance(route_json, str) else route_json
|
||||
except ValueError:
|
||||
return None
|
||||
return value.get("mode") if isinstance(value, dict) else None
|
||||
|
||||
|
||||
def _assemble(
|
||||
root: Path,
|
||||
project: dict[str, Any],
|
||||
segments: list[dict[str, Any]],
|
||||
choices: dict[int, Any],
|
||||
edits: dict[int, dict[str, Any]] | None = None,
|
||||
) -> tuple[list[dict[str, Any] | None], list[dict[str, Any]] | None, list[list[float]]]:
|
||||
"""(연도 저장본 · 이은 누가 · 이은 중심선) — 저장본이 하나라도 없으면 None · 빈 선.
|
||||
|
||||
`edits` = 연도 → 고친 노선(`{mode, planned, metrics}`) — 누가 · 선만 고친 길이로(저장본은 엔진
|
||||
그대로).
|
||||
"""
|
||||
results = [analyze.read_segment(root, seg) for seg in segments]
|
||||
if any(result is None for result in results):
|
||||
return results, None, []
|
||||
view = analyze.with_edits(results, edits or {}) # type: ignore[arg-type]
|
||||
rows = analyze.chain(view, choices, float(project["route_start_m"] or 0.0))
|
||||
return results, rows, analyze.joined_line(view, rows)
|
||||
|
||||
|
||||
def _saved_edits(
|
||||
root: Path, segments: list[dict[str, Any]], choices: dict[int, Any]
|
||||
) -> dict[int, dict[str, Any]]:
|
||||
"""저장된 꺾임점 고침 — 저장 갈래 그대로 · 분석 지문이 지금과 같은 것만."""
|
||||
out = {}
|
||||
for seg in segments:
|
||||
saved = plan_edit.saved_edit(seg["route_json"])
|
||||
result = analyze.read_segment(root, seg)
|
||||
if (
|
||||
saved
|
||||
and result
|
||||
and saved["base_mode"] == choices.get(seg["year"])
|
||||
and saved["base_fingerprint"] == result.get("fingerprint")
|
||||
and saved.get("planned")
|
||||
):
|
||||
out[seg["year"]] = saved
|
||||
return out
|
||||
|
||||
|
||||
def _edit_view(saved: dict[str, Any]) -> dict[str, Any]:
|
||||
return {key: saved[key] for key in (*plan_edit.INPUT_KEYS, "metrics")}
|
||||
|
||||
|
||||
def _query_points(points: str | None) -> list[PlanPoint] | None:
|
||||
if not points:
|
||||
return None
|
||||
try:
|
||||
return [PlanPoint(**item) for item in json.loads(points)]
|
||||
except (ValueError, TypeError) as error:
|
||||
raise HTTPException(status_code=400, detail="points 꼴이 다릅니다.") from error
|
||||
|
||||
|
||||
@router.get("/{project_id}/plan/analysis", response_model=None)
|
||||
async def plan_analysis(project_id: UUID, points: str | None = None) -> dict[str, Any]:
|
||||
"""분석 저장본(지점이 지금과 같은 것만) + 고른 갈래(저장값 · 없으면 첫 성공 갈래).
|
||||
|
||||
`points` = 끈 지점 JSON(`[{year, start, end}]` · 작업 좌표계 m) — 그 지점의 저장본을 봄.
|
||||
"""
|
||||
root, project, segments = await _plan_inputs(project_id, _query_points(points))
|
||||
choices = {seg["year"]: _saved_mode(seg["route_json"]) for seg in segments}
|
||||
saved = await asyncio.to_thread(_saved_edits, root, segments, choices)
|
||||
edits = {year: {"mode": e["base_mode"], **e} for year, e in saved.items()}
|
||||
results, rows, line = await asyncio.to_thread(
|
||||
_assemble, root, project, segments, choices, edits
|
||||
)
|
||||
return {
|
||||
"status": "success",
|
||||
"years": [
|
||||
{
|
||||
"year": seg["year"],
|
||||
"start": seg["start"],
|
||||
"end": seg["end"],
|
||||
"analysis": result,
|
||||
# 저장된 꺾임점 고침(입력 + 그때 지표) — 분석 지문이 같을 때만
|
||||
"edit": _edit_view(saved[seg["year"]]) if seg["year"] in saved else None,
|
||||
}
|
||||
for seg, result in zip(segments, results)
|
||||
],
|
||||
"chain": rows,
|
||||
"line": line,
|
||||
"start_chainage_m": float(project["route_start_m"] or 0.0),
|
||||
# 등록 목표 끝 누가(요구서 3-1) — 화면이 이은 끝 누가와 차이를 보임
|
||||
"end_chainage_target_m": (
|
||||
float(project["route_end_m"]) if project["route_end_m"] is not None else None
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
async def _store_points(project_id: UUID, segments: list[dict[str, Any]]) -> None:
|
||||
"""끈 지점 → D02 정본 규칙(경위도 · 잇기 · 바뀐 연도만 · 그 연도 노선 비움)."""
|
||||
epsg = segments[0]["epsg"]
|
||||
|
||||
def lonlat(xy: tuple[float, float]) -> base.LonLat:
|
||||
lon, lat = tm_to_lonlat(xy[0], xy[1], epsg)
|
||||
return base.LonLat(lon=lon, lat=lat)
|
||||
|
||||
project, old = await base._load(project_id)
|
||||
start = int(project["project_year"])
|
||||
end = int(project["project_year_end"] or start)
|
||||
items = [
|
||||
base.YearPoints(year=s["year"], start=lonlat(s["start"]), end=lonlat(s["end"]))
|
||||
for s in segments
|
||||
]
|
||||
new_epsg, rows = base.with_tm(base.chain_years(start, end, items))
|
||||
if new_epsg != epsg:
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail="끈 지점이 다른 좌표계 띠로 넘어갔습니다 — 1단계에서 지정하세요.",
|
||||
)
|
||||
writes = [row for row in rows if base.changed(row, old.get(row["year"]), epsg)]
|
||||
if writes:
|
||||
await base._store(project_id, epsg, writes, [])
|
||||
|
||||
|
||||
@router.put("/{project_id}/plan/analysis", response_model=None)
|
||||
async def save_plan_analysis(
|
||||
project_id: UUID,
|
||||
request: ChoiceRequest,
|
||||
session: dict[str, Any] = Depends(require_company),
|
||||
) -> dict[str, Any]:
|
||||
"""[저장] — 끈 지점(있으면 정본으로) · 연도별 고른 갈래를 서버 저장본으로 다시 이어 씀."""
|
||||
if (request.points or request.edits) and not is_forest_service(session):
|
||||
raise HTTPException(status_code=403, detail="예정노선 지점은 산림청만 고칠 수 있습니다.")
|
||||
root, project, segments = await _plan_inputs(project_id, request.points)
|
||||
bad = [m for m in request.choices.values() if m not in analyze.PLAN_MODES]
|
||||
if bad:
|
||||
raise HTTPException(status_code=400, detail=f"모르는 갈래: {bad}")
|
||||
choices = {seg["year"]: _saved_mode(seg["route_json"]) for seg in segments}
|
||||
choices.update(request.choices)
|
||||
results, rows, _ = await asyncio.to_thread(_assemble, root, project, segments, choices)
|
||||
if rows is None:
|
||||
raise HTTPException(
|
||||
status_code=409, detail="분석 저장본이 없습니다 — [분석]을 먼저 하세요."
|
||||
)
|
||||
lost = [r["year"] for r in rows if choices.get(r["year"]) not in (None, r["mode"])]
|
||||
if lost:
|
||||
raise HTTPException(status_code=409, detail=f"고른 갈래에 노선이 없습니다: {lost}년")
|
||||
edited: dict[int, dict[str, Any]] = {}
|
||||
if request.edits:
|
||||
saved = {seg["year"]: plan_edit.saved_edit(seg["route_json"]) for seg in segments}
|
||||
asked = {year: body.model_dump() for year, body in request.edits.items()}
|
||||
try:
|
||||
edited = await asyncio.to_thread(
|
||||
plan_edit.build_edits, root, results, rows, asked, saved
|
||||
)
|
||||
except plan_edit.EditError as error:
|
||||
raise HTTPException(status_code=409, detail=str(error)) from error
|
||||
_, rows, _ = await asyncio.to_thread(_assemble, root, project, segments, choices, edited)
|
||||
if request.points:
|
||||
await _store_points(project_id, segments)
|
||||
async with get_db_pool().acquire() as connection:
|
||||
await connection.begin()
|
||||
try:
|
||||
async with connection.cursor() as cursor:
|
||||
for result, row in zip(results, rows):
|
||||
route = edited.get(row["year"]) or result["routes"][row["mode"]]
|
||||
body = {**row, **{k: route[k] for k in ("planned", "nodes", "curves")}}
|
||||
if row["year"] in edited: # 꺾임점 고침 — 입력 · 지표 함께(GET 이 돌려줌)
|
||||
body.update(metrics=route["metrics"], edit=route["edit"])
|
||||
await cursor.execute(
|
||||
"UPDATE plan_route_years SET route_json = %s, fingerprint = %s "
|
||||
"WHERE project_id = %s AND year = %s",
|
||||
(
|
||||
json.dumps(body, ensure_ascii=False),
|
||||
result["fingerprint"],
|
||||
str(project_id),
|
||||
row["year"],
|
||||
),
|
||||
)
|
||||
await connection.commit()
|
||||
except Exception:
|
||||
await connection.rollback()
|
||||
raise
|
||||
return {"status": "success", "chain": rows}
|
||||
|
||||
@@ -149,9 +149,6 @@ SHEET_SURFACE_METHODS = [
|
||||
).split(",")
|
||||
if method.strip()
|
||||
]
|
||||
# D 예정노선 [분석] 연도 구간을 나란히 셀 프로세스 수 상한(프로세스 하나 약 0.8 GB) — 여럿이 동시에
|
||||
# 돌리면 줄임.
|
||||
PLAN_ANALYZE_WORKERS = int(os.getenv("PLAN_ANALYZE_WORKERS", "3"))
|
||||
# 확정에 쓸 기본 방식 — 라플라스 (2026-08-30 사용자 확정).
|
||||
SHEET_SURFACE_DEFAULT_METHOD = os.getenv("SHEET_SURFACE_DEFAULT_METHOD", "laplace")
|
||||
# 자동 전처리가 만드는 도엽 방식 — 기본 하나뿐이다. 여섯 방식을 매번 다 만들면
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
-- 021_plan_network.sql
|
||||
-- 예정노선 2단계 노선 정본 — 손으로 그은 선 · 갈래 · 구역 마크 · 조각(구역 × 갈래) (요구 389 · 391 · PLAN 562-2)
|
||||
--
|
||||
-- 추가만 함 — plan_route_years 는 조각에서 만든 연도 행으로 그대로 둠(아래 단계가 읽음).
|
||||
|
||||
USE aislo_db;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS plan_networks (
|
||||
project_id CHAR(36) NOT NULL PRIMARY KEY COMMENT 'projects.id (PLAN)',
|
||||
network_json JSON NOT NULL COMMENT '{epsg, lines, marks, pieces} — 길이 · 측점 · 높이는 저장 안 함',
|
||||
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||||
CONSTRAINT fk_plan_networks_project FOREIGN KEY (project_id)
|
||||
REFERENCES projects (id) ON DELETE CASCADE
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='예정노선 노선 정본';
|
||||
|
||||
CREATE TABLE IF NOT EXISTS plan_route_pieces (
|
||||
project_id CHAR(36) NOT NULL COMMENT 'projects.id (PLAN)',
|
||||
piece_id VARCHAR(64) NOT NULL COMMENT '조각 id(network_json.pieces.id)',
|
||||
line_id VARCHAR(64) NOT NULL COMMENT '선 id(본선 · 갈래)',
|
||||
year INT NOT NULL,
|
||||
from_m DOUBLE NOT NULL COMMENT '선 위 시작 거리(선마다 No.0 부터)',
|
||||
to_m DOUBLE NOT NULL,
|
||||
length_m DOUBLE NOT NULL,
|
||||
epsg INT NOT NULL,
|
||||
route_json JSON NOT NULL COMMENT '{planned: [[x, y]…], stations: [m…]}',
|
||||
fingerprint CHAR(40) NOT NULL COMMENT '조각 내용 지문(sha1)',
|
||||
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||||
PRIMARY KEY (project_id, piece_id),
|
||||
KEY idx_plan_route_pieces_year (project_id, year),
|
||||
CONSTRAINT fk_plan_route_pieces_project FOREIGN KEY (project_id)
|
||||
REFERENCES projects (id) ON DELETE CASCADE
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='예정노선 조각(구역 × 갈래)';
|
||||
|
||||
-- 업체 지정 = 조각 단위(요구 391) — NULL = 옛 연도 단위 지정
|
||||
ALTER TABLE plan_route_assignees
|
||||
ADD COLUMN IF NOT EXISTS piece_id VARCHAR(64) NULL COMMENT 'plan_route_pieces.piece_id' AFTER year,
|
||||
ADD INDEX IF NOT EXISTS idx_plan_route_assignees_piece (project_id, piece_id);
|
||||
@@ -77,6 +77,7 @@ from B09_Estimation.B09_Estimation_Router import router as b09_estimation_router
|
||||
from D01_PlanRegister.D01_PlanRegister_Router import dashboard_router as d01_dashboard_router
|
||||
from D01_PlanRegister.D01_PlanRegister_Router import router as d01_plan_register_router
|
||||
from D02_PlanBase.D02_PlanBase_Router import router as d02_plan_base_router
|
||||
from D03_PlanAnalysis.D03_PlanAnalysis_Network import router as d03_plan_network_router
|
||||
from D03_PlanAnalysis.D03_PlanAnalysis_Router import router as d03_plan_analysis_router
|
||||
from D04_PlanConfirm.D04_PlanConfirm_Link import dashboard_router as d04_link_dashboard_router
|
||||
from D04_PlanConfirm.D04_PlanConfirm_Link import router as d04_plan_link_router
|
||||
@@ -496,6 +497,7 @@ app.include_router(b05_route_router, dependencies=protected_with_company)
|
||||
app.include_router(b05_route_lifecycle_router, dependencies=protected_with_company)
|
||||
app.include_router(d02_plan_base_router, dependencies=protected_with_company)
|
||||
app.include_router(d03_plan_analysis_router, dependencies=protected_with_company)
|
||||
app.include_router(d03_plan_network_router, dependencies=protected_with_company)
|
||||
app.include_router(d04_plan_confirm_router, dependencies=protected_with_company)
|
||||
app.include_router(d04_plan_link_router, dependencies=protected_with_company)
|
||||
app.include_router(e02_stab_ground_router, dependencies=protected_with_company)
|
||||
|
||||
@@ -1,168 +1,258 @@
|
||||
"""D 예정노선 2단계 분석(PLAN 530-2) — 합성 지형 두 구간 · 경계 · 누가 · 저장본 · 복도 인자.
|
||||
"""D 예정노선 2단계 노선 정본(PLAN 562-2) — 손 그은 선 · 갈래 · 마크 · 조각 → 서버가 다시 셈 ·
|
||||
연도 행 · 옛 연도 노선을 선으로 · GET · PUT(산림청만 · 틀린 꼴 거절 · 업체 지정된 조각 지움 막음).
|
||||
|
||||
옛 추천 엔진 시험(합성 지형 갈래 넷 · 저장본 · 나란히 셈)은 엔진과 함께 걷음.
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_analyze.py -q
|
||||
"""
|
||||
|
||||
import numpy as np
|
||||
import json
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS, Terrain
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Search import CORRIDOR_M, SearchGrid
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze
|
||||
from common_util.common_util_auth import require_company
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Network as net
|
||||
|
||||
CRITERIA = {"max_grade": 0.14, "min_radius_m": 12.0, "design_speed": 20}
|
||||
SEGMENTS = [
|
||||
{"year": 2026, "start": (60.0, 200.0), "end": (560.123456789, 230.987654321)},
|
||||
{"year": 2027, "start": (560.123456789, 230.987654321), "end": (1040.0, 160.0)},
|
||||
]
|
||||
PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
|
||||
MAIN = [[0.0, 0.0], [30.0, 40.0], [30.0, 100.0], [90.0, 180.0]] # 50 + 60 + 100 = 210 m
|
||||
BRANCH = [[30.0, 100.0], [30.0, 150.0]] # 본선 점 2 에서 50 m
|
||||
|
||||
|
||||
def _hill() -> Terrain:
|
||||
x = np.arange(0.0, 1102.0, 2.0)
|
||||
y = np.arange(0.0, 402.0, 2.0)
|
||||
xx, yy = np.meshgrid(x, y)
|
||||
z = 30.0 * np.exp(-((xx - 300.0) ** 2 + (yy - 210.0) ** 2) / 80.0**2) + 0.02 * xx
|
||||
return Terrain(x, y, z, np.ones(xx.shape, dtype=bool))
|
||||
def _network(**over) -> dict:
|
||||
body = {
|
||||
"epsg": 5186,
|
||||
"lines": [
|
||||
{"id": "main", "vertices": MAIN, "branch": None},
|
||||
{"id": "b1", "vertices": BRANCH, "branch": {"line": "main", "vertex": 2}},
|
||||
],
|
||||
"marks": [{"line": "main", "at_m": 75.0}],
|
||||
"pieces": [
|
||||
{"id": "p1", "line": "main", "from_m": 0, "to_m": 75.0, "year": 2026},
|
||||
{"id": "p2", "line": "main", "from_m": 75.004, "to_m": 210.0, "year": 2027},
|
||||
{"id": "p3", "line": "b1", "from_m": 0, "to_m": 50.0, "year": 2026},
|
||||
],
|
||||
}
|
||||
body.update(over)
|
||||
return body
|
||||
|
||||
|
||||
def hill_loader(root, selection, models_dir, cache_dir):
|
||||
"""나란히 셈 프로세스가 이름으로 부르는 지형(모듈 맨 위 함수)."""
|
||||
return _hill(), []
|
||||
def test_derive_pieces_lengths_stations() -> None:
|
||||
got = net.derive(_network(), 2026)
|
||||
assert [(x["id"], x["length_m"]) for x in got["lines"]] == [("main", 210), ("b1", 50)]
|
||||
assert got["lines"][0]["stations"][-2:] == [200, 210] and got["lines"][1]["stations"] == [
|
||||
0,
|
||||
20,
|
||||
40,
|
||||
50,
|
||||
]
|
||||
pieces = {p["id"]: p for p in got["pieces"]}
|
||||
assert [p["id"] for p in got["pieces"]] == ["p1", "p2", "p3"] # 선 차례 · 앞에서부터
|
||||
# 경계 = 마크 그대로(보낸 75.004 아님)
|
||||
assert pieces["p2"]["from_m"] == 75.0 and pieces["p2"]["length_m"] == 135.0
|
||||
assert pieces["p1"]["planned"] == [[0, 0], [30, 40], [30, 65]]
|
||||
assert pieces["p1"]["stations"] == [0, 20, 40, 60, 75]
|
||||
assert pieces["p2"]["stations"][:2] == [75, 80]
|
||||
assert pieces["p3"]["planned"] == BRANCH # 갈래 측점 = 갈래마다 No.0
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def hill() -> Terrain:
|
||||
return _hill()
|
||||
@pytest.mark.parametrize(
|
||||
"over",
|
||||
[
|
||||
{"marks": [{"line": "main", "at_m": 210.0}]}, # 마크가 끝
|
||||
{"marks": [{"line": "zz", "at_m": 10.0}]},
|
||||
{"marks": [{"line": "main", "at_m": 75.0}, {"line": "main", "at_m": 75.0005}]},
|
||||
{"pieces": [{"id": "p1", "line": "main", "from_m": 0, "to_m": 80.0, "year": 2026}]},
|
||||
{"pieces": [{"id": "p1", "line": "main", "from_m": 0, "to_m": 75.0, "year": 2025}]},
|
||||
{
|
||||
"pieces": [
|
||||
{"id": "p1", "line": "main", "from_m": 0, "to_m": 75.0, "year": 2026},
|
||||
{"id": "p9", "line": "main", "from_m": 0, "to_m": 75.0, "year": 2027},
|
||||
]
|
||||
},
|
||||
{
|
||||
"lines": [
|
||||
{"id": "b1", "vertices": BRANCH, "branch": {"line": "main", "vertex": 2}},
|
||||
{"id": "main", "vertices": MAIN, "branch": None},
|
||||
],
|
||||
"marks": [],
|
||||
"pieces": [],
|
||||
}, # 부모가 뒤
|
||||
{
|
||||
"lines": [
|
||||
{"id": "main", "vertices": MAIN, "branch": None},
|
||||
{"id": "b1", "vertices": BRANCH, "branch": {"line": "main", "vertex": 1}},
|
||||
],
|
||||
}, # 갈래 첫 점 ≠ 부모 점
|
||||
{"lines": [{"id": "main", "vertices": [[0, 0], [float("nan"), 1]], "branch": None}]},
|
||||
],
|
||||
)
|
||||
def test_derive_rejects(over) -> None:
|
||||
with pytest.raises(net.NetworkError):
|
||||
net.derive(_network(**over), 2026)
|
||||
|
||||
|
||||
def test_year_rows_group_pieces() -> None:
|
||||
rows = net.year_rows(net.derive(_network(), 2026)["pieces"])
|
||||
assert list(rows) == [2026, 2027]
|
||||
first = rows[2026]["route"]
|
||||
assert first["mode"] == "manual" and [p["id"] for p in first["pieces"]] == ["p1", "p3"]
|
||||
assert first["planned"] == [[0, 0], [30, 40], [30, 65]] and first["length_m"] == 125.0
|
||||
assert rows[2026]["start"] == [0, 0] and rows[2026]["end"] == [30, 65]
|
||||
assert rows[2027]["route"]["start_chainage_m"] == 75.0
|
||||
|
||||
|
||||
def test_legacy_rows_become_one_line() -> None:
|
||||
rows = [
|
||||
{"year": 2026, "epsg": 5186, "route_json": json.dumps({"planned": [[0, 0], [30, 40]]})},
|
||||
{"year": 2027, "epsg": 5186, "route_json": None},
|
||||
{"year": 2028, "epsg": 5186, "route_json": {"planned": [[30, 40], [30, 100]]}},
|
||||
]
|
||||
got = net.legacy_network(rows)
|
||||
assert got["lines"][0]["vertices"] == [[0, 0], [30, 40], [30, 100]]
|
||||
assert got["marks"] == [{"line": "main", "at_m": 50.0}]
|
||||
assert [(p["year"], p["from_m"], p["to_m"]) for p in got["pieces"]] == [
|
||||
(2026, 0, 50),
|
||||
(2028, 50, 110),
|
||||
]
|
||||
assert net.derive(got)["pieces"][1]["planned"] == [[30, 40], [30, 100]]
|
||||
assert net.legacy_network([{"year": 2026, "epsg": 5186, "route_json": None}]) is None
|
||||
|
||||
|
||||
class _Cursor:
|
||||
def __init__(self, db):
|
||||
self.db = db
|
||||
self.last = ""
|
||||
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc):
|
||||
return False
|
||||
|
||||
async def execute(self, sql, args=None):
|
||||
self.last = sql
|
||||
self.db.log.append((sql, args))
|
||||
|
||||
async def fetchone(self):
|
||||
return self._rows()[0] if self._rows() else None
|
||||
|
||||
async def fetchall(self):
|
||||
return self._rows()
|
||||
|
||||
def _rows(self):
|
||||
for key, rows in self.db.answers.items():
|
||||
if key in self.last:
|
||||
return rows
|
||||
return []
|
||||
|
||||
|
||||
class _Connection:
|
||||
def __init__(self, db):
|
||||
self.db = db
|
||||
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc):
|
||||
return False
|
||||
|
||||
async def begin(self):
|
||||
pass
|
||||
|
||||
async def commit(self):
|
||||
self.db.log.append(("COMMIT", None))
|
||||
|
||||
async def rollback(self):
|
||||
self.db.log.append(("ROLLBACK", None))
|
||||
|
||||
def cursor(self, *args):
|
||||
return _Cursor(self.db)
|
||||
|
||||
|
||||
class _Db:
|
||||
def __init__(self):
|
||||
self.log: list = []
|
||||
self.answers: dict = {"FROM projects": [{"project_year": 2026}]}
|
||||
|
||||
def acquire(self):
|
||||
return _Connection(self)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def run(tmp_path, hill):
|
||||
_POOLS.clear()
|
||||
calls = []
|
||||
|
||||
def loader(root, selection, models_dir, cache_dir):
|
||||
calls.append((models_dir.name, cache_dir.name))
|
||||
return hill, []
|
||||
|
||||
def go(criteria=CRITERIA, corridor_m=150.0):
|
||||
return analyze.analyze_plan(
|
||||
tmp_path, SEGMENTS, criteria, loader=loader, corridor_m=corridor_m, workers=1
|
||||
)
|
||||
|
||||
go.calls = calls
|
||||
yield go
|
||||
_POOLS.clear()
|
||||
def client(monkeypatch):
|
||||
db = _Db()
|
||||
monkeypatch.setattr(net, "get_db_pool", lambda: db)
|
||||
app = FastAPI()
|
||||
app.include_router(net.router)
|
||||
session = {"role": "USER", "email": "a@forest.go.kr"}
|
||||
app.dependency_overrides[require_company] = lambda: session
|
||||
test = TestClient(app)
|
||||
test.db = db
|
||||
test.session = session
|
||||
return test
|
||||
|
||||
|
||||
def test_boundary_chainage_and_line(run) -> None:
|
||||
results = run()
|
||||
# 연도 구간마다 제 DEM · 제 비용면 캐시
|
||||
assert run.calls == [("models_2026", "cost_2026"), ("models_2027", "cost_2027")]
|
||||
for seg, result in zip(SEGMENTS, results):
|
||||
assert set(result["routes"]) == set(analyze.PLAN_MODES)
|
||||
for route in result["routes"].values():
|
||||
assert "error" not in route
|
||||
# 시종점 그대로 — 연도 경계가 정확히 맞물림
|
||||
assert route["planned"][0] == result["start"] == pytest.approx(seg["start"], abs=1e-4)
|
||||
assert route["planned"][-1] == result["end"] == pytest.approx(seg["end"], abs=1e-4)
|
||||
assert results[0]["routes"]["gradeline"]["planned"][-1] == results[1]["start"]
|
||||
|
||||
rows = analyze.chain(results, {2026: "earthwork", 2027: None}, 1234.5)
|
||||
assert [r["mode"] for r in rows] == ["earthwork", "gradeline"]
|
||||
assert rows[0]["start_chainage_m"] == 1234.5
|
||||
assert rows[1]["start_chainage_m"] == rows[0]["end_chainage_m"]
|
||||
for result, row in zip(results, rows):
|
||||
length = result["routes"][row["mode"]]["metrics"]["length_m"]
|
||||
assert row["end_chainage_m"] - row["start_chainage_m"] == pytest.approx(length, abs=1e-3)
|
||||
|
||||
line = analyze.joined_line(results, rows)
|
||||
first = results[0]["routes"]["earthwork"]["planned"]
|
||||
second = results[1]["routes"]["gradeline"]["planned"]
|
||||
assert len(line) == len(first) + len(second) - 1 # 경계점 한 번만(꺾인 채)
|
||||
assert line[len(first) - 1] == results[0]["end"] == results[1]["start"]
|
||||
URL = f"/api/projects/{PID}/plan/network"
|
||||
|
||||
|
||||
def test_same_input_reads_store(run, monkeypatch) -> None:
|
||||
first = run()
|
||||
assert [r["stored"] for r in first] == [False, False]
|
||||
|
||||
def boom(*args, **kwargs):
|
||||
raise AssertionError("저장본이 있는데 다시 셈")
|
||||
|
||||
monkeypatch.setattr(analyze, "generate_initial_route", boom)
|
||||
again = run()
|
||||
assert [r["stored"] for r in again] == [True, True]
|
||||
assert again[1]["routes"] == first[1]["routes"]
|
||||
# 입력(복도 · 기준)이 바뀌면 다시 셈
|
||||
with pytest.raises(AssertionError, match="다시 셈"):
|
||||
run(corridor_m=120.0)
|
||||
with pytest.raises(AssertionError, match="다시 셈"):
|
||||
run(criteria={**CRITERIA, "max_grade": 0.12})
|
||||
def test_get_saved_legacy_or_empty(client) -> None:
|
||||
assert client.get(URL).json()["network"] is None
|
||||
client.db.answers["FROM plan_route_years"] = [
|
||||
{"year": 2026, "epsg": 5186, "route_json": '{"planned": [[0, 0], [30, 40]]}'}
|
||||
]
|
||||
got = client.get(URL).json()
|
||||
assert got["legacy"] is True and got["derived"]["pieces"][0]["length_m"] == 50
|
||||
client.db.answers["FROM plan_networks"] = [{"network_json": json.dumps(_network())}]
|
||||
got = client.get(URL).json()
|
||||
assert got["legacy"] is False and len(got["derived"]["pieces"]) == 3
|
||||
|
||||
|
||||
def test_parallel_years_match_sequential(tmp_path) -> None:
|
||||
"""연도 구간 나란히(프로세스 둘) = 차례로 — 갈래 넷 노선 좌표 · 잰 값 차 0(531-6)."""
|
||||
_POOLS.clear()
|
||||
seq = analyze.analyze_plan(
|
||||
tmp_path / "a", SEGMENTS, CRITERIA, loader=hill_loader, corridor_m=150.0, workers=1
|
||||
def test_put_stores_network_pieces_years(client) -> None:
|
||||
got = client.put(URL, json=_network())
|
||||
assert got.status_code == 200 and got.json()["derived"]["pieces"][1]["from_m"] == 75.0
|
||||
log = client.db.log
|
||||
assert log[-1][0] == "COMMIT"
|
||||
pieces = [a for s, a in log if s.startswith("INSERT INTO plan_route_pieces")]
|
||||
assert [(a[1], a[2], a[3]) for a in pieces] == [
|
||||
("p1", "main", 2026),
|
||||
("p2", "main", 2027),
|
||||
("p3", "b1", 2026),
|
||||
]
|
||||
years = [a for s, a in log if s.startswith("INSERT INTO plan_route_years")]
|
||||
assert [a[1] for a in years] == [2026, 2027]
|
||||
assert json.loads(years[0][7])["pieces"][1]["id"] == "p3"
|
||||
assert ("UPDATE projects SET project_year_end = %s WHERE id = %s", (2027, PID)) in log
|
||||
stale = [a for s, a in log if "year NOT IN" in s]
|
||||
assert stale == [(PID, 2026, 2027)]
|
||||
|
||||
|
||||
def test_put_reset_clears_year_routes(client) -> None:
|
||||
got = client.put(URL, json={"epsg": 5186, "lines": [], "marks": [], "pieces": []})
|
||||
assert got.status_code == 200 and got.json()["derived"] == {"lines": [], "pieces": []}
|
||||
sqls = [s for s, _ in client.db.log]
|
||||
assert any(s.startswith("UPDATE plan_route_years SET route_json = NULL") for s in sqls)
|
||||
assert not any("project_year_end" in s for s in sqls)
|
||||
|
||||
|
||||
def test_put_rejects(client) -> None:
|
||||
bad = _network(marks=[{"line": "main", "at_m": 500.0}])
|
||||
assert client.put(URL, json=bad).status_code == 400
|
||||
assert (
|
||||
client.put(URL, json=_network(lines=[{"id": "a b", "vertices": MAIN}])).status_code == 422
|
||||
)
|
||||
_POOLS.clear()
|
||||
steps = []
|
||||
par = analyze.analyze_plan(
|
||||
tmp_path / "b",
|
||||
SEGMENTS,
|
||||
CRITERIA,
|
||||
loader=hill_loader,
|
||||
corridor_m=150.0,
|
||||
workers=2,
|
||||
on_progress=lambda p, m: steps.append(m),
|
||||
# 업체가 지정된 조각 · 옛 연도를 지우면 막음
|
||||
client.db.answers["piece_id IS NOT NULL"] = [("p2",)]
|
||||
one = _network(
|
||||
lines=[{"id": "main", "vertices": MAIN}],
|
||||
marks=[],
|
||||
pieces=[{"id": "p1", "line": "main", "from_m": 0, "to_m": 210, "year": 2026}],
|
||||
)
|
||||
assert [r["routes"] for r in par] == [r["routes"] for r in seq]
|
||||
assert [r["fingerprint"] for r in par] == [r["fingerprint"] for r in seq]
|
||||
assert "나란히 2" in steps[0] and steps[-1] == "분석 끝"
|
||||
# 저장본은 제 프로세스에서 지문만 보고 그대로
|
||||
again = analyze.analyze_plan(
|
||||
tmp_path / "b", SEGMENTS, CRITERIA, loader=hill_loader, corridor_m=150.0, workers=2
|
||||
)
|
||||
assert [r["stored"] for r in again] == [True, True]
|
||||
_POOLS.clear()
|
||||
|
||||
|
||||
def test_near_streams_keeps_every_nearest_line(hill) -> None:
|
||||
"""계류 거르기 — 지형 상자 안 점의 계류 거리 · 건너기가 통째와 같고 먼 선은 빠짐(531-6)."""
|
||||
import shapely
|
||||
from shapely.geometry import LineString, MultiLineString
|
||||
|
||||
rng = np.random.default_rng(7)
|
||||
streams = []
|
||||
for _ in range(80): # 지형(1.1 × 0.4 km) 둘레 5 km 에 흩은 계류
|
||||
start = rng.uniform(-2000.0, 3000.0, 2)
|
||||
streams.append([tuple(start + step) for step in np.cumsum(rng.normal(0, 60, (6, 2)), 0)])
|
||||
kept = analyze.near_streams(hill, streams)
|
||||
assert 0 < len(kept) < len(streams)
|
||||
points = shapely.points(rng.uniform((0.0, 0.0), (1100.0, 400.0), (4000, 2)))
|
||||
whole, part = MultiLineString(streams), MultiLineString(kept)
|
||||
assert np.array_equal(shapely.distance(points, whole), shapely.distance(points, part))
|
||||
route = LineString([(0.0, 0.0), (1100.0, 400.0)])
|
||||
assert route.intersection(whole).equals(route.intersection(part))
|
||||
assert analyze.near_streams(hill, []) == []
|
||||
|
||||
|
||||
def test_corridor_argument_keeps_b_default(hill) -> None:
|
||||
line = [SEGMENTS[0]["start"], SEGMENTS[0]["end"]]
|
||||
base = SearchGrid(line, hill, [], stream_offset_m=10.0)
|
||||
wide = SearchGrid(line, hill, [], stream_offset_m=10.0, corridor_m=180.0)
|
||||
assert base.corridor_m == CORRIDOR_M == 100.0
|
||||
assert wide.cells > base.cells
|
||||
|
||||
|
||||
def test_criteria_from_project() -> None:
|
||||
from B05_Profile.B05_Profile_Engine_Grade import (
|
||||
legal_grade_limit_pct,
|
||||
legal_plan_radius_min_m,
|
||||
)
|
||||
|
||||
got = analyze.plan_criteria("work", 99) # 못 고르는 속도 → 종류 기본
|
||||
assert got["design_speed"] == 20
|
||||
assert got["max_grade"] == legal_grade_limit_pct("work", "special", False, 20) / 100.0
|
||||
assert got["min_radius_m"] == legal_plan_radius_min_m(20, "special")
|
||||
assert client.put(URL, json=one).status_code == 409
|
||||
client.db.answers["piece_id IS NOT NULL"] = []
|
||||
client.db.answers["piece_id IS NULL"] = [(2027,)]
|
||||
assert client.put(URL, json=one).status_code == 409
|
||||
assert not [s for s, _ in client.db.log if s.startswith("INSERT")]
|
||||
client.session["email"] = "a@gmail.com"
|
||||
assert client.put(URL, json=_network()).status_code == 403
|
||||
|
||||
@@ -1,303 +1,36 @@
|
||||
"""D 분석 API(PLAN 530-2 · 530-3) — GET 이은 누가 · PUT [저장] = 서버 저장본으로 다시 셈 · 모르는 갈래
|
||||
거절 · 끈 지점(points) = 그 지점으로 보기 · 안 이어지면 거절 · [저장] 때만 지점 정본(산림청만) ·
|
||||
꺾임점 고침(543-3) = 서버가 마디로 선 · 지표를 다시 셈 · 누가가 고친 길이로 · GET 은 지문 같은 고침만.
|
||||
"""D 지형 API(PLAN 530-1) — 끈 지점(points) = 연도 같음 · 직전 종점 = 다음 시점 아니면 거절.
|
||||
|
||||
옛 분석 API 시험(GET · PUT analysis · 꺾임점 고침)은 옛 추천 엔진과 함께 걷음(PLAN 562-2) ·
|
||||
노선 정본 시험 = `test_530_2_plan_analyze.py`.
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_router.py -q
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from fastapi import HTTPException
|
||||
|
||||
from common_util.common_util_auth import require_company
|
||||
from common_util.common_util_json import atomic_write_json
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Edit as plan_edit
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Router as router_mod
|
||||
|
||||
PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
|
||||
P0, P1, P2 = (0.0, 0.0), (300.0, 400.0), (900.0, 400.0)
|
||||
|
||||
|
||||
def _route(points: list, length: float) -> dict:
|
||||
return {"planned": points, "nodes": [], "curves": [], "metrics": {"length_m": length}}
|
||||
|
||||
|
||||
class _Cursor:
|
||||
def __init__(self, log):
|
||||
self.log = log
|
||||
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc):
|
||||
return False
|
||||
|
||||
async def execute(self, sql, args=None):
|
||||
self.log.append((sql, args))
|
||||
|
||||
|
||||
class _Connection(_Cursor):
|
||||
async def begin(self):
|
||||
pass
|
||||
|
||||
async def commit(self):
|
||||
self.log.append(("COMMIT", None))
|
||||
|
||||
async def rollback(self):
|
||||
pass
|
||||
|
||||
def cursor(self, *args):
|
||||
return _Cursor(self.log)
|
||||
|
||||
|
||||
class _Pool:
|
||||
def __init__(self, log):
|
||||
self.log = log
|
||||
|
||||
def acquire(self):
|
||||
return _Connection(self.log)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(tmp_path, monkeypatch):
|
||||
segments = [
|
||||
{"year": 2026, "start": P0, "end": P1, "epsg": 5186, "route_json": None},
|
||||
{"year": 2027, "start": P1, "end": P2, "epsg": 5186, "route_json": '{"mode": "gentle"}'},
|
||||
]
|
||||
for seg, routes in (
|
||||
(segments[0], {"gradeline": _route([[0, 0], [100, 200], [300, 400]], 530.0)}),
|
||||
(
|
||||
segments[1],
|
||||
{
|
||||
"gradeline": _route([[300, 400], [900, 400]], 600.0),
|
||||
"gentle": _route([[300, 400], [600, 500], [900, 400]], 640.0),
|
||||
"earthwork": {"error": "길 없음"},
|
||||
},
|
||||
),
|
||||
):
|
||||
atomic_write_json(
|
||||
analyze.derived_dir(tmp_path) / f"route_{seg['year']}.json",
|
||||
{
|
||||
"year": seg["year"],
|
||||
"start": list(seg["start"]),
|
||||
"end": list(seg["end"]),
|
||||
"fingerprint": "f",
|
||||
"criteria": {"min_radius_m": 12.0, "max_grade_pct": 14.0},
|
||||
"routes": routes,
|
||||
},
|
||||
)
|
||||
project = {
|
||||
"road_type": "work",
|
||||
"design_speed_kph": None,
|
||||
"route_start_m": 1000.0,
|
||||
"route_end_m": 2500.0,
|
||||
}
|
||||
|
||||
async def inputs(project_id, points=None):
|
||||
return tmp_path, project, router_mod._override(segments, points) if points else segments
|
||||
|
||||
log: list = []
|
||||
monkeypatch.setattr(router_mod, "_plan_inputs", inputs)
|
||||
monkeypatch.setattr(router_mod, "get_db_pool", lambda: _Pool(log))
|
||||
app = FastAPI()
|
||||
app.include_router(router_mod.router)
|
||||
session = {"role": "USER", "email": "a@forest.go.kr"}
|
||||
app.dependency_overrides[require_company] = lambda: session
|
||||
stored: list = []
|
||||
|
||||
async def store_points(project_id, segs):
|
||||
stored.append([(s["year"], s["start"], s["end"]) for s in segs])
|
||||
|
||||
monkeypatch.setattr(router_mod, "_store_points", store_points)
|
||||
# 꺾임점 고침 지표 — 지형 없이(선 길이만)
|
||||
loads: list = []
|
||||
monkeypatch.setattr(
|
||||
analyze, "_load", lambda root, year, loader: loads.append(year) or (None, [])
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
plan_edit,
|
||||
"edit_metrics",
|
||||
lambda line, result, terrain, streams: {
|
||||
"length_m": round(analyze.line_length(line.vertices), 2),
|
||||
"max_grade_pct": 9.5,
|
||||
"over_limit": {"total": 1},
|
||||
},
|
||||
)
|
||||
test = TestClient(app)
|
||||
test.segments = segments
|
||||
test.loads = loads
|
||||
test.log = log
|
||||
test.session = session
|
||||
test.stored = stored
|
||||
return test
|
||||
|
||||
|
||||
def test_get_chains_saved_choice(client) -> None:
|
||||
body = client.get(f"/api/projects/{PID}/plan/analysis").json()
|
||||
chain = body["chain"]
|
||||
assert [r["mode"] for r in chain] == ["gradeline", "gentle"] # 저장값 · 없으면 첫 성공
|
||||
assert chain[0]["start_chainage_m"] == 1000.0
|
||||
assert chain[1]["start_chainage_m"] == chain[0]["end_chainage_m"] == 1530.0
|
||||
assert chain[1]["end_chainage_m"] == 2170.0
|
||||
assert body["line"] == [[0, 0], [100, 200], [300, 400], [600, 500], [900, 400]]
|
||||
assert body["end_chainage_target_m"] == 2500.0 # 등록 목표 — 화면이 차이를 보임
|
||||
|
||||
|
||||
def test_put_saves_server_routes(client) -> None:
|
||||
got = client.put(f"/api/projects/{PID}/plan/analysis", json={"choices": {"2027": "gradeline"}})
|
||||
assert got.status_code == 200
|
||||
assert got.json()["chain"][1]["end_chainage_m"] == 2130.0
|
||||
updates = [args for sql, args in client.log if sql.startswith("UPDATE plan_route_years")]
|
||||
saved = {args[3]: json.loads(args[0]) for args in updates}
|
||||
assert saved[2027]["mode"] == "gradeline" and saved[2027]["planned"] == [[300, 400], [900, 400]]
|
||||
assert saved[2026]["start_chainage_m"] == 1000.0 and updates[0][1] == "f"
|
||||
assert client.log[-1][0] == "COMMIT"
|
||||
|
||||
|
||||
def test_put_rejects_unknown_or_empty_mode(client) -> None:
|
||||
url = f"/api/projects/{PID}/plan/analysis"
|
||||
assert client.put(url, json={"choices": {"2027": "follow"}}).status_code == 400
|
||||
assert client.put(url, json={"choices": {"2027": "earthwork"}}).status_code == 409
|
||||
assert not [sql for sql, _ in client.log if sql.startswith("UPDATE")]
|
||||
SEGMENTS = [
|
||||
{"year": 2026, "start": P0, "end": P1, "epsg": 5186, "route_json": None},
|
||||
{"year": 2027, "start": P1, "end": P2, "epsg": 5186, "route_json": None},
|
||||
]
|
||||
|
||||
|
||||
def _points(p1) -> list:
|
||||
return [
|
||||
{"year": 2026, "start": list(P0), "end": list(p1)},
|
||||
{"year": 2027, "start": list(p1), "end": list(P2)},
|
||||
router_mod.PlanPoint(year=2026, start=P0, end=p1),
|
||||
router_mod.PlanPoint(year=2027, start=p1, end=P2),
|
||||
]
|
||||
|
||||
|
||||
def test_points_override_sees_matching_store_only(client) -> None:
|
||||
url = f"/api/projects/{PID}/plan/analysis"
|
||||
same = client.get(url, params={"points": json.dumps(_points(P1))}).json()
|
||||
assert same["chain"] is not None and same["line"][-1] == [900, 400]
|
||||
moved = client.get(url, params={"points": json.dumps(_points((310.0, 400.0)))}).json()
|
||||
assert moved["chain"] is None and moved["years"][0]["analysis"] is None # 끈 지점 = 저장본 없음
|
||||
|
||||
|
||||
def test_points_must_chain_and_cover_years(client) -> None:
|
||||
url = f"/api/projects/{PID}/plan/analysis"
|
||||
def test_points_must_chain_and_cover_years() -> None:
|
||||
moved = router_mod._override(SEGMENTS, _points((310.0, 400.0)))
|
||||
assert moved[0]["end"] == (310.0, 400.0) and moved[0]["saved"] == (P0, P1)
|
||||
broken = _points(P1)
|
||||
broken[1]["start"] = [301.0, 400.0]
|
||||
assert client.get(url, params={"points": json.dumps(broken)}).status_code == 400
|
||||
assert client.get(url, params={"points": json.dumps(_points(P1)[:1])}).status_code == 400
|
||||
|
||||
|
||||
def test_put_with_points_stores_points_forest_only(client) -> None:
|
||||
url = f"/api/projects/{PID}/plan/analysis"
|
||||
body = {"choices": {"2027": "gradeline"}, "points": _points(P1)}
|
||||
assert client.put(url, json=body).status_code == 200
|
||||
assert client.stored == [[(2026, P0, P1), (2027, P1, P2)]]
|
||||
assert client.put(url, json={"choices": {}}).status_code == 200 and len(client.stored) == 1
|
||||
client.session["email"] = "a@gmail.com"
|
||||
assert client.put(url, json=body).status_code == 403
|
||||
moved = {"choices": {}, "points": _points((310.0, 400.0))}
|
||||
client.session["email"] = "a@forest.go.kr"
|
||||
assert client.put(url, json=moved).status_code == 409 and len(client.stored) == 1
|
||||
|
||||
|
||||
def _edit(nodes, fingerprint="f") -> dict:
|
||||
return {
|
||||
"base_mode": "gentle",
|
||||
"base_fingerprint": fingerprint,
|
||||
"nodes": nodes,
|
||||
"curve_flags": [True] * len(nodes),
|
||||
"radii": [None] * len(nodes),
|
||||
}
|
||||
|
||||
|
||||
def test_put_edit_server_redraws_and_get_returns(client) -> None:
|
||||
url = f"/api/projects/{PID}/plan/analysis"
|
||||
edit = _edit([[300, 400], [600, 560], [900, 400]])
|
||||
got = client.put(url, json={"choices": {}, "edits": {"2027": edit}})
|
||||
assert got.status_code == 200, got.text
|
||||
updates = {args[3]: json.loads(args[0]) for sql, args in client.log if sql.startswith("UPDATE")}
|
||||
body = updates[2027]
|
||||
line = plan_edit.edited_line(edit, 12.0)
|
||||
assert len(body["planned"]) == len(line.vertices) and body["edit"]["nodes"] == edit["nodes"]
|
||||
assert body["planned"][0] == [300, 400] and body["planned"][-1] == [900, 400]
|
||||
assert body["metrics"]["max_grade_pct"] == 9.5 and len(body["curves"]) == 1
|
||||
length = analyze.line_length(body["planned"])
|
||||
assert got.json()["chain"][1]["length_m"] == round(length, 3) != 640.0
|
||||
assert "edit" not in updates[2026] and client.loads == [2027]
|
||||
|
||||
# GET — 저장된 고침 + 고친 길이로 이은 누가 · 지문이 바뀌면 안 돌려줌
|
||||
client.segments[1]["route_json"] = json.dumps(body)
|
||||
view = client.get(url).json()
|
||||
assert view["years"][1]["edit"]["nodes"] == edit["nodes"]
|
||||
assert view["years"][1]["edit"]["metrics"]["over_limit"]["total"] == 1
|
||||
assert view["chain"][1]["length_m"] == round(length, 3)
|
||||
assert view["years"][1]["analysis"]["routes"]["gentle"]["metrics"]["length_m"] == 640.0
|
||||
# 같은 입력 다시 [저장] = 저장된 선 · 지표 그대로(지형 안 읽음)
|
||||
assert client.put(url, json={"choices": {}, "edits": {"2027": edit}}).status_code == 200
|
||||
assert client.loads == [2027]
|
||||
client.segments[1]["route_json"] = json.dumps(
|
||||
{**body, "edit": {**body["edit"], "base_fingerprint": "old"}}
|
||||
)
|
||||
stale = client.get(url).json()
|
||||
assert stale["years"][1]["edit"] is None and stale["chain"][1]["length_m"] == 640.0
|
||||
|
||||
|
||||
def test_put_edit_rejects(client) -> None:
|
||||
url = f"/api/projects/{PID}/plan/analysis"
|
||||
off_end = _edit([[300, 400], [600, 560], [901, 400]])
|
||||
stale = _edit([[300, 400], [600, 560], [900, 400]], "old")
|
||||
other = {**stale, "base_fingerprint": "f", "base_mode": "gradeline"}
|
||||
for bad in (off_end, stale, other):
|
||||
assert client.put(url, json={"choices": {}, "edits": {"2027": bad}}).status_code == 409
|
||||
client.session["email"] = "a@gmail.com"
|
||||
good = _edit([[300, 400], [600, 560], [900, 400]])
|
||||
assert client.put(url, json={"choices": {}, "edits": {"2027": good}}).status_code == 403
|
||||
assert not [sql for sql, _ in client.log if sql.startswith("UPDATE")]
|
||||
|
||||
|
||||
def _analyze_moved(client, tmp_path, monkeypatch, p1, length=700.0) -> None:
|
||||
"""끈 지점 분석(`POST …/plan/analyze` 와 같은 길) — 엔진 · 지문은 가짜."""
|
||||
monkeypatch.setattr(analyze, "segment_fingerprint", lambda seg, *a, **k: "g")
|
||||
monkeypatch.setattr(
|
||||
analyze,
|
||||
"generate_initial_route",
|
||||
lambda expected, *a, mode, **k: _route([list(expected[0]), list(expected[1])], length),
|
||||
)
|
||||
points = [router_mod.PlanPoint(**p) for p in _points(p1)]
|
||||
segs = router_mod._override(client.segments, points)
|
||||
analyze.analyze_plan(tmp_path, segs, analyze.plan_criteria("work", None), workers=1)
|
||||
|
||||
|
||||
def test_moved_analysis_keeps_saved_store(client, tmp_path, monkeypatch) -> None:
|
||||
"""543-4 — 끈 지점 분석은 지점 열쇠 파일로 · 저장 지점 GET 그대로 · 끈 지점 GET 은 그 분석 ·
|
||||
[저장] 뒤 정본 = 끈 지점 분석본 · 옛 꼴 파일은 읽기만 · 쌓임 거르기는 저장 지점 것을 남김."""
|
||||
url = f"/api/projects/{PID}/plan/analysis"
|
||||
before = client.get(url).json()
|
||||
moved = (310.0, 400.0)
|
||||
_analyze_moved(client, tmp_path, monkeypatch, moved)
|
||||
folder = analyze.derived_dir(tmp_path)
|
||||
assert (folder / "route_2026.json").is_file() and (folder / "route_2027.json").is_file()
|
||||
|
||||
assert client.get(url).json() == before # 저장 지점 = 옛 분석 그대로
|
||||
view = client.get(url, params={"points": json.dumps(_points(moved))}).json()
|
||||
assert [y["analysis"]["fingerprint"] for y in view["years"]] == ["g", "g"]
|
||||
assert view["chain"][1]["end_chainage_m"] == 2400.0
|
||||
|
||||
# [저장] = 끈 지점 정본 → 저장 지점 GET 이 그 분석
|
||||
body = {"choices": {"2027": "gentle"}, "points": _points(moved)}
|
||||
assert client.put(url, json=body).status_code == 200
|
||||
for seg, (start, end) in zip(client.segments, ((P0, moved), (moved, P2))):
|
||||
seg.update(start=start, end=end)
|
||||
after = client.get(url).json()
|
||||
assert [y["analysis"]["fingerprint"] for y in after["years"]] == ["g", "g"]
|
||||
|
||||
# 쌓임 — 저장 지점 것은 가장 옛것이어도 남음 · 연도마다 최근 KEEP_PER_YEAR 개
|
||||
for path in folder.glob("route_*_*.json"):
|
||||
os.utime(path, (1000.0, 1000.0))
|
||||
for n in range(analyze.KEEP_PER_YEAR + 1):
|
||||
_analyze_moved(client, tmp_path, monkeypatch, (320.0 + n, 400.0))
|
||||
keyed = sorted(p.name for p in folder.glob("route_2026_*.json"))
|
||||
assert len(keyed) == analyze.KEEP_PER_YEAR + 1
|
||||
assert analyze.segment_path(tmp_path, client.segments[0]).name in keyed
|
||||
assert client.get(url).json() == after
|
||||
broken[1] = router_mod.PlanPoint(year=2027, start=(301.0, 400.0), end=P2)
|
||||
for points in (broken, _points(P1)[:1]):
|
||||
with pytest.raises(HTTPException) as error:
|
||||
router_mod._override(SEGMENTS, points)
|
||||
assert error.value.status_code == 400
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
"""D03 화면 계산 거울(PLAN 530-3) — 파이썬 `D03_PlanAnalysis_Analyze` ↔ TS `logic/D03_PlanAnalysis_Plan.ts`.
|
||||
"""D03 화면 계산(PLAN 530-3) — TS `logic/D03_PlanAnalysis_Plan.ts`.
|
||||
|
||||
짝: `pick_mode` · `chain` · `joined_line` ↔ `pickMode` · `chainRows` · `joinedLine`.
|
||||
TS 만: 측점 누가(경계 = 측점 · 간격 배수) · 앞붙임 선(선 위 누가 = 등록 시작 누가) · 받은 지형 판정.
|
||||
측점 누가 앞붙임 선(선 위 누가 = 등록 시작 누가) · 받은 지형 판정.
|
||||
파이썬 짝(`D03_PlanAnalysis_Analyze`)은
|
||||
옛 추천 엔진과 함께 걷음(PLAN 562-2).
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_530_3_plan_logic.py -q
|
||||
"""
|
||||
@@ -18,8 +19,6 @@ ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze # noqa: E402
|
||||
|
||||
TS_FILE = ROOT / "D03_PlanAnalysis" / "logic" / "D03_PlanAnalysis_Plan.ts"
|
||||
TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc"
|
||||
|
||||
@@ -115,24 +114,6 @@ def _run_ts(tmp_path: Path) -> dict:
|
||||
return json.loads(result.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def test_ts_matches_python_chain(tmp_path: Path) -> None:
|
||||
got = _run_ts(tmp_path)
|
||||
for case, ts in zip(_CASES, got["out"]):
|
||||
choices = {int(k): v for k, v in case["choices"].items()}
|
||||
rows = analyze.chain(_RESULTS, choices, case["start"])
|
||||
assert ts["picks"] == [analyze.pick_mode(r, choices.get(r["year"])) for r in _RESULTS]
|
||||
assert ts["rows"] == rows
|
||||
assert ts["line"] == analyze.joined_line(_RESULTS, rows)
|
||||
|
||||
# 측점 = 경계(1530) · 20 배수 · 끝 — 첫 연도 시작 1000 · 끝 2170.333
|
||||
stations = got["out"][0]["stations"]
|
||||
assert stations[:3] == [1000.0, 1020.0, 1040.0]
|
||||
assert 1530.0 in stations and 1520.0 in stations and 1540.0 in stations
|
||||
assert stations[-2:] == [2160.0, 2170.333]
|
||||
# 시작 17.25 → 첫 규칙 측점 20
|
||||
assert got["out"][1]["stations"][:2] == [17.25, 20.0]
|
||||
|
||||
|
||||
def test_lead_in_and_reach(tmp_path: Path) -> None:
|
||||
got = _run_ts(tmp_path)
|
||||
back = got["lead"][0]
|
||||
|
||||
@@ -1,210 +0,0 @@
|
||||
"""D03 꺾임점 고침 거울(PLAN 543-1 · 543-3) — 화면 `logic/D03_PlanAnalysis_Edit.ts` ↔ 서버
|
||||
`D03_PlanAnalysis_Edit.py` · 이은 누가 짝 `withEdits` ↔ `with_edits`.
|
||||
|
||||
- 끌기 · 넣기 · 지우기 · 손잡이 뒤 처음 · 끝 마디 그대로 · 끝 마디 잠금 · R ≥ 법정 최소
|
||||
- 고친 선 길이 = `lineLength` · 이은 누가가 고친 길이로 이어짐(파이썬 = TS)
|
||||
- 브라우저 선 = 서버 선(1 mm) — Node 길 · 파이썬 폴백 둘 다
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_543_plan_edit.py -q
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from common_util.common_util_route_polyline import build_planned_polyline # noqa: E402
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze # noqa: E402
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Edit as edit_mod # noqa: E402
|
||||
|
||||
TS_FILE = ROOT / "D03_PlanAnalysis" / "logic" / "D03_PlanAnalysis_Edit.ts"
|
||||
TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc"
|
||||
MIN_R = 12.0
|
||||
NODES = [(0.0, 0.0), (120.0, 10.0), (200.0, 120.0), (330.0, 140.0), (420.0, 60.0)]
|
||||
|
||||
_RUNNER = """
|
||||
import { readFileSync, writeFileSync } from "node:fs";
|
||||
import * as ed from "./D03_PlanAnalysis/logic/D03_PlanAnalysis_Edit.js";
|
||||
import * as plan from "./D03_PlanAnalysis/logic/D03_PlanAnalysis_Plan.js";
|
||||
const job = JSON.parse(readFileSync(process.argv[2], "utf8"));
|
||||
const R = job.minR;
|
||||
const base = ed.editFromRoute(job.route, "gentle", "f1");
|
||||
const step = (s) =>
|
||||
"blocked" in s ? { blocked: s.blocked } : { edit: s.edit, line: s.line.vertices };
|
||||
const moved = ed.moveNode(base, 2, [215, 150], R);
|
||||
const handle = ed.moveHandle(base, 1, "end", job.handleTo, R);
|
||||
const inserted = ed.insertNode(base, 0, [60, 5], R);
|
||||
const removed = ed.removeNode(base, 1, R);
|
||||
const out = {
|
||||
base,
|
||||
baseLine: ed.buildEdit(base, R).line.vertices,
|
||||
moved: step(moved),
|
||||
handle: step(handle),
|
||||
inserted: step(inserted),
|
||||
removed: step(removed),
|
||||
endLocked: step(ed.moveNode(base, 0, [5, 5], R)),
|
||||
endRemove: step(ed.removeNode(base, base.nodes.length - 1, R)),
|
||||
movedLength: "edit" in moved ? plan.lineLength(moved.line.vertices) : null,
|
||||
};
|
||||
const edits = { 2026: { mode: "gentle", planned: out.moved.line } };
|
||||
const results = plan.withEdits(job.results, edits);
|
||||
out.rows = plan.chainRows(results, {}, 1000);
|
||||
out.line = plan.joinedLine(results, out.rows);
|
||||
writeFileSync(process.argv[3], JSON.stringify(out));
|
||||
"""
|
||||
|
||||
|
||||
def _route() -> dict:
|
||||
line = build_planned_polyline(list(NODES), min_radius_m=MIN_R, simplify=False)
|
||||
return {
|
||||
"planned": [list(v) for v in line.vertices],
|
||||
"nodes": [n.as_dict() for n in line.nodes],
|
||||
"curves": [c.as_dict() for c in line.curves],
|
||||
"metrics": {"length_m": 999.0},
|
||||
}
|
||||
|
||||
|
||||
def _results(route: dict) -> list:
|
||||
end = list(NODES[-1])
|
||||
return [
|
||||
{
|
||||
"year": 2026,
|
||||
"start": list(NODES[0]),
|
||||
"end": end,
|
||||
"fingerprint": "f1",
|
||||
"criteria": {"min_radius_m": MIN_R, "max_grade_pct": 14.0},
|
||||
"routes": {"gentle": route, "earthwork": {"error": "길 없음"}},
|
||||
},
|
||||
{
|
||||
"year": 2027,
|
||||
"start": end,
|
||||
"end": [600.0, 60.0],
|
||||
"routes": {"gentle": {"planned": [end, [600.0, 60.0]], "metrics": {"length_m": 180.0}}},
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def ts(tmp_path_factory) -> dict:
|
||||
tmp = tmp_path_factory.mktemp("543")
|
||||
out = tmp / "js"
|
||||
subprocess.run( # noqa: S603 — 고정 실행 파일
|
||||
[
|
||||
"node",
|
||||
str(TSC),
|
||||
str(TS_FILE),
|
||||
str(ROOT / "D03_PlanAnalysis" / "logic" / "D03_PlanAnalysis_Plan.ts"),
|
||||
"--outDir",
|
||||
str(out),
|
||||
"--rootDir",
|
||||
str(ROOT),
|
||||
"--module",
|
||||
"esnext",
|
||||
"--target",
|
||||
"es2022",
|
||||
"--moduleResolution",
|
||||
"bundler",
|
||||
"--ignoreConfig",
|
||||
],
|
||||
cwd=str(ROOT),
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
for emitted in out.rglob("*.js"):
|
||||
emitted.write_text(
|
||||
re.sub(r'(from\s+"\.\.?/[^"]+)"', r'\1.js"', emitted.read_text(encoding="utf-8")),
|
||||
encoding="utf-8",
|
||||
)
|
||||
(out / "runner.mjs").write_text(_RUNNER, encoding="utf-8")
|
||||
route = _route()
|
||||
job = {"minR": MIN_R, "route": route, "results": _results(route), "handleTo": None}
|
||||
# 손잡이 = 첫 곡선 끝 접선점을 나가는 직선 따라 교각점에서 1.8 배 자리로(접선 길이 ↑ = R ↑)
|
||||
curve = next(c for c in route["curves"] if c["node_first"] == 1)
|
||||
apex, end = curve["apex"], curve["end"]
|
||||
job["handleTo"] = [apex[0] + 1.8 * (end[0] - apex[0]), apex[1] + 1.8 * (end[1] - apex[1])]
|
||||
(tmp / "job.json").write_text(json.dumps(job), encoding="utf-8")
|
||||
subprocess.run( # noqa: S603
|
||||
["node", str(out / "runner.mjs"), str(tmp / "job.json"), str(tmp / "out.json")],
|
||||
cwd=str(ROOT),
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
return {**json.loads((tmp / "out.json").read_text(encoding="utf-8")), "route": route}
|
||||
|
||||
|
||||
def _close(a: list, b: list, tol: float = 1e-3) -> bool:
|
||||
return len(a) == len(b) and all(
|
||||
abs(p[0] - q[0]) <= tol and abs(p[1] - q[1]) <= tol for p, q in zip(a, b)
|
||||
)
|
||||
|
||||
|
||||
def test_ends_kept_and_locked(ts) -> None:
|
||||
for name in ("moved", "handle", "inserted", "removed"):
|
||||
step = ts[name]
|
||||
assert "edit" in step, (name, step)
|
||||
assert step["edit"]["nodes"][0] == list(NODES[0])
|
||||
assert step["edit"]["nodes"][-1] == list(NODES[-1])
|
||||
assert _close([step["line"][0], step["line"][-1]], [list(NODES[0]), list(NODES[-1])])
|
||||
assert "연도 경계" in ts["endLocked"]["blocked"]
|
||||
assert "연도 경계" in ts["endRemove"]["blocked"]
|
||||
count = len(ts["base"]["nodes"])
|
||||
assert len(ts["inserted"]["edit"]["nodes"]) == count + 1
|
||||
assert len(ts["removed"]["edit"]["nodes"]) == count - 1
|
||||
# 손잡이 = 그 마디 R 이 바뀜 · 법정 최소 밑 아님
|
||||
assert ts["handle"]["edit"]["radii"][1] >= MIN_R
|
||||
assert ts["handle"]["edit"]["radii"][1] > ts["base"]["radii"][1] * 1.5
|
||||
|
||||
|
||||
def test_unedited_base_redraws_engine_line(ts) -> None:
|
||||
assert _close(ts["baseLine"], ts["route"]["planned"])
|
||||
|
||||
|
||||
def test_length_and_chain_mirror(ts) -> None:
|
||||
line = ts["moved"]["line"]
|
||||
assert abs(ts["movedLength"] - analyze.line_length(line)) < 1e-6
|
||||
results = analyze.with_edits(_results(ts["route"]), {2026: {"mode": "gentle", "planned": line}})
|
||||
rows = analyze.chain(results, {}, 1000.0)
|
||||
assert ts["rows"] == rows
|
||||
assert rows[0]["length_m"] == round(analyze.line_length(line), 3)
|
||||
assert rows[1]["start_chainage_m"] == rows[0]["end_chainage_m"]
|
||||
assert ts["line"] == analyze.joined_line(results, rows)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("node_path", [True, False])
|
||||
def test_browser_line_equals_server_line(ts, monkeypatch, node_path) -> None:
|
||||
if not node_path:
|
||||
monkeypatch.setattr(edit_mod, "node_polyline", lambda *a, **k: None)
|
||||
for name in ("moved", "handle", "inserted", "removed"):
|
||||
step = ts[name]
|
||||
server = edit_mod.edited_line(step["edit"], MIN_R)
|
||||
assert _close([list(v) for v in server.vertices], step["line"]), name
|
||||
|
||||
|
||||
def test_check_edit_rejects() -> None:
|
||||
result = _results(_route())[0]
|
||||
good = {
|
||||
"base_mode": "gentle",
|
||||
"base_fingerprint": "f1",
|
||||
"nodes": [list(p) for p in NODES],
|
||||
"curve_flags": [True] * len(NODES),
|
||||
"radii": [None] * len(NODES),
|
||||
}
|
||||
edit_mod.check_edit(good, result, "gentle")
|
||||
for bad, mode in (
|
||||
({**good, "base_fingerprint": "old"}, "gentle"),
|
||||
(good, "earthwork"),
|
||||
({**good, "nodes": [[1.0, 0.0], *good["nodes"][1:]]}, "gentle"),
|
||||
({**good, "radii": [None, -3.0, None, None, None]}, "gentle"),
|
||||
({**good, "curve_flags": [True]}, "gentle"),
|
||||
):
|
||||
with pytest.raises(edit_mod.EditError):
|
||||
edit_mod.check_edit(bad, result, mode)
|
||||
Reference in New Issue
Block a user