Aislo/B08_Quantity/B08_Quantity_Engine_StructureSummary.py

"""B08 **구조물 집계표** — 측점별 구조물 한 줄 · 종류별 표 (PLAN 2장 · 사용자 확정 ④).

한 줄 = 측점 + 종류 + 실치수(제원 칸) + 개소·연장. 실무 「돌-골막이치수」 치수조서가
실치수를 들고 구조도 탭이 끌어 쓰는 모양을 **한 탭으로 겸함**.

값이 어디서 오나 — 칸마다 출처를 붙임
    auto     정본에 적힌 값 — `structures.json`(B05) · `pipe_points.json`(계곡 통과 시설) ·
             관 연장은 B06 횡단 `design.pipe_length_m`
    user     이 표에서 사람이 고쳐 **정본에 적은** 값 — 산출 조건에 「손댄 칸」 표만 둠(브레인 판정)
    library  정본 칸이 비어 **양식 기본값**으로 선 칸(구조물도 양식 `vars.*.default`)
    empty    정본도 양식도 값이 없음 — 계산 쪽이 막히거나 기준값으로 돎
⚠ 사용자 손 값은 덮개층이 아니라 **정본에 씀** — 도면·수량이 한 값(지침 5장 「쪼개지 않음」).
⚠ 손댄 칸의 정본 값이 뒤에 B05 에서 바뀌면 그 칸은 자동으로 돌아가되 **조용히 말고** 줄에 알림.
⚠ 값을 여기서 짓지 않음 — 읽어 줄 세우기만. 연장이 없는 관은 빈칸(0 으로 안 채움).
⚠ 관 지점 정본 타입(`managed_by`)은 `structures.json` 에 있어도 안 셈 — `pipe_points.json` 이 주인.
"""

from __future__ import annotations

from collections.abc import Iterable
from typing import Any

SOURCE_AUTO = "auto"
SOURCE_USER = "user"
SOURCE_LIBRARY = "library"
SOURCE_EMPTY = "empty"
#: 산출 조건 자리 — 손댄 칸 `{줄 id: {칸: {"value": 적은 값, "was": 고치기 전 정본 값}}}`.
USER_CELLS_KEY = "structure_summary_user_cells"

#: 계곡 통과 시설 `facility` → 레지스트리 종류. 빈 값은 배관(`PipePoint` 기본).
FACILITY_TYPES = {
    "pipe": "pipe",
    "box_culvert": "box_culvert",
    "ford_pavement": "ford_pavement",
    "ford_bridge": "ford_bridge",
    "revetment": "revetment",
}


def _blank(value: Any) -> bool:
    return value is None or value == ""


def _number(value: Any) -> float | None:
    if isinstance(value, bool) or _blank(value):
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _defaults(template: dict[str, Any] | None) -> dict[str, Any]:
    """양식 제원 칸 기본값 `{옵션 키: 기본값}` — 정본 칸이 비었을 때 계산이 쓰는 값."""
    return {
        str(spec["option"]): spec["default"]
        for spec in ((template or {}).get("vars") or {}).values()
        if spec.get("option") and "default" in spec and not _blank(spec["default"])
    }


def _same(left: Any, right: Any) -> bool:
    """정본 값과 적은 값이 같은가 — 숫자는 3.0 과 "3" 을 같게 봄."""
    a, b = _number(left), _number(right)
    return a == b if a is not None and b is not None else left == right


def _cells(
    keys: Iterable[str],
    options: dict[str, Any],
    defaults: dict[str, Any],
    marks: dict[str, Any],
) -> dict:
    cells = {}
    for key in keys:
        current = None if _blank(options.get(key)) else options[key]
        mark = marks.get(key)
        if mark is not None and not _same(current, mark.get("value")):
            # 사람이 적은 값을 B05 가 바꿈 — 자동으로 돌아가되 옛 값을 함께 실어 화면이 알림.
            cells[key] = {
                "value": current,
                "source": SOURCE_EMPTY if current is None else SOURCE_AUTO,
                "replaced_user_value": mark.get("value"),
            }
        elif mark is not None:
            cells[key] = {"value": current, "source": SOURCE_USER, "was": mark.get("was")}
        elif current is not None:
            cells[key] = {"value": options[key], "source": SOURCE_AUTO}
        elif key in defaults:
            cells[key] = {"value": defaults[key], "source": SOURCE_LIBRARY}
        else:
            cells[key] = {"value": None, "source": SOURCE_EMPTY}
    return cells


