Aislo/B07_DesignDetail/B07_DesignDetail_Router_Support_Plan.py

"""B07 계획평면도 종류 · 정보 조각 파일 (PLAN 60장).

저장 자리(프로젝트 저장소 `B07_DesignDetail/`):
    plan_kinds/{종류 id}.json     종류 하나 = {name, pieces, order} — M02 양식 층처럼 파일 하나씩
    plan_pieces/{장}/{조각}.json  정보 조각 하나 = 그 장 범위의 기하(사업지 좌표 m)
    plan_marks.json               사용자가 정한 대피소 · 사토장 자리

`plan_kinds/` 가 없으면 기본 넷(지형 · 노선배치도 · 대피소 · 라이다)을 쓴다. 처음 고칠 때
기본 넷을 파일로 내려 두고 그 위에서 더하고 지운다 — 기본 넷도 지울 수 있다.
조각은 서버가 도면을 열 때마다 원자료(B05 노선 · 곡선표 · 구조물, B06 횡단 설계, B04 배경)
에서 다시 계산해 파일로 쓰고, 도면은 그 파일을 읽어 고른 조각만 얹는다.
"""

import re
import uuid
from pathlib import Path
from typing import Any

from B05_Profile.B05_Profile_Engine_Sections import (
    _load_design_curves,
    cross_station,
    find_cross_file,
)
from B05_Profile.B05_Profile_Structures_Repository import load_structures
from B06_Section.B06_Section_Engine_Design import compute_cross_design, curve_widening_args
from B07_DesignDetail.B07_DesignDetail_Engine_Cad import infer_station_interval, station_no_label
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Plan import plan_area_mm, plan_chunks
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_PlanKinds import (
    DEFAULT_KINDS,
    PIECE_IDS,
    curve_pieces,
    route_frame,
    slope_end,
    slope_pieces,
)
from B07_DesignDetail.B07_DesignDetail_Router_Support_Basin import (
    _chunk_route,
    _hillshade_for_box,
    _sheet_box,
    map_background,
    plan_stations,
)
from B07_DesignDetail.B07_DesignDetail_Router_Support_Io import _design_root, _read_json
from common_util.common_util_json import atomic_write_json
from common_util.common_util_route_profile import design_elevation_from_longitudinal
from config.config_system import DRAWING_SCALE_PLAN

KIND_ID = re.compile(r"^[a-z][a-z0-9]{0,31}$")
PLAN_KIND_ID = re.compile(r"^plan_([a-z][a-z0-9]{0,31})(?:_(\d+))?$")
_NAME_MAX = 40
_REFUGE_TYPES = {"refuge": "대피소", "spoil_bank": "사토장"}
_REFUGE_OFFSET_M = 8.0  # 대피소 · 사토장 라벨을 중심선에서 띄우는 거리(좌측)


def plan_title(name: str) -> str:
    return f"계획평면도({name})"


# ── 종류 ─────────────────────────────────────────────


def _kinds_dir(project_root: Path) -> Path:
    return _design_root(project_root) / "plan_kinds"


def list_kinds(project_root: Path) -> list[dict[str, Any]]:
    """종류 목록 [{id, name, pieces}] — 차례대로. 파일이 없으면 기본 넷."""
    folder = _kinds_dir(project_root)
    if not folder.is_dir():
        return [{**kind, "pieces": list(kind["pieces"])} for kind in DEFAULT_KINDS]
    rows: list[tuple[int, str, dict[str, Any]]] = []
    for path in folder.glob("*.json"):
        if not KIND_ID.fullmatch(path.stem):
            continue
        try:
            document = _read_json(path)
        except (OSError, ValueError):
            continue
        pieces = [piece for piece in document.get("pieces") or [] if piece in PIECE_IDS]
        rows.append(
            (
                int(document.get("order") or 0),
                path.stem,
                {"id": path.stem, "name": str(document.get("name") or path.stem), "pieces": pieces},
            )
        )
    return [row for _order, _id, row in sorted(rows, key=lambda item: item[:2])]


def _materialize(project_root: Path) -> list[dict[str, Any]]:
    """기본 넷을 파일로 내려 둔다(이미 있으면 그대로) — 고치기 전에 한 번."""
    kinds = list_kinds(project_root)
    folder = _kinds_dir(project_root)
    if not folder.is_dir():
        for order, kind in enumerate(kinds):
            atomic_write_json(
                folder / f"{kind['id']}.json",
                {"name": kind["name"], "pieces": kind["pieces"], "order": order},
            )
    return kinds


