Aislo/B08_Quantity/B08_Quantity_Engine_HaulSummary.py

"""운반 가중평균 — 내역 줄이 되는 4줄과 그 근거 (B08 일감 5 · PLAN 8-3·8-7).

무엇을 내나
    실무는 **(운반수단 × 지반유형)별 가중평균 1개**를 내역에 올린다. 울진 실측 —
    「도자 토사 1,170㎥ 평균 43.66m · 도자 암 1,554㎥ 39.07m · 덤프 토사 1,667㎥ 293.78m ·
    덤프 암 1,714㎥ 318.6m」로 **4줄**이다. 오솔길도 분류별 가중평균 1개를 낸다
    (거창 무대: 10,399 ÷ 871 = 11.94m).

    개별 구간 줄은 버리지 않고 **근거**로 함께 낸다 — 어느 구간이 그 평균을 만들었는지
    되짚을 수 있어야 한다.

가중평균 = Σ(토량 × 거리) ÷ Σ(토량)
    실무 산출서가 「토량 × 거리」를 쌓아 나누는 그 식이다. 단순평균이 아니다.

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

⚠⚠ 상태(狀態)가 두 개다 — **거리는 다짐, 수량은 자연** (2026-09-09)
    「운반거리의 산정 시에 모든 수량은 다짐상태로 환산하여 계산하고, 내역서에 적용하는
     수량은 자연상태로 한다」(설계실무 요령 — `config_system_design` 5-4-3 인용문).
    유토곡선은 다짐상태로 쌓으므로 **가중평균 거리는 그대로 두고**, 내역에 오르는 수량만
    `natural_m3`(÷C)로 낸다. **환산은 이 파일 한 곳에서 한 번만** 한다 — 받는 쪽(집계표·인계
    줄)은 고르기만 한다. `L`(팽창률)은 쓰지 않는다: 품셈 10-11·10-12 가 `f = 1/L` 을 식 안에서
    스스로 곱하므로 우리는 **자연상태 물량만 정확히 넘기면 된다**.

입력은 `HaulPlan` 이다 (이미 있는 값 — 다시 세지 않는다)
    띠(`bands`)마다 `equipment` · `haul_distance_m` · 지반유형별 물량(`ea_m3`·`rr_m3`·`br_m3`)이
    들어 있다. 떨어진 구간끼리 옮기는 `transfers` 도 같은 모양이라 함께 센다.
"""

from __future__ import annotations

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

from config.config_system_design import EARTHWORK_CONVERSION_FACTORS

# 지반유형 키 ↔ 표기. `HaulPlan` 이 절토 구간 구성비로 안분해 둔 세 갈래다.
GROUND_LABELS = {"ea_m3": "토사", "rr_m3": "리핑암", "br_m3": "발파암"}
# 표기 ↔ 환산계수 이름(`EARTHWORK_CONVERSION_FACTORS` 의 키).
GROUND_KIND_OF = {"토사": "soil", "리핑암": "ripping_rock", "발파암": "blasting_rock"}


#: 계수 묶음의 모양 — `{지반유형: {"compacted": C}}`. 안 주면 정본 기본값이 선다.
Factors = dict[str, dict[str, float]]


def _factor_of(ground: str, factors: Factors | None = None) -> float | None:
    """그 갈래의 다짐 환산계수 `C`. 모르면 `None`(받는 쪽이 환산했는지 되짚는 데 쓴다).

    `factors` 는 프로젝트가 고른 계수다(`earthwork_conversion_factors`) — 안 주면 정본.
    """
    kind = GROUND_KIND_OF.get(ground)
    table = factors or EARTHWORK_CONVERSION_FACTORS
    entry = table.get(kind) if kind else None
    return float(entry["compacted"]) if entry else None


