Aislo/B08_Quantity/B08_Quantity_Engine_ObservedUnit.py

"""콘크리트 구조물 — **관측 원단위표** 조회 (B08 일감 ⑩ · PLAN 8-6·8-8).

왜 전개식이 아니라 관측값인가
    지식DB 가 못 박아 둔 사실이다 — **구조물별 표준 물량표는 품셈에 없다**
    (`구조물_수량.md` 마지막 줄 · `배수공_수량.md` §2). 콘크리트 구조물의 물량은
    **설계 표준도**에서 나오는데 그 표준도가 원문(법·품셈)에 없다. 그래서 옹벽·집수정처럼
    치수가 표준화된 것은 **실무 설계원본에서 뽑은 관측값**이 유일한 원천이다.

두 근거가 한 표에 섞인다 — 그래서 줄마다 `basis` 를 단다
    · `derived` — 저장된 치수에서 **식으로** 나온 값(돌쌓기 계열, `..._Engine_UnitQuantity`).
    · `observed` — 실무 관측 원단위표에서 **규격을 맞춰 꺼낸** 값(이 모듈).
    섞어 두고 근거를 안 적으면, 나중에 「이 값이 왜 이런가」를 아무도 못 되짚는다.

⚠⚠ **보간하지 않는다**
    관측값은 **그 규격에서만** 맞다. `반중력식 H=2.0` 의 콘크리트 1.35 ㎥/m 를 H=1.6 으로
    줄여 쓰면 틀린다 — 기초·벽 두께는 높이에 비례하지 않는다. 규격이 표에 없으면
    **「원단위 미확보」로 드러낸다.** 가까운 값을 갖다 쓰는 길을 두지 않는다.

⚠ 치수를 지어내지 않는다
    BOX암거는 `structures.json` 에 **벽·저판·상판 두께가 없어** 전개식조차 못 세운다.
    두께를 가정하면 그 값이 콘크리트·거푸집·철근으로 **번져 나간다**. 미확보로 낸다.

⚠ 이중계상 규칙은 그대로다
    ㉢ 배합을 분해하지 않는다(콘크리트 ㎥·모르터 ㎥ 까지). ㉠ 할증은 자재총괄 한 곳뿐.
    터파기·되메우기·잔토는 `destination: earthwork` 로 토공에 합산된다.
"""

from __future__ import annotations

import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

DATASET_DIR = Path(__file__).resolve().parents[1] / "resources" / "data_structure_unit"
DATASET_PREFIX = "structure_unit_observed_"

#: 값이 어디서 왔나 — 한 표에 섞이므로 줄마다 단다.
BASIS_DERIVED = "derived"  # 저장된 치수에서 식으로
BASIS_OBSERVED = "observed"  # 실무 관측 원단위표에서

NOTE_UNIT_MISSING = "원단위 미확보"


def _latest_dataset_path(directory: Path | None = None) -> Path | None:
    folder = directory or DATASET_DIR
    if not folder.is_dir():
        return None
    files = sorted(folder.glob(DATASET_PREFIX + "*.json"))
    return files[-1] if files else None


def _same(left: Any, right: Any) -> bool:
    """규격 한 칸 비교. 숫자는 값으로, 나머지는 글자로 **정확히** 본다.

    `"800"` 과 `800` 은 같게 보되(입력 폼이 문자열을 준다), `2.0` 과 `1.6` 은 다르다 —
    가까운 값을 같다고 보는 길은 두지 않는다.
    """
    if isinstance(left, (int, float)) and not isinstance(left, bool):
        try:
            return abs(float(left) - float(right)) < 1e-9
        except (TypeError, ValueError):
            return False
    return str(left).strip() == str(right).strip()


