"""추천 후보 초기 종단(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)