Aislo/B07_DesignDetail/B07_DesignDetail_Engine_Cad_Development.py

"""B07 전개도 자료 · 묶음 — 종단 길이를 가진 구조물을 (종류 · 형태)마다 한 장으로(요구 241-2 · 3).

서버 단독(CLAUDE.md 5장 ③) · 저장 안 함 · 값을 짓지 않고 정본을 모으기만:
    관보호공      배관 유입 · 유출(`{쪽}_type == 기슭막이`) `{쪽}_revet_*`
                  + 추가 흙막이 `soil_add_{n}_*`
    BOX · 세월교   유입 · 유출 구조 == 흙막이(B08 `_guard`) + 추가 흙막이
    흙막이 · 옹벽  C군 벽(표 형상 wall) start_m ~ end_m · 높이 height_m · 추가 단 = 별도 박스
                  옹벽 형태 = `wall_form`(`form` 은 형식) · 숨김 벽 종류 = 흙막이 묶음
    포장 · 쇄석    G군 `pavement_specs_from` 구간 · 폭 B = `width_m`(빈칸 = 횡단 노폭 평균)
흙막이 길이 · 빈 높이 = B08 구조물집계표와 같은 규칙(`_revet_length` · 관경 최소 높이 · 박스 높이) ·
시점 = 기준 − 전 · 종점 = 기준 + 후(둘 다 비면 길이 반씩) · 거리 = 종점 − 시점 · 면적 = 거리 × H(B).
추가 흙막이 단 수 = 지형이 세운 유출 다단까지(`container_rows` · 설계가 있을 때만).
그림 = `B07_DesignDetail_Engine_Cad_Development_Draw.py`.
"""

from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any

from B06_Section.B06_Section_Engine_Culvert import pipe_wall_height_m, pipe_wall_min_height_m
from B06_Section.B06_Section_Engine_Structures_Wall import tier_extras
from B07_DesignDetail.B07_DesignDetail_Engine_Cad import infer_station_interval
from B07_DesignDetail.B07_DesignDetail_Schema import DesignDrawingItem
from B08_Quantity.B08_Quantity_Router_Summary import (
    FORMS,
    GUARD_SOIL,
    SOIL,
    _guard,
    _num,
    _revet_length,
)
from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
from common_util.common_util_structure_catalog import base_type, catalog_row
from common_util.common_util_structure_pavement import pavement_specs_from
from common_util.common_util_summary_item import GROUP_JOIN
from M02_MasterTemplete import M02_Table_Fill as fill
from M02_MasterTemplete.M02_Table_Fill_Spec import container_rows, soil_tiers_from_designs

DEVELOPMENT_ID = re.compile(r"^development_([a-z]+\d?)(?:_(\d+))?$")
#: 묶음 차례 · 이름 — 포장은 형태 없이 한 묶음
GROUPS = {
    "pipe": "관보호공",
    "box": "BOX암거 흙막이",
    "ford": "세월교 흙막이",
    "soil": "흙막이",
    "wall": "옹벽",
    "concrete": "콘크리트포장",
    "gravel": "쇄석포설",
}
PAVEMENTS = {"pavement_concrete": "concrete", "gravel_surfacing": "gravel"}
_BODY_GROUP = {"pipe": "pipe", "box_culvert": "box", "ford_bridge": "ford"}
#: 한 장에 담는 박스 수 — 넘치면 `_2` 장(우하단 표가 작도 영역 안에 들게)
MAX_ROWS = 30
_EMPTY_FORM = 9  # 형태 빈칸 id 자리


