eomsangdonandClaude Opus 5.5
4753d88983
- 분석본 = route_<연도>_<지점 열쇠>.json · GET 은 DB 지점 분석본 · 끈 지점 GET 은 그 분석 - [저장] 뒤 정본 = 끈 지점 분석본(DB 지점 따라감) - 쌓임 거르기 = 연도마다 최근 3 개 + 저장 지점 것 - 옛 꼴 route_<연도>.json 은 읽기만(공유 storage · 임시 배포 옛 판 서버가 읽음 — 옮기지 않음) - 시험 test_530_2 끈 지점 분석 · 저장 · 쌓임 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
401 lines
16 KiB
Python
401 lines
16 KiB
Python
"""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
|