def natural_m3(
    compacted_volume_m3: float, ground: str, factors: Factors | None = None
) -> float | None:
    """**다짐상태 → 자연상태**(÷ C). 내역서에 오르는 수량은 자연상태다.

    근거 — `config_system_design` 5-4-3 에 이미 적혀 있던 문장이다.

        「운반거리의 산정 시에 모든 수량은 다짐상태로 환산하여 계산하고,
         내역서에 적용하는 수량은 자연상태로 한다.」
        (2021년도 국도건설공사 설계실무 요령 / 표준품셈 계열)

    ⚠ **나누기다.** `C = 다짐 ÷ 자연` 이므로 되돌리려면 나눠야 한다. 곱하면 토사가
      1.111배가 아니라 0.9배가 되어 **방향이 뒤집힌다**(거울 시험이 이 방향을 잠근다).
    ⚠ **`L`(팽창률 1.3·1.35·1.625)을 쓰지 않는다.** 우리 곡선은 `×C` 로 쌓았으니
      되돌리는 것도 `C` 다. 품셈 10-11·10-12 는 `f = 1/L` 을 **식 안에서 스스로** 곱하므로
      우리가 `L` 을 또 들면 두 번 환산이 된다.
    ⚠ **환산은 내보내는 이 자리에서 한 번만.** 곡선 안쪽(띠·이동·잔량)은 다짐상태 그대로 둔다
      — 성토 배분은 다짐으로 세는 것이 맞다.
    ⚠ 갈래를 모르면 `None` 이다 — 토사 계수로 눅이면 근거 없이 금액이 움직인다.
    """
    kind = GROUND_KIND_OF.get(ground)
    table = factors or EARTHWORK_CONVERSION_FACTORS
    entry = table.get(kind) if kind else None
    if not entry:
        return None
    factor = float(entry["compacted"])
    return compacted_volume_m3 / factor if factor > 0 else None


# 무대 — 품에 포함이라 내역 줄이 되지 않는다.
FREE_HAUL_KEY = "free_haul"


@dataclass(slots=True)
class HaulLeg:
    """근거 줄 하나 — 어느 구간을 얼마나 몇 m 옮겼나."""

    equipment: str
    ground: str
    volume_m3: float
    distance_m: float
    from_m: float
    to_m: float
    source: str  # `band` 또는 `transfer`


@dataclass(slots=True)
class HaulSummaryRow:
    """내역 줄 — (운반수단 × 지반유형) 하나."""

    equipment: str
    ground: str
    volume_m3: float = 0.0
    work_m3m: float = 0.0  # Σ(토량 × 거리) — 가중평균의 분자
    legs: int = 0
    in_bill: bool = True

    @property
    def average_distance_m(self) -> float:
        return self.work_m3m / self.volume_m3 if self.volume_m3 else 0.0


def _legs_of(plan: dict[str, Any]) -> list[HaulLeg]:
    """`HaulPlan` → 근거 줄 목록. 띠와 장거리 이동을 같은 모양으로 편다."""
    legs: list[HaulLeg] = []

    def push(
        item: dict[str, Any],
        equipment: str | None,
        distance: Any,
        source: str,
        from_m: Any,
        to_m: Any,
    ) -> None:
        if not equipment or not isinstance(distance, (int, float)):
            return
        for key, label in GROUND_LABELS.items():
            volume = item.get(key)
            if not isinstance(volume, (int, float)) or volume <= 0:
                continue
            legs.append(
                HaulLeg(
                    equipment=str(equipment),
                    ground=label,
                    volume_m3=float(volume),
                    distance_m=float(distance),
                    from_m=float(from_m or 0.0),
                    to_m=float(to_m or 0.0),
                    source=source,
                )
            )

    for block in plan.get("blocks") or []:
        for band in block.get("bands") or []:
            push(
                band,
                band.get("equipment"),
                band.get("haul_distance_m"),
                "band",
                band.get("haul_from_m"),
                band.get("haul_to_m"),
            )
    for transfer in plan.get("transfers") or []:
        push(
            transfer,
            transfer.get("equipment"),
            transfer.get("haul_distance_m"),
            "transfer",
            transfer.get("from_m"),
            transfer.get("to_m"),
        )
    return legs


def summarize(legs: Iterable[HaulLeg]) -> list[HaulSummaryRow]:
    """(운반수단 × 지반유형)별 가중평균. 실무 내역이 이 줄들을 그대로 쓴다."""
    grouped: dict[tuple[str, str], HaulSummaryRow] = {}
    for leg in legs:
        key = (leg.equipment, leg.ground)
        row = grouped.get(key)
        if row is None:
            row = HaulSummaryRow(
                equipment=leg.equipment,
                ground=leg.ground,
                in_bill=leg.equipment != FREE_HAUL_KEY,
            )
            grouped[key] = row
        row.volume_m3 += leg.volume_m3
        row.work_m3m += leg.volume_m3 * leg.distance_m
        row.legs += 1
    # 수단 → 지반유형 순으로 안정 정렬 — 화면·내역 줄 순서가 매번 같아야 한다.
    order = {FREE_HAUL_KEY: 0, "dozer": 1, "dump_truck": 2}
    labels = list(GROUND_LABELS.values())
    return sorted(
        grouped.values(),
        key=lambda row: (
            order.get(row.equipment, 9),
            labels.index(row.ground) if row.ground in labels else 9,
        ),
    )