def add_kind(project_root: Path, name: str, pieces: list[str]) -> dict[str, Any]:
    text = str(name or "").strip()
    if not text or len(text) > _NAME_MAX:
        raise ValueError(f"종류 이름은 1~{_NAME_MAX}자로 적어 주세요.")
    chosen = [piece for piece in PIECE_IDS if piece in set(pieces or [])]
    if not chosen:
        raise ValueError("얹을 정보를 하나 이상 고르세요.")
    kinds = _materialize(project_root)
    if any(kind["name"] == text for kind in kinds):
        raise ValueError(f"같은 이름의 종류가 이미 있습니다: {text}")
    kind_id = f"k{uuid.uuid4().hex[:8]}"
    atomic_write_json(
        _kinds_dir(project_root) / f"{kind_id}.json",
        {"name": text, "pieces": chosen, "order": len(kinds)},
    )
    return {"id": kind_id, "name": text, "pieces": chosen}


def delete_kind(project_root: Path, kind_id: str) -> None:
    if not KIND_ID.fullmatch(kind_id):
        raise ValueError("올바르지 않은 종류 id 입니다.")
    _materialize(project_root)
    path = _kinds_dir(project_root) / f"{kind_id}.json"
    if not path.is_file():
        raise FileNotFoundError("지울 종류를 찾을 수 없습니다.")
    path.unlink()


# ── 대피소 · 사토장 자리(사용자 지정) ─────────────────


def _marks_path(project_root: Path) -> Path:
    return _design_root(project_root) / "plan_marks.json"


def read_marks(project_root: Path) -> list[dict[str, Any]]:
    path = _marks_path(project_root)
    if not path.is_file():
        return []
    try:
        marks = _read_json(path).get("marks")
    except (OSError, ValueError):
        return []
    return [mark for mark in marks or [] if isinstance(mark, dict)]


def write_marks(project_root: Path, marks: list[dict[str, Any]]) -> list[dict[str, Any]]:
    clean: list[dict[str, Any]] = []
    for mark in marks:
        chainage = mark.get("chainage_m")
        if not isinstance(chainage, (int, float)) or chainage < 0:
            raise ValueError("대피소 · 사토장 자리의 측점(m)이 올바르지 않습니다.")
        refuge, spoil = bool(mark.get("refuge")), bool(mark.get("spoil"))
        if refuge or spoil:
            clean.append(
                {"chainage_m": round(float(chainage), 2), "refuge": refuge, "spoil": spoil}
            )
    clean.sort(key=lambda mark: mark["chainage_m"])
    atomic_write_json(_marks_path(project_root), {"marks": clean})
    return clean


# ── 정보 조각 ─────────────────────────────────────────


def _station_design(
    longitudinal_path: Path,
    longitudinal: dict[str, Any],
    source: dict[str, Any],
    stored: dict[str, Any] | None,
) -> dict[str, Any] | None:
    """그 측점의 횡단 설계 — B06 저장분 우선, 없으면 기본값으로 계산(횡단면도와 같은 규칙)."""
    if isinstance(stored, dict) and stored.get("design_line") and stored.get("road_edges"):
        return stored
    try:
        return compute_cross_design(
            source.get("samples", []),
            design_elevation_from_longitudinal(longitudinal, float(source.get("chainage_m", 0.0))),
            ground_type="soil",
            section_mode="left_cut",
            **curve_widening_args(source),
        )
    except (ValueError, KeyError):
        return None