@dataclass
class ObservedUnitTable:
    """관측 원단위표 한 판."""

    effective_date: str = ""
    entries: list[dict[str, Any]] = field(default_factory=list)
    sources: dict[str, Any] = field(default_factory=dict)
    not_found: dict[str, Any] = field(default_factory=dict)
    #: 값을 바꾸는 **설계 조건**인데 우리 제원에 칸이 없는 것 — 화면이 보이게 한다.
    pending_choices: dict[str, Any] = field(default_factory=dict)

    def find(self, type_id: str, spec: dict[str, Any]) -> dict[str, Any] | None:
        """규격이 **모두** 맞는 줄만 돌려준다. 하나라도 어긋나면 없는 것으로 본다."""
        for entry in self.entries:
            if entry.get("type_id") != type_id:
                continue
            wanted = entry.get("spec") or {}
            if all(key in spec and _same(value, spec[key]) for key, value in wanted.items()):
                return entry
        return None

    def specs_for(self, type_id: str) -> list[dict[str, Any]]:
        """그 종류로 표에 있는 규격 목록 — 「무엇이 있는지」를 화면이 보이게."""
        return [
            entry.get("spec") or {} for entry in self.entries if entry.get("type_id") == type_id
        ]


def load_observed_table(path: Path | None = None) -> ObservedUnitTable:
    """관측 원단위표를 읽는다. 파일이 없으면 **빈 표** — 전부 「미확보」로 드러난다."""
    target = path or _latest_dataset_path()
    if target is None or not target.is_file():
        return ObservedUnitTable()
    payload = json.loads(target.read_text(encoding="utf-8"))
    return ObservedUnitTable(
        effective_date=str(payload.get("effective_date") or ""),
        entries=list(payload.get("entries") or []),
        sources=payload.get("sources") or {},
        not_found=payload.get("not_found") or {},
        pending_choices=payload.get("pending_choices") or {},
    )


def scale_for(entry: dict[str, Any], structure: dict[str, Any]) -> tuple[float, str]:
    """관측값에 곱할 수 — 단위가 `m` 면 연장, `㎡` 면 면적, `개소` 면 1.

    ⚠ **규격을 늘리는 것이 아니라 개수를 세는 것**이다. `H=2.0 옹벽 10m` 는 같은 단면이
      10m 이어진 것이라 곱해도 되지만, `H=1.6` 으로 바꾸는 것은 단면이 달라지므로 안 된다.
    """
    options = structure.get("options") or {}
    unit = str(entry.get("unit") or "개소")
    if unit == "m":
        length = options.get("length_m")
        if length is None:
            start, end = structure.get("start_m"), structure.get("end_m")
            length = (
                abs(float(end) - float(start)) if start is not None and end is not None else 0.0
            )
        return float(length or 0.0), f"연장 {float(length or 0.0):g} m"
    if unit == "㎡":
        width = options.get("ford_width_m")
        length = options.get("length_m") or 0.0
        area = float(width or 0.0) * float(length or 0.0)
        return area, f"면적 {area:g} ㎡"
    return 1.0, "1 개소"


def billing_of(
    type_id: str,
    spec: dict[str, Any],
    structure: dict[str, Any],
    table: ObservedUnitTable | None = None,
) -> tuple[str, float] | None:
    """(내역 단위, 그 단위로 센 수량). 표에 없으면 `None`.

    ⚠ **왜 있나** — 관측 원단위는 「개소당」·「㎡당」으로도 온다. 그런데 인계 줄이 늘
      「m · 연장」으로 나가고 있어, **집수정 한 개소가 「연장 2m」면 값이 두 배로 실렸다**
      (2026-09-08 ㉕ 실증에서 드러남). 성분은 개소 기준으로 맞게 서는데 **줄의 축만
      어긋나** 있어서 아무 시험도 안 잡았다.
    """
    found = (table or load_observed_table()).find(type_id, spec)
    if found is None:
        return None
    scale, _note = scale_for(found, structure)
    return str(found.get("unit") or "개소"), scale


#: 단면으로 터파기를 세운 줄의 사유 — 표 사유·구조물도 그림이 **같은 말**(브레인 판정 2026-09-14).
SECTION_TRENCH_NOTE = (
    "터파기는 기초 폭 + 양쪽 여유 · 옆면 1:0.5(판 깊이 기준)"
    " — 벽 높이 몫은 지반선이 제원에 없어 제외"
)


