Merge remote-tracking branch 'origin/dev' into sub_laptop_4

# Conflicts:
#	resources/tester/test_530_3_plan_logic.py
This commit is contained in:
2026-10-27 20:17:00 +09:00
12 changed files with 1055 additions and 1618 deletions
@@ -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
-169
View File
@@ -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,436 @@
"""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`).
전체 시점 · 종점(`base`) = D02 정본(`plan_base.load_base`) — 비어 옛 연도 행을 접어 보이던 것은
[저장] 전에 칸에 굳힘(연도 행이 바뀌어도 그대로).
계산 자리 = ① 파이썬·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 D02_PlanBase import D02_PlanBase_Router as plan_base
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, base: dict[str, Any], year_start: int | None
) -> dict[str, Any]:
return {
"status": "success",
"base": base,
"year_start": year_start,
"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]:
base = await plan_base.load_base(project_id)
async with get_db_pool().acquire() as connection:
network, legacy = await _load(connection, project_id)
async with connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"SELECT project_year FROM projects WHERE id = %s", (str(project_id),)
)
row = await cursor.fetchone()
try:
return _view(network, legacy, base, row["project_year"] if row else None)
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,),
)
async def _freeze_base(connection: Any, project_id: UUID, base: dict[str, Any]) -> None:
"""옛 연도 행을 접은 시점 · 종점 → D02 정본 칸(비었을 때만)."""
stored = {
"epsg": base["epsg"],
**{key: {"x": base[key]["x"], "y": base[key]["y"]} for key in ("start", "end")},
}
async with connection.cursor() as cursor:
await cursor.execute(
"UPDATE projects SET plan_base_json = %s WHERE id = %s AND plan_base_json IS NULL",
(json.dumps(stored), str(project_id)),
)
@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()
base = await plan_base.load_base(project_id) # 예정노선 프로젝트가 아니면 404 · 400
if base["epsg"] is not None and int(base["epsg"]) != network["epsg"]:
raise HTTPException(status_code=400, detail="좌표계가 기준 지정과 다릅니다.")
async with get_db_pool().acquire() as connection:
async with connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"SELECT project_year, plan_base_json FROM projects WHERE id = %s",
(str(project_id),),
)
project = await cursor.fetchone()
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:
if project["plan_base_json"] is None and base["start"] and base["end"]:
await _freeze_base(connection, project_id, base)
await _store(connection, project_id, network, derived)
await connection.commit()
except Exception:
await connection.rollback()
raise
return {**_view(network, False, base, project["project_year"]), "derived": derived}
+64 -324
View File
@@ -1,17 +1,12 @@
"""D 예정노선 2단계 분석 API (요구 345 · PLAN 530-1 · 530-2).
"""D 예정노선 2단계 지형 API (요구 345 · 389 · PLAN 530-1 · 561-2).
- `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 /api/projects/{id}/plan/terrain/extent` — 받기 전 범위(상자 · 도엽 · 칸 수 · 받은 범위)
- `POST …/plan/terrain` — 사업 전체 지형 한 벌 받기(뒤에서 · 누를 때만) · 끝나면 `surface_models` 행
- `GET …/plan/progress` — 일의 진행률
지점 정본은 `plan_route_years` 시점 · 종점 칸(529-3 몫). D03 화면에서 끈 지점(캐시 · PLAN 530-3)은
`points`(작업 좌표계 m)로 받아 그 지점으로 받기 · 분석 · 보기 · [저장] 때만 D02 규칙으로 정본에 씀.
계산 자리 = ③ 서버 단독 — [저장]은 노선을 받지 않고 갈래 이름만 받아 서버 저장본으로 다시 셈.
꺾임점 고침(543)은 마디 · 켬끔 · R 만 받아 서버가 선 · 지표를 다시 셈(`D03_PlanAnalysis_Edit`) ·
GET 은 분석 지문이 같은 저장 고침만 돌려줌.
범위 = 전체 시점 · 종점(D02 정본 `load_base`) + 그은 점(`extra`) 상자 + 여유.
노선 정본(손 그은 선)은
`D03_PlanAnalysis_Network`(PLAN 562-2 · 옛 추천 엔진 · 꺾임점 R 걷음).
"""
from __future__ import annotations
@@ -23,18 +18,13 @@ 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__)
@@ -46,98 +36,35 @@ _TASKS: set[asyncio.Task] = set()
KINDS = ("terrain", "analyze")
class PlanPoint(BaseModel):
"""끈 지점(작업 좌표계 m) — 연도 하나의 시점 · 종점."""
class TerrainBody(BaseModel):
"""지형 받기 — 여유(m · 빈 값 = 처음 값) · 더 덮을 점(그은 꺾임점 · 작업 좌표계 m)."""
year: int
start: tuple[float, float]
end: tuple[float, float]
margin_m: float | None = None
extra: list[tuple[float, float]] = []
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}
if set(by_year) != {seg["year"] for seg in segments}:
raise HTTPException(status_code=400, detail="끈 지점의 연도가 사업연도와 다릅니다.")
out = [
{
**seg,
"start": tuple(by_year[seg["year"]].start),
"end": tuple(by_year[seg["year"]].end),
"saved": (seg["start"], seg["end"]), # 저장 지점 — 그 분석본은 쌓임 거르기에서 남김
}
for seg in segments
]
for prev, seg in zip(out, out[1:]):
if max(abs(a - b) for a, b in zip(prev["end"], seg["start"])) > base.SAME_M:
raise HTTPException(
status_code=400, detail=f"{seg['year']}년 시점이 직전 종점과 다릅니다."
)
return out
async def _plan_inputs(
project_id: UUID, points: list[PlanPoint] | None = None
) -> tuple[Path, dict[str, Any], list[dict[str, Any]]]:
"""(실경로, 프로젝트 칸, 연도 구간) — 빈 연도 · 좌표계 섞임 거절 · `points` = 끈 지점."""
async def _plan_ends(project_id: UUID) -> tuple[Path, int, list[tuple[float, float]]]:
"""(실경로, 좌표계, [전체 시점, 전체 종점]) — D02 정본(빈 칸이면 옛 연도 행 접음)."""
plan = await base.load_base(project_id)
if not plan["start"] or not plan["end"]:
raise HTTPException(
status_code=400, detail="전체 시점 · 종점이 없습니다 — 1단계에서 지정하세요."
)
async with get_db_pool().acquire() as connection:
stored = await get_project_storage_relative_path(connection, project_id)
async with connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"SELECT road_type, design_speed_kph, route_start_m, route_end_m FROM projects"
" WHERE id = %s",
(str(project_id),),
)
project = await cursor.fetchone()
await cursor.execute(
"SELECT year, start_x, start_y, end_x, end_y, epsg, route_json, fingerprint "
"FROM plan_route_years WHERE project_id = %s ORDER BY year",
(str(project_id),),
)
rows = await cursor.fetchall()
if not stored or not project:
if not stored:
raise HTTPException(status_code=404, detail="프로젝트 저장 경로를 찾을 수 없습니다.")
if not rows:
raise HTTPException(
status_code=400, detail="연도별 지점이 없습니다 — 1단계에서 지정하세요."
)
for row in rows:
if any(row[k] is None for k in ("start_x", "start_y", "end_x", "end_y", "epsg")):
raise HTTPException(status_code=400, detail=f"{row['year']}년 지점이 비었습니다.")
if len({int(row["epsg"]) for row in rows}) != 1:
raise HTTPException(status_code=400, detail="연도마다 좌표계가 다릅니다.")
segments = [
{
"year": int(row["year"]),
"start": (float(row["start_x"]), float(row["start_y"])),
"end": (float(row["end_x"]), float(row["end_y"])),
"epsg": int(row["epsg"]),
"route_json": row["route_json"],
}
for row in rows
]
if points:
segments = _override(segments, points)
return Path(resolve_stored_project_path(stored)), project, segments
ends = [(float(p["x"]), float(p["y"])) for p in (plan["start"], plan["end"])]
return Path(resolve_stored_project_path(stored)), int(plan["epsg"]), ends
def _query_xy(extra: str | None) -> list[tuple[float, float]]:
if not extra:
return []
try:
return [(float(x), float(y)) for x, y in json.loads(extra)]
except (ValueError, TypeError) as error:
raise HTTPException(status_code=400, detail="extra 꼴이 다릅니다.") from error
def _start_job(project_id: UUID, kind: str, work: Any) -> dict[str, Any]:
@@ -166,62 +93,59 @@ def _start_job(project_id: UUID, kind: str, work: Any) -> dict[str, Any]:
return {"status": "accepted", **job}
async def _save_models(project_id: UUID, years: dict[int, list[dict[str, Any]]], epsg: int) -> None:
"""연도별 지형면 → `surface_models` 행(같은 파일 · 같은 지문이면 id 살림 · 다른 행 그대로)."""
async def _save_models(project_id: UUID, models: list[dict[str, Any]], epsg: int) -> None:
"""지형면 → `surface_models` 행(같은 파일 · 같은 지문이면 id 살림 · 다른 행 그대로)."""
from B04_PreProcess.B04_PreProcess_Repository import _existing_models, _upsert_surface_model
async with get_db_pool().acquire() as connection:
await connection.begin()
try:
existing = await _existing_models(connection, project_id)
for models in years.values():
for model in models:
await _upsert_surface_model(
connection,
existing,
project_id=project_id,
model=model,
source_file_id=None,
processed_cloud_id=None,
crs_epsg=epsg,
)
for model in models:
await _upsert_surface_model(
connection,
existing,
project_id=project_id,
model=model,
source_file_id=None,
processed_cloud_id=None,
crs_epsg=epsg,
)
await connection.commit()
except Exception:
await connection.rollback()
raise
@router.get("/{project_id}/plan/terrain/extent", response_model=None)
async def plan_terrain_extent(
project_id: UUID, margin_m: float | None = None, extra: str | None = None
) -> dict[str, Any]:
"""받기 전 범위 — 전체 시점 · 종점(+ `extra` 그은 점 `[[x, y]]`) 상자 + 여유 ∪ 받은 범위."""
root, epsg, ends = await _plan_ends(project_id)
view = await asyncio.to_thread(
terrain.plan_extent, root, [*ends, *_query_xy(extra)], epsg, margin_m
)
return {"status": "success", **view}
@router.post("/{project_id}/plan/terrain", response_model=None)
async def plan_terrain(project_id: UUID, body: PointsBody | None = None) -> dict[str, Any]:
"""지점 기준 수치지형도 · 위성 · 연도별 DEM 받기(뒤에서) — 진행은 `GET …/plan/progress`."""
root, _, segments = await _plan_inputs(project_id, body.points if body else None)
epsg = segments[0]["epsg"]
async def plan_terrain(project_id: UUID, body: TerrainBody | None = None) -> dict[str, Any]:
"""사업 전체 수치지형도 · 위성 · DEM 받기(뒤에서) — 진행은 `GET …/plan/progress`."""
body = body or TerrainBody()
root, epsg, ends = await _plan_ends(project_id)
points = [*ends, *body.extra]
async def work(report: Any) -> None:
got = await asyncio.to_thread(terrain.ensure_plan_terrain, root, segments, epsg, report)
await _save_models(project_id, got["years"], epsg)
got = await asyncio.to_thread(
terrain.ensure_plan_terrain, root, points, epsg, body.margin_m, report
)
await _save_models(project_id, got["models"], epsg)
report(100, f"지형 받기 끝 — 새 도엽 {len(got['sheets']['downloaded'])}매")
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 +153,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}
+139 -69
View File
@@ -1,12 +1,14 @@
"""D 예정노선 지형 받기 — 지점 기준 수치지형도 · 위성 · 연도 구간별 DEM (요구 345 · PLAN 530-1).
"""D 예정노선 지형 받기 — 사업 전체 한 벌(도엽 · 위성 · DEM) (요구 345 · 389 · PLAN 530-1 · 561-2).
지형은 D 프로젝트 storage 안 `B04_PreProcess/` 배치 그대로(지형 코드 주인 = B04 · 고칠 곳 0):
- 도엽 = 모든 지점 + 구간 표본(`SAMPLE_STEP_M`) 각 3×3 → `ensure_sheets`(없는 것만 · prune 없음)
- 범위 = 전체 시점 · 종점(+ 그은 점) 상자 + 여유 · 받은 범위가 있으면 그 범위와 합침(줄지 않음)
- 도엽 = 범위 상자에 닿는 도엽 모두 → `ensure_sheets`(없는 것만 · prune 없음)
- 병합 = 지난번 병합 도엽 ∪ 이번 도엽(옛 도엽 유지 · 열쇠 같으면 건너뜀)
- 위성 = 기준 도엽 도곽 ∪ 옛 위성 범위(덮으면 건너뜀)
- DEM = 연도 구간마다 `models_<연도>/`(긴 노선 메모리 · 비용면 한도) · 범위 = 구간 상자 + 탐색 복도
- 위성 = 범위 도엽 도곽 ∪ 옛 위성 범위(덮으면 건너뜀)
- DEM = 한 벌 `models_plan/`(연도는 D03 나눔 뒤에 정해짐) · 받은 범위 = `plan_terrain.json`
- 작업 좌표계 = 지점 epsg · `B03_FileInput/input/prj/plan.prj` 한 장(B04 · B05 좌표계 창구가 읽음)
- 받기 전 범위 보기 = `plan_extent`(내려받기 없음) · 받기는 사용자가 누를 때만
계산 자리 = ③ 서버 단독. DB 행(`surface_models`)은 라우터가 씀.
"""
@@ -22,7 +24,7 @@ from typing import Any, Callable
import numpy as np
from B04_PreProcess.B04_PreProcess_Engine_MapSheet import (
neighbors_for_points,
SHEET5K_SIZE_DEG,
sheet5k_to_bounds,
sheets_for_points,
)
@@ -35,50 +37,119 @@ from B04_PreProcess.B04_PreProcess_Engine_SheetStore import (
from B04_PreProcess.B04_PreProcess_Engine_SheetSurface import build_sheet_surface_model
from B04_PreProcess.B04_PreProcess_Engine_VWorld import download_vworld_satellite_map
from common_util.common_util_json import atomic_write_json
from config.config_system_terrain import SHEET_SURFACE_DEFAULT_METHOD, SHEET_SURFACE_MARGIN_M
from config.config_system_terrain import (
SHEET_SURFACE_DEFAULT_METHOD,
SHEET_SURFACE_GRID_M,
SHEET_SURFACE_MARGIN_M,
)
logger = logging.getLogger(__name__)
# 구간 표본 간격 — 도엽(≈2.8 × 2.3 km) 3×3 묶음이 끊기지 않게
SAMPLE_STEP_M = 1000.0
# 탐색 복도 반폭(D 만 · B 기본 100 m) — 실측(530-2): 100 · 200 m 는 못 지킨 곳 남음 · 300 m 0~1
PLAN_CORRIDOR_M = 300.0
# DEM 이 복도 밖으로 더 덮는 폭(복도 끝 칸 기울기 · 횡단 표본 ±20 m)
DEM_EXTRA_M = 50.0
# 여유 처음 값 = 옛 연도 구간 DEM 폭(탐색 복도 300 m + 횡단 표본 50 m) · 바닥 = B 절취 여유
DEFAULT_MARGIN_M = 350.0
MAX_MARGIN_M = 5000.0
STAGE = Path("B04_PreProcess")
PLAN_MODELS = "models_plan"
TERRAIN_META = "plan_terrain.json"
PRJ_NAME = "plan.prj"
Progress = Callable[[int, str], None]
Box = tuple[float, float, float, float] # x0, y0, x1, y1(작업 좌표계 m)
XY = tuple[float, float]
def models_dir_of(project_root: Path, year: int) -> Path:
"""연도 구간 DEM 자리."""
return project_root / STAGE / f"models_{int(year)}"
def models_dir_of(project_root: Path) -> Path:
"""사업 전체 DEM 자리."""
return project_root / STAGE / PLAN_MODELS
def segment_samples(
start: tuple[float, float], end: tuple[float, float]
) -> list[tuple[float, float]]:
"""구간 위 표본(끝점 둘 포함 · `SAMPLE_STEP_M` 간격 이하)."""
length = math.hypot(end[0] - start[0], end[1] - start[1])
count = max(1, math.ceil(length / SAMPLE_STEP_M))
return [
(start[0] + (end[0] - start[0]) * i / count, start[1] + (end[1] - start[1]) * i / count)
for i in range(count + 1)
def clamp_margin(margin_m: float | None) -> float:
"""여유 — 빈 값 = 처음 값 · B 절취 여유 ~ 5 km."""
if margin_m is None or not math.isfinite(margin_m):
return DEFAULT_MARGIN_M
return min(MAX_MARGIN_M, max(SHEET_SURFACE_MARGIN_M, float(margin_m)))
def received_box(project_root: Path) -> Box | None:
"""받은 DEM 범위 — 받은 적 없거나 읽지 못하면 None."""
meta = _read_json(models_dir_of(project_root) / TERRAIN_META) or {}
try:
x0, y0, x1, y1 = (float(v) for v in meta["box"])
except (KeyError, TypeError, ValueError):
return None
return x0, y0, x1, y1
def wanted_box(points: list[XY], margin_m: float, have: Box | None) -> Box:
"""받을 범위 = 점 상자 + 여유 ∪ 받은 범위."""
xs, ys = [p[0] for p in points], [p[1] for p in points]
box = (min(xs) - margin_m, min(ys) - margin_m, max(xs) + margin_m, max(ys) + margin_m)
if have is None:
return box
return min(box[0], have[0]), min(box[1], have[1]), max(box[2], have[2]), max(box[3], have[3])
def covers(have: Box | None, box: Box) -> bool:
return (
have is not None
and all(have[i] <= box[i] for i in (0, 1))
and all(have[i] >= box[i] for i in (2, 3))
)
def box_ring(box: Box, epsg: int) -> list[list[float]]:
"""상자 네 귀 경위도([경도, 위도] · 왼아래부터 반시계)."""
from common_util.common_util_crs import tm_to_lonlat
x0, y0, x1, y1 = box
return [list(tm_to_lonlat(x, y, epsg)) for x, y in ((x0, y0), (x1, y0), (x1, y1), (x0, y1))]
def sheets_for_box(box: Box, epsg: int) -> list[str]:
"""상자에 닿는 도엽 — 경위도 외접 상자를 도엽 반 칸 간격으로 짚음."""
ring = box_ring(box, epsg)
lons, lats = [p[0] for p in ring], [p[1] for p in ring]
step = SHEET5K_SIZE_DEG / 2
def ticks(lo: float, hi: float) -> list[float]:
return [lo + step * i for i in range(int((hi - lo) / step) + 1)] + [hi]
grid = [
(lat, lon) for lat in ticks(min(lats), max(lats)) for lon in ticks(min(lons), max(lons))
]
return sorted(set(sheets_for_points(grid)))
def reference_points(segments: list[dict[str, Any]], epsg: int) -> list[tuple[float, float]]:
"""도엽 기준 좌표(위도, 경도) — 모든 지점 + 구간 표본."""
from pyproj import Transformer
to_wgs = Transformer.from_crs(f"EPSG:{epsg}", "EPSG:4326", always_xy=True)
points: list[tuple[float, float]] = []
for seg in segments:
for x, y in segment_samples(seg["start"], seg["end"]):
lon, lat = to_wgs.transform(x, y)
points.append((lat, lon))
return points
def plan_extent(
project_root: Path, points: list[XY], epsg: int, margin_m: float | None = None
) -> dict[str, Any]:
"""받기 전 범위 보기 — 내려받기 없음. `points` = 전체 시점 · 종점(+ 그은 점 · 작업 좌표계 m)."""
margin = clamp_margin(margin_m)
have = received_box(project_root)
box = wanted_box(points, margin, have)
store = get_project_map_sheets_dir(project_root)
sheets = []
for no in sheets_for_box(box, epsg):
lon0, lat0, lon1, lat1 = sheet5k_to_bounds(no)
sheets.append(
{
"no": no,
"ring": [[lon0, lat0], [lon1, lat0], [lon1, lat1], [lon0, lat1]],
"stored": get_sheet_path(store, no) is not None,
}
)
return {
"epsg": int(epsg),
"margin_m": margin,
"box_m": list(box),
"box": box_ring(box, epsg),
"have_m": list(have) if have else None,
"have": box_ring(have, epsg) if have else None,
"covered": covers(have, box),
"sheets": sheets,
# DEM 칸 수 어림 — 받기 시간 · 메모리 가늠
"cells": int((box[2] - box[0]) * (box[3] - box[1]) / SHEET_SURFACE_GRID_M**2),
}
def write_plan_prj(project_root: Path, epsg: int) -> Path:
@@ -93,15 +164,6 @@ def write_plan_prj(project_root: Path, epsg: int) -> Path:
return path
def dem_outline(start: tuple[float, float], end: tuple[float, float]) -> np.ndarray:
"""DEM 상자 꼭짓점 둘 — B 여유(`SHEET_SURFACE_MARGIN_M`)를 더하면 복도 + `DEM_EXTRA_M`."""
pad = max(0.0, PLAN_CORRIDOR_M + DEM_EXTRA_M - SHEET_SURFACE_MARGIN_M)
xs, ys = (start[0], end[0]), (start[1], end[1])
return np.array(
[[min(xs) - pad, min(ys) - pad], [max(xs) + pad, max(ys) + pad]], dtype=np.float64
)
def _read_json(path: Path) -> dict[str, Any] | None:
try:
return json.loads(path.read_text(encoding="utf-8"))
@@ -158,29 +220,30 @@ def _satellite(processed: Path, prj: Path, points: list[tuple[float, float]], ep
def ensure_plan_terrain(
project_root: Path,
segments: list[dict[str, Any]],
points: list[XY],
epsg: int,
margin_m: float | None = None,
on_progress: Progress | None = None,
) -> dict[str, Any]:
"""지점 기준 지형 받기 — `segments` = [{year, start: (x, y), end: (x, y)}](작업 좌표계 m).
"""사업 전체 지형 받기 — `points` = 전체 시점 · 종점(+ 그은 점 · 작업 좌표계 m).
돌려줌 = 도엽(받은 · 실패 · 병합) · 위성 상태 · 연도별 모델 등록용 dict 목록.
돌려줌 = 도엽(받은 · 실패 · 병합) · 위성 상태 · 받은 범위 · 모델 등록용 dict 목록.
"""
def report(percent: int, message: str) -> None:
if on_progress is not None:
on_progress(percent, message)
if not segments:
raise ValueError("연도 구간이 없습니다.")
if not points:
raise ValueError("시점 · 종점이 없습니다.")
processed = project_root / STAGE / "processed"
processed.mkdir(parents=True, exist_ok=True)
prj = write_plan_prj(project_root, epsg)
crs = f"EPSG:{int(epsg)}"
margin = clamp_margin(margin_m)
box = wanted_box(points, margin, received_box(project_root))
report(5, "수치지형도 도엽 고르는 중")
points = reference_points(segments, epsg)
wanted = neighbors_for_points(points)
wanted = sheets_for_box(box, epsg)
store = get_project_map_sheets_dir(project_root)
before = {s for s in wanted if get_sheet_path(store, s) is not None}
report(10, f"수치지형도 도엽 {len(wanted) - len(before)}매 받는 중")
@@ -190,24 +253,30 @@ def ensure_plan_terrain(
report(45, "도엽 레이어 병합 중")
merged = _merge_sheets(processed, store, list(got["available"]))
report(60, "위성 지도 받는 중")
satellite = _satellite(processed, prj, points, epsg)
satellite = _satellite(processed, prj, [(p[1], p[0]) for p in box_ring(box, epsg)], epsg)
years: dict[int, list[dict[str, Any]]] = {}
for i, seg in enumerate(segments):
year = int(seg["year"])
report(65 + 30 * i // len(segments), f"{year}년 구간 지형면 만드는 중")
models_dir = models_dir_of(project_root, year)
models_dir.mkdir(parents=True, exist_ok=True)
years[year] = build_sheet_surface_model(
project_root,
processed,
models_dir,
dem_outline(seg["start"], seg["end"]),
crs,
[SHEET_SURFACE_DEFAULT_METHOD],
report(65, "사업 지형면 만드는 중")
models_dir = models_dir_of(project_root)
models_dir.mkdir(parents=True, exist_ok=True)
# B 절취 = 넘긴 꼭짓점 상자 + `SHEET_SURFACE_MARGIN_M` → 그만큼 안쪽으로 넘김(여유 ≥ 그 값)
inner = SHEET_SURFACE_MARGIN_M
outline = np.array(
[[box[0] + inner, box[1] + inner], [box[2] - inner, box[3] - inner]], dtype=np.float64
)
models = build_sheet_surface_model(
project_root,
processed,
models_dir,
outline,
f"EPSG:{int(epsg)}",
[SHEET_SURFACE_DEFAULT_METHOD],
)
if not models:
raise ValueError(
"사업 지형면을 만들지 못했습니다 — 범위 안 도엽 등고선 높이 확인"
f"(병합 도엽 {len(merged)}매)"
)
if not years[year]:
raise ValueError(f"{year}년 구간 지형면을 만들지 못했습니다 — 도엽 등고선 확인")
atomic_write_json(models_dir / TERRAIN_META, {"box": list(box), "margin_m": margin})
report(100, "지형 받기 끝")
return {
"sheets": {
@@ -217,5 +286,6 @@ def ensure_plan_terrain(
"merged": merged,
},
"satellite": satellite,
"years": years,
"box_m": list(box),
"models": models,
}
-3
View File
@@ -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")
# 자동 전처리가 만드는 도엽 방식 — 기본 하나뿐이다. 여섯 방식을 매번 다 만들면
+37
View File
@@ -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);
+2
View File
@@ -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)
+63 -34
View File
@@ -1,10 +1,10 @@
"""D 예정노선 지형 받기(PLAN 530-1) — 가짜 내려받기 · 옛 도엽 유지 · 새 도엽만 · 빠짐 없음.
"""D 예정노선 지형 받기(PLAN 530-1 · 561-2) — 사업 전체 한 벌 · 범위 보기 = 내려받기 없음 ·
옛 도엽 유지 · 새 도엽만 · 범위는 줄지 않음 · 빠짐 없음.
./venv/Scripts/python.exe -m pytest resources/tester/test_530_1_plan_terrain.py -q
"""
import json
import math
from pathlib import Path
import pytest
@@ -60,56 +60,85 @@ def _zips(root):
return sorted(p.stem for p in store_mod.get_project_map_sheets_dir(root).glob("*.zip"))
def test_more_points_keep_old_sheets_and_fetch_only_new(fake) -> None:
def _dem_box(seen) -> list[float]:
"""가짜 DEM 이 받은 꼭짓점 상자 + B 절취 여유 = 실제 DEM 상자."""
(x0, y0), (x1, y1) = seen["dem"][-1][1]
m = SHEET_SURFACE_MARGIN_M
return [x0 - m, y0 - m, x1 + m, y1 + m]
def test_one_set_and_extend_keeps_old(fake) -> None:
root, seen = fake
first = terrain.ensure_plan_terrain(root, [{"year": 2026, "start": A, "end": B}], EPSG)
first = terrain.ensure_plan_terrain(root, [A, B], EPSG)
old = _zips(root)
assert sorted(seen["download"]) == old == sorted(first["sheets"]["downloaded"])
assert first["satellite"] == "done" and list(first["years"]) == [2026]
assert first["satellite"] == "done" and len(first["models"]) == 1
assert seen["dem"][-1][0] == "models_plan" # 연도 없는 한 벌
m = terrain.DEFAULT_MARGIN_M
box = [min(A[0], B[0]) - m, min(A[1], B[1]) - m, max(A[0], B[0]) + m, max(A[1], B[1]) + m]
assert _dem_box(seen) == pytest.approx(box) == first["box_m"]
assert terrain.received_box(root) == pytest.approx(tuple(box))
# 작업 좌표계 .prj — B 좌표계 창구가 읽음
assert resolve_project_crs(root) == f"EPSG:{EPSG}"
# 그은 점이 범위 밖 — 더 받기 = 옛 범위 ∪ 새 범위 · 새 도엽만
seen["download"].clear()
segs = [{"year": 2026, "start": A, "end": B}, {"year": 2027, "start": B, "end": C}]
second = terrain.ensure_plan_terrain(root, segs, EPSG)
second = terrain.ensure_plan_terrain(root, [A, B, C], EPSG)
now = _zips(root)
assert set(old) <= set(now) # 옛 도엽 그대로(prune 없음)
assert sorted(seen["download"]) == sorted(set(now) - set(old)) # 새 도엽만 받음
assert sorted(seen["download"]) == sorted(set(now) - set(old))
assert seen["merge"][-1] == now # 병합 = 옛 ∪ 새
# 위성 = 옛 범위 ∪ 새 범위
first_box, union = seen["satellite"][0], seen["satellite"][-1]
assert union["x"][0] <= first_box["x"][0] and union["x"][1] >= first_box["x"][1]
assert union["y"][0] <= first_box["y"][0] and union["y"][1] >= first_box["y"][1]
assert [d[0] for d in seen["dem"][-2:]] == ["models_2026", "models_2027"]
grown = second["box_m"]
assert grown[0] <= box[0] and grown[1] <= box[1] and grown[2] >= C[0] + m
first_sat, union = seen["satellite"][0], seen["satellite"][-1]
assert union["x"][0] <= first_sat["x"][0] and union["x"][1] >= first_sat["x"][1]
assert second["sheets"]["failed"] == {}
# 같은 지점 다시 — 받기 0 · 위성 건너뜀 · 병합 열쇠 같음
# 같은 범위 다시 — 받기 0 · 위성 건너뜀 · 병합 열쇠 같음 · 작은 점이어도 범위 줄지 않음
seen["download"].clear()
merges = len(seen["merge"])
third = terrain.ensure_plan_terrain(root, segs, EPSG)
third = terrain.ensure_plan_terrain(root, [A, B], EPSG)
assert seen["download"] == [] and third["satellite"] == "skip"
assert len(seen["merge"]) == merges
assert len(seen["merge"]) == merges and third["box_m"] == grown
def test_far_points_leave_no_gap(fake) -> None:
def test_extent_no_download_and_covered(fake) -> None:
root, seen = fake
before = terrain.plan_extent(root, [A, B], EPSG)
assert seen["download"] == [] and before["have"] is None and before["covered"] is False
assert before["margin_m"] == terrain.DEFAULT_MARGIN_M and len(before["box"]) == 4
x0, y0, x1, y1 = before["box_m"]
assert before["cells"] == int((x1 - x0) * (y1 - y0))
assert all(not s["stored"] and len(s["ring"]) == 4 for s in before["sheets"])
terrain.ensure_plan_terrain(root, [A, B], EPSG)
after = terrain.plan_extent(root, [A, B], EPSG)
assert after["covered"] and after["have_m"] == after["box_m"]
assert all(s["stored"] for s in after["sheets"])
assert {s["no"] for s in after["sheets"]} == {s["no"] for s in before["sheets"]}
assert not terrain.plan_extent(root, [A, C], EPSG)["covered"] # 그은 점이 밖 = 더 받기
# 여유 — 바닥 B 절취 여유 · 빈 값 = 처음 값
assert terrain.clamp_margin(100.0) == SHEET_SURFACE_MARGIN_M
assert terrain.clamp_margin(None) == terrain.DEFAULT_MARGIN_M
assert terrain.clamp_margin(1e9) == terrain.MAX_MARGIN_M
def test_box_sheets_leave_no_gap(fake) -> None:
root, _ = fake
far = (A[0] + 9000.0, A[1] + 7000.0) # 11 km — 3×3 두 벌이 안 닿는 거리
got = terrain.ensure_plan_terrain(root, [{"year": 2026, "start": A, "end": far}], EPSG)
far = (A[0] + 9000.0, A[1] + 7000.0) # 11 km
got = terrain.ensure_plan_terrain(root, [A, far], EPSG)
x0, y0, x1, y1 = got["box_m"]
to_wgs = Transformer.from_crs(f"EPSG:{EPSG}", "EPSG:4326", always_xy=True)
on_line = []
for i in range(201):
lon, lat = to_wgs.transform(
A[0] + (far[0] - A[0]) * i / 200, A[1] + (far[1] - A[1]) * i / 200
)
on_line.append((lat, lon))
assert set(sheets_for_points(on_line)) <= set(got["sheets"]["merged"])
inside = []
for i in range(41):
for j in range(41):
lon, lat = to_wgs.transform(x0 + (x1 - x0) * i / 40, y0 + (y1 - y0) * j / 40)
inside.append((lat, lon))
assert set(sheets_for_points(inside)) <= set(got["sheets"]["merged"])
def test_dem_outline_covers_corridor() -> None:
box = terrain.dem_outline(A, B)
reach = terrain.PLAN_CORRIDOR_M + terrain.DEM_EXTRA_M
assert box[0][0] - SHEET_SURFACE_MARGIN_M <= min(A[0], B[0]) - reach + 1e-9
assert box[1][1] + SHEET_SURFACE_MARGIN_M >= max(A[1], B[1]) + reach - 1e-9
samples = terrain.segment_samples(A, C)
steps = [math.dist(p, q) for p, q in zip(samples, samples[1:])]
assert samples[0] == A and samples[-1] == C and max(steps) <= terrain.SAMPLE_STEP_M
def test_empty_surface_raises_and_keeps_no_box(fake, monkeypatch) -> None:
root, _ = fake
monkeypatch.setattr(terrain, "build_sheet_surface_model", lambda *a, **k: [])
with pytest.raises(ValueError, match="사업 지형면을 만들지 못했습니다"):
terrain.ensure_plan_terrain(root, [A, B], EPSG)
assert terrain.received_box(root) is None
+252 -142
View File
@@ -1,168 +1,278 @@
"""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, "plan_base_json": "{}"}]}
def acquire(self):
return _Connection(self)
@pytest.fixture
def run(tmp_path, hill):
_POOLS.clear()
calls = []
def client(monkeypatch):
db = _Db()
monkeypatch.setattr(net, "get_db_pool", lambda: db)
base = {"epsg": 5186, "start": {"x": 0.0, "y": 0.0}, "end": {"x": 90.0, "y": 180.0}}
def loader(root, selection, models_dir, cache_dir):
calls.append((models_dir.name, cache_dir.name))
return hill, []
async def load_base(project_id):
return base
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()
monkeypatch.setattr(net.plan_base, "load_base", load_base)
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
test.base = base
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
assert got["base"]["end"] == {"x": 90.0, "y": 180.0} and got["year_start"] == 2026
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
assert got.json()["year_start"] == 2026
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)]
assert not [s for s, _ in log if "plan_base_json = %s" in s] # 정본 칸 있음 = 안 굳힘
def test_put_freezes_folded_base_first(client) -> None:
client.db.answers["FROM projects"] = [{"project_year": 2026, "plan_base_json": None}]
assert client.put(URL, json=_network()).status_code == 200
sqls = [(s, a) for s, a in client.db.log if s.startswith("UPDATE projects SET plan_base_json")]
assert len(sqls) == 1 and json.loads(sqls[0][1][0]) == client.base
first_year = next(i for i, (s, _) in enumerate(client.db.log) if "plan_route_years" in s)
assert client.db.log.index(sqls[0]) < first_year
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")]
assert client.put(URL, json=_network(epsg=5187)).status_code == 400 # 기준 지정과 다른 좌표계
client.session["email"] = "a@gmail.com"
assert client.put(URL, json=_network()).status_code == 403
+62 -267
View File
@@ -1,303 +1,98 @@
"""D 분석 API(PLAN 530-2 · 530-3) — GET 이은 누가 · PUT [저장] = 서버 저장본으로 다시 셈 · 모르는 갈래
거절 · 끈 지점(points) = 그 지점으로 보기 · 안 이어지면 거절 · [저장] 때만 지점 정본(산림청만) ·
꺾임점 고침(543-3) = 서버가 마디로 선 · 지표를 다시 셈 · 누가가 고친 길이로 · GET 은 지문 같은 고침만.
"""D 지형 API(PLAN 530-1 · 561-2) — 범위 = 전체 시점 · 종점(D02 정본) 상자 + 여유 · 받기 = 한 벌.
옛 분석 API 시험(GET · PUT analysis · 꺾임점 고침)은 옛 추천 엔진과 함께 걷음(PLAN 562-2) ·
노선 정본 시험 = `test_530_2_plan_analyze.py` · 지형 셈 시험 = `test_530_1_plan_terrain.py`.
./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_router.py -q
"""
import asyncio
import json
import os
from contextlib import asynccontextmanager
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
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)
P0, P2 = (0.0, 0.0), (900.0, 400.0)
URL = f"/api/projects/{PID}/plan/terrain"
class _Pool:
def __init__(self, log):
self.log = log
def acquire(self):
return _Connection(self.log)
@asynccontextmanager
async def acquire(self):
yield None
@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,
plan = {
"epsg": 5186,
"start": {"lon": 0, "lat": 0, "x": P0[0], "y": P0[1]},
"end": {"lon": 0, "lat": 0, "x": P2[0], "y": P2[1]},
}
async def inputs(project_id, points=None):
return tmp_path, project, router_mod._override(segments, points) if points else segments
async def load_base(project_id):
return plan
log: list = []
monkeypatch.setattr(router_mod, "_plan_inputs", inputs)
monkeypatch.setattr(router_mod, "get_db_pool", lambda: _Pool(log))
async def stored(connection, project_id):
return "1/3/p"
monkeypatch.setattr(router_mod.base, "load_base", load_base)
monkeypatch.setattr(router_mod, "get_db_pool", lambda: _Pool())
monkeypatch.setattr(router_mod, "get_project_storage_relative_path", stored)
monkeypatch.setattr(router_mod, "resolve_stored_project_path", lambda s: str(tmp_path))
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
test.plan = plan
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_extent_uses_whole_ends(client) -> None:
got = client.get(f"{URL}/extent?margin_m=400").json()
m = 400.0
assert got["margin_m"] == m and got["have"] is None and got["covered"] is False
# 전체 시점 · 종점(D02 정본) — 연도 구간 아님
assert got["box_m"] == [P0[0] - m, P0[1] - m, P2[0] + m, P2[1] + m]
extra = client.get(f"{URL}/extent", params={"extra": json.dumps([[2000, 0]])}).json()
assert extra["box_m"][2] == 2000 + extra["margin_m"]
assert client.get(f"{URL}/extent", params={"extra": "[1]"}).status_code == 400
client.plan["end"] = None # 종점 빔 — 1단계로
assert client.get(f"{URL}/extent").status_code == 400
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_post_runs_one_set(client, monkeypatch) -> None:
calls: list = []
saved: list = []
works: list = []
def ensure(root, points, epsg, margin_m, report):
calls.append((points, epsg, margin_m))
return {"sheets": {"downloaded": ["1"]}, "models": [{"model_file_path": "m"}]}
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")]
async def save(project_id, models, epsg):
saved.append((models, epsg))
def start(project_id, kind, work):
works.append(work)
return {"status": "accepted", "kind": kind}
def _points(p1) -> list:
return [
{"year": 2026, "start": list(P0), "end": list(p1)},
{"year": 2027, "start": list(p1), "end": list(P2)},
]
def post(body):
res = client.post(URL, json=body)
asyncio.run(works.pop()(lambda percent, message: None))
return res
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"
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
monkeypatch.setattr(router_mod.terrain, "ensure_plan_terrain", ensure)
monkeypatch.setattr(router_mod, "_save_models", save)
monkeypatch.setattr(router_mod, "_start_job", start)
res = post({"margin_m": 500, "extra": [[5, 6]]})
assert res.json() == {"status": "accepted", "kind": "terrain"}
assert calls == [([P0, P2, (5.0, 6.0)], 5186, 500.0)]
assert saved == [([{"model_file_path": "m"}], 5186)]
# 옛 화면 꼴(points 몸) — 여유 처음 값 · 전체 시점 · 종점
post({"points": []})
assert calls[-1] == ([P0, P2], 5186, None)
-210
View File
@@ -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)