def build_table(plan: dict[str, Any] | None, factors: Factors | None = None) -> dict[str, Any]:
    """화면·API 가 그대로 쓰는 모양. 내역 줄과 근거 줄을 함께 낸다.

    `factors` 는 프로젝트가 고른 토량환산계수다 — 다짐 → 자연 되돌리기가 이 값에 걸린다.
    """
    legs = _legs_of(plan or {})
    rows = summarize(legs)
    return {
        "method": "volume_weighted_average",
        "rows": [
            {
                "equipment": row.equipment,
                "ground": row.ground,
                # ⚠ 이 칸은 **다짐상태**다 — 운반거리를 낸 그 상태 그대로(검산도 이 값으로 한다).
                "volume_m3": row.volume_m3,
                "volume_basis": "compacted",
                # 내역서에 오르는 수량 = **자연상태**(÷C). 갈래를 모르면 `None`.
                "natural_m3": natural_m3(row.volume_m3, row.ground, factors),
                "natural_volume_basis": "natural",
                "conversion_c": _factor_of(row.ground, factors),
                "average_distance_m": row.average_distance_m,
                "work_m3m": row.work_m3m,
                "legs": row.legs,
                "in_bill": row.in_bill,
            }
            for row in rows
        ],
        # 근거 — 어느 구간이 그 평균을 만들었나. 내역에는 안 오른다.
        "legs": [
            {
                "equipment": leg.equipment,
                "ground": leg.ground,
                "volume_m3": leg.volume_m3,
                "distance_m": leg.distance_m,
                "from_m": leg.from_m,
                "to_m": leg.to_m,
                "source": leg.source,
            }
            for leg in legs
        ],
        "bill_row_count": sum(1 for row in rows if row.in_bill),
    }


def summary_input_rows(table: dict[str, Any]) -> list[dict[str, Any]]:
    """토공집계표가 받는 모양으로 줄인다 — 집계표는 근거 줄을 안 쓴다.

    ⚠ **두 상태를 함께 넘긴다** — 집계표·내역은 `natural_m3`(자연상태)를 쓰고,
      `volume_m3`(다짐상태)는 검산·되짚기용이다. 받는 쪽이 또 환산하지 않게 칸 이름으로 가른다.
    """
    return [
        {
            "equipment": row["equipment"],
            "ground": row["ground"],
            "volume_m3": row["volume_m3"],
            "volume_basis": row.get("volume_basis") or "compacted",
            "natural_m3": row.get("natural_m3"),
            "conversion_c": row.get("conversion_c"),
            "average_distance_m": row["average_distance_m"],
        }
        for row in table.get("rows") or []
    ]


@dataclass(slots=True)
class HaulCheck:
    """검산 — 무대를 안 내면 이 대조가 죽는다(PLAN 8-7 ㉡)."""

    hauled_total_m3: float = 0.0
    plan_total_m3: float = 0.0
    difference_m3: float = 0.0
    details: dict[str, float] = field(default_factory=dict)


def check_against_plan(table: dict[str, Any], plan: dict[str, Any] | None) -> HaulCheck:
    """`무대 + 도자 + 덤프` 합이 `HaulPlan` 의 총 운반량과 맞는가.

    ⚠ **다짐상태끼리 비교한다** — 계획(`HaulPlan`)이 다짐이라 자연상태로 환산한 값을 대면
      늘 어긋난다. 검산은 환산 전 값(`volume_m3`)으로 하는 것이 맞다.
    """
    hauled = sum(float(row.get("volume_m3") or 0.0) for row in table.get("rows") or [])
    plan = plan or {}
    planned = float(plan.get("hauled_m3") or 0.0) + float(plan.get("transferred_m3") or 0.0)
    by_equipment: dict[str, float] = {}
    for row in table.get("rows") or []:
        key = str(row.get("equipment"))
        by_equipment[key] = by_equipment.get(key, 0.0) + float(row.get("volume_m3") or 0.0)
    return HaulCheck(
        hauled_total_m3=hauled,
        plan_total_m3=planned,
        difference_m3=hauled - planned,
        details=by_equipment,
    )