diff --git a/B05_Profile/B05_Profile_Engine_InitialProfile.py b/B05_Profile/B05_Profile_Engine_InitialProfile.py new file mode 100644 index 000000000..ae629a60f --- /dev/null +++ b/B05_Profile/B05_Profile_Engine_InitialProfile.py @@ -0,0 +1,263 @@ +"""추천 후보 초기 종단(476 · 요구 276) — [적용]하면 작업본 · 초기값에 설 종단을 미리 셈. + +[적용] 체인(`B05_Profile_Route_Save.run_route_save_chain`)과 같은 조각을 메모리에서 부름(파일 안 +씀): + 후보 노드 → 화면 `flattenServerPlan` 꼴(`draft_nodes`) → `build_planned_polyline`(편집 갈래 · csv + 4자리) → 관 · 구조물 · 추가 측점 따라감(`followed_pipes` · `followed_structures` · + `followed_plus_rows` · 새 세류선 교차 관 포함) → 측점 틀 · 1차 계획선 438(`build_sections`) → + 「종단 반영」 켬이면 1 m 균형 종단을 꼭짓점에 얹음(`seat_base_pvi` → `rebuild_alignment_profile`). +[적용]은 이 값을 받아 적지 않음 — 서버가 체인에서 같은 조각으로 다시 셈(CLAUDE.md 5). + +저장본 = `B05_Profile/route_recommend/initial/<결과 키>.json` `{fingerprint, profile}` — 결과 정리 +(`*.json` glob)가 안 보는 하위 폴더 · 결과 짝 없는 파일은 쓸 때 지움 · 지문 다르면 다시 셈. +응답 칸 `initial_profile`(화면 476-2) = `{stations_m, ground_m, plan_m, line: [[s, z]…], pvi_count, +curve_count, straightened, max_grade_pct, curve_length_m, warnings, with_profile, seconds}` · 실패 = +`{error}`. +""" + +from __future__ import annotations + +import hashlib +import json +import math +from pathlib import Path +from typing import Any + +from common_util.common_util_json import atomic_write_json + +INITIAL_DIR = "B05_Profile/route_recommend/initial" +# 초기 종단을 낳는 코드(저장소 뿌리 기준) — 바뀌면 저장본 다시 셈 +SOURCE_GLOBS = ( + "B05_Profile/B05_Profile_Engine_InitialProfile.py", + "B05_Profile/B05_Profile_Engine_Grade*.py", + "B05_Profile/B05_Profile_Engine_Sections*.py", + "B05_Profile/B05_Profile_Engine_RouteInitial_Profile.py", + "B05_Profile/B05_Profile_Route_Save_Parts.py", + "B05_Profile/B05_Profile_Plus_Stations.py", + "common_util/common_util_route_polyline.py", + "common_util/common_util_route_follow.py", + "common_util/common_util_structure_station.py", + "common_util/common_util_drainage_pipes.py", + "config/config_system_design.py", +) +# 직선 위 꼭짓점(438 직선화) 판정 — 앞뒤 기울기 차(비율) +_STRAIGHT_EPS = 1e-5 + + +def draft_nodes( + nodes: list[dict[str, Any]], curves: list[dict[str, Any]] +) -> tuple[list[tuple[float, float]], list[bool], list[float | None]]: + """후보 노드 · 곡선 → [적용] 초안(노드 · 곡선 켬 · R) — 화면 `flattenServerPlan`(편집 없음) · + `snapshotOf` · `routeDraftOf` 와 같은 꼴(곡선 = 교각점 하나 · R 0.01 m 반올림 = Math.round).""" + if not curves: + return ( + [(float(n["x"]), float(n["y"])) for n in nodes], + [n.get("curve") is not False for n in nodes], + [n.get("radius_m") for n in nodes], + ) + starts = {int(c["node_first"]): c for c in curves} + inside = {i for c in curves for i in range(int(c["node_first"]) + 1, int(c["node_last"]) + 1)} + points: list[tuple[float, float]] = [] + flags: list[bool] = [] + radii: list[float | None] = [] + for index, node in enumerate(nodes): + if index in inside: + continue + curve = starts.get(index) + points.append( + (float(curve["apex"][0]), float(curve["apex"][1])) + if curve + else (float(node["x"]), float(node["y"])) + ) + flags.append(curve is not None) + radii.append(math.floor(float(curve["radius_m"]) * 100 + 0.5) / 100 if curve else None) + return points, flags, radii + + +def candidate_line( + nodes: list[dict[str, Any]], curves: list[dict[str, Any]], radius_m: float +) -> tuple[list[tuple[float, float]], list[dict[str, Any]]]: + """[적용]이 쓸 노선(csv 4자리 = 풀린 노선 geojson) · 곡선표 — `_write_planned_polyline` 짜기.""" + from common_util.common_util_route_polyline import build_planned_polyline + + points, flags, radii = draft_nodes(nodes, curves) + outline = build_planned_polyline( + points, min_radius_m=radius_m, simplify=False, curve_flags=flags, radii=radii + ) + line = [(round(x, 4), round(y, 4)) for x, y in outline.vertices] + return line, [curve.as_dict() for curve in outline.curves] + + +def followed_facilities( + project_root: Path, + line: list[tuple[float, float]], + old_route_data_path: str | None, + old_xy: list[list[float]], + old_pipes: list, + streams: list, +) -> tuple[list, list, list[float] | None]: + """체인과 같은 따라감(파일 안 씀) → (관 · 구조물 · 추가 측점 누가거리 | 없음 None).""" + from B05_Profile.B05_Profile_Plus_Stations import followed_plus_rows, load_plus_stations + from B05_Profile.B05_Profile_Route_Save import followed_pipes + from B05_Profile.B05_Profile_Structures_Repository import load_structures + from B05_Profile.B05_Profile_Structures_Router import followed_structures + from common_util.common_util_drainage_pipes import ( + ensure_pipe_ids, + fill_pipe_coordinates, + parse_pipe_points, + ) + + points = [{"x": x, "y": y} for x, y in line] + items = load_structures(str(project_root))[1] + if items: + followed_structures(project_root, items, old_route_data_path, points) + rows = load_plus_stations(project_root) + plus = ( + sorted({float(r["chainage_m"]) for r in followed_plus_rows(rows, line)["rows"]}) + if rows + else None + ) + pairs, _crossing, vertices = followed_pipes(old_pipes, old_xy, points, streams) + pipes = [pipe for _, pipe in pairs] + fill_pipe_coordinates(pipes, vertices) + ensure_pipe_ids(pipes) + # 체인은 관 정본을 쓰고 다시 읽음(지문 같음 = 저장분 그대로) — 같은 왕복 + return parse_pipe_points([pipe.as_dict() for pipe in pipes]), items, plus + + +def initial_longitudinal( + project_root: Path, + line: list[tuple[float, float]], + curves: list[dict[str, Any]], + sampler: Any, + facilities: tuple[list, list, list[float] | None], + *, + options: Any, + grade_options: Any, + source_snapshot: dict[str, Any], + crs: str | None, + target: tuple[list[float], list[float]] | None, +) -> dict[str, Any]: + """[적용] 체인이 남길 종단(측점 · 계획선) — 측점 생성 · 균형 얹기와 같은 조각.""" + from B05_Profile.B05_Profile_Engine_Grade_Profile import rebuild_alignment_profile + from B05_Profile.B05_Profile_Engine_Sections import build_sections, resolve_extra_stations + from B05_Profile.B05_Profile_Router_RouteInitial import seat_base_pvi + + polyline = [[x, y, 0.0] for x, y in line] + pipes, structures, plus = facilities + extras = resolve_extra_stations(project_root, pipes, polyline, structures=structures, plus=plus) + result, _summary = build_sections( + polyline, + sampler, + pipes, + extras, + options=options, + grade_options=grade_options, + source_snapshot=source_snapshot, + crs=crs, + design_curves=curves, + station_snap=True, # [적용] = 스냅 켬 · 측점 No.(`saveRecommendedRoute`) + station_snap_step=None, + ) + longitudinal = result["longitudinal"] + if target is not None and longitudinal.get("profile_alignment"): + seat_base_pvi(longitudinal, *target) + alignment, profile = rebuild_alignment_profile(longitudinal, {}) + longitudinal["profile_alignment"] = alignment + longitudinal["design_profiles"][0] = profile + return longitudinal + + +def straightened_count(base: list[dict[str, Any]]) -> int: + """앞뒤 기울기가 같은 안쪽 꼭짓점 수 = 직선 위(438 직선화 · 종단곡선 없음).""" + s = [float(p["chainage_m"]) for p in base] + z = [float(p["elevation_m"]) for p in base] + grades = [(z[i + 1] - z[i]) / (s[i + 1] - s[i]) for i in range(len(s) - 1)] + return sum(abs(grades[i] - grades[i - 1]) < _STRAIGHT_EPS for i in range(1, len(grades))) + + +def profile_summary(longitudinal: dict[str, Any]) -> dict[str, Any]: + """응답 칸 `initial_profile` — 측점 · 지반 · 계획고 · 계획선(종단곡선 포함) · 수.""" + alignment = longitudinal.get("profile_alignment") + if not isinstance(alignment, dict) or not alignment.get("base_pvi"): + return {"error": "계획선을 세우지 못했습니다(지반고 종단)."} + summary = longitudinal["design_profiles"][0]["summary"] + rows = alignment.get("stations") or [] + return { + "stations_m": [round(float(r["chainage_m"]), 3) for r in rows], + "ground_m": [round(float(r["ground_elevation_m"]), 3) for r in rows], + "plan_m": [round(float(r["plan_elevation_m"]), 4) for r in rows], + "line": [ + [round(float(p["chainage_m"]), 3), round(float(p["elevation_m"]), 4)] + for p in alignment.get("samples") or [] + ], + "pvi_count": len(alignment["base_pvi"]), + "curve_count": int(summary.get("vertical_curve_count") or 0), + "straightened": straightened_count(alignment["base_pvi"]), + "max_grade_pct": round(float(summary.get("max_grade_pct") or 0.0), 2), + "curve_length_m": (alignment.get("policy") or {}).get("default_curve_length_m"), + "warnings": list(alignment.get("warnings") or []), + } + + +def source_fingerprint() -> str: + """초기 종단 코드 지문 — `SOURCE_GLOBS` 내용(줄끝 LF).""" + root = Path(__file__).resolve().parents[1] + digest = hashlib.blake2b(digest_size=8) + for path in sorted({p for pattern in SOURCE_GLOBS for p in root.glob(pattern)}): + digest.update(path.name.encode()) + digest.update(path.read_bytes().replace(b"\r\n", b"\n")) + return digest.hexdigest() + + +def file_marks(paths: list[Path]) -> list[str]: + """지문에 드는 파일 자국 — 내용 해시(작은 정본) · 없음 「-」.""" + marks = [] + for path in paths: + try: + marks.append(hashlib.blake2b(path.read_bytes(), digest_size=8).hexdigest()) + except OSError: + marks.append("-") + return marks + + +def stat_marks(paths: list[Path]) -> list[str]: + """큰 자료(지형 npz · 도엽) 자국 — 이름 · 크기 · 수정시각.""" + marks = [] + for path in sorted(paths): + try: + info = path.stat() + marks.append(f"{path.name}:{info.st_size}:{info.st_mtime_ns}") + except OSError: + marks.append(f"{path.name}:-") + return marks + + +def fingerprint(parts: dict[str, Any]) -> str: + text = json.dumps({**parts, "source": source_fingerprint()}, sort_keys=True, default=str) + return hashlib.blake2b(text.encode(), digest_size=16).hexdigest() + + +def _path(project_root: Path, key: str) -> Path: + return Path(project_root) / INITIAL_DIR / f"{key}.json" + + +def read_initial(project_root: Path, key: str, mark: str) -> dict[str, Any] | None: + """저장본 — 없거나 지문 다르면 None.""" + try: + data = json.loads(_path(project_root, key).read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + return ( + data.get("profile") if isinstance(data, dict) and data.get("fingerprint") == mark else None + ) + + +def write_initial(project_root: Path, key: str, mark: str, profile: dict[str, Any]) -> None: + """한 결과에 하나(덮어씀) · 결과 저장본 짝 없는 파일 지움.""" + path = _path(project_root, key) + atomic_write_json(path, {"fingerprint": mark, "profile": profile}) + results = path.parent.parent + for other in path.parent.glob("*.json"): + if not (results / other.name).exists(): + other.unlink(missing_ok=True) diff --git a/B05_Profile/B05_Profile_Router_RouteInitial.py b/B05_Profile/B05_Profile_Router_RouteInitial.py index 19579feec..bc894ebe2 100644 --- a/B05_Profile/B05_Profile_Router_RouteInitial.py +++ b/B05_Profile/B05_Profile_Router_RouteInitial.py @@ -66,6 +66,7 @@ async def compute_initial( request_fingerprint, write_stored, ) + from B05_Profile.B05_Profile_Router_RouteInitial_Initial import initial_profile_of from common_util.common_util_surface_confirmation import ( analysis_terrain_params, get_surface_confirmation_params, @@ -175,6 +176,13 @@ async def compute_initial( profile.update(cover_shortfall(profile, line, pipes)) except (FileNotFoundError, ValueError) as error: return JSONResponse(status_code=409, content={"status": "error", "message": str(error)}) + # [적용]하면 설 초기 종단(476) — 같은 조각으로 미리 셈 · 저장본 · 실패 = {error}(추천은 그대로) + result = { + **result, + "initial_profile": await initial_profile_of( # 476 + project_id, project_root, paths[1], result, request, selection, key + ), + } logger.info( "초기 계획노선 %s(%s): project_id=%s %.1fs 연장 %.0fm 이격 평균 %.1fm", request.mode, diff --git a/B05_Profile/B05_Profile_Router_RouteInitial_Initial.py b/B05_Profile/B05_Profile_Router_RouteInitial_Initial.py new file mode 100644 index 000000000..a18f11767 --- /dev/null +++ b/B05_Profile/B05_Profile_Router_RouteInitial_Initial.py @@ -0,0 +1,185 @@ +"""추천 후보 초기 종단 모으기(476) — `POST /route/initial` 결과에 `initial_profile` 을 붙임. + +입력 모으기(DB · 파일) · 지문 · 저장본 = 여기 · 셈 조각 = `B05_Profile_Engine_InitialProfile`(머리에 +흐름 · 응답 칸). `B05_Profile_Router_RouteInitial`(700줄)에서 나눔 — `compute_initial` 이 부름. +""" + +from __future__ import annotations + +import asyncio +import functools +import logging +import time +from pathlib import Path +from typing import Any +from uuid import UUID + +from fastapi.responses import JSONResponse + +from config.config_db import get_db_pool + +logger = logging.getLogger(__name__) + + +async def initial_profile_of( + project_id: UUID, + project_root: Path, + stored_path: str, + result: dict[str, Any], + request: Any, + selection: dict[str, Any], + key: str, +) -> dict[str, Any]: + """추천 후보 초기 종단(476 · `B05_Profile_Engine_InitialProfile` 머리) — [적용] 체인과 같은 + 입력 · 같은 조각 · 지문 같으면 저장본 꺼냄. 실패 = `{error}`.""" + from dataclasses import asdict + + import aiomysql + + from B03_FileInput.B03_FileInput_Service_Chain import redesign_solve_request + from B04_PreProcess.B04_PreProcess_Router_Watershed import STREAM_FILE, _prepare, _sheet_dir + from B05_Profile import B05_Profile_Engine_InitialProfile as ip + from B05_Profile import B05_Profile_Router_Replan as replan + from B05_Profile.B05_Profile_Engine_Sections import _MODELS_SUBDIR + from B05_Profile.B05_Profile_Plus_Stations import plus_stations_path + from B05_Profile.B05_Profile_Repository import get_latest_route + from B05_Profile.B05_Profile_Route_Save import _xy, section_seed + from B05_Profile.B05_Profile_Router_RouteInitial import balanced_profile_of, project_standard + from B05_Profile.B05_Profile_Structures_Repository import structures_file_path + from B06_Section.B06_Section_Repository import ( + get_latest_section_options, + get_route_generation_source, + ) + from B06_Section.B06_Section_Router_Regenerate_Options import ( + _regeneration_grade_options, + _regeneration_options, + ) + from common_util.common_util_drainage_pipes import pipe_points_path, read_pipe_points_file + from common_util.common_util_surface_sampler import build_surface_sampler + from common_util.common_util_workflow_state import get_workflow_state + + started = time.perf_counter() + try: + radius_m, _, _ = await replan._plan_criteria(project_id) # [적용] 노선 짜기와 같은 R + line, curves = ip.candidate_line(result["nodes"], result.get("curves") or [], radius_m) + pool = get_db_pool() + async with pool.acquire() as connection: + latest = await get_latest_route(connection, project_id) + source = ( + await get_route_generation_source(connection, project_id, int(latest["id"])) + if latest + else None + ) + old_options = await get_latest_section_options(connection, project_id) + async with connection.cursor(aiomysql.DictCursor) as cursor: + state = await get_workflow_state(cursor, str(project_id)) + if not latest: + return {"error": "저장된 노선이 없어 초기 종단을 셀 수 없습니다(자동설계 먼저)."} + stage2 = next((s for s in state.get("stages", []) if int(s.get("stage_no", -1)) == 2), None) + params = dict((stage2 or {}).get("params") or {}) + first, last = line[0], line[-1] + # 체인 = 새 노선 점으로 stage 2 를 세움 — 측점 간격 · 계획선 기준엔 무관 + params["points"] = { + "bp": {"x": first[0], "y": first[1]}, + "ep": {"x": last[0], "y": last[1]}, + } + model_id = int(latest.get("surface_model_id") or 0) + seed = section_seed(redesign_solve_request(params, selection, model_id), old_options) + options = _regeneration_options(seed, params, float(seed["cross_half_width_m"])) + grade_options = _regeneration_grade_options(params) + epsg = (source or {}).get("crs_epsg") + crs = f"EPSG:{epsg}" if epsg is not None else None + surface = {k: selection.get(k) for k in ("source_filter", "method", "smooth")} + models = project_root / _MODELS_SUBDIR + route_path = str(latest.get("route_data_path") or "") + mark = ip.fingerprint( + { + "key": key, + "nodes": result["nodes"], + "curves": result.get("curves"), + "radius": radius_m, + "surface": surface, + "models": ip.stat_marks(list(models.glob(f"*_{surface['source_filter']}*.npz"))), + "streams": ip.stat_marks([_sheet_dir(stored_path) / STREAM_FILE]), + "files": ip.file_marks( + [ + pipe_points_path(stored_path), + Path(structures_file_path(str(project_root))), + plus_stations_path(project_root), + project_root / route_path, + ] + ), + "options": asdict(options), + "grade": grade_options.as_dict() if grade_options else None, + "crs": crs, + "with_profile": request.with_profile, + "criteria": request.criteria.model_dump() if request.criteria else None, + "standard": await project_standard(project_id), + "road": await replan._road_settings(project_id), + } + ) + stored = await asyncio.to_thread(ip.read_initial, project_root, key, mark) + if stored is not None: + return stored + prepared = await _prepare(project_id) + streams = [] if isinstance(prepared, JSONResponse) else prepared["streams"] + facilities = await asyncio.to_thread( + ip.followed_facilities, + project_root, + line, + route_path or None, + _xy(project_root, route_path), + read_pipe_points_file(pipe_points_path(stored_path)), + streams, + ) + target = None + if request.with_profile: # [적용] `balanced_target` 과 같은 셈(노선 정점 · 새 곡선표) + points = [{"x": x, "y": y} for x, y in line] + balanced = await balanced_profile_of( + project_id, project_root, points, request.criteria, curves + ) + target = None if balanced is None else (balanced["stations_m"], balanced["plan_m"]) + sampler = await asyncio.to_thread( + build_surface_sampler, + models, + str(surface["source_filter"]), + str(surface["method"] or "dtm"), + bool(surface["smooth"]), + ) + longitudinal = await asyncio.to_thread( + functools.partial( + ip.initial_longitudinal, + project_root, + line, + curves, + sampler, + facilities, + options=options, + grade_options=grade_options, + source_snapshot={ + "filter": surface["source_filter"], + "method": surface["method"], + "smooth": bool(surface["smooth"]), + }, + crs=crs, + target=target, + ) + ) + summary = { + **ip.profile_summary(longitudinal), + "with_profile": bool(request.with_profile), + "seconds": round(time.perf_counter() - started, 2), + } + await asyncio.to_thread(ip.write_initial, project_root, key, mark, summary) + logger.info( + "추천 초기 종단 셈: project_id=%s %s 꼭짓점 %s 직선화 %s %.2fs", + project_id, + request.mode, + summary.get("pvi_count"), + summary.get("straightened"), + summary["seconds"], + ) + return summary + except Exception as error: # noqa: BLE001 — 추천은 그대로 · 화면이 이유 한 줄 + logger.exception("추천 초기 종단 셈 실패: project_id=%s", project_id) + return {"error": f"초기 종단을 셈하지 못했습니다: {error}"} diff --git a/B05_Profile/logic/B05_Profile_UI_RouteEdit_Edits.ts b/B05_Profile/logic/B05_Profile_UI_RouteEdit_Edits.ts index abafe855d..359bd14f1 100644 --- a/B05_Profile/logic/B05_Profile_UI_RouteEdit_Edits.ts +++ b/B05_Profile/logic/B05_Profile_UI_RouteEdit_Edits.ts @@ -345,6 +345,9 @@ export interface RouteNodeEdit { * * `edits`(430 · 노드 수와 같을 때만 서버가 줌)가 있으면 곡선 켬끔 · R · 고정 · 고정 길이는 **넣은 값**으로 — * 그린 R 로 열면 접선 자리에 눌린 R 이 넣은 값으로 굳고 R · 길이 고정이 [저장] 뒤 사라짐. + * + * 짝(edits 없음) = 파이썬 `B05_Profile_Engine_InitialProfile.draft_nodes`(추천 초기 종단 476) · + * 거울 `test_476_1_recommend_initial`. */ export function flattenServerPlan( nodes: ServerNode[], diff --git a/docs/wiki/concepts/design_page/calculations.md b/docs/wiki/concepts/design_page/calculations.md index f69e093a8..c24929aaf 100644 --- a/docs/wiki/concepts/design_page/calculations.md +++ b/docs/wiki/concepts/design_page/calculations.md @@ -57,6 +57,7 @@ source: ["docs/raw/조사/2026-10-19_472_구조다시세움_목록.md", "PLAN.md | D3 | 3D 길 모형 | `B05_Profile_UI_Corridor_Build.ts` · Worker | — | co | ②(`B05_Profile_Corridor_Prebuild.py`) | | D4 | 평면 발자국 | `logic/B05_Profile_PlanFootprint.ts:160` · `logic/B05_Profile_Footprint_Stage.ts` | `B07_DesignDetail_Engine_Cad_PlanSlope.py`(B07 따로) | — | 브라우저 · `test_346_2_plan_footprint_mirror` | | N1 | 노선 추천 | `B05_Profile_UI_RouteEdit_Initial_Model.ts`(보기만) | `B05_Profile_Engine_RouteInitial*` | — | 서버 · `test_419_1_recommend_engine` | +| N2 | 추천 후보 초기 종단(476) — [적용] 체인과 같은 조각을 메모리에서(노선 짜기 · 시설 따라감 · `build_sections` 438 · 켬 = 균형 얹기) · 저장본 `route_recommend/initial` · 화면 펴기 짝 = `flattenServerPlan` ↔ `draft_nodes` | 그리기만(476-2) · `logic/B05_Profile_UI_RouteEdit_Edits.ts` flattenServerPlan | `B05_Profile_Engine_InitialProfile.py` · `B05_Profile_Router_RouteInitial_Initial.py` | — | 서버 · [적용]은 체인이 다시 셈 · `test_476_1_recommend_initial`(체인 대조 0 mm) | ## 짝인데 값이 갈리는 자리 diff --git a/docs/wiki/concepts/design_page/rules.md b/docs/wiki/concepts/design_page/rules.md index 4854979b6..113a2e6cf 100644 --- a/docs/wiki/concepts/design_page/rules.md +++ b/docs/wiki/concepts/design_page/rules.md @@ -30,7 +30,7 @@ source: ["CLAUDE.md §5", "docs/raw/조사/2026-10-19_465_요구272_값안변함 | 측점 열쇠 = `common_util_station_key` 한 벌 · 노선 바뀜 = 관 · 구조물 좌표(x · y) 따라감 · 누가거리 다시 매김 | `common_util_station_key.ts:32` · `common_util_route_follow.ts:291` | `test_344_3_station_key_mirror` · `test_410_follow_xy` | | 노선 바뀜 넘김 = 열쇠마다 지반 · 측구 종류 · 측구 선택 · 사면 쉬프트 · 소단 · 소단 깊이 · 자리 칸(`SEAT_KEYS` = 암 경계 · 암 절토 경사 · 2단 경사 · 손 상단측 = 단면 유형 · 측구 방향)은 같은 자리(`sameSeat` · 중심 0.01 m · 축 1e-3) 측점만 · 새 자리 = 기본값 · 세션 값은 쓴 때 자리(`@seat`)로 판정(↶ ↷ 유지) · 끌기 갈아끼움 · 미리보기 · Node 넘김 · 서버 노선 저장 한 규칙 | `logic/B06_Section_Station_Swap.ts` SEAT_KEYS · unseated · `logic/B06_Section_Route_Carry.ts` carryKeyedStates · `B05_Profile/logic/B05_Profile_Pipeline_Detail.ts` crossAtSeat · `B05_Profile_Route_Save_Parts.py` carry_cross_inputs | `test_475_3_seat_carry` | | 노선 초안 ↶(버림) = 차례 하나 · 버림 동안 종단도 그리기 멈춤 · 낡음 다시 셈 미룸 → 되돌림 · 저장 기준 맞춤 · 뒤 차례 끝에 한 번 그림(사이 섞인 유토곡선 없음) | `design_pipeline_page.ts:642 · 403` · `B06_Section_Section_Store.ts:316` | `test_472_5c_drop_once` | -| 추천 [적용] 종단 = 새 규칙 초기 종단 — 끔 · 균형 셈 실패 = 1차 계획선(438 · L 20) · 켬 = 균형 종단을 꼭짓점에 얹음(438) · 옛 계획고 · 편집 · L 안 옮김 · 손 노선 저장만 옮김 | `B05_Profile_Route_Save.py` run_route_save_chain `initial` | `test_476_1_apply_initial` | +| 추천 [적용] 종단 = 새 규칙 초기 종단 — 끔 · 균형 셈 실패 = 1차 계획선(438 · L 20) · 켬 = 균형 종단을 꼭짓점에 얹음(438) · 옛 계획고 · 편집 · L 안 옮김 · 손 노선 저장만 옮김 | `B05_Profile_Route_Save.py` run_route_save_chain `initial` · 미리 보기 = `B05_Profile_Engine_InitialProfile`(같은 조각) | `test_476_1_apply_initial` · `test_476_1_recommend_initial` | | B05 · B06 캐시 = 한 벌 · [저장] 한 번에 함께 나감 | `design_shared.ts:395-448` · `design_save.ts:237` | — | | 관 · 구조물 주인 = 가게 칸 하나 — 화면 사본은 가게 알림으로 맞춤(제 쓰기는 같은 값 · 남 고침은 다시 실음) | `design_view_profile.ts` 가게 구독 · `design_view_cross.ts` onShop · `B06_Section_UI_Page_Structures_Panel.ts` seen | `test_472_5_shop_owner` | | 정본 자리 = 값을 낳은 주인 — 노선 시설 = `pipe_points.json` · `structures.json` · 횡단 값 = 횡단 패치 | `B05_Profile_Api_Pipes_Draft.ts:30-37` · `B05_Profile_Structures_Router.py:386-405` · `B06_Section_Section_Store.ts:449` | — | diff --git a/resources/tester/helper_476_1_flatten.cjs b/resources/tester/helper_476_1_flatten.cjs new file mode 100644 index 000000000..55402bede --- /dev/null +++ b/resources/tester/helper_476_1_flatten.cjs @@ -0,0 +1,39 @@ +/* 476-1 추천 [적용] 초안 — 화면 `flattenServerPlan`(logic/B05_Profile_UI_RouteEdit_Edits.ts) + + * `routeDraftOf` 칸 꼴(곡선 켬 = !== false · R ?? null)을 Node 로 돌려 JSON 한 줄 + * (`test_476_1_recommend_initial.py` 가 파이썬 `draft_nodes` 와 견줌). + * node helper_476_1_flatten.cjs <후보 {nodes, curves} JSON 파일> + */ +const fs = require("fs"); +const path = require("path"); +const ts = require(path.join(__dirname, "..", "..", "config", "node_modules", "typescript")); + +const ROOT = path.join(__dirname, "..", ".."); +const cache = new Map(); + +function loadTs(file) { + const full = path.resolve(file); + if (cache.has(full)) return cache.get(full); + const js = ts.transpileModule(fs.readFileSync(full, "utf8"), { + compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 }, + }).outputText; + const box = { exports: {} }; + cache.set(full, box.exports); + const localRequire = (request) => + request.startsWith(".") + ? loadTs(path.join(path.dirname(full), `${request}.ts`)) + : require(request); + new Function("exports", "module", "require", js)(box.exports, box, localRequire); + cache.set(full, box.exports); + return box.exports; +} + +const edits = loadTs(path.join(ROOT, "B05_Profile", "logic", "B05_Profile_UI_RouteEdit_Edits.ts")); +const body = JSON.parse(fs.readFileSync(process.argv[2], "utf8")); +const flat = edits.flattenServerPlan(body.nodes, body.curves); +process.stdout.write( + JSON.stringify({ + nodes: flat.planned.map(([x, y]) => [x, y]), + curveOn: flat.planned.map((_, i) => flat.curveOn[i] !== false), + curveRadius: flat.planned.map((_, i) => flat.curveRadius[i] ?? null), + }), +); diff --git a/resources/tester/test_476_1_recommend_initial.py b/resources/tester/test_476_1_recommend_initial.py new file mode 100644 index 000000000..a8d923301 --- /dev/null +++ b/resources/tester/test_476_1_recommend_initial.py @@ -0,0 +1,409 @@ +# -*- coding: utf-8 -*- +"""476-1 추천 후보 초기 종단 — 미리 셈 = [적용] 체인 종단 · 화면 펴기 짝 · 응답 칸 · 저장본 지문. + +고정 자료 = `fixtures/golden_472`(사본) · 체인 쪽 = `run_route_save_chain` 파일 길 차례(노선 csv → +구조물 · 추가 측점 따라감 → 노선 geojson → 관 따라감 · 정본 → `run_section_generation` → +켬이면 `_set_base_pvi` + `_apply_alignment_edits`) · 미리 셈 쪽 = 메모리 조각(파일 안 씀). +화면 펴기 = TS `flattenServerPlan`(Node `helper_476_1_flatten.cjs`) ↔ 파이썬 `draft_nodes`. +""" + +from __future__ import annotations + +import asyncio +import copy +import json +import subprocess +import sys +from pathlib import Path +from uuid import uuid4 + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +TESTER = Path(__file__).resolve().parent +for path in (ROOT, TESTER): + if str(path) not in sys.path: + sys.path.insert(0, str(path)) + +import helper_472_1_paths as P # noqa: E402 +from test_224_3_recommend_store import _ask, wired # noqa: E402, F401 — 가짜 세계 fixture + +from B05_Profile import B05_Profile_Engine_InitialProfile as ip # noqa: E402 + +RADIUS = 12.0 +HELPER = TESTER / "helper_476_1_flatten.cjs" +NODES_CSV = "B05_Profile/route/planned_route_initial_nodes.csv" +WALL = { + "structure_id": "wall600", + "type_id": "retaining_wall", + "placement": "interval", + "chainage_m": 600.0, + "start_m": 595.0, + "end_m": 605.0, + "options": {"wall_form": "콘크리트", "length_m": 10, "height_m": 2}, + "memo": "", + "placement_source": "manual", + "status": "draft", + "revision": 0, + "geometry": None, + "summary_item": None, + "group_items": None, + "x": 208785.614, + "y": 367845.1028, +} + + +def _candidate(root: Path, shift: float = 0.0) -> dict: + """엔진 결과 꼴 후보(`nodes` · `curves` as_dict) — 교점 9 를 동으로 `shift` m.""" + from B05_Profile.B05_Profile_Router_Replan import _vertices_of + from common_util.common_util_route_polyline import build_planned_polyline + + nodes = [tuple(p) for p in _vertices_of(root / NODES_CSV)] + nodes[9] = (nodes[9][0] + shift, nodes[9][1]) + outline = build_planned_polyline(nodes, min_radius_m=RADIUS, simplify=False) + return { + "nodes": [n.as_dict() for n in outline.nodes], + "curves": [c.as_dict() for c in outline.curves], + } + + +def _browser_draft(candidate: dict, work: Path) -> dict: + if not (ROOT / "config" / "node_modules" / "typescript").is_dir(): + pytest.skip("프론트엔드 의존성 미설치") + body = work / "candidate.json" + body.write_text(json.dumps(candidate), encoding="utf-8") + out = subprocess.run( + ["node", str(HELPER), str(body)], capture_output=True, text=True, check=True + ) + return json.loads(out.stdout) + + +def _settings() -> dict: + from B05_Profile.B05_Profile_Engine_Grade import GradeDesignOptions + from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions + + summary = P.read("summary.json")["longitudinal"]["data"] + options = {k: v for k, v in summary["options"].items() if k != "extra_stations"} + return { + "options": SectionGenerationOptions(**options), + "grade_options": GradeDesignOptions(**summary["grade_options"]), + "crs": f"EPSG:{P.read('context.json')['crs_epsg']}", + } + + +def _old_xy(root: Path) -> list[list[float]]: + coords = json.loads((root / P.ROUTE).read_text("utf-8"))["geometry"]["coordinates"] + return [[c[0], c[1]] for c in coords] + + +def _seed(root: Path, extras: bool) -> None: + from B05_Profile.B05_Profile_Plus_Stations import save_plus_stations + from common_util.common_util_json import atomic_write_json + + if extras: + atomic_write_json( + root / "B05_Profile" / "route" / "structures.json", + {"revision": 1, "structures": [copy.deepcopy(WALL)], "definition_version": None}, + ) + save_plus_stations(root, [510], _old_xy(root)) + + +def _chain(root: Path, draft: dict, target) -> dict: + """`run_route_save_chain` 파일 길 차례(DB 없이) — [적용] 뒤 종단 정본.""" + from B05_Profile.B05_Profile_Engine import _route_geojson + from B05_Profile.B05_Profile_Engine_AsPlanned import _vertex_z + from B05_Profile.B05_Profile_Engine_Sections import run_section_generation + from B05_Profile.B05_Profile_Plus_Stations import follow_plus_stations + from B05_Profile.B05_Profile_Route_Save import followed_pipes + from B05_Profile.B05_Profile_Router import _apply_alignment_edits + from B05_Profile.B05_Profile_Router_Replan import _vertices_of, _write_planned_polyline + from B05_Profile.B05_Profile_Router_RouteInitial import _set_base_pvi + from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route + from common_util.common_util_drainage_pipes import ( + ensure_pipe_ids, + fill_pipe_coordinates, + pipe_points_path_in, + read_pipe_points_file, + route_signature, + ) + from common_util.common_util_json import atomic_write_json + from common_util.common_util_route_geometry import planned_route_working_path + + old_xy = _old_xy(root) + old_pipes = read_pipe_points_file(pipe_points_path_in(root)) + working = planned_route_working_path(root) + count = len(draft["nodes"]) + _write_planned_polyline( + working, + [tuple(n) for n in draft["nodes"]], + RADIUS, + simplify=False, + curve_flags=draft["curveOn"], + radii=draft["curveRadius"], + station_snap=True, + station_snap_step=None, + locks=[None] * count, + arcs=[None] * count, + ) + points = [{"x": x, "y": y} for x, y in _vertices_of(working)] + follow_structures_to_route(root, P.ROUTE, points) + follow_plus_stations(root, [(p["x"], p["y"]) for p in points]) + xy = [(p["x"], p["y"]) for p in points] + z = _vertex_z(root, *P.SURFACE, xy) + atomic_write_json(root / P.ROUTE, _route_geojson([[x, y, h] for (x, y), h in zip(xy, z)])) + pairs, _crossing, vertices = followed_pipes(old_pipes, old_xy, points, []) + pipes = [pipe for _, pipe in pairs] + fill_pipe_coordinates(pipes, vertices) + ensure_pipe_ids(pipes) + atomic_write_json( + pipe_points_path_in(root), + {"route_signature": route_signature(vertices), "points": [p.as_dict() for p in pipes]}, + ) + settings = _settings() + run_section_generation(root, P.ROUTE, *P.SURFACE, **settings) + if target is not None: + _set_base_pvi(root / P.LONG, *target) + _apply_alignment_edits(root, P.LONG, {}) + return json.loads((root / P.LONG).read_text("utf-8")) + + +def _preview(root: Path, candidate: dict, target) -> tuple[dict, list, list]: + from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file + from common_util.common_util_surface_sampler import build_surface_sampler + + line, curves = ip.candidate_line(candidate["nodes"], candidate["curves"], RADIUS) + facilities = ip.followed_facilities( + root, + line, + P.ROUTE, + _old_xy(root), + read_pipe_points_file(pipe_points_path_in(root)), + [], + ) + sampler = build_surface_sampler(root / "B04_PreProcess" / "models", *P.SURFACE) + longitudinal = ip.initial_longitudinal( + root, + line, + curves, + sampler, + facilities, + source_snapshot={"filter": "classification", "method": "dtm", "smooth": True}, + target=target, + **_settings(), + ) + return longitudinal, line, curves + + +def _target(root: Path, line, curves): + from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import route_profile + from common_util.common_util_surface_sampler import build_surface_sampler + + sampler = build_surface_sampler(root / "B04_PreProcess" / "models", *P.SURFACE) + profile = route_profile(line, sampler, max_grade=0.14, design_speed=20, design_curves=curves) + return profile["stations_m"], profile["plan_m"] + + +CASES = { + "same_off": {"shift": 0.0, "extras": False, "balanced": False}, + "moved_off": {"shift": 5.0, "extras": True, "balanced": False}, + "moved_on": {"shift": 5.0, "extras": True, "balanced": True}, +} + + +@pytest.fixture(scope="module", params=list(CASES)) +def pair(request, tmp_path_factory): + case = CASES[request.param] + work = tmp_path_factory.mktemp(request.param) + chain_root = P.stage_project(work / "chain") + preview_root = P.stage_project(work / "preview") + for root in (chain_root, preview_root): + _seed(root, case["extras"]) + candidate = _candidate(preview_root, case["shift"]) + draft = _browser_draft(candidate, work) + target = None + if case["balanced"]: + line, curves = ip.candidate_line(candidate["nodes"], candidate["curves"], RADIUS) + target = _target(preview_root, line, curves) + preview, line, curves = _preview(preview_root, candidate, target) + chain = _chain(chain_root, draft, target) + return {"preview": preview, "chain": chain, "line": line, "curves": curves, "root": chain_root} + + +def test_화면_펴기_짝(tmp_path): + root = P.stage_project(tmp_path / "p") + candidate = _candidate(root, 5.0) + draft = _browser_draft(candidate, tmp_path) + points, flags, radii = ip.draft_nodes(candidate["nodes"], candidate["curves"]) + assert [list(p) for p in points] == draft["nodes"] + assert flags == draft["curveOn"] and radii == draft["curveRadius"] + + +def test_노선_곡선표_같음(pair): + from B05_Profile.B05_Profile_Engine_Sections import _load_design_curves + from B05_Profile.B05_Profile_Router_Replan import _vertices_of + from common_util.common_util_route_geometry import planned_route_working_path + + root = pair["root"] + assert [tuple(p) for p in _vertices_of(planned_route_working_path(root))] == pair["line"] + assert _load_design_curves(root) == pair["curves"] + + +def test_미리_셈_종단_체인과_1mm(pair): + preview, chain = pair["preview"], pair["chain"] + got = [float(s["chainage_m"]) for s in preview["stations"]] + want = [float(s["chainage_m"]) for s in chain["stations"]] + assert got == pytest.approx(want, abs=1e-6) + a = preview["profile_alignment"]["base_pvi"] + b = chain["profile_alignment"]["base_pvi"] + assert len(a) == len(b) + for p, q in zip(a, b): + assert abs(p["chainage_m"] - q["chainage_m"]) <= 1e-3 + assert abs(p["elevation_m"] - q["elevation_m"]) <= 1e-3 + assert ip.profile_summary(preview)["plan_m"] == pytest.approx( + ip.profile_summary(chain)["plan_m"], abs=1e-3 + ) + + +def test_초기_종단_438_규칙(pair): + summary = ip.profile_summary(pair["preview"]) + assert summary["curve_length_m"] == 20.0 # 432 L 20 + assert summary["straightened"] > 0 # R > 500 꼭짓점 묶음 = 직선 위(438) + curves = pair["preview"]["profile_alignment"]["curves"] + assert not [c for c in curves if not c["omitted"] and c["r_m"] > 500.0 + 1e-6] + assert len(summary["line"]) > len(summary["stations_m"]) # 종단곡선 포함 선 + + +# ── 응답 칸 · 저장본 지문 ────────────────────────────────────────────────────── + + +class _Pool: + def acquire(self): + pool = self + + class _Acquire: + async def __aenter__(self): + return pool + + async def __aexit__(self, *exc): + return False + + return _Acquire() + + def cursor(self, *args): + return self.acquire() + + +def _async(value): + async def fake(*args, **kwargs): + return value() if callable(value) else value + + return fake + + +@pytest.fixture +def gather(tmp_path, monkeypatch): + import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed + import B05_Profile.B05_Profile_Repository as b05_repo + import B05_Profile.B05_Profile_Router_Replan as replan + import B05_Profile.B05_Profile_Router_RouteInitial as route_initial + import B05_Profile.B05_Profile_Router_RouteInitial_Initial as gatherer + import B06_Section.B06_Section_Repository as b06_repo + import common_util.common_util_drainage_pipes as drainage + import common_util.common_util_workflow_state as workflow + + root = P.stage_project(tmp_path / "project") + summary = P.read("summary.json")["longitudinal"]["data"] + grade = summary["grade_options"] + params = { + "options": { + "grade_class": grade.get("grade_class") or "trunk", + "terrain_type": grade.get("terrain_type") or "special", + "design_speed_kph": grade.get("design_speed_kph"), + }, + "max_grade_pct": grade.get("max_grade_pct"), + } + state = {"stages": [{"stage_no": 2, "params": params}]} + pipes_file = root / "B04_PreProcess" / "drainage" / "edits" / "pipe_points.json" + monkeypatch.setattr(gatherer, "get_db_pool", lambda: _Pool()) + monkeypatch.setattr(replan, "_plan_criteria", _async((RADIUS, 0.0, 0.0))) + monkeypatch.setattr(replan, "_road_settings", _async(("trunk", 20, "special"))) + monkeypatch.setattr(b05_repo, "get_latest_route", _async({"id": 1, "route_data_path": P.ROUTE})) + monkeypatch.setattr(b06_repo, "get_route_generation_source", _async({"crs_epsg": 5186})) + monkeypatch.setattr(b06_repo, "get_latest_section_options", _async(summary["options"])) + monkeypatch.setattr(workflow, "get_workflow_state", _async(lambda: copy.deepcopy(state))) + monkeypatch.setattr(route_initial, "project_standard", _async(None)) + monkeypatch.setattr(watershed, "_prepare", _async({"streams": []})) + monkeypatch.setattr(watershed, "_sheet_dir", lambda stored: root / "none") + monkeypatch.setattr(drainage, "pipe_points_path", lambda stored: pipes_file) + request = replan.RouteInitialRequest(mode="follow", with_profile=False) + results = root / "B05_Profile" / "route_recommend" + results.mkdir(parents=True, exist_ok=True) + (results / "k1.json").write_text("{}", "utf-8") # 결과 저장본 짝(없으면 초기 종단 파일 지움) + candidate = _candidate(root) + selection = {"source_filter": "classification", "method": "dtm", "smooth": True} + + def run(key: str = "k1") -> dict: + return asyncio.run( + gatherer.initial_profile_of(uuid4(), root, "p", candidate, request, selection, key) + ) + + return {"run": run, "root": root, "state": state, "pipes": pipes_file} + + +def test_응답_칸(gather): + got = gather["run"]() + assert "error" not in got, got + for name in ( + "stations_m", + "ground_m", + "plan_m", + "line", + "pvi_count", + "curve_count", + "straightened", + "max_grade_pct", + "curve_length_m", + "warnings", + ): + assert name in got, name + assert got["with_profile"] is False and got["curve_length_m"] == 20.0 + assert len(got["stations_m"]) == len(got["plan_m"]) == len(got["ground_m"]) + + +def test_지문_같으면_꺼냄_다르면_다시(gather, monkeypatch): + first = gather["run"]() + assert "error" not in first, first + + def broken(*args, **kwargs): + raise RuntimeError("다시 셈") + + monkeypatch.setattr(ip, "initial_longitudinal", broken) + assert gather["run"]() == first # 같은 입력 = 저장본 + gather["state"]["stages"][0]["params"]["max_grade_pct"] = 9.0 # 페이지 설정 최대 기울기 + assert "다시 셈" in gather["run"]()["error"] + gather["state"]["stages"][0]["params"].pop("max_grade_pct") + doc = json.loads(gather["pipes"].read_text("utf-8")) + doc["points"][0]["x"] = float(doc["points"][0]["x"]) + 1.0 # 관 하나 옮김 + gather["pipes"].write_text(json.dumps(doc), "utf-8") + assert "다시 셈" in gather["run"]()["error"] + + +def test_코드_바뀌면_다시(gather, monkeypatch): + assert "error" not in gather["run"]() + monkeypatch.setattr(ip, "source_fingerprint", lambda: "other") + monkeypatch.setattr(ip, "initial_longitudinal", lambda *a, **k: {}) + assert "error" in gather["run"]() # 438 등 소스 바뀜 = 다시 셈(빈 종단 = 실패 글) + + +def test_추천_응답에_initial_profile(wired, monkeypatch): + import B05_Profile.B05_Profile_Router_RouteInitial_Initial as gatherer + + seen: list = [] + + async def fake(project_id, root, stored, result, request, selection, key): + seen.append(key) + return {"pvi_count": 3} + + monkeypatch.setattr(gatherer, "initial_profile_of", fake) + body = _ask(mode="earthwork") # 224-3 가짜 세계 + assert body["initial_profile"] == {"pvi_count": 3} and seen # 옛 코드 = 칸 없음