def slope_records(
    longitudinal_path: Path,
    longitudinal: dict[str, Any],
    designs: dict[int | float, dict[str, Any]],
    start_m: float,
    end_m: float,
) -> list[dict[str, Any]]:
    """측점별 노견 끝 · 비탈 끝 오프셋(좌 + · 우 -) — 장 범위와 앞뒤 한 측점."""
    cross_dir = longitudinal_path.parent.parent / "cross_sections"
    chainages = sorted(chainage for chainage, _x, _y in plan_stations(longitudinal))
    inside = [i for i, c in enumerate(chainages) if start_m <= c <= end_m]
    if not inside:
        return []
    chosen = chainages[max(inside[0] - 1, 0) : inside[-1] + 2]
    records: list[dict[str, Any]] = []
    for chainage in chosen:
        path = find_cross_file(cross_dir, chainage)
        if not path.is_file():
            continue
        source = _read_json(path)
        design = _station_design(
            longitudinal_path, longitudinal, source, designs.get(cross_station(chainage))
        )
        edges = (design or {}).get("road_edges") or {}
        try:
            left_edge = float(edges["left"]["offset_m"])
            right_edge = float(edges["right"]["offset_m"])
        except (KeyError, TypeError, ValueError):
            continue
        line, samples = (design or {}).get("design_line") or [], source.get("samples") or []
        record: dict[str, Any] = {
            "chainage_m": float(chainage),
            "left_edge_m": left_edge,
            "left_end_m": slope_end(line, samples, left_edge, "left"),
            "right_edge_m": right_edge,
            "right_end_m": slope_end(line, samples, right_edge, "right"),
        }
        frame = source.get("frame") or {}
        origin, axis = frame.get("origin") or {}, frame.get("left_xy")
        if isinstance(axis, list) and len(axis) == 2 and "x" in origin and "y" in origin:
            record.update(x=float(origin["x"]), y=float(origin["y"]), nx=axis[0], ny=axis[1])
        records.append(record)
    return records


def _refuge_items(
    project_root: Path, route: list[tuple[float, float, float]], start_m: float, end_m: float
) -> dict[str, Any]:
    """대피소 · 사토장 = 구조물 정본의 그 종류 + 사용자가 정한 자리. 한 자리에 둘 다 가능."""
    spots: dict[float, set[str]] = {}
    _revision, structures = load_structures(str(project_root))
    for structure in structures:
        label = _REFUGE_TYPES.get(structure.type_id)
        if not label:
            continue
        chainage = structure.chainage_m
        if chainage is None and structure.start_m is not None and structure.end_m is not None:
            chainage = (structure.start_m + structure.end_m) / 2.0
        if chainage is not None:
            spots.setdefault(round(float(chainage), 1), set()).add(label)
    for mark in read_marks(project_root):
        key = round(float(mark["chainage_m"]), 1)
        if mark.get("refuge"):
            spots.setdefault(key, set()).add("대피소")
        if mark.get("spoil"):
            spots.setdefault(key, set()).add("사토장")
    items = []
    for chainage, labels in sorted(spots.items()):
        if not (start_m <= chainage <= end_m) or len(route) < 2:
            continue
        x, y, nx, ny = route_frame(route, chainage)
        items.append(
            {
                "chainage_m": chainage,
                "label": " · ".join(name for name in ("대피소", "사토장") if name in labels),
                "x": x + nx * _REFUGE_OFFSET_M,
                "y": y + ny * _REFUGE_OFFSET_M,
            }
        )
    return {"items": items}