def build_summary(
    structures: Iterable[dict[str, Any]],
    pipe_points: Iterable[dict[str, Any]],
    types: dict[str, Any],
    templates: dict[str, dict[str, Any]] | None = None,
    pipe_lengths: dict[float, float] | None = None,
    user_cells: dict[str, dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """종류별 표 목록(레지스트리 차례) — 표마다 칸 정의·줄·개소·연장 합·평균치수.

    `types` 는 `structure_type_map()` · `templates` 는 종류별 양식(프로젝트에 박힌 것 → 기본) ·
    `pipe_lengths` 는 B06 횡단의 `{측점: 관 연장}`(`pipe_lengths_from_designs`) ·
    `user_cells` 는 이 표에서 손댄 칸(`USER_CELLS_KEY` 모양).
    """
    # 관 길이 찾기는 배수관 물량과 **같은 규칙**(허용 0.5m) — 두 표가 같은 관에 다른 연장을 안 적게.
    from B08_Quantity.B08_Quantity_Engine_Pipe import _nearest

    templates = templates or {}
    lengths = pipe_lengths or {}
    rows_by_type: dict[str, list[dict[str, Any]]] = {}
    notes: list[str] = []

    def add(type_id: str, row: dict[str, Any]) -> None:
        definition = types.get(type_id)
        if definition is None:
            notes.append(f"{type_id}: 레지스트리에 없는 종류라 안 셈")
            return
        options = row.pop("options")
        keys = [field.key for field in definition.options]
        marks = (user_cells or {}).get(str(row["id"])) or {}
        row["cells"] = _cells(keys, options, _defaults(templates.get(type_id)), marks)
        replaced = [key for key, cell in row["cells"].items() if "replaced_user_value" in cell]
        if replaced:
            labels = {field.key: field.label for field in definition.options}
            row["replaced"] = [labels[key] for key in replaced]
        if definition.design_owner:
            row["note"] = f"{definition.design_owner}가 수량을 셈 — 여기선 자리만"
        elif definition.reference_only:
            row["note"] = "전문 상세설계 대상 — 배치까지만"
        rows_by_type.setdefault(type_id, []).append(row)

    for item in structures:
        type_id = str(item.get("type_id") or "")
        definition = types.get(type_id)
        if definition is not None and definition.managed_by:
            continue  # 관 지점 정본이 주인 — 옛 저장분이 남아 있어도 두 번 안 셈
        options = dict(item.get("options") or {})
        start, end = item.get("start_m"), item.get("end_m")
        span = abs(float(end) - float(start)) if start is not None and end is not None else None
        stated = _number(options.get("length_m"))
        add(
            type_id,
            {
                "id": item.get("structure_id"),
                "origin": "structures",
                "chainage_m": item.get("chainage_m"),
                "start_m": start,
                "end_m": end,
                "count": 1,
                # 원단위 전개와 같은 규칙 — 제원 연장이 있으면 그것, 없으면 시·종점 거리.
                "length_m": stated if stated else span,
                "length_basis": "제원 연장" if stated else ("시·종점" if span else ""),
                "memo": item.get("memo") or "",
                "options": options,
            },
        )

    for point in pipe_points:
        facility = str(point.get("facility") or "pipe")
        type_id = FACILITY_TYPES.get(facility, facility)
        chainage = float(point.get("chainage_m") or 0.0)
        length = _nearest(lengths, chainage) if type_id == "pipe" else None
        add(
            type_id,
            {
                "id": f"pipe@{chainage:.3f}",
                "origin": "pipe_points",
                "chainage_m": chainage,
                "start_m": point.get("start_m"),
                "end_m": point.get("end_m"),
                "count": 1,
                "length_m": length,
                "length_basis": "B06 횡단 관 연장" if length else "",
                "memo": "",
                "options": dict(point.get("options") or {}),
            },
        )

    tables = []
    for type_id, definition in types.items():
        rows = rows_by_type.get(type_id)
        if not rows:
            continue
        rows.sort(key=lambda row: float(row.get("chainage_m") or row.get("start_m") or 0.0))
        columns = [
            {"key": f.key, "label": f.label, "unit": f.unit or "", "input": f.input}
            for f in definition.options
        ]
        averages = {}
        for column in columns:
            values = [_number(row["cells"][column["key"]]["value"]) for row in rows]
            present = [v for v in values if v is not None]
            if present and column["input"] == "number":
                averages[column["key"]] = sum(present) / len(present)
        replaced = sum(1 for row in rows if row.get("replaced"))
        if replaced:
            notes.append(
                f"{definition.name} {replaced}줄 — 이 표에서 고친 값을 B05 가 바꿔 "
                "자동값으로 돌아감"
            )
        measured = [row["length_m"] for row in rows if row["length_m"]]
        # 연장이 뜻 있는 표만 빈 연장을 셈 — 반사경·표지판 같은 점 시설은 연장이 없는 것이 정상.
        needs_length = definition.placement == "interval" or type_id == "pipe"
        tables.append(
            {
                "type_id": type_id,
                "name": definition.name,
                "group": definition.group,
                "placement": definition.placement,
                "columns": columns,
                "rows": rows,
                "count": len(rows),
                "length_total_m": sum(measured) if measured else None,
                "length_missing": len(rows) - len(measured) if needs_length else 0,
                "averages": averages,
            }
        )
    return {"tables": tables, "notes": notes}


def pipe_lengths_from_designs(designs: Iterable[dict[str, Any]]) -> dict[float, float]:
    """저장된 횡단 설계 → `{측점: 관 연장}` — 배수관 물량과 같은 읽기."""
    from B08_Quantity.B08_Quantity_Engine_Pipe import _length_by_chainage

    return _length_by_chainage(list(designs or []), "pipe_length_m")