Aislo/B05_Profile/B05_Profile_Engine_InitialProfile.py

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