def earthwork_missing_note(entry: dict[str, Any]) -> str | None:
    """관측 줄에 토공 줄이 없고 단면으로도 못 세우면 한 줄 — 표 사유·구조물도 그림이 같은 말."""
    if any(item.get("destination") == "earthwork" for item in entry.get("components") or []):
        return None
    if (entry.get("section") or {}).get("trench"):
        return SECTION_TRENCH_NOTE
    return "⚠ 터파기·되메우기·잔토가 안 섬 — 관측 원단위 표에 그 줄이 없음(값을 지어내지 않음)"


def weep_component(
    entry: dict[str, Any], options: dict[str, Any], scale: float, scale_note: str
) -> dict[str, Any] | None:
    """물구멍관 — 단면이 있는 관측 줄은 **실무 식에 제원 칸을 넣어** 다시 셈(붙박이 값 대신).

    식 `벽 높이 ÷ 개소당 벽면적 × 관 길이`(소광리 「옹벽2.0」 T31). 관 길이는 실무 관측값이라
    `basis_kind` 는 그대로 `observed`. 규격 `Ø지름` 을 달아 돌쌓기 물구멍관과 (이름+규격)으로 합침.
    """
    from B08_Quantity.B08_Quantity_Engine_UnitQuantity_StoneSpec import weep_hole_spec

    section = entry.get("section") or {}
    weep = section.get("weep")
    if not weep:
        return None
    area, diameter, spec_basis = weep_hole_spec(options)
    height, length = section["wall"]["height_m"], weep["pipe_length_m"]
    per_unit = height / area * length
    return {
        "name": "물구멍관",
        "unit": "m",
        "amount": per_unit * scale,
        "destination": "material",
        "basis": (
            f"관측 원단위 식(소광리 「옹벽2.0」 T31) 벽 높이 {height:g} ÷ {area:g}㎡/개소"
            f" × 관 {length:g} m = {per_unit:.3f}/{entry.get('unit')} × {scale_note}"
            f" · 관 길이는 관측값 · {spec_basis}"
        ),
        "basis_kind": BASIS_OBSERVED,
        "source": str(entry.get("source") or ""),
        "spec": f"Ø{diameter}",
    }


def section_trench_components(
    entry: dict[str, Any], rubble_thickness_m: float, scale: float
) -> tuple[list[dict[str, Any]], float] | None:
    """단면이 있는 관측 줄의 터파기·되메우기·잔토(`destination: earthwork`)와 파는 깊이(m).

    ⚠ 같은 파일 식생옹벽블럭 기초 터파기 식 — 밑폭 = 기초 폭 + 양쪽 여유 0.3 · 옆면 1:0.5
      (윗폭 = 밑폭 + 0.5 × 판 깊이 × 2) · 깊이 = 기초 + 버림 + 기초잡석 · 벽 높이 몫은 안 셈
      (지반선이 제원에 없음). 브레인 판정 2026-09-14 — 처음 수직에서 1:0.5 로 고침.
    ⚠ 되메우기 = 터파기 − 그 안에 든 것(기초·전단키 콘크리트 + 버림 + 기초잡석) — 잔토 = 든 것.
    """
    section = entry.get("section") or {}
    trench = section.get("trench")
    if not trench:
        return None
    footing, key, blinding = section["footing"], section["key"], section["blinding"]
    clearance, slope = trench["clearance_m"], trench.get("side_slope", 0.0)
    width = footing["width_m"] + 2 * clearance
    depth = footing["thickness_m"] + blinding["thickness_m"] + rubble_thickness_m
    top_width = width + 2 * slope * depth
    key_below = max(key["depth_m"] - blinding["thickness_m"] - rubble_thickness_m, 0.0)
    excavation = (width + top_width) / 2 * depth + key["width_m"] * key_below
    layer_width = footing["width_m"] + 2 * blinding["overhang_m"] - key["width_m"]
    filled = (
        footing["width_m"] * footing["thickness_m"]
        + key["width_m"] * key["depth_m"]
        + layer_width * (blinding["thickness_m"] + rubble_thickness_m)
    )
    backfill = excavation - filled
    basis = (
        f"(밑폭 {width:.2f}(기초 {footing['width_m']:g} + 여유 {clearance:g}×2)"
        f" + 윗폭 {top_width:.2f}(1:{slope:g})) ÷ 2 × 깊이 {depth:.2f}"
        f"(기초 {footing['thickness_m']:g} + 버림 {blinding['thickness_m']:g}"
        f" + 기초잡석 {rubble_thickness_m:g})"
        f" · {SECTION_TRENCH_NOTE} · 여유·기울기는 소광리 식생옹벽블럭 기초 터파기"
    )
    rows = [
        ("터파기", excavation, basis),
        ("되메우기", backfill, f"터파기 − 든 것 {filled:.3f}(기초·전단키 + 버림 + 기초잡석)"),
        ("잔토처리", filled, "터파기 − 되메우기"),
    ]
    return [
        {
            "name": name,
            "unit": "㎥",
            "amount": amount * scale,
            "destination": "earthwork",
            "basis": f"{text} = {amount:.3f}/{entry.get('unit')}",
            "basis_kind": BASIS_DERIVED,
            "source": str(entry.get("source") or ""),
        }
        for name, amount, text in rows
    ], depth


