Aislo/B09_Estimation/B09_Estimation_MachineCost.py

"""B09 원가계산 — 기계경비 (PLAN 9-3 의 `S` → `X` 두 단계).

실무·교본이 같은 두 단계다 (신규 문서 5장 라-1 예제도 같은 모양):

    S 취득가(천원)  ──(손료계수)──►  X 시간당 중기사용료
                                        = 손료(경비) + 연료(재료) + 운전사(노무)

이 모듈이 내는 것은 **시간당 사용료 한 시간분**이고, 3분할(재료·노무·경비)로 낸다.
`B09_Estimation_PriceBook` 의 `MACHINE_BASE`(S) · `MACHINE_HOURLY`(X) 층에 그대로 앉는다.

데이터 (`resources/data_cost_input_value/mach_base_2026.json`)
  - `mach_price` **613 기종** — `machine_code` + `specification` 으로 규격이 갈린다.
  - `mach_loss_coef` **387건** — 시간당 손료계수·내용시간·연간표준시간.
  - ⚠ `mach_fuel_rate` **0건** · `mach_operator_map` **0건** — **비어 있다.**
    연료소모량과 기종별 운전사 직종은 품셈 본문에 있고, 아직 뽑히지 않았다.

⚠ **그래서 이 모듈은 손료(경비)까지만 채우고, 연료·운전사는 「공백」으로 표시한다.**
0 으로 때우면 **시간당 사용료가 절반 이하로 나오고 그대로 총액에 섞인다** — 실측
비중이 노무 53 % · 재료 20 % · 경비 27 % 라 손료만으로는 4분의 1 남짓이다.
"""

from __future__ import annotations

import json
import os
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Any

from B09_Estimation.B09_Estimation_Guards import (
    check_operator_hours_basis,
    reject_efficiency_in_hourly_rate,
)
from B09_Estimation.B09_Estimation_PriceBook import Money3

_CATALOG_SUBPATH = ("resources", "data_cost_input_value")
_THOUSAND = Decimal(1000)


#: 조종원 노임을 시간당으로 바꿀 때 나누는 시간.
#: 근거 — 품셈 8-1-6 기계손료 단서 「**관리비는 1일 8시간 초과해도 8시간으로 계산**」이
#: 1일을 8시간으로 잡는다. 노임표도 `hours_per_day: 8` 이다.
#: ⚠ **실무 관측은 이보다 짧은 시간으로 나눈 사례가 있다** — STC 2024 관측
#: 굴착기 0.7㎥ 노무 55,700 원/hr 은 건설기계운전사 일당 283,297 을 **약 5.09시간**으로
#: 나눈 값이다(작업효율 `E`(실작업시간율)를 시간당 사용료에 미리 반영한 것으로 보임).
#: 품셈에서 작업효율은 **작업량 산정**(`Q = n·q·f·E`, 8-1-4)에 들어가지 시간당 사용료에
#: 들어가지 않으므로 **8시간을 유지한다.** 관측에 맞추려 나눗수를 바꾸지 않는다
#: (★법대로 PLAN 8-10). TODO(미결 PLAN 9-6): 발주처가 실가동시간을 요구하는 사례 확인.
OPERATOR_HOURS_PER_DAY = 8


class MachineCostError(LookupError):
    """기계경비를 세울 수 없는 경우. 0 으로 때우지 않고 멈춘다."""


def _project_root() -> str:
    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def _read_json(file_name: str) -> dict[str, Any]:
    with open(os.path.join(_project_root(), *_CATALOG_SUBPATH, file_name), encoding="utf-8") as h:
        return json.load(h)


@dataclass(frozen=True)
class MachineSpec:
    """기종 한 줄 — 규격까지 붙어야 한 대가 정해진다."""

    machine_code: str
    name: str
    specification: str
    price_thousand_krw: Decimal
    loss_coefficient_per_hour: Decimal | None = None
    economic_life_hours: int | None = None
    annual_standard_hours: int | None = None

    @property
    def display_name(self) -> str:
        return f"{self.name} {self.specification}".strip()


@dataclass
class MachineCatalog:
    """613 기종. **이름만으로는 못 고른다** — 규격이 있어야 한 대가 정해진다."""

    machines: dict[str, MachineSpec] = field(default_factory=dict)

    def by_name(self, name: str) -> list[MachineSpec]:
        return [m for m in self.machines.values() if m.name == name]

    def resolve(self, name: str, specification: str) -> MachineSpec | None:
        """이름 + 규격으로 한 대를 고른다. 규격이 없으면 **고르지 않는다**."""
        found = self.by_name(name)
        if not found:
            return None
        if len(found) == 1 and not specification:
            return found[0]
        narrowed = [m for m in found if m.specification == specification]
        return narrowed[0] if len(narrowed) == 1 else None

    def get(self, machine_code: str) -> MachineSpec:
        try:
            return self.machines[machine_code]
        except KeyError as exc:
            raise MachineCostError(f"기종 카탈로그에 없는 코드입니다: {machine_code}") from exc


