diff --git a/B05_Profile/B05_Profile_Engine_RouteInitial.py b/B05_Profile/B05_Profile_Engine_RouteInitial.py index f2ae4b30d..0e30d03a6 100644 --- a/B05_Profile/B05_Profile_Engine_RouteInitial.py +++ b/B05_Profile/B05_Profile_Engine_RouteInitial.py @@ -132,10 +132,16 @@ _TERRAINS: dict[str, tuple[Terrain, list[list[tuple[float, float]]]]] = {} def load_terrain_and_streams( - project_root: Path, selection: dict[str, Any] + project_root: Path, + selection: dict[str, Any], + models_dir: Path | None = None, + cache_dir: Path | None = None, ) -> tuple[Terrain, list[list[tuple[float, float]]]]: """지반 격자(탐색 비용면 캐시 그대로)와 이격을 따질 계류(사업지 좌표계)를 읽는다. + `models_dir` · `cache_dir` 없으면 B 자리(`B04_PreProcess/models` · `B05_Profile/route`) — + D 예정노선은 연도 구간마다 따로 줌(530-2). + 같은 입력(비용면 서명 · 계류 파일 지문 · 좌표계)이면 메모리 기억을 그대로 돌려준다 — 점 놓을 때마다 npz · geojson 을 다시 읽지 않는다. 돌려준 배열은 고치지 않는다. """ @@ -156,7 +162,7 @@ def load_terrain_and_streams( filter_key = str(selection.get("source_filter")) method = str(selection.get("method") or "dtm") smooth = bool(selection.get("smooth")) - models_dir = project_root / _MODELS_SUBDIR + models_dir = project_root / _MODELS_SUBDIR if models_dir is None else Path(models_dir) stream_path = project_root / "B04_PreProcess" / "processed" / STREAM_FILE crs = resolve_project_crs(project_root) try: @@ -177,7 +183,7 @@ def load_terrain_and_streams( return _TERRAINS[memo_key] x, y, z, valid, *_ = _load_or_build_cost_surface( - project_root, models_dir, filter_key, method, smooth + project_root, models_dir, filter_key, method, smooth, cache_dir ) to_metric = Transformer.from_crs("EPSG:4326", crs, always_xy=True) features = _load_features(stream_path.parent, STREAM_FILE) @@ -302,6 +308,7 @@ def generate_initial_route( pipes: list[tuple[float, float, float]] | None = None, with_profile: bool = True, standard: dict[str, Any] | None = None, + corridor_m: float = CORRIDOR_M, ) -> dict[str, Any]: """갈래 하나로 초기 계획노선을 만들고 비교값을 붙여 돌려준다. `max_grade` 는 비율(0.14). @@ -313,6 +320,7 @@ def generate_initial_route( 몇 개에만 균형 종단(`..._Profile` · PLAN 63-4)을 붙여 고른다(419) — 편집 창 추천 점수 · 미리 보기에만 쓰고 확정 종단과 무관. `pipes`(x, y, 덮개 여유)는 덮개 부족 수만 센다. `with_profile` 끄면(종단 반영 끔) 종단 없이 계획고 지반고 횡단 토공으로 고른다. + `corridor_m` = 탐색 복도 반폭(기본 B 값 · D 예정노선만 넓힘 · 530-2). """ if mode not in MODES: raise ValueError(f"모르는 갈래: {mode}") @@ -431,7 +439,15 @@ def generate_initial_route( pool: Pool | None = None target = _weighted_end(used) if mode == "weighted" else mode key = _pool_key( - expected, terrain, streams, max_grade, min_radius_m, stream_offset, grade_smooth, goal_grade + expected, + terrain, + streams, + max_grade, + min_radius_m, + stream_offset, + grade_smooth, + goal_grade, + *([] if corridor_m == CORRIDOR_M else [corridor_m]), # 기본 복도 = B 열쇠 그대로 ) if target == "follow": # 비중 예상 100 = 추적 그 선(PLAN 23-3e) · 묶음은 짜지 않음(빠른 길) outline, metrics = follow_line() @@ -445,6 +461,7 @@ def generate_initial_route( stream_offset_m=stream_offset, grade_smooth_m=grade_smooth, target_grade=goal_grade, + corridor_m=corridor_m, ) graph = RouteGraph(grid, expected, limit=max_grade, min_radius_m=min_radius_m) while len(_POOLS) >= POOL_CACHE_SIZE: @@ -513,7 +530,7 @@ def generate_initial_route( "min_radius_m": round(min_radius_m, 2), "stream_offset_m": round(stream_offset, 2), "road_width_m": ROAD_WIDTH_M, - "corridor_m": CORRIDOR_M, + "corridor_m": corridor_m, "grade_smooth_m": round(grade_smooth, 2), "target_grade_pct": round(goal_grade * 100.0, 2), # 기준값 확정(PLAN 20장, 2026-09-29) — 더는 후보가 아니다. 항상 빈 목록. diff --git a/B05_Profile/B05_Profile_Engine_RouteInitial_Search.py b/B05_Profile/B05_Profile_Engine_RouteInitial_Search.py index adfaeb6bc..64e811022 100644 --- a/B05_Profile/B05_Profile_Engine_RouteInitial_Search.py +++ b/B05_Profile/B05_Profile_Engine_RouteInitial_Search.py @@ -111,8 +111,10 @@ class SearchGrid: stream_offset_m: float, grade_smooth_m: float = GRADE_SMOOTH_M, target_grade: float | None = None, + corridor_m: float = CORRIDOR_M, ) -> None: step = max(1, int(round(SEARCH_CELL_M / _cell_size(terrain.x)))) + self.corridor_m = corridor_m # 복도 반폭 — 기본 B 값 · D 예정노선만 넓힘(530-2) self.target_grade = target_grade # 영선 노선 목표 기울기(비율) · 없으면 링크 상한 self.xs, self.ys = terrain.x[::step], terrain.y[::step] # 기울기 · 옆경사는 비교값과 같은 고른 지반에서 잰다(`smoothed_ground`) @@ -122,7 +124,7 @@ class SearchGrid: self.d_expected = _distance_grid([expected], self.xs, self.ys, self.cell) self.d_stream = _distance_grid(streams, self.xs, self.ys, self.cell) valid = np.asarray(terrain.valid[::step, ::step], dtype=bool) - self.corridor = (self.d_expected <= CORRIDOR_M) & valid + self.corridor = (self.d_expected <= corridor_m) & valid self.grad_y, self.grad_x = np.gradient(self.z, self.cell) finite = np.where(np.isfinite(self.d_stream), self.d_stream, 0.0) self.away_y, self.away_x = np.gradient(finite, self.cell) # 계류에서 멀어지는 쪽 @@ -214,7 +216,7 @@ class SearchGrid: along = (sample(self.d_stream, mid) < self.stream_offset_m) & ~across & (size > 1e-6) terms = np.stack( [ - np.minimum(sample(self.d_expected, mid), CORRIDOR_M), + np.minimum(sample(self.d_expected, mid), self.corridor_m), earth, (grade / limit) ** 2, over / limit, diff --git a/B05_Profile/B05_Profile_Engine_Solver.py b/B05_Profile/B05_Profile_Engine_Solver.py index 4058c979b..3157d0fc8 100644 --- a/B05_Profile/B05_Profile_Engine_Solver.py +++ b/B05_Profile/B05_Profile_Engine_Solver.py @@ -212,11 +212,18 @@ def _build_cost_surface(models_dir: Path, filter_key: str, method: str, smooth: def _load_or_build_cost_surface( - project_root: Path, models_dir: Path, filter_key: str, method: str, smooth: bool + project_root: Path, + models_dir: Path, + filter_key: str, + method: str, + smooth: bool, + cache_dir: Path | None = None, ): - """비용면을 반환한다(서명 일치 시 캐시 재사용, 아니면 재빌드·재캐시).""" + """비용면을 반환한다(서명 일치 시 캐시 재사용, 아니면 재빌드·재캐시). + + `cache_dir` 없으면 `B05_Profile/route` — D 예정노선은 연도 구간마다 따로 줌(530-2).""" check_surface_keys(filter_key, method) - cache_dir = project_root / _ROUTE_CACHE_SUBDIR + cache_dir = project_root / _ROUTE_CACHE_SUBDIR if cache_dir is None else Path(cache_dir) suffix = "_smooth" if smooth else "" cache_path = cache_dir / f"cost_surface_{filter_key}_{method}{suffix}.npz" signature = _cost_surface_signature(models_dir, filter_key, method, smooth) diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Analyze.py b/D03_PlanAnalysis/D03_PlanAnalysis_Analyze.py new file mode 100644 index 000000000..5e6296638 --- /dev/null +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Analyze.py @@ -0,0 +1,208 @@ +"""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`(입력 지문 같으면 다시 안 셈) + +계산 자리 = ③ 서버 단독(B 추천과 같음). +""" + +from __future__ import annotations + +import hashlib +import json +import logging +from pathlib import Path +from typing import Any, Callable + +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 ( + 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 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 + +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 _same_point(a: Any, b: Any) -> bool: + return abs(float(a[0]) - float(b[0])) < 1e-6 and abs(float(a[1]) - float(b[1])) < 1e-6 + + +def read_segment(project_root: Path, seg: dict[str, Any]) -> dict[str, Any] | None: + """저장본 — 지점이 지금과 같을 때만(입력 지문은 분석 때 다시 봄).""" + try: + stored = json.loads( + (derived_dir(project_root) / f"route_{int(seg['year'])}.json").read_text("utf-8") + ) + except (OSError, ValueError): + return None + if _same_point(stored["start"], seg["start"]) and _same_point(stored["end"], seg["end"]): + return stored + return None + + +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 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 = [tuple(map(float, seg["start"])), tuple(map(float, seg["end"]))] + inputs = { + "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"], + } + keys = [ + request_fingerprint(expected, terrain, streams, mode=mode, **inputs)[0] + for mode in PLAN_MODES + ] + fingerprint = hashlib.blake2b( + repr((keys, float(corridor_m), ALGO)).encode(), digest_size=16 + ).hexdigest() + 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(derived_dir(project_root) / f"route_{int(seg['year'])}.json", result) + return {**result, "stored": False} + + +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, +) -> list[dict[str, Any]]: + """연도 구간마다 지형(연도 DEM · 연도 비용면 캐시)을 읽어 `analyze_segment`.""" + results = [] + for i, seg in enumerate(segments): + year = int(seg["year"]) + if on_progress is not None: + on_progress(100 * i // len(segments), f"{year}년 구간 노선 추천 중") + terrain, streams = loader( + project_root, + SELECTION, + models_dir_of(project_root, year), + derived_dir(project_root) / f"cost_{year}", + ) + results.append(analyze_segment(project_root, seg, terrain, streams, criteria, corridor_m)) + if on_progress is not None: + on_progress(100, "분석 끝") + return results + + +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 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 diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Router.py b/D03_PlanAnalysis/D03_PlanAnalysis_Router.py new file mode 100644 index 000000000..0a620fc6e --- /dev/null +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Router.py @@ -0,0 +1,250 @@ +"""D 예정노선 2단계 분석 API (요구 345 · PLAN 530-1 · 530-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` + +지점 정본은 `plan_route_years` 시점 · 종점 칸(529-3 몫)을 읽기만 함. 계산 자리 = ③ 서버 단독 — +[저장]은 브라우저 값을 받지 않고 갈래 이름만 받아 서버 저장본으로 다시 셈. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from pathlib import Path +from typing import Any +from uuid import UUID + +import aiomysql +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel + +from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path +from common_util.common_util_storage import resolve_stored_project_path +from config.config_db import get_db_pool +from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze +from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/projects", tags=["D03 Plan Analysis"]) + +# 진행 — (프로젝트, 일) → {running, percent, message, error} · 프로세스 메모리 +_JOBS: dict[tuple[str, str], dict[str, Any]] = {} +_TASKS: set[asyncio.Task] = set() +KINDS = ("terrain", "analyze") + + +class ChoiceRequest(BaseModel): + choices: dict[int, str] # 연도 → 갈래 + + +async def _plan_inputs(project_id: UUID) -> tuple[Path, dict[str, Any], list[dict[str, Any]]]: + """(프로젝트 실경로, 프로젝트 칸, 연도 구간) — 지점이 빈 연도 · 좌표계 섞임은 거절.""" + 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 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: + 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 + ] + return Path(resolve_stored_project_path(stored)), project, segments + + +def _start_job(project_id: UUID, kind: str, work: Any) -> dict[str, Any]: + key = (str(project_id), kind) + if (_JOBS.get(key) or {}).get("running"): + raise HTTPException(status_code=409, detail="이미 돌고 있습니다.") + job = {"running": True, "percent": 0, "message": "시작", "error": None} + _JOBS[key] = job + + def report(percent: int, message: str) -> None: + job.update(percent=int(percent), message=message) + + async def run() -> None: + try: + await work(report) + job.update(percent=100) + except Exception as error: # noqa: BLE001 — 진행률에 까닭을 그대로 올림 + logger.exception("D %s 실패: project_id=%s", kind, project_id) + job["error"] = str(getattr(error, "detail", None) or error) + finally: + job["running"] = False + + task = asyncio.create_task(run(), name=f"D {kind} {project_id}") + _TASKS.add(task) + task.add_done_callback(_TASKS.discard) + return {"status": "accepted", **job} + + +async def _save_models(project_id: UUID, years: dict[int, 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, + ) + await connection.commit() + except Exception: + await connection.rollback() + raise + + +@router.post("/{project_id}/plan/terrain", response_model=None) +async def plan_terrain(project_id: UUID) -> dict[str, Any]: + """지점 기준 수치지형도 · 위성 · 연도별 DEM 받기(뒤에서) — 진행은 `GET …/plan/progress`.""" + root, _, segments = await _plan_inputs(project_id) + epsg = segments[0]["epsg"] + + 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) + 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) -> dict[str, Any]: + """연도 구간마다 노선 추천 갈래 넷(뒤에서 · 같은 입력이면 저장본).""" + root, project, segments = await _plan_inputs(project_id) + 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} + return { + "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] +) -> tuple[list[dict[str, Any] | None], list[dict[str, Any]] | None, list[list[float]]]: + """(연도 저장본 · 이은 누가 · 이은 중심선) — 저장본이 하나라도 없으면 None · 빈 선.""" + results = [analyze.read_segment(root, seg) for seg in segments] + if any(result is None for result in results): + return results, None, [] + rows = analyze.chain(results, choices, float(project["route_start_m"] or 0.0)) + return results, rows, analyze.joined_line(results, rows) + + +@router.get("/{project_id}/plan/analysis", response_model=None) +async def plan_analysis(project_id: UUID) -> dict[str, Any]: + """분석 저장본(지점이 지금과 같은 것만) + 고른 갈래(저장값 · 없으면 첫 성공 갈래).""" + root, project, segments = await _plan_inputs(project_id) + choices = {seg["year"]: _saved_mode(seg["route_json"]) for seg in segments} + results, rows, line = await asyncio.to_thread(_assemble, root, project, segments, choices) + return { + "status": "success", + "years": [ + {"year": seg["year"], "start": seg["start"], "end": seg["end"], "analysis": result} + for seg, result in zip(segments, results) + ], + "chain": rows, + "line": line, + } + + +@router.put("/{project_id}/plan/analysis", response_model=None) +async def save_plan_analysis(project_id: UUID, request: ChoiceRequest) -> dict[str, Any]: + """[저장] — 연도별 고른 갈래를 서버 저장본으로 다시 이어 `plan_route_years` 에 씀.""" + root, project, segments = await _plan_inputs(project_id) + 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}년") + 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 = result["routes"][row["mode"]] + body = {**row, **{k: route[k] for k in ("planned", "nodes", "curves")}} + 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} diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Terrain.py b/D03_PlanAnalysis/D03_PlanAnalysis_Terrain.py new file mode 100644 index 000000000..63027aac1 --- /dev/null +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Terrain.py @@ -0,0 +1,221 @@ +"""D 예정노선 지형 받기 — 지점 기준 수치지형도 · 위성 · 연도 구간별 DEM (요구 345 · PLAN 530-1). + +지형은 D 프로젝트 storage 안 `B04_PreProcess/` 배치 그대로(지형 코드 주인 = B04 · 고칠 곳 0): + +- 도엽 = 모든 지점 + 구간 표본(`SAMPLE_STEP_M`) 각 3×3 → `ensure_sheets`(없는 것만 · prune 없음) +- 병합 = 지난번 병합 도엽 ∪ 이번 도엽(옛 도엽 유지 · 열쇠 같으면 건너뜀) +- 위성 = 기준 도엽 도곽 ∪ 옛 위성 범위(덮으면 건너뜀) +- DEM = 연도 구간마다 `models_<연도>/`(긴 노선 메모리 · 비용면 한도) · 범위 = 구간 상자 + 탐색 복도 +- 작업 좌표계 = 지점 epsg · `B03_FileInput/input/prj/plan.prj` 한 장(B04 · B05 좌표계 창구가 읽음) + +계산 자리 = ③ 서버 단독. DB 행(`surface_models`)은 라우터가 씀. +""" + +from __future__ import annotations + +import json +import logging +import math +from pathlib import Path +from typing import Any, Callable + +import numpy as np + +from B04_PreProcess.B04_PreProcess_Engine_MapSheet import ( + neighbors_for_points, + sheet5k_to_bounds, + sheets_for_points, +) +from B04_PreProcess.B04_PreProcess_Engine_SheetParser import parse_and_merge_sheets +from B04_PreProcess.B04_PreProcess_Engine_SheetStore import ( + ensure_sheets, + get_project_map_sheets_dir, + get_sheet_path, +) +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 + +logger = logging.getLogger(__name__) + +# 구간 표본 간격 — 도엽(≈2.8 × 2.3 km) 3×3 묶음이 끊기지 않게 +SAMPLE_STEP_M = 1000.0 +# 탐색 복도 반폭(D 만 · B 기본 100 m) — 실측(530-2): 노선이 복도 끝에 닿지 않는 폭 +PLAN_CORRIDOR_M = 300.0 +# DEM 이 복도 밖으로 더 덮는 폭(복도 끝 칸 기울기 · 횡단 표본 ±20 m) +DEM_EXTRA_M = 50.0 +STAGE = Path("B04_PreProcess") +PRJ_NAME = "plan.prj" + +Progress = Callable[[int, str], None] + + +def models_dir_of(project_root: Path, year: int) -> Path: + """연도 구간 DEM 자리.""" + return project_root / STAGE / f"models_{int(year)}" + + +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 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 write_plan_prj(project_root: Path, epsg: int) -> Path: + """작업 좌표계 .prj — 내용이 같으면 그대로.""" + from pyproj import CRS + + path = project_root / "B03_FileInput" / "input" / "prj" / PRJ_NAME + text = CRS.from_epsg(int(epsg)).to_wkt("WKT1_ESRI") + if not path.is_file() or path.read_text(encoding="utf-8", errors="ignore") != text: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + 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")) + except (OSError, ValueError): + return None + + +def _merge_sheets(processed: Path, store: Path, available: list[str]) -> list[str]: + """병합 = 지난번 병합 도엽(zip 남은 것) ∪ 이번 도엽 · 열쇠 같으면 건너뜀.""" + key_path = processed / "sheet_merge.key.json" + old = (_read_json(key_path) or {}).get("sheets") or [] + sheets = sorted({*available, *(s for s in old if get_sheet_path(store, s) is not None)}) + key = {"sheets": sheets} + if sheets and (_read_json(key_path) != key or not any(processed.glob("도엽_*.geojson"))): + parse_and_merge_sheets(store, sheets, processed) + atomic_write_json(key_path, key) + return sheets + + +def _satellite(processed: Path, prj: Path, points: list[tuple[float, float]], epsg: int) -> str: + """위성 = 기준 도엽 도곽 ∪ 옛 범위 · 이미 덮으면 건너뜀. 돌려줌 = skip · done · failed.""" + from pyproj import Transformer + + to_metric = Transformer.from_crs("EPSG:4326", f"EPSG:{epsg}", always_xy=True) + xs: list[float] = [] + ys: list[float] = [] + for sheet in sheets_for_points(points): + lon0, lat0, lon1, lat1 = sheet5k_to_bounds(sheet) + for lon, lat in ((lon0, lat0), (lon0, lat1), (lon1, lat0), (lon1, lat1)): + x, y = to_metric.transform(lon, lat) + xs.append(x) + ys.append(y) + need = {"x": [min(xs), max(xs)], "y": [min(ys), max(ys)]} + meta = _read_json(processed / "vworld_satellite_meta.json") + try: + old = [float(meta[k]) for k in ("x_min", "x_max", "y_min", "y_max")] if meta else None + except (KeyError, TypeError, ValueError): + old = None + if old and old[0] <= need["x"][0] and old[1] >= need["x"][1]: + if old[2] <= need["y"][0] and old[3] >= need["y"][1]: + return "skip" + if old: + need = { + "x": [min(old[0], need["x"][0]), max(old[1], need["x"][1])], + "y": [min(old[2], need["y"][0]), max(old[3], need["y"][1])], + } + try: + done = download_vworld_satellite_map(prj, need, processed, layer_name="Satellite") + except Exception as exc: # noqa: BLE001 — 위성은 배경 · 지형 받기를 막지 않음 + logger.warning("D 위성 지도 받기 실패: %s", exc) + return "failed" + return "done" if done else "failed" + + +def ensure_plan_terrain( + project_root: Path, + segments: list[dict[str, Any]], + epsg: int, + on_progress: Progress | None = None, +) -> dict[str, Any]: + """지점 기준 지형 받기 — `segments` = [{year, start: (x, y), end: (x, y)}](작업 좌표계 m). + + 돌려줌 = 도엽(받은 · 실패 · 병합) · 위성 상태 · 연도별 모델 등록용 dict 목록. + """ + + def report(percent: int, message: str) -> None: + if on_progress is not None: + on_progress(percent, message) + + if not segments: + 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)}" + + report(5, "수치지형도 도엽 고르는 중") + points = reference_points(segments, epsg) + wanted = neighbors_for_points(points) + 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)}매 받는 중") + got = ensure_sheets(store, wanted) + if not got["available"]: + raise ValueError(f"수치지형도 도엽을 하나도 받지 못했습니다: {got['failed']}") + report(45, "도엽 레이어 병합 중") + merged = _merge_sheets(processed, store, list(got["available"])) + report(60, "위성 지도 받는 중") + satellite = _satellite(processed, prj, points, 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], + ) + if not years[year]: + raise ValueError(f"{year}년 구간 지형면을 만들지 못했습니다 — 도엽 등고선 확인") + report(100, "지형 받기 끝") + return { + "sheets": { + "wanted": len(wanted), + "downloaded": sorted(set(got["available"]) - before), + "failed": got["failed"], + "merged": merged, + }, + "satellite": satellite, + "years": years, + } diff --git a/main.py b/main.py index d9042e54b..40d04b8b5 100644 --- a/main.py +++ b/main.py @@ -72,6 +72,7 @@ from B08_Quantity.B08_Quantity_Router_Summary import router as b08_summary_route from B08_Quantity.B08_Quantity_Router_Material import router as b08_material_router from B08_Quantity.B08_Quantity_Router_Milestone import router as b08_milestone_router from B09_Estimation.B09_Estimation_Router import router as b09_estimation_router +from D03_PlanAnalysis.D03_PlanAnalysis_Router import router as d03_plan_analysis_router from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router from M02_MasterTemplete.M02_Labor_Router import router as m02_labor_router from M02_MasterTemplete.M02_MasterTemplete_Router import router as m02_master_template_router @@ -468,6 +469,7 @@ app.include_router(b04_basins_router, dependencies=protected_with_company) app.include_router(tiles_router, dependencies=protected_with_company) app.include_router(b05_route_router, dependencies=protected_with_company) app.include_router(b05_route_lifecycle_router, dependencies=protected_with_company) +app.include_router(d03_plan_analysis_router, dependencies=protected_with_company) app.include_router(b05_corridor_router, dependencies=protected_with_company) app.include_router(b05_route_replan_router, dependencies=protected_with_company) app.include_router(b05_route_terrain_router, dependencies=protected_with_company) diff --git a/resources/tester/test_530_1_plan_terrain.py b/resources/tester/test_530_1_plan_terrain.py new file mode 100644 index 000000000..3958cbb58 --- /dev/null +++ b/resources/tester/test_530_1_plan_terrain.py @@ -0,0 +1,115 @@ +"""D 예정노선 지형 받기(PLAN 530-1) — 가짜 내려받기 · 옛 도엽 유지 · 새 도엽만 · 빠짐 없음. + +./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 +from pyproj import Transformer + +from B04_PreProcess import B04_PreProcess_Engine_SheetStore as store_mod +from B04_PreProcess.B04_PreProcess_Engine_MapSheet import sheets_for_points +from common_util.common_util_crs import resolve_project_crs +from config.config_system_terrain import SHEET_SURFACE_MARGIN_M +from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain + +EPSG = 5186 +A, B, C = (230000.0, 450000.0), (231500.0, 451200.0), (238000.0, 455000.0) + + +@pytest.fixture +def fake(tmp_path, monkeypatch): + seen = {"download": [], "merge": [], "satellite": [], "dem": []} + + def download_sheet(store_dir, sheet_no, version=2): + seen["download"].append(sheet_no) + store_dir = Path(store_dir) + store_dir.mkdir(parents=True, exist_ok=True) + (store_dir / f"{sheet_no}.zip").write_bytes(b"PK") + index = store_mod._load_index(store_dir) + index["sheets"][sheet_no] = {"file": f"{sheet_no}.zip"} + store_mod._save_index(store_dir, index) + + def merge(store_dir, sheets, output_dir, layers=None): + seen["merge"].append(list(sheets)) + (output_dir / "도엽_등고선.geojson").write_text("{}", encoding="utf-8") + return {} + + def satellite(prj, bounds, output_dir, layer_name="Satellite", ext="jpeg"): + seen["satellite"].append(bounds) + meta = {"x_min": bounds["x"][0], "x_max": bounds["x"][1]} + meta.update({"y_min": bounds["y"][0], "y_max": bounds["y"][1]}) + (output_dir / "vworld_satellite_meta.json").write_text(json.dumps(meta), "utf-8") + return output_dir / "vworld_satellite.png" + + def dem(root, processed, models_dir, route_xy, crs, methods=None): + seen["dem"].append((models_dir.name, route_xy.tolist(), crs)) + return [{"model_file_path": f"B04_PreProcess/{models_dir.name}/dtm.npz"}] + + monkeypatch.setattr(store_mod, "download_sheet", download_sheet) + monkeypatch.setattr(terrain, "parse_and_merge_sheets", merge) + monkeypatch.setattr(terrain, "download_vworld_satellite_map", satellite) + monkeypatch.setattr(terrain, "build_sheet_surface_model", dem) + return tmp_path, seen + + +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: + root, seen = fake + first = terrain.ensure_plan_terrain(root, [{"year": 2026, "start": A, "end": B}], EPSG) + old = _zips(root) + assert sorted(seen["download"]) == old == sorted(first["sheets"]["downloaded"]) + assert first["satellite"] == "done" and list(first["years"]) == [2026] + # 작업 좌표계 .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) + now = _zips(root) + assert set(old) <= set(now) # 옛 도엽 그대로(prune 없음) + 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"] + assert second["sheets"]["failed"] == {} + + # 같은 지점 다시 — 받기 0 · 위성 건너뜀 · 병합 열쇠 같음 + seen["download"].clear() + merges = len(seen["merge"]) + third = terrain.ensure_plan_terrain(root, segs, EPSG) + assert seen["download"] == [] and third["satellite"] == "skip" + assert len(seen["merge"]) == merges + + +def test_far_points_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) + 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"]) + + +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 diff --git a/resources/tester/test_530_2_plan_analyze.py b/resources/tester/test_530_2_plan_analyze.py new file mode 100644 index 000000000..5d364e0af --- /dev/null +++ b/resources/tester/test_530_2_plan_analyze.py @@ -0,0 +1,111 @@ +"""D 예정노선 2단계 분석(PLAN 530-2) — 합성 지형 두 구간 · 경계 · 누가 · 저장본 · 복도 인자. + +./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_analyze.py -q +""" + +import numpy as np +import pytest + +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 + +CRITERIA = {"max_grade": 0.14, "min_radius_m": 12.0, "design_speed": 20} +SEGMENTS = [ + {"year": 2026, "start": (60.0, 200.0), "end": (560.0, 230.0)}, + {"year": 2027, "start": (560.0, 230.0), "end": (1040.0, 160.0)}, +] + + +@pytest.fixture(scope="module") +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)) + + +@pytest.fixture +def run(tmp_path, hill): + _POOLS.clear() + calls = [] + + def loader(root, selection, models_dir, cache_dir): + calls.append((models_dir.name, cache_dir.name)) + return hill, [] + + def go(criteria=CRITERIA, corridor_m=150.0): + return analyze.analyze_plan( + tmp_path, SEGMENTS, criteria, loader=loader, corridor_m=corridor_m + ) + + go.calls = calls + yield go + _POOLS.clear() + + +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] == list(seg["start"]) + assert route["planned"][-1] == list(seg["end"]) + 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] == list(SEGMENTS[0]["end"]) + + +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_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") diff --git a/resources/tester/test_530_2_plan_router.py b/resources/tester/test_530_2_plan_router.py new file mode 100644 index 000000000..f8755420c --- /dev/null +++ b/resources/tester/test_530_2_plan_router.py @@ -0,0 +1,127 @@ +"""D 분석 API(PLAN 530-2) — GET 이은 누가 · PUT [저장] = 서버 저장본으로 다시 셈 · 모르는 갈래 거절. + +./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_router.py -q +""" + +import json + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +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_Router as router_mod + +PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2" +P0, P1, P2 = (0.0, 0.0), (300.0, 400.0), (900.0, 400.0) + + +def _route(points: list, length: float) -> dict: + return {"planned": points, "nodes": [], "curves": [], "metrics": {"length_m": length}} + + +class _Cursor: + def __init__(self, log): + self.log = log + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def execute(self, sql, args=None): + self.log.append((sql, args)) + + +class _Connection(_Cursor): + async def begin(self): + pass + + async def commit(self): + self.log.append(("COMMIT", None)) + + async def rollback(self): + pass + + def cursor(self, *args): + return _Cursor(self.log) + + +class _Pool: + def __init__(self, log): + self.log = log + + def acquire(self): + return _Connection(self.log) + + +@pytest.fixture +def client(tmp_path, monkeypatch): + segments = [ + {"year": 2026, "start": P0, "end": P1, "epsg": 5186, "route_json": None}, + {"year": 2027, "start": P1, "end": P2, "epsg": 5186, "route_json": '{"mode": "gentle"}'}, + ] + for seg, routes in ( + (segments[0], {"gradeline": _route([[0, 0], [100, 200], [300, 400]], 530.0)}), + ( + segments[1], + { + "gradeline": _route([[300, 400], [900, 400]], 600.0), + "gentle": _route([[300, 400], [600, 500], [900, 400]], 640.0), + "earthwork": {"error": "길 없음"}, + }, + ), + ): + atomic_write_json( + analyze.derived_dir(tmp_path) / f"route_{seg['year']}.json", + { + "year": seg["year"], + "start": list(seg["start"]), + "end": list(seg["end"]), + "fingerprint": "f", + "routes": routes, + }, + ) + project = {"road_type": "work", "design_speed_kph": None, "route_start_m": 1000.0} + + async def inputs(project_id): + return tmp_path, project, segments + + log: list = [] + monkeypatch.setattr(router_mod, "_plan_inputs", inputs) + monkeypatch.setattr(router_mod, "get_db_pool", lambda: _Pool(log)) + app = FastAPI() + app.include_router(router_mod.router) + test = TestClient(app) + test.log = log + 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]] + + +def test_put_saves_server_routes(client) -> None: + got = client.put(f"/api/projects/{PID}/plan/analysis", json={"choices": {"2027": "gradeline"}}) + assert got.status_code == 200 + assert got.json()["chain"][1]["end_chainage_m"] == 2130.0 + updates = [args for sql, args in client.log if sql.startswith("UPDATE plan_route_years")] + saved = {args[3]: json.loads(args[0]) for args in updates} + assert saved[2027]["mode"] == "gradeline" and saved[2027]["planned"] == [[300, 400], [900, 400]] + assert saved[2026]["start_chainage_m"] == 1000.0 and updates[0][1] == "f" + assert client.log[-1][0] == "COMMIT" + + +def test_put_rejects_unknown_or_empty_mode(client) -> None: + url = f"/api/projects/{PID}/plan/analysis" + assert client.put(url, json={"choices": {"2027": "follow"}}).status_code == 400 + assert client.put(url, json={"choices": {"2027": "earthwork"}}).status_code == 409 + assert not [sql for sql, _ in client.log if sql.startswith("UPDATE")]