def expand_observed(
    type_id: str,
    spec: dict[str, Any],
    structure: dict[str, Any],
    table: ObservedUnitTable | None = None,
) -> tuple[list[dict[str, Any]], list[str]]:
    """(성분 목록, 알림). 규격이 표에 없으면 **빈 목록 + 미확보 알림**을 낸다."""
    from B08_Quantity.B08_Quantity_Wording import spec_missing

    found = (table or load_observed_table()).find(type_id, spec)
    if found is None:
        known = (table or load_observed_table()).specs_for(type_id)
        # ⚠ 키 이름을 화면에 내보내지 않는다 — 사용자는 `retaining_wall` 을 모른다.
        return [], [spec_missing(type_id, known)]

    scale, scale_note = scale_for(found, structure)
    if scale <= 0:
        from B08_Quantity.B08_Quantity_Wording import type_label

        where = (
            "면적(월류 폭 × 포장 길이)이 0 — 두 칸을 적으면 섬"
            if found.get("unit") == "㎡"
            else "연장이 0 — 연장을 적으면 섬"
        )
        return [], [f"{type_label(type_id)}의 {where}"]

    source_key = str(found.get("source") or "")
    options = structure.get("options") or {}
    components: list[dict[str, Any]] = []
    for item in found.get("components") or []:
        note = str(item.get("basis_note") or "")
        weep = (
            weep_component(found, options, scale, scale_note)
            if item["name"] == "물구멍관"
            else None
        )
        if weep is not None:
            components.append(weep)
            continue
        components.append(
            {
                "name": item["name"],
                "unit": item["unit"],
                "amount": float(item["amount"]) * scale,
                "destination": item.get("destination") or "material",
                # 근거를 값 옆에 붙인다 — 관측값임을 화면·인계에서 바로 알아야 한다.
                "basis": f"관측 원단위 {item['amount']:g}/{found.get('unit')} × {scale_note}"
                + (f" ({note})" if note else ""),
                "basis_kind": BASIS_OBSERVED,
                "source": source_key,
                # 규격은 이름과 따로(명세 13장 Ⓒ) — 「이형철근」 + 「D13」(2026-09-14 가르기).
                "spec": str(item.get("spec") or ""),
            }
        )
    notes = [f"관측 원단위 적용 — {found.get('source_note') or source_key}"]
    # 줄에 붙은 사유(원문끼리 다름 등) — 구조물도 그림도 같은 글자를 씀.
    notes.extend(str(note) for note in found.get("notes") or [])
    earthwork_note = earthwork_missing_note(found)
    if earthwork_note:
        notes.append(earthwork_note)
    return components, notes