def load_machine_catalog(file_name: str = "mach_base_2026.json") -> MachineCatalog:
    """취득가 613건에 손료계수 387건을 붙여 카탈로그 한 벌을 만든다."""
    variables = _read_json(file_name)["variables"]
    coefficients = {
        r["machine_code"]: r for r in variables.get("mach_loss_coef", {}).get("records", [])
    }

    catalog = MachineCatalog()
    for row in variables.get("mach_price", {}).get("records", []):
        code = row["machine_code"]
        coefficient = coefficients.get(code, {})
        catalog.machines[code] = MachineSpec(
            machine_code=code,
            name=row["machine_name"],
            specification=str(row.get("specification", "")),
            price_thousand_krw=Decimal(str(row["price_thousand_krw"])),
            loss_coefficient_per_hour=(
                Decimal(str(coefficient["loss_coefficient_per_hour"]))
                if "loss_coefficient_per_hour" in coefficient
                else None
            ),
            economic_life_hours=coefficient.get("economic_life_hours"),
            annual_standard_hours=coefficient.get("annual_standard_hours"),
        )
    return catalog


def hourly_loss_cost(machine: MachineSpec) -> Decimal:
    """시간당 손료 = 취득가 × 손료계수.

    취득가가 **천원 단위**라 원으로 환산한다 — 이 단위를 놓치면 1,000배 틀린다.
    """
    if machine.loss_coefficient_per_hour is None:
        raise MachineCostError(
            f"{machine.display_name}: 손료계수가 없습니다 (취득가만 있는 기종 226건 중 하나)"
        )
    return machine.price_thousand_krw * _THOUSAND * machine.loss_coefficient_per_hour


@dataclass
class HourlyMachineCost:
    """시간당 중기사용료 — 3분할과 **채우지 못한 성분 목록**을 함께 낸다."""

    machine: MachineSpec
    money: Money3
    gaps: list[str] = field(default_factory=list)

    @property
    def is_complete(self) -> bool:
        return not self.gaps


def hourly_machine_cost(
    machine: MachineSpec,
    *,
    fuel_liters_per_hour: Decimal | None = None,
    fuel_price_per_liter: Decimal | None = None,
    operator_daily_wage: Decimal | None = None,
    operator_hours_per_day: int = OPERATOR_HOURS_PER_DAY,
    efficiency_factor: Decimal | None = None,
) -> HourlyMachineCost:
    """시간당 사용료 한 시간분.

    손료(경비)는 카탈로그로 바로 나온다. **연료(재료)·운전사(노무)는 값을 주지 않으면
    0 으로 때우지 않고 `gaps` 에 적어 돌려준다** — 빠진 채로 총액에 섞이는 것이
    이 자리에서 제일 위험하다.
    """
    # ㉣ 작업효율이 이 자리에 들어오면 멈춘다 — 효율은 작업량 산정에만 들어간다.
    reject_efficiency_in_hourly_rate(
        efficiency_factor, where=f"{machine.display_name} 시간당 사용료"
    )

    gaps: list[str] = []
    expense = hourly_loss_cost(machine)

    material = Decimal(0)
    if fuel_liters_per_hour is None or fuel_price_per_liter is None:
        # TODO(미결 PLAN 9-6): `mach_fuel_rate` 0건 — 연료소모량이 품셈 본문에만 있다.
        gaps.append("연료소모량(L/hr) 미확보 — 재료비 성분 비어 있음")
    else:
        material = fuel_liters_per_hour * fuel_price_per_liter

    labor = Decimal(0)
    if operator_daily_wage is None:
        # TODO(미결 PLAN 9-6): `mach_operator_map` 0건 — 기종별 운전사 직종이 품셈 본문에만 있다.
        gaps.append("운전사 직종 매핑 미확보 — 노무비 성분 비어 있음")
    else:
        labor = operator_daily_wage / Decimal(operator_hours_per_day)
        # ㉣ 보조 — 나눗수를 몰래 줄이면 효율을 사용료에 넣은 것이 된다.
        check_operator_hours_basis(
            labor_per_hour=labor,
            daily_wage=operator_daily_wage,
            hours_per_day=operator_hours_per_day,
        )

    return HourlyMachineCost(
        machine=machine,
        money=Money3(material=material, labor=labor, expense=expense),
        gaps=gaps,
    )


def catalog_gaps(catalog: MachineCatalog) -> dict[str, int]:
    """카탈로그 자체의 공백 — 몇 기종이 손료계수를 못 가졌나."""
    missing = [m for m in catalog.machines.values() if m.loss_coefficient_per_hour is None]
    return {"machines": len(catalog.machines), "without_loss_coefficient": len(missing)}