diff --git a/B05_Profile/B05_Profile_Engine_RouteInitial.py b/B05_Profile/B05_Profile_Engine_RouteInitial.py new file mode 100644 index 00000000..8c5cd017 --- /dev/null +++ b/B05_Profile/B05_Profile_Engine_RouteInitial.py @@ -0,0 +1,464 @@ +"""초기 계획노선 갈래 — 예상노선 추종 · 토공 최소 · 기울기 순한 노선 · 비중(PLAN 20장). + +계산 자리 = 서버 단독(③). 결과는 화면 캐시용 — 이 모듈은 파일을 쓰지 않는다. 적용은 화면이 +받은 노드로 기존 `POST /route/replan`([확인])을 부른다. + +갈래(`mode`): + · follow — 지금 로직 그대로: 예상노선 점 묶음을 폴리라인화(`build_planned_polyline`). + `B05_Profile_Router_Replan._ensure_planned_initial` 과 같은 입력 · 같은 결과. + · earthwork — 절 · 성토 합 최소(DEM 옆경사로 가늠한 단면 토량 + 상한 넘는 지반 기울기). + · gentle — 종단기울기 최소(지반 기울기 제곱 합). + · weighted — 예상노선 이격 · 토공 · 기울기 세 목표를 **정규화해 가중합**(합 100). + 정규화 = 목표마다 복도 안 링크 평균값으로 나눔 — 단위가 다른 세 값을 1 m 당 + 「보통 크기 1」로 맞춘 뒤 비중을 곱한다. +탐색 = 예상노선 둘레 복도(`CORRIDOR_M`) 안 `SEARCH_CELL_M` 격자의 8-연결 Dijkstra(scipy csgraph). +탐색 격자 경로는 follow 와 같은 폴리라인화(꺾임점 뽑기 + 원호 끼우기)를 지난다. + +공통 제약(모든 갈래) — 종단기울기 상한 · 최소 곡선반지름 · 계류 이격: + · 탐색 비용 — 상한 넘는 지반 기울기 링크 · 계류 버퍼 안 링크에 벌점(막지 않음 — 길이 없으면 + 노선이 끊기므로). + · 폴리라인화 — 최소 곡선반지름으로 원호를 끼우고 못 끼우면 위반 표시. + · 결과 — `metrics.over_limit` 에 제약별 초과 구간 수. + +⚠ 사용자 확정 대기 — 아래 임시값과 호출부가 넘기는 종단기울기 상한 · 최소 곡선반지름은 지식DB +(`resources/knowledge/technical_info/01_임도`) 후보일 뿐이다(PLAN 20장 기준값 후보 표). +확정되면 이 머리의 상수만 고친다. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import numpy as np +import shapely +from scipy import ndimage +from scipy.interpolate import RegularGridInterpolator +from scipy.sparse import csr_matrix +from scipy.sparse.csgraph import dijkstra +from shapely.geometry import LineString, MultiLineString + +from common_util.common_util_route_polyline import build_planned_polyline + +# ── 사용자 확정 대기(지식DB 후보 · PLAN 20장) ───────────────────────────────────── +# 계류 이격(m) — 법정 이격 수치 없음. 타당성평가 「산지계류 100 m 이내 구간 비율」(별표1 비고, +# 1/25,000 주요 수계)을 후보로 씀. +STREAM_OFFSET_M = 100.0 +# 이격을 따질 계류 — 도엽 하천중심선 `구분`. 「주요 수계」 가늠으로 세류 · 연결선은 뺌. +STREAM_CLASSES = ("소하천", "지방하천", "국가하천") +# 토공 가늠 폭(m) = 유효너비 3 + 길어깨 0.5 × 2(별표2 간선). +ROAD_WIDTH_M = 4.0 +# 탐색 복도 반폭(m) — 예상노선에서 이보다 멀리는 가지 않음(법령 값 아님 · 알고리즘 값). +CORRIDOR_M = 100.0 +PENDING = ("max_grade_pct", "min_radius_m", "stream_offset_m", "road_width_m", "corridor_m") + +# ── 탐색 · 측정 값 ─────────────────────────────────────────────────────────────── +MODES = ("follow", "earthwork", "gentle", "weighted") +DEFAULT_WEIGHTS = {"expected": 50.0, "earthwork": 25.0, "grade": 25.0} +MODE_WEIGHTS = { + "follow": {"expected": 100.0, "earthwork": 0.0, "grade": 0.0}, + "earthwork": {"expected": 0.0, "earthwork": 100.0, "grade": 0.0}, + "gentle": {"expected": 0.0, "earthwork": 0.0, "grade": 100.0}, +} +SEARCH_CELL_M = 4.0 # 탐색 격자 칸(m) — 비용면(2 m)을 솎아 씀 +SAMPLE_M = 2.0 # 결과 측정 간격(m) +STATION_M = 20.0 # 종단기울기 측정 간격(m) — 타당성평가 20 m 간격 · 측점 체계 +LENGTH_COST = 0.05 # 1 m 당 거리 비용 — 목표가 평평한 곳에서 헤매지 않게 +# 지반 기울기가 상한을 넘는 링크 1 m 당 벌점 = 이 값 × (넘은 기울기 / 상한). 세게 두면(20) +# 제약만 남고 비중이 안 먹음(랩탑_보조 실측: 비중 100/0/0 · 80/10/10 · 50/25/25 결과 같음). +# 2 에서 비중을 올릴수록 이격 · 토공이 한 방향으로 바뀜(이격 4 → 16 m · 토공 4.9 → 4.0천 m³). +GRADE_OVER_COST = 2.0 +STREAM_COST = 3.0 # 계류 버퍼 안 링크 1 m 당 벌점 — 건너기는 되고 나란히 달리기는 피함 +TURN_COST = 1.0 # 꺾을 때 벌점 = 최소 곡선반지름 원호 길이(R·θ) × 이 값 +MAX_TURN_STEPS = 2 # 한 걸음에 꺾는 방향 칸 수 한도(16 방향 중 ±2 = 약 45°) +# 16 방향(행, 열) — +x 에서 반시계. 나이트 걸음(1, 2) 등을 넣어 22.5° 안팎 방향을 곧게 달림. +DIRECTIONS = ( + (0, 1), + (1, 2), + (1, 1), + (2, 1), + (1, 0), + (2, -1), + (1, -1), + (1, -2), + (0, -1), + (-1, -2), + (-1, -1), + (-2, -1), + (-1, 0), + (-2, 1), + (-1, 1), + (-1, 2), +) +# 계류 버퍼 안 한 구간이 건너기로 인정되는 길이 — 45° 이상으로 가로지를 때의 버퍼 안 길이. +STREAM_CROSS_MAX_M = 2.0 * STREAM_OFFSET_M * math.sqrt(2.0) + + +@dataclass +class Terrain: + """지반 격자 — 탐색 비용면(`B05_Profile_Engine_Solver._load_or_build_cost_surface`) 그대로.""" + + x: np.ndarray # 열 좌표(오름차순) + y: np.ndarray # 행 좌표(오름차순) + z: np.ndarray # [행, 열] 표고 + valid: np.ndarray # [행, 열] 지형 있음 + + +def load_terrain_and_streams( + project_root: Path, selection: dict[str, Any] +) -> tuple[Terrain, list[list[tuple[float, float]]]]: + """지반 격자(탐색 비용면 캐시 그대로)와 이격을 따질 계류(사업지 좌표계)를 읽는다.""" + from pyproj import Transformer + + from B04_PreProcess.B04_PreProcess_Router_Watershed import ( + STREAM_FILE, + _load_features, + _reproject_features, + ) + from B05_Profile.B05_Profile_Engine_Solver import _MODELS_SUBDIR, _load_or_build_cost_surface + from common_util.common_util_crs import resolve_project_crs + + x, y, z, valid, *_ = _load_or_build_cost_surface( + project_root, + project_root / _MODELS_SUBDIR, + str(selection.get("source_filter")), + str(selection.get("method") or "dtm"), + bool(selection.get("smooth")), + ) + to_metric = Transformer.from_crs("EPSG:4326", resolve_project_crs(project_root), always_xy=True) + features = _load_features(project_root / "B04_PreProcess" / "processed", STREAM_FILE) + streams = stream_lines(_reproject_features(features, to_metric)) + return Terrain(np.asarray(x), np.asarray(y), np.asarray(z), np.asarray(valid)), streams + + +def resolve_weights(mode: str, weights: dict[str, float] | None) -> dict[str, float]: + """갈래의 비중(합 100). weighted 가 아니면 갈래가 정한 값.""" + if mode != "weighted": + return dict(MODE_WEIGHTS[mode]) + merged = {**DEFAULT_WEIGHTS, **(weights or {})} + return {key: float(merged[key]) for key in DEFAULT_WEIGHTS} + + +def stream_lines(features: list[dict[str, Any]]) -> list[list[tuple[float, float]]]: + """이격을 따질 계류만 좌표 목록으로(사업지 좌표계로 이미 옮긴 피처).""" + lines: list[list[tuple[float, float]]] = [] + for feature in features: + if (feature.get("properties") or {}).get("구분") not in STREAM_CLASSES: + continue + geometry = feature.get("geometry") or {} + parts = geometry.get("coordinates") or [] + if geometry.get("type") == "LineString": + parts = [parts] + elif geometry.get("type") != "MultiLineString": + continue + for part in parts: + points = [(float(p[0]), float(p[1])) for p in part if len(p) >= 2] + if len(points) >= 2: + lines.append(points) + return lines + + +def generate_initial_route( + expected: list[tuple[float, float]], + terrain: Terrain, + streams: list[list[tuple[float, float]]], + *, + mode: str, + weights: dict[str, float] | None, + max_grade: float, + min_radius_m: float, +) -> dict[str, Any]: + """갈래 하나로 초기 계획노선을 만들고 비교값을 붙여 돌려준다. `max_grade` 는 비율(0.14).""" + if mode not in MODES: + raise ValueError(f"모르는 갈래: {mode}") + if len(expected) < 2: + raise ValueError("예상노선 정점이 2개 미만입니다.") + used = resolve_weights(mode, weights) + points = ( + list(expected) + if mode == "follow" + else _search_route(expected, terrain, streams, used, max_grade, min_radius_m) + ) + outline = build_planned_polyline(points, min_radius_m=min_radius_m, simplify=True) + metrics = route_metrics( + outline.vertices, + expected, + terrain, + streams, + max_grade=max_grade, + radius_runs=sum(1 for curve in outline.curves if curve.violations), + ) + return { + "mode": mode, + "weights": used, + "planned": [[round(x, 4), round(y, 4)] for x, y in outline.vertices], + "nodes": [node.as_dict() for node in outline.nodes], + "curves": [curve.as_dict() for curve in outline.curves], + "metrics": metrics, + "criteria": { + "max_grade_pct": round(max_grade * 100.0, 2), + "min_radius_m": round(min_radius_m, 2), + "stream_offset_m": STREAM_OFFSET_M, + "road_width_m": ROAD_WIDTH_M, + "corridor_m": CORRIDOR_M, + "pending": list(PENDING), + }, + } + + +# ── 탐색 ───────────────────────────────────────────────────────────────────────── + + +def _cell_size(coords: np.ndarray) -> float: + return float((coords[-1] - coords[0]) / (len(coords) - 1)) if len(coords) > 1 else 1.0 + + +def _distance_grid( + lines: list[list[tuple[float, float]]], xs: np.ndarray, ys: np.ndarray, cell: float +) -> np.ndarray: + """격자 칸마다 선까지 거리(m). 선이 격자에 안 닿으면 모두 무한.""" + marked = np.zeros((len(ys), len(xs)), dtype=bool) + for line in lines: + for (x0, y0), (x1, y1) in zip(line, line[1:]): + count = max(2, int(math.hypot(x1 - x0, y1 - y0) / (cell * 0.5)) + 1) + px = np.linspace(x0, x1, count) + py = np.linspace(y0, y1, count) + cols = np.rint((px - xs[0]) / cell).astype(int) + rows = np.rint((py - ys[0]) / cell).astype(int) + inside = (cols >= 0) & (cols < len(xs)) & (rows >= 0) & (rows < len(ys)) + marked[rows[inside], cols[inside]] = True + if not marked.any(): + return np.full(marked.shape, np.inf) + return ndimage.distance_transform_edt(~marked) * cell + + +def _search_route( + expected: list[tuple[float, float]], + terrain: Terrain, + streams: list[list[tuple[float, float]]], + weights: dict[str, float], + max_grade: float, + min_radius_m: float, +) -> list[tuple[float, float]]: + """복도 안 격자에서 진행 방향을 기억하는 Dijkstra — 시점 · 종점은 예상노선 양 끝 그대로. + + 상태 = (칸, 들어온 방향 16 가지). 한 걸음에 22.5° · 45° 까지만 꺾고, 꺾을 때마다 최소 + 곡선반지름 원호 길이만큼 벌점을 매겨 곧게 달리게 한다. 16 방향(체스 나이트 걸음 포함)이라 + 8 방향 격자가 급한 옆경사에서 등고선을 못 따라 계단처럼 꺾이던 것을 피한다. + """ + step = max(1, int(round(SEARCH_CELL_M / _cell_size(terrain.x)))) + xs, ys = terrain.x[::step], terrain.y[::step] + z = np.asarray(terrain.z[::step, ::step], dtype=np.float64) + cell = _cell_size(xs) + d_expected = _distance_grid([expected], xs, ys, cell) + d_stream = _distance_grid(streams, xs, ys, cell) + corridor = (d_expected <= CORRIDOR_M) & np.asarray(terrain.valid[::step, ::step], dtype=bool) + grad_y, grad_x = np.gradient(z, cell) + rows, cols = z.shape + index = np.full((rows, cols), -1, dtype=np.int64) + cells = int(corridor.sum()) + index[corridor] = np.arange(cells) + + # 방향마다 링크(칸 → 칸)와 세 목표 · 벌점(1 m 당) + links: list[tuple[np.ndarray, np.ndarray, float, np.ndarray, np.ndarray]] = [] + for dr, dc in DIRECTIONS: + first = (slice(max(0, -dr), rows - max(0, dr)), slice(max(0, -dc), cols - max(0, dc))) + second = (slice(max(0, dr), rows - max(0, -dr)), slice(max(0, dc), cols - max(0, -dc))) + a, b = index[first], index[second] + ok = (a >= 0) & (b >= 0) + norm = math.hypot(dr, dc) + grade = np.abs(z[second][ok] - z[first][ok]) / (cell * norm) + over = np.maximum(0.0, grade - max_grade) + # 옆경사 = 진행 방향에 수직인 지반 기울기(두 칸 평균) + cross = np.abs( + (grad_x[first][ok] + grad_x[second][ok]) * -dr + + (grad_y[first][ok] + grad_y[second][ok]) * dc + ) / (2.0 * norm) + # 토공 가늠(1 m 당 m³) — 옆경사 위 반절 · 반성 단면 W²·s/4 + 상한 넘는 기울기가 측점 + # 반 간격 동안 쌓는 높이 차 단면 + earth = ROAD_WIDTH_M**2 * cross / 4.0 + ROAD_WIDTH_M * over * STATION_M / 2.0 + offset = 0.5 * (d_expected[first][ok] + d_expected[second][ok]) + in_buffer = 0.5 * ( + (d_stream[first][ok] < STREAM_OFFSET_M).astype(float) + + (d_stream[second][ok] < STREAM_OFFSET_M) + ) + penalty = GRADE_OVER_COST * over / max_grade + STREAM_COST * in_buffer + links.append( + ( + a[ok], + b[ok], + cell * norm, + np.stack([offset, earth, (grade / max_grade) ** 2]), + penalty, + ) + ) + if not any(link[0].size for link in links): + raise ValueError("예상노선 둘레에 탐색할 지형이 없습니다.") + + # 세 목표 정규화 — 복도 안 링크 평균이 1 이 되게(길이 가중). 평균이 0 이면 그 목표는 0. + lengths = np.concatenate([np.full(link[0].size, link[2]) for link in links]) + stacked = np.hstack([link[3] for link in links]) + scale = (stacked * lengths).sum(axis=1) / max(lengths.sum(), 1e-9) + order = ("expected", "earthwork", "grade") + share = np.array([weights[key] for key in order]) / max(sum(weights.values()), 1e-9) + factor = np.divide(share, scale, out=np.zeros_like(share), where=scale > 1e-12) + + count = len(DIRECTIONS) + angles = [math.atan2(dr, dc) for dr, dc in DIRECTIONS] + heads, tails, costs = [], [], [] + for out, (a, b, length, term, penalty) in enumerate(links): + move = length * (LENGTH_COST + factor @ term + penalty) + for turn in range(-MAX_TURN_STEPS, MAX_TURN_STEPS + 1): + came = (out + turn) % count + bend = abs((angles[out] - angles[came] + math.pi) % (2.0 * math.pi) - math.pi) + heads.append(a * count + came) + tails.append(b * count + out) + costs.append(move + TURN_COST * min_radius_m * bend) + graph = csr_matrix( + (np.concatenate(costs), (np.concatenate(heads), np.concatenate(tails))), + shape=(cells * count, cells * count), + ) + cell_rows, cell_cols = np.nonzero(corridor) + centres = np.column_stack([xs[cell_cols], ys[cell_rows]]) + + def nearest(point: tuple[float, float]) -> int: + return int(np.argmin(np.hypot(centres[:, 0] - point[0], centres[:, 1] - point[1]))) + + start, end = nearest(expected[0]), nearest(expected[-1]) + distance, predecessors, _ = dijkstra( + graph, + indices=[start * count + d for d in range(count)], + return_predecessors=True, + min_only=True, + ) + state = end * count + int(np.argmin(distance[end * count : (end + 1) * count])) + if not np.isfinite(distance[state]): + raise ValueError("복도 안에서 시점과 종점을 잇는 길을 찾지 못했습니다.") + path = [state] + while predecessors[path[-1]] >= 0: + path.append(int(predecessors[path[-1]])) + points = [ + (float(centres[s // count, 0]), float(centres[s // count, 1])) for s in reversed(path) + ] + # 양 끝은 격자 칸 가운데가 아니라 예상노선 끝점 그대로 — 기점 · 종점은 옮기지 않는다. + points[0], points[-1] = tuple(expected[0]), tuple(expected[-1]) + return points + + +# ── 비교값 ─────────────────────────────────────────────────────────────────────── + + +def _resample(vertices: list[tuple[float, float]], step: float) -> tuple[np.ndarray, np.ndarray]: + """폴리라인을 `step` 간격 점으로 — (점 [n,2], 누가거리 [n]).""" + xy = np.asarray(vertices, dtype=np.float64) + seg = np.hypot(*np.diff(xy, axis=0).T) + chain = np.concatenate([[0.0], np.cumsum(seg)]) + total = float(chain[-1]) + at = np.append(np.arange(0.0, total, step), total) if total > 0 else np.array([0.0]) + return np.column_stack([np.interp(at, chain, xy[:, 0]), np.interp(at, chain, xy[:, 1])]), at + + +def _runs(flags: np.ndarray) -> list[tuple[int, int]]: + """참이 이어진 구간들의 (첫, 끝) 번호.""" + runs: list[tuple[int, int]] = [] + start = None + for i, flag in enumerate(flags): + if flag and start is None: + start = i + elif not flag and start is not None: + runs.append((start, i - 1)) + start = None + if start is not None: + runs.append((start, len(flags) - 1)) + return runs + + +def grade_line(ground: np.ndarray, step: float, max_grade: float) -> np.ndarray: + """상한 안의 계획고 가늠 — 앞으로 · 뒤로 기울기를 누른 두 선의 평균(둘 다 상한 안 → 평균도).""" + rise = max_grade * step + forward, backward = ground.copy(), ground.copy() + for i in range(1, len(ground)): + forward[i] = min(max(ground[i], forward[i - 1] - rise), forward[i - 1] + rise) + for i in range(len(ground) - 2, -1, -1): + backward[i] = min(max(ground[i], backward[i + 1] - rise), backward[i + 1] + rise) + return 0.5 * (forward + backward) + + +def route_metrics( + vertices: list[tuple[float, float]], + expected: list[tuple[float, float]], + terrain: Terrain, + streams: list[list[tuple[float, float]]], + *, + max_grade: float, + radius_runs: int, +) -> dict[str, Any]: + """연장 · 최대/평균 종단기울기(지반) · 절성토량(가늠) · 기준 초과 구간 수 · 예상노선 이격.""" + sampler = RegularGridInterpolator( + (terrain.y, terrain.x), terrain.z, bounds_error=False, fill_value=None + ) + points, chain = _resample(vertices, SAMPLE_M) + length = float(chain[-1]) + ground = sampler(points[:, ::-1]) + + # 종단기울기 — 측점(20 m) 사이 지반 기울기 + stations = np.append(np.arange(0.0, length, STATION_M), length) + station_z = np.interp(stations, chain, ground) + spans = np.diff(stations) + keep = spans > 0.5 + grades = np.abs(np.diff(station_z))[keep] / spans[keep] + max_grade_run = float(grades.max()) if grades.size else 0.0 + avg_grade = float((grades * spans[keep]).sum() / spans[keep].sum()) if grades.size else 0.0 + + # 절 · 성토(가늠) — 상한 안 계획고와 옆경사 단면을 폭 W 로 적분(비탈면 제외) + grad_y, grad_x = np.gradient(terrain.z, _cell_size(terrain.y), _cell_size(terrain.x)) + gx = RegularGridInterpolator( + (terrain.y, terrain.x), grad_x, bounds_error=False, fill_value=None + ) + gy = RegularGridInterpolator( + (terrain.y, terrain.x), grad_y, bounds_error=False, fill_value=None + ) + tangent = np.gradient(points, axis=0) + tangent /= np.maximum(np.hypot(tangent[:, 0], tangent[:, 1]), 1e-9)[:, None] + lookup = points[:, ::-1] + cross = np.abs(gx(lookup) * -tangent[:, 1] + gy(lookup) * tangent[:, 0]) + height = grade_line(ground, SAMPLE_M, max_grade) - ground + across = np.linspace(-ROAD_WIDTH_M / 2.0, ROAD_WIDTH_M / 2.0, 9) + depth = cross[:, None] * across[None, :] - height[:, None] # +: 지반이 노면 위 = 절토 + weight = np.gradient(chain) if len(chain) > 1 else np.zeros_like(chain) + cut = float((np.maximum(depth, 0.0).mean(axis=1) * ROAD_WIDTH_M * weight).sum()) + fill = float((np.maximum(-depth, 0.0).mean(axis=1) * ROAD_WIDTH_M * weight).sum()) + + # 예상노선 이격 + route_points = shapely.points(points) + offsets = shapely.distance(route_points, LineString(expected)) + + # 계류 이격 — 버퍼 안 구간 중 봐줄 길이를 넘는 것. 봐줌 = 건너기(계류에 닿음) 한 번의 + # 버퍼 안 길이 + 기점 · 종점이 버퍼 안이면 45° 로 빠져나가는 길이(끝점은 옮길 수 없음) + stream_runs = 0 + if streams: + near = shapely.distance(route_points, MultiLineString(streams)) + for first, last in _runs(near < STREAM_OFFSET_M): + ends = int(first == 0) + int(last == len(near) - 1) + crosses = float(near[first : last + 1].min()) <= SAMPLE_M + allowance = ends * STREAM_OFFSET_M * math.sqrt(2.0) + crosses * STREAM_CROSS_MAX_M + if chain[last] - chain[first] > allowance: + stream_runs += 1 + grade_runs = len(_runs(grades > max_grade)) + return { + "length_m": round(length, 2), + "max_grade_pct": round(max_grade_run * 100.0, 2), + "avg_grade_pct": round(avg_grade * 100.0, 2), + "cut_m3": round(cut, 1), + "fill_m3": round(fill, 1), + "over_limit": { + "grade": grade_runs, + "radius": int(radius_runs), + "stream": stream_runs, + "total": grade_runs + int(radius_runs) + stream_runs, + }, + "offset_avg_m": round(float(offsets.mean()), 2), + "offset_max_m": round(float(offsets.max()), 2), + } diff --git a/B05_Profile/B05_Profile_Router_Replan.py b/B05_Profile/B05_Profile_Router_Replan.py index 50bc3103..0297f3b3 100644 --- a/B05_Profile/B05_Profile_Router_Replan.py +++ b/B05_Profile/B05_Profile_Router_Replan.py @@ -26,6 +26,7 @@ GET /api/projects/{id}/route/plan → 예상노선·계획노선 정점(사업지 좌표계) POST /api/projects/{id}/route/replan → 고친 계획노선으로 갈아 끼우고 재계산 POST /api/projects/{id}/route/replan/reset → 계획노선을 예상노선으로 되돌리고 재계산 + POST /api/projects/{id}/route/initial → 초기 계획노선 갈래 하나를 만들어 돌려줌(파일 안 씀) """ import asyncio @@ -35,7 +36,7 @@ import json import logging import time from pathlib import Path -from typing import Any +from typing import Any, Literal from uuid import UUID from fastapi import APIRouter @@ -84,6 +85,21 @@ class RouteVertexInput(BaseModel): """못박을 곡선 반지름(m). 없으면 예정노선에 맞추거나 법정 하한을 쓴다.""" +class RouteInitialWeights(BaseModel): + """비중 갈래의 세 목표 비중 — 합 100(PLAN 20장 · 기본 50/25/25).""" + + expected: float = Field(50.0, ge=0, le=100) + earthwork: float = Field(25.0, ge=0, le=100) + grade: float = Field(25.0, ge=0, le=100) + + +class RouteInitialRequest(BaseModel): + """초기 계획노선 갈래 — follow(예상노선 추종) · earthwork · gentle · weighted.""" + + mode: Literal["follow", "earthwork", "gentle", "weighted"] = "follow" + weights: RouteInitialWeights | None = None + + class RouteReplanRequest(BaseModel): """고친 계획노선. 정점은 시점 → 종점 순서(사업지 좌표계 m).""" @@ -155,14 +171,25 @@ async def _plan_criteria(project_id: UUID) -> tuple[float, float, float]: 있는 것은 `road_type` 뿐). 못 읽으면 가장 완화된 조건으로 떨어진다 — 막지 않고 위반 표시만 하기 때문이다. """ - import aiomysql - from B05_Profile.B05_Profile_Engine_Grade import ( legal_plan_radius_min_m, plan_curve_length_limit_m, plan_radius_limit_m, resolve_design_speed, ) + + grade_class, design_speed, terrain = await _road_settings(project_id) + return ( + legal_plan_radius_min_m(resolve_design_speed(grade_class, design_speed), terrain), + plan_radius_limit_m(grade_class, design_speed, terrain), + plan_curve_length_limit_m(grade_class), + ) + + +async def _road_settings(project_id: UUID) -> tuple[str, int | None, str]: + """(임도 종류, 설계속도, 지형) — 못 읽으면 가장 완화된 조건(작업 · 특수).""" + import aiomysql + from common_util.common_util_workflow_state import get_workflow_state grade_class, design_speed, terrain = "work", None, "special" @@ -189,11 +216,7 @@ async def _plan_criteria(project_id: UUID) -> tuple[float, float, float]: terrain = str(params["terrain_type"]) except Exception: # noqa: BLE001 — 설정을 못 읽어도 폴리라인화는 이어 간다 logger.exception("최소곡선반지름 설정을 못 읽어 기본값을 씁니다: %s", project_id) - return ( - legal_plan_radius_min_m(resolve_design_speed(grade_class, design_speed), terrain), - plan_radius_limit_m(grade_class, design_speed, terrain), - plan_curve_length_limit_m(grade_class), - ) + return grade_class, design_speed, terrain def _nodes_path(path: Path) -> Path: @@ -564,3 +587,73 @@ async def reset_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse: status_code=409, content={"status": "error", "message": result["error"]} ) return {"status": "success", "project_id": str(project_id), **result} + + +@router.post("/{project_id}/route/initial", response_model=None) +async def initial_route( + project_id: UUID, request: RouteInitialRequest +) -> dict[str, Any] | JSONResponse: + """초기 계획노선 갈래 하나를 만들어 비교값과 함께 돌려준다(PLAN 20장). + + **파일을 쓰지 않는다** — 화면 캐시용. 적용은 받은 `nodes` 로 `POST /route/replan`([확인]). + """ + from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct + from B05_Profile.B05_Profile_Engine_RouteInitial import ( + generate_initial_route, + load_terrain_and_streams, + ) + from common_util.common_util_surface_confirmation import get_surface_confirmation_params + + weights = request.weights.model_dump() if request.weights else None + if request.mode == "weighted" and weights and abs(sum(weights.values()) - 100.0) > 0.5: + return JSONResponse( + status_code=400, + content={"status": "error", "message": "비중 세 값의 합은 100 이어야 합니다."}, + ) + paths = await _project_paths(project_id) + if paths is None: + return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING) + project_root, _ = paths + expected = await asyncio.to_thread(_vertices_of, expected_route_csv_path(project_root)) + if not expected: + expected = await asyncio.to_thread(_vertices_of, design_route_csv_path(project_root)) + if len(expected) < 2: + return JSONResponse( + status_code=400, + content={"status": "error", "message": "예상노선이 없어 초기노선을 만들 수 없습니다."}, + ) + radius_m, _, _ = await _plan_criteria(project_id) + grade_class, design_speed, terrain_type = await _road_settings(project_id) + # 비포장 기준 — 포장 예외(18 %)는 포장 여부가 정해진 뒤 종단 설계가 따진다(사용자 확정 대기). + max_grade = legal_grade_limit_pct(grade_class, terrain_type, False, design_speed) / 100.0 + pool = get_db_pool() + async with pool.acquire() as connection: + selection = await get_surface_confirmation_params(connection, str(project_id)) + started = time.perf_counter() + try: + terrain, streams = await asyncio.to_thread( + load_terrain_and_streams, project_root, selection + ) + result = await asyncio.to_thread( + functools.partial( + generate_initial_route, + [(x, y) for x, y in expected], + terrain, + streams, + mode=request.mode, + weights=weights, + max_grade=max_grade, + min_radius_m=radius_m, + ) + ) + except (FileNotFoundError, ValueError) as error: + return JSONResponse(status_code=409, content={"status": "error", "message": str(error)}) + logger.info( + "초기 계획노선 %s: project_id=%s %.1fs 연장 %.0fm 이격 평균 %.1fm", + request.mode, + project_id, + time.perf_counter() - started, + result["metrics"]["length_m"], + result["metrics"]["offset_avg_m"], + ) + return {"status": "success", "project_id": str(project_id), **result} diff --git a/resources/tester/test_route_initial.py b/resources/tester/test_route_initial.py new file mode 100644 index 00000000..bf3b850d --- /dev/null +++ b/resources/tester/test_route_initial.py @@ -0,0 +1,126 @@ +"""초기 계획노선 갈래(PLAN 20장) — `B05_Profile_Engine_RouteInitial` 합성 지형 시험. + +./venv/Scripts/python.exe -m pytest resources/tester/test_route_initial.py -q +""" + +import numpy as np +import pytest + +from B05_Profile.B05_Profile_Engine_RouteInitial import ( + MODES, + PENDING, + STREAM_OFFSET_M, + Terrain, + generate_initial_route, + grade_line, + resolve_weights, + stream_lines, +) + +GRADE = 0.14 +RADIUS = 12.0 + + +def _terrain(z_of) -> Terrain: + x = np.arange(0.0, 802.0, 2.0) + y = np.arange(0.0, 402.0, 2.0) + xx, yy = np.meshgrid(x, y) + return Terrain(x, y, z_of(xx, yy), np.ones(xx.shape, dtype=bool)) + + +def _line(y: float) -> list[tuple[float, float]]: + return [(float(x), y) for x in np.arange(50.0, 752.0, 4.0)] + + +@pytest.fixture(scope="module") +def hill() -> Terrain: + """가운데 솟은 언덕(높이 40 m) — 곧은 예상노선이 꼭대기를 넘는다.""" + return _terrain( + lambda xx, yy: 40.0 * np.exp(-((xx - 400.0) ** 2 + (yy - 200.0) ** 2) / 90.0**2) + ) + + +def _run(terrain: Terrain, mode: str, streams=(), weights=None) -> dict: + return generate_initial_route( + _line(200.0), + terrain, + list(streams), + mode=mode, + weights=weights, + max_grade=GRADE, + min_radius_m=RADIUS, + ) + + +def test_follow_keeps_expected_route(hill: Terrain) -> None: + result = _run(hill, "follow") + assert result["metrics"]["offset_max_m"] < 1.0 + assert result["weights"] == {"expected": 100.0, "earthwork": 0.0, "grade": 0.0} + assert result["criteria"]["pending"] == list(PENDING) + + +@pytest.mark.parametrize("mode", ["earthwork", "gentle", "weighted"]) +def test_search_modes_beat_follow_on_hill(hill: Terrain, mode: str) -> None: + follow = _run(hill, "follow")["metrics"] + result = _run(hill, mode) + metrics = result["metrics"] + # 언덕을 돌아가 기울기 · 토공이 준다 + assert metrics["max_grade_pct"] < follow["max_grade_pct"] + assert metrics["cut_m3"] + metrics["fill_m3"] < follow["cut_m3"] + follow["fill_m3"] + assert metrics["over_limit"]["grade"] <= follow["over_limit"]["grade"] + # 기점 · 종점은 옮기지 않고 복도(100 m) 밖으로 안 나감 + assert result["planned"][0] == [50.0, 200.0] + assert result["planned"][-1] == [750.0, 200.0] + assert metrics["offset_max_m"] <= result["criteria"]["corridor_m"] + 4.0 + assert result["nodes"] and result["curves"] is not None + + +def test_weighted_expected_share_pulls_route_back(hill: Terrain) -> None: + near = _run(hill, "weighted", weights={"expected": 90, "earthwork": 5, "grade": 5}) + far = _run(hill, "weighted", weights={"expected": 10, "earthwork": 45, "grade": 45}) + assert near["metrics"]["offset_avg_m"] < far["metrics"]["offset_avg_m"] + + +def test_stream_offset_is_common_constraint() -> None: + """평지 · 예상노선 20 m 옆을 나란히 흐르는 소하천 — 추종은 초과 1 · 탐색은 비켜 감.""" + flat = _terrain(lambda xx, yy: np.zeros_like(xx)) + stream = [[(0.0, 220.0), (800.0, 220.0)]] + follow = _run(flat, "follow", streams=stream)["metrics"] + moved = _run(flat, "earthwork", streams=stream)["metrics"] + assert follow["over_limit"]["stream"] == 1 + assert moved["over_limit"]["stream"] == 0 + assert moved["offset_avg_m"] > STREAM_OFFSET_M - 20.0 - 10.0 + + +def test_stream_crossing_is_allowed() -> None: + flat = _terrain(lambda xx, yy: np.zeros_like(xx)) + crossing = [[(400.0, 0.0), (400.0, 400.0)]] + assert _run(flat, "follow", streams=crossing)["metrics"]["over_limit"]["stream"] == 0 + + +def test_grade_line_stays_within_limit() -> None: + ground = np.array([0.0, 0.0, 5.0, 10.0, 10.0, 2.0, 0.0]) + line = grade_line(ground, 2.0, GRADE) + assert np.all(np.abs(np.diff(line)) <= GRADE * 2.0 + 1e-9) + + +def test_weights_and_stream_filter() -> None: + assert resolve_weights("weighted", None) == {"expected": 50.0, "earthwork": 25.0, "grade": 25.0} + assert resolve_weights("gentle", {"expected": 99})["grade"] == 100.0 + assert set(MODES) == {"follow", "earthwork", "gentle", "weighted"} + features = [ + { + "properties": {"구분": "세류"}, + "geometry": {"type": "LineString", "coordinates": [[0, 0], [1, 1]]}, + }, + { + "properties": {"구분": "소하천"}, + "geometry": {"type": "LineString", "coordinates": [[0, 0], [1, 1]]}, + }, + ] + assert len(stream_lines(features)) == 1 + + +def test_unknown_mode_rejected(hill: Terrain) -> None: + with pytest.raises(ValueError): + _run(hill, "shortest")