Aislo/B08_Quantity/B08_Quantity_Engine_EarthworkSummary.py

"""토공집계표 — 토적표·사면표를 공종별 총량으로 모은다 (B08 일감 4 · PLAN 8-11).

열 구성 (거창 실무 `토공집계표` 시트 그대로)
    `구분 · 공종 · 규격 · 단위 · 계 · 비고`

암 갈래는 **개수를 코드에 박지 않는다** (PLAN 8-13)
    울진 2갈래(연암·발파암) · 거창 5갈래(토사·풍화암·연암·보통암·경암) · 오솔길 BOM 1갈래로
    공사마다 다르다. 프로젝트 설정의 세트를 받아 그만큼 줄을 낸다.

    ⚠ 측점별 암질 판정에 기대지 않는다(PLAN 8-1 사용자 확정) — 절토량은 기하에서 나오고
    **암/토사 나눔과 갈래 비율은 설계자 입력**이다. 그래서 여기서는 토적표의 「암」 총량을
    설계자가 준 비율(%)로 나눠 줄을 만든다.

⚠ 반영률은 법정값이 아니다 (PLAN 8-11)
    실무 시트가 「성토면 80 % 반영」처럼 비고란에 손으로 적어 둔 값이다. **기본 100 %** 이고
    설계자가 바꾼다. 실무 관측치(80/50/80)는 참고이지 기본값이 아니다.

⚠ 무대(소운반 20m)는 집계에는 오르되 **내역 줄이 되지 않는다** (PLAN 8-7 ㉡)
    품셈 1-2-7 「소운반 20 m 이내는 품에 포함」. 켜도 붙일 단가가 품셈에 없다.
    그래서 `in_bill=False` 로 표시해 넘긴다 — 값은 검산(`무대+도자+덤프 = 총 운반토량`)에 쓴다.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Iterable

# 지반 구분이 붙는 공종 — 실무 시트가 이 셋을 각각 암 갈래만큼 늘려 적는다.
GROUND_SPLIT_GROUPS = ("흙깎기", "측구터파기", "구조물터파기")

# 반영률 키 ↔ 집계 공종. 값은 프로젝트 설정에서 온다(기본 100 %).
RATIO_OF_ROW = {
    "성토면다짐": "fill_slope_compaction",
    "초류종자살포": "seed_spray_fill",  # 성토면 몫에만 걸린다 — 절토면은 별도 키
    "지장목제거": "obstacle_removal",
}


@dataclass(slots=True)
class SummaryRow:
    """집계표 한 줄. 이름·단위는 거창 실무 시트 문구를 따른다."""

    group: str  # 구분 (흙깎기·성토·…)
    item: str = ""  # 공종 (토사·연암·…)
    spec: str = ""  # 규격 (기계(굴삭기)·백호우·…)
    unit: str = "㎥"
    amount: float = 0.0
    note: str = ""
    # 내역서 줄이 되는가 — 무대처럼 품에 포함된 것은 False (PLAN 8-7 ㉡).
    in_bill: bool = True


@dataclass(slots=True)
class SummaryInput:
    """집계에 필요한 값 묶음. 토적표·사면표·운반계획에서 이미 나온 것만 받는다."""

    earthwork_totals: dict[str, float] = field(default_factory=dict)
    slope_totals: dict[str, float] = field(default_factory=dict)
    haul_rows: list[dict[str, Any]] = field(default_factory=list)
    rock_classes: list[str] = field(default_factory=list)
    rock_ratios_pct: dict[str, float] = field(default_factory=dict)
    application_ratios: dict[str, float] = field(default_factory=dict)


def _ratio(source: SummaryInput, key: str) -> float:
    """반영률(0~1). 없으면 1.0 — 실무 관측치를 기본값으로 쓰지 않는다."""
    value = source.application_ratios.get(key)
    return float(value) if isinstance(value, (int, float)) else 1.0


def _split_by_rock(total: float, source: SummaryInput) -> list[tuple[str, float]]:
    """암 총량을 설계자가 준 비율(%)로 갈래별로 나눈다.

    비율이 아직 없으면 **나누지 않고 「암」 한 줄로** 낸다 — 지어낸 비율로 쪼개지 않는다.
    """
    classes = [name for name in source.rock_classes if name != "토사"]
    ratios = {name: float(source.rock_ratios_pct.get(name, 0) or 0) for name in classes}
    given = sum(ratios.values())
    if given <= 0:
        return [("암", total)]
    return [(name, total * ratios[name] / given) for name in classes if ratios[name] > 0]


def build_rows(source: SummaryInput) -> list[SummaryRow]:
    """토공집계표 줄 목록. 값이 0 인 갈래도 줄은 남긴다(실무 시트가 그렇다)."""
    earth = source.earthwork_totals
    slope = source.slope_totals
    rows: list[SummaryRow] = []

    # ── 흙깎기 · 측구터파기 — 토사 한 줄 + 암 갈래만큼 ──────────────
    for group, soil_key, rock_key in (
        ("흙깎기", "cut_soil_volume_m3", "cut_rock_volume_m3"),
        ("측구터파기", "ditch_soil_volume_m3", "ditch_rock_volume_m3"),
    ):
        rows.append(
            SummaryRow(
                group=group, item="토사", spec="기계(굴삭기)", amount=earth.get(soil_key, 0.0)
            )
        )
        for name, amount in _split_by_rock(earth.get(rock_key, 0.0), source):
            rows.append(SummaryRow(group=group, item=name, spec="굴삭기+브레카", amount=amount))

    rows.append(SummaryRow(group="보정량계", amount=earth.get("adjusted_total_m3", 0.0)))
    rows.append(SummaryRow(group="성토", amount=earth.get("fill_volume_m3", 0.0)))

    # ── 운반 — 수단별. 무대는 집계에 오르되 내역 줄이 아니다 ────────
    rows.extend(_haul_rows(source))

    # ── 사면 계열 — 반영률이 여기서 걸린다 ────────────────────────
    fill_face = slope.get("face_dressing_fill", 0.0)
    cut_face = slope.get("face_dressing_cut", 0.0)
    rows.append(
        SummaryRow(
            group="성토면다짐",
            unit="㎡",
            amount=fill_face * _ratio(source, "fill_slope_compaction"),
            note=_ratio_note(source, "fill_slope_compaction", "성토면"),
        )
    )
    seed = fill_face * _ratio(source, "seed_spray_fill") + cut_face * _ratio(
        source, "seed_spray_cut"
    )
    rows.append(
        SummaryRow(
            group="초류종자살포",
            spec="씨드스프레이",
            unit="㎡",
            amount=seed,
            note=_seed_note(source),
        )
    )
    removal = slope.get("tree_removal_fill", 0.0) + slope.get("tree_removal_cut", 0.0)
    rows.append(
        SummaryRow(
            group="지장목제거",
            unit="㎡",
            amount=removal * _ratio(source, "obstacle_removal"),
            note=_ratio_note(source, "obstacle_removal", "성토면+절토면"),
        )
    )
    rows.append(
        SummaryRow(
            group="층따기", spec="백호우", unit="㎡", amount=slope.get("bench_cut_fill", 0.0)
        )
    )
    return rows


def _ratio_note(source: SummaryInput, key: str, base: str) -> str:
    ratio = _ratio(source, key)
    return "" if abs(ratio - 1.0) < 1e-9 else f"{base} {ratio * 100:g} % 반영"


def _seed_note(source: SummaryInput) -> str:
    fill = _ratio(source, "seed_spray_fill")
    cut = _ratio(source, "seed_spray_cut")
    if abs(fill - 1.0) < 1e-9 and abs(cut - 1.0) < 1e-9:
        return ""
    return f"성토면 {fill * 100:g} % 반영 + 절토면 {cut * 100:g} % 반영"


# 운반수단 표기 — `HaulPlan` 의 키를 실무 시트 문구로 옮긴다.
HAUL_LABELS = {"free_haul": "무대(종방향유용토)", "dozer": "도자운반", "dump_truck": "덤프운반"}


def _haul_rows(source: SummaryInput) -> list[SummaryRow]:
    """운반 — (운반수단 × 지반유형)별 가중평균 줄 (PLAN 8-3).

    무대는 `in_bill=False` — 품셈 1-2-7 로 품에 포함돼 단가가 없다. 값은 검산에 쓴다.
    """
    rows: list[SummaryRow] = []
    for item in source.haul_rows:
        key = str(item.get("equipment") or "")
        label = HAUL_LABELS.get(key, key or "운반")
        ground = str(item.get("ground") or "")
        distance = item.get("average_distance_m")
        note = f"평균운반거리 {float(distance):.2f} m" if isinstance(distance, (int, float)) else ""
        if key == "free_haul":
            note = (note + " · 내역 제외(품에 포함)").strip(" ·")
        rows.append(
            SummaryRow(
                group=label,
                item=ground,
                amount=float(item.get("volume_m3") or 0.0),
                note=note,
                in_bill=key != "free_haul",
            )
        )
    return rows


def build_table(source: SummaryInput) -> dict[str, Any]:
    """화면·API 가 그대로 쓰는 모양."""
    rows = build_rows(source)
    return {
        "columns": ["구분", "공종", "규격", "단위", "계", "비고"],
        "rock_classes": list(source.rock_classes),
        "rock_ratios_pct": dict(source.rock_ratios_pct),
        "application_ratios": dict(source.application_ratios),
        "rows": [
            {
                "group": row.group,
                "item": row.item,
                "spec": row.spec,
                "unit": row.unit,
                "amount": row.amount,
                "note": row.note,
                "in_bill": row.in_bill,
            }
            for row in rows
        ],
        "row_count": len(rows),
    }


def haul_check(source: SummaryInput, earthwork_totals: dict[str, float]) -> dict[str, Any]:
    """검산 — `무대 + 도자 + 덤프` 합이 총 운반토량과 맞는가 (PLAN 8-7 ㉡).

    무대를 안 내면 이 검산이 안 된다. 그래서 값은 내되 내역 줄만 빼는 것이다.
    """
    hauled = sum(float(item.get("volume_m3") or 0.0) for item in source.haul_rows)
    diverted = float(earthwork_totals.get("diverted_m3") or 0.0)
    return {
        "hauled_total_m3": hauled,
        "diverted_total_m3": diverted,
        "difference_m3": hauled - diverted,
    }


def totals_by_unit(rows: Iterable[SummaryRow]) -> dict[str, float]:
    """단위별 합계 — ㎥ 와 ㎡ 를 섞어 더하지 않는다."""
    result: dict[str, float] = {}
    for row in rows:
        result[row.unit] = result.get(row.unit, 0.0) + row.amount
    return result