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)

    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 {},
    )


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 expand_observed(
    type_id: str,
    spec: dict[str, Any],
    structure: dict[str, Any],
    table: ObservedUnitTable | None = None,
) -> tuple[list[dict[str, Any]], list[str]]:
    """(성분 목록, 알림). 규격이 표에 없으면 **빈 목록 + 미확보 알림**을 낸다."""
    found = (table or load_observed_table()).find(type_id, spec)
    if found is None:
        known = (table or load_observed_table()).specs_for(type_id)
        detail = f" — 표에 있는 규격: {known}" if known else ""
        return [], [f"{NOTE_UNIT_MISSING} ({type_id} {spec}){detail}"]

    scale, scale_note = scale_for(found, structure)
    if scale <= 0:
        return [], [f"{NOTE_UNIT_MISSING} — 곱할 연장·면적이 0 ({type_id})"]

    source_key = str(found.get("source") or "")
    components: list[dict[str, Any]] = []
    for item in found.get("components") or []:
        note = str(item.get("basis_note") or "")
        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,
            }
        )
    return components, [f"관측 원단위 적용 — {found.get('source_note') or source_key}"]