def station_text(chainage_m: float, interval_m: float) -> str:
    """전개도 측점 — `No.04+08` · 소수면 `No.22+05.5`(그림 241 표기)."""
    safe = interval_m if interval_m > 0 else 20.0
    number = int((chainage_m + 1e-6) // safe)
    rest = round(chainage_m - number * safe, 1)
    if rest >= safe - 0.05:
        number, rest = number + 1, 0.0
    if abs(rest - round(rest)) < 0.05:
        return f"No.{number:02d}+{int(round(rest)):02d}"
    return f"No.{number:02d}+{rest:04.1f}"


def _row(group: str, form: Any, start: float | None, end: float | None, height: Any, key: str):
    """박스 한 줄 — 시점 · 종점 중 하나라도 없으면 길이 미입력."""
    if start is not None and end is not None and end < start:
        start, end = end, start
    return {
        "group": group,
        "form": str(form or ""),
        "start_m": start,
        "end_m": end,
        "height_m": _num(height),
        "key": key,
    }


def _span(at: float, options: dict[str, Any], head: str, unit: str):
    """기준 측점 + 전 · 후(둘 다 비면 길이 반씩) → (시점, 종점) · 길이도 없으면 (None, None)."""
    before = _num(options.get(f"{head}before{unit}"))
    after = _num(options.get(f"{head}after{unit}"))
    if before is None and after is None:
        length = _revet_length(options, head, unit)
        if not length:
            return None, None
        before = after = length / 2.0
    return at - (before or 0.0), at + (after or 0.0)


def _empty_heights(type_id: str, item: dict[str, Any]) -> dict[str, float | None]:
    """빈 높이 — 관 = 관경 최소(유입 + 0.7) · 박스 = 박스 높이 · 세월교 = 미입력(B08 같은 규칙)."""
    options = item["options"]
    if type_id == "pipe":
        diameter = _num(options.get("pipe_diameter_mm"))
        if not diameter:
            return {"inlet": None, "outlet": None, "soil_add": None}
        meters = diameter / 1000.0
        wall = pipe_wall_min_height_m(meters)
        return {"inlet": pipe_wall_min_height_m(meters, "inlet"), "outlet": wall, "soil_add": wall}
    height = _num(options.get("body_height_m")) if type_id == "box_culvert" else None
    return {"inlet": height, "outlet": height, "soil_add": height}


def _body_rows(item: dict[str, Any], raw: dict[str, Any]) -> list[dict[str, Any]]:
    """배관 · 박스 · 세월교 한 곳 — 유입 · 유출 흙막이 + 추가 흙막이 n."""
    type_id = item["type_id"]
    group = _BODY_GROUP[type_id]
    at = float(item["chainage_m"])
    empty = _empty_heights(type_id, item)
    diameter = _num(item["options"].get("pipe_diameter_mm"))
    if type_id == "pipe":
        sides = [side for side in ("inlet", "outlet") if raw.get(f"{side}_type") == SOIL]
    else:
        sides = [
            s for s in ("inlet", "outlet") if _guard(type_id, item["options"], s) == GUARD_SOIL
        ]
    rows = []
    for side in sides:
        height = _num(raw.get(f"{side}_revet_height_m"))
        if type_id == "pipe" and diameter:
            height = pipe_wall_height_m(height, diameter / 1000.0)
        start, end = _span(at, raw, f"{side}_revet_", "_m")
        form = raw.get(f"{side}_revet_form")
        key = f"{item['복사본']}:{side}"
        rows.append(_row(group, form, start, end, empty[side] if height is None else height, key))
    for extra in container_rows(item, {}):
        link = extra["복사본"].partition(GROUP_JOIN)[2]
        if not link.startswith("gsoiladd"):
            continue
        head = f"soil_add_{link[len('gsoiladd') :]}_"
        height = _num(raw.get(f"{head}height"))
        start, end = _span(at, raw, head, "")
        height = empty["soil_add"] if height is None else height
        rows.append(_row(group, raw.get(f"{head}form"), start, end, height, extra["복사본"]))
    return rows


def _wall_rows(item: dict[str, Any]) -> list[dict[str, Any]]:
    """C군 벽 — 본 단 + 추가 단(별도 박스 · 같은 구간)."""
    type_id = item["type_id"]
    options = item["options"]
    wall = base_type(type_id) == "retaining_wall"
    group = "wall" if wall else "soil"
    form = (
        options.get("wall_form")
        if wall
        else options.get("form")
        or ((catalog_row(type_id) or {}).get("shapeOpts") or {}).get("form")
    )
    start, end = _num(item.get("start_m")), _num(item.get("end_m"))
    key = item["복사본"]
    rows = [_row(group, form, start, end, options.get("height_m"), key)]
    for n, extra in enumerate(tier_extras(type_id, options), start=1):
        tier_form = extra["form"] or form
        rows.append(_row(group, tier_form, start, end, extra["height_m"], f"{key}:tier{n}"))
    return rows


def _road_width(designs: dict[float, Any], start: float, end: float) -> float | None:
    """구간 안 측점 횡단 노폭(`carriageway_width_m`) 평균 — 없으면 None."""
    widths = [
        width
        for chainage, design in (designs or {}).items()
        if start - 0.02 <= float(chainage) <= end + 0.02
        and isinstance(design, dict)
        and (width := _num(design.get("carriageway_width_m")))
    ]
    return round(sum(widths) / len(widths), 2) if widths else None


def _pavement_rows(item: dict[str, Any], designs: dict[float, Any]) -> list[dict[str, Any]]:
    """콘크리트 포장 · 쇄석 포설 — 구간 × 폭 B."""
    specs = pavement_specs_from([item], {item["type_id"]})
    group = PAVEMENTS[base_type(item["type_id"])]
    if not specs:
        return [_row(group, "", None, None, None, item["복사본"])]
    start, end = specs[0]["start_m"], specs[0]["end_m"]
    width = _num(item["options"].get("width_m")) or _road_width(designs, start, end)
    if end <= start:
        start = end = None
    return [_row(group, "", start, end, width, item["복사본"])]


def development_rows(
    project_root: str | Path, designs: dict[float, Any] | None = None
) -> list[dict[str, Any]]:
    """설계 정본 → 전개도 박스 줄 전부(묶음 · 형태 · 시점 · 종점 · 높이(폭)).

    `designs` = `{측점: 횡단 설계}` — 추가 흙막이 단 수 · 포장 빈 폭에 씀(없으면 저장된 것 그대로).
    """
    rows_in = [{"chainage_m": k, "design": v} for k, v in (designs or {}).items()]
    items = fill.collect_items(
        project_root, soil_tiers=soil_tiers_from_designs(rows_in) if designs else None
    )
    raw = {
        (point.facility or "pipe", f"pipe@{round(float(point.chainage_m), 3)}"): point.options
        for point in read_pipe_points_file(pipe_points_path_in(Path(project_root)))
    }
    rows: list[dict[str, Any]] = []
    for item in items:
        owner, _, link = item["복사본"].partition(GROUP_JOIN)
        if link:
            continue  # 그룹 · 컨테이너 줄은 본체 줄에서 다시 셈(단 수 거름)
        type_id = item["type_id"]
        if type_id in _BODY_GROUP:
            rows += _body_rows(item, dict(raw.get((type_id, owner)) or {}))
        elif (catalog_row(type_id) or {}).get("shape") == "wall":
            rows += _wall_rows(item)
        elif base_type(type_id) in PAVEMENTS:
            rows += _pavement_rows(item, designs or {})
    return rows


def _form_index(form: str) -> int:
    return FORMS.index(form) if form in FORMS else _EMPTY_FORM


def development_sheets(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """(묶음 · 형태)마다 한 장 · 넘치면 `_2` — `{id, kind_id, title, group, rows, missing}`."""
    found: dict[tuple[int, int, str], list[dict[str, Any]]] = {}
    order = list(GROUPS)
    for row in rows:
        paved = row["group"] in ("concrete", "gravel")
        form = "" if paved else row["form"]
        key = (order.index(row["group"]), _form_index(form) if not paved else 0, form)
        found.setdefault(key, []).append(row)
    sheets = []
    for (group_at, form_at, form), members in sorted(found.items()):
        group = order[group_at]
        paved = group in ("concrete", "gravel")
        kind_id = f"development_{group}" if paved else f"development_{group}{form_at}"
        title = (
            f"{GROUPS[group]} 전개도"
            if paved
            else f"{GROUPS[group]} {form or '형태 미입력'} 전개도"
        )
        drawn = sorted(
            (r for r in members if r["start_m"] is not None), key=lambda r: (r["start_m"], r["key"])
        )
        missing = len(members) - len(drawn)
        chunks = [drawn[i : i + MAX_ROWS] for i in range(0, len(drawn), MAX_ROWS)] or [[]]
        for index, chunk in enumerate(chunks, start=1):
            sheets.append(
                {
                    "id": kind_id if index == 1 else f"{kind_id}_{index}",
                    "kind_id": kind_id,
                    "title": title if index == 1 else f"{title} ({index})",
                    "group": group,
                    "form": form,
                    "rows": chunk,
                    "missing": missing if index == 1 else 0,
                }
            )
    return sheets


def table_rows(rows: list[dict[str, Any]], interval_m: float) -> list[dict[str, Any]]:
    """우하단 표 줄 — 측점 범위 · 거리 · 높이(폭) · 면적(높이 빈칸이면 빈칸)."""
    out = []
    for row in rows:
        distance = round(row["end_m"] - row["start_m"], 3)
        height = row["height_m"]
        out.append(
            {
                "station": f"{station_text(row['start_m'], interval_m)}~"
                f"{station_text(row['end_m'], interval_m)}",
                "distance": distance,
                "height": height,
                "area": None if height is None else round(distance * height, 3),
            }
        )
    return out


def _read_json(path: Path) -> dict[str, Any]:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}


def development_list(
    project_root: Path, manifest_drawings: dict[str, Any]
) -> list[DesignDrawingItem]:
    """좌측 「전개도」 카드 단추 — 장마다 한 줄 · 하위 묶음(plan_kind) = 구조물 종류."""
    return [
        DesignDrawingItem(
            id=sheet["id"],
            kind="development",
            label=sheet["title"],
            confirmed=bool(manifest_drawings.get(sheet["id"], {}).get("confirmed")),
            plan_kind=f"development_{sheet['group']}",
            plan_kind_name=GROUPS[sheet["group"]],
        )
        for sheet in development_sheets(development_rows(project_root))
    ]


def read_development(
    project_root: Path, longitudinal_path: Path, drawing_id: str, designs: Any
) -> tuple[str, str, dict[str, Any], bool, None]:
    """전개도 한 장 — 목록 뒤 설계로 단 수가 줄어 장이 사라졌으면 빈 장 + 안내."""
    from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Development_Draw import (
        build_development_drawing,
    )

    designs = designs if isinstance(designs, dict) else {}
    sheets = development_sheets(development_rows(project_root, designs))
    sheet = next((s for s in sheets if s["id"] == drawing_id), None)
    if sheet is None:
        match = DEVELOPMENT_ID.fullmatch(drawing_id)
        if not match:
            raise ValueError("올바르지 않은 도면 ID입니다.")
        base = next((s for s in sheets if s["kind_id"] == f"development_{match.group(1)}"), None)
        sheet = {**(base or {"title": "전개도", "group": "soil", "form": ""}), "rows": []}
        sheet["missing"] = 0
    longitudinal = _read_json(longitudinal_path)
    interval = infer_station_interval(longitudinal.get("stations") or []) or 20.0
    drawing = build_development_drawing(drawing_id, sheet, interval)
    return "development", sheet["title"], drawing, False, None