def build_pieces(
    context: Any,
    longitudinal: dict[str, Any],
    longitudinal_path: Path,
    designs: dict[int | float, dict[str, Any]],
    number: int,
    wanted: list[str],
) -> tuple[dict[str, dict[str, Any]], tuple[float, float, float, float], dict[str, Any], int]:
    """그 장의 조각들을 계산해 파일로 쓰고 (조각들, 도곽 범위, 장 구간, 전체 장수)를 낸다."""
    route_xy, chunk, total = _chunk_route(context, longitudinal, number)
    if not route_xy:
        raise FileNotFoundError("계획평면도에 그릴 노선이 없습니다.")
    box = _sheet_box(route_xy)
    start_m, end_m = float(chunk["start_m"]), float(chunk["end_m"])
    project_root = Path(context.project_root)
    route = sorted(
        (float(vertex.chainage_m), float(vertex.x), float(vertex.y)) for vertex in context.vertices
    )
    stations = plan_stations(longitudinal)
    interval = infer_station_interval(longitudinal.get("stations") or [])
    pieces: dict[str, dict[str, Any]] = {}

    if {"contour", "stream"} & set(wanted):
        background = map_background(
            project_root, context.crs, DRAWING_SCALE_PLAN, plan_area_mm(), route_xy
        )
        pieces["contour"] = {"lines": background["contours"]}
        pieces["stream"] = {"lines": background["streams"]}
    if "centerline" in wanted:
        pieces["centerline"] = {"lines": [route_xy]}
    if {"slope_outline", "slope_ticks"} & set(wanted):
        records = slope_records(longitudinal_path, longitudinal, designs, start_m, end_m)
        pieces["slope_outline"], pieces["slope_ticks"] = slope_pieces(
            route, records, start_m, end_m
        )
    if {"ip", "radius"} & set(wanted):
        pieces["ip"], pieces["radius"] = curve_pieces(_load_design_curves(project_root), route, box)
    if "station" in wanted:
        items = []
        for chainage, x, y in stations:
            if not (start_m <= chainage <= end_m):
                continue
            _x, _y, nx, ny = route_frame(route, chainage)
            label = station_no_label(chainage, interval).replace("No.", "NO.")
            number_only = "+" not in label
            major = number_only and int(label[3:]) % 5 == 0
            items.append(
                {
                    "chainage_m": chainage,
                    "x": x,
                    "y": y,
                    "nx": nx,
                    "ny": ny,
                    "label": label,
                    "major": major,
                }
            )
        pieces["station"] = {"items": items}
    if "endpoints" in wanted and stations:
        items = []
        first, last = stations[0], stations[-1]
        length = last[0] - first[0]
        for (chainage, x, y), word in ((first, "시점"), (last, "종점")):
            if not (start_m <= chainage <= end_m):
                continue
            _x, _y, nx, ny = route_frame(route, chainage)
            label = f"{word} {station_no_label(chainage, interval)}"
            if word == "종점":
                label += f"  L={length:.1f}m"
            items.append({"label": label, "x": x, "y": y, "nx": nx, "ny": ny})
        pieces["endpoints"] = {"items": items}
    if "refuge" in wanted:
        pieces["refuge"] = _refuge_items(project_root, route, start_m, end_m)
    if "lidar" in wanted:
        try:
            image, image_box = _hillshade_for_box(context, box)
            pieces["lidar"] = {"image": image, "box": list(image_box)}
        except (FileNotFoundError, ValueError, OSError):
            pieces["lidar"] = {"image": None, "box": None}

    folder = _design_root(project_root) / "plan_pieces" / str(number)
    for piece, data in pieces.items():
        atomic_write_json(folder / f"{piece}.json", data)
    return {piece: pieces[piece] for piece in wanted if piece in pieces}, box, chunk, total


def plan_kind_source(
    context: Any,
    longitudinal: dict[str, Any],
    longitudinal_path: Path,
    designs: dict[int | float, dict[str, Any]],
    drawing_id: str,
) -> dict[str, Any]:
    """종류 도면 한 장의 입력 — 그 종류가 고른 조각 + 도곽 범위 + 도면명."""
    match = PLAN_KIND_ID.fullmatch(drawing_id)
    if not match:
        raise ValueError("올바르지 않은 계획평면도 ID입니다.")
    kind = next(
        (item for item in list_kinds(Path(context.project_root)) if item["id"] == match.group(1)),
        None,
    )
    if kind is None:
        raise FileNotFoundError("계획평면도 종류를 찾을 수 없습니다.")
    number = int(match.group(2)) if match.group(2) else 1
    pieces, box, chunk, total = build_pieces(
        context, longitudinal, longitudinal_path, designs, number, kind["pieces"]
    )
    title = plan_title(kind["name"])
    return {
        "label": title if total <= 1 else f"{title} {chunk['number']}장",
        "box": box,
        "pieces": pieces,
    }


def plan_kind_items(project_root: Path, longitudinal: dict[str, Any]) -> list[dict[str, Any]]:
    """도면 목록에 실을 종류 × 장 [{id, label, plan_kind}]."""
    sheets = plan_chunks(plan_stations(longitudinal))
    items = []
    for kind in list_kinds(project_root):
        title = plan_title(kind["name"])
        for chunk in sheets:
            single = len(sheets) <= 1
            items.append(
                {
                    "id": f"plan_{kind['id']}"
                    if single
                    else f"plan_{kind['id']}_{chunk['number']}",
                    "label": title if single else f"{title} {chunk['number']}장",
                    "plan_kind": kind["id"],
                    "plan_kind_name": kind["name"],
                }
            )
    return items