dodil/erp-brick-manufacturing

DODIL ERP brick: manufacturing

src/brick_manufacturing/domain/routing.py

"""Routings and work centres (MFG-01, MFG-03): the steps to make a product, on work centres, and what they cost.

A routing is versioned and dated like a BOM: written once (never edited), valid from its valid_from until the next version's, so two
versions never overlap. An operation says how long it takes per batch (`run_minutes` per `per_quantity`); `basis` says what a batch is
counted in: the product made (`output`) or what goes into it (`input`, the BOM's components in `unit_code`). An operation without a
work centre is waiting time: it moves the date and costs nothing. Pure arithmetic: nothing here reads a rate or a product.
"""
from dataclasses import dataclass
from datetime import date
from decimal import ROUND_CEILING, Decimal
from typing import Optional

from brick_manufacturing.domain.errors import OutOfBounds, RuleBroken
from brick_manufacturing.domain.values import money, q4

RATE_KINDS = ("machine", "labour", "overhead")


@dataclass
class Operation:
    op_no: int
    description: str
    work_centre_id: str                   # "" = waiting time (no work centre)
    setup_minutes: Decimal
    run_minutes: Decimal
    per_quantity: Decimal
    basis: str                            # output | input
    unit_code: str                        # the unit per_quantity is counted in (input basis: which components)
    wait_hours: Decimal


def check_operations(ops: list) -> None:
    if not ops:
        raise RuleBroken("A routing has at least one operation", gate="operations")
    seen = set()
    for o in ops:
        if o.op_no in seen:
            raise RuleBroken(f"Operation {o.op_no} is written twice", gate="operations")
        seen.add(o.op_no)
        if o.work_centre_id and not (o.setup_minutes > 0 or o.run_minutes > 0):
            raise RuleBroken(f"Operation {o.op_no} runs on {o.work_centre_id} but takes no time", gate="operations")
        if o.basis == "input" and not o.unit_code:
            raise OutOfBounds(f"Operation {o.op_no} counts its batch in the input: say in which unit (unit_code)")
        if o.per_quantity <= 0:
            raise OutOfBounds(f"Operation {o.op_no}: per_quantity is more than zero")


def valid_version(versions: list, day: date) -> Optional[dict]:
    """The routing row in force on `day`: the latest whose valid_from is not after it (a later version ends the earlier one)."""
    held = [v for v in versions if v["valid_from"] <= day]
    return max(held, key=lambda v: v["routing_version"]) if held else None


def batches(op: dict, basis_qty: Decimal) -> int:
    per = Decimal(op["per_quantity"])
    return int((basis_qty / per).to_integral_value(rounding=ROUND_CEILING)) if basis_qty > 0 else 0


def basis_quantity(op: dict, scale: Decimal, frozen_lines: list, output_base: Decimal) -> Decimal:
    """What one run of the BOM, times `scale`, counts for this operation: the output, or the components in the operation's unit."""
    if op["basis"] == "input":
        return sum((Decimal(x["quantity_base"]) for x in frozen_lines
                    if x["kind"] == "component" and x["base_unit_code"] == op["unit_code"]), Decimal("0")) * scale
    return output_base * scale


def plan_operation(op: dict, basis_qty: Decimal, hourly: Decimal) -> dict:
    """{op_no, work_centre_id, batches, minutes, wait_hours, cost}: setup once, run per batch; waiting costs nothing."""
    if not op["work_centre_id"]:
        return {"op_no": op["op_no"], "work_centre_id": "", "batches": 0, "minutes": "0", "wait_hours": str(op["wait_hours"]), "cost": Decimal("0")}
    n = batches(op, basis_qty)
    minutes = Decimal(op["setup_minutes"]) + n * Decimal(op["run_minutes"])
    return {"op_no": op["op_no"], "work_centre_id": op["work_centre_id"], "batches": n, "minutes": f"{minutes:f}",
            "wait_hours": str(op["wait_hours"]), "cost": minutes / Decimal(60) * hourly}


def rate_on(rows: list, work_centre_id: str, kind: str, day: date) -> Decimal:
    """The hourly rate of one kind in force on `day` (dated rows, never edited): 0 when none is."""
    held = [r for r in rows if r["work_centre_id"] == work_centre_id and r["rate_kind"] == kind and r["valid_from"] <= day]
    return Decimal(max(held, key=lambda r: r["valid_from"])["rate_amount"]) if held else Decimal("0")


def hourly(rows: list, work_centre_id: str, day: date) -> Decimal:
    return sum((rate_on(rows, work_centre_id, k, day) for k in RATE_KINDS), Decimal("0"))


def activity_cost(rows: list, work_centre_id: str, day: date, machine_minutes: Decimal, labour_minutes: Decimal) -> Decimal:
    """Machine time at the machine and overhead rates, labour time at the labour rate: 160 min x (45.0000 + 22.5000) / 60 = 180.00."""
    machine = rate_on(rows, work_centre_id, "machine", day) + rate_on(rows, work_centre_id, "overhead", day)
    labour = rate_on(rows, work_centre_id, "labour", day)
    return money(machine_minutes / Decimal(60) * machine + labour_minutes / Decimal(60) * labour, "the activity cost")


def lead_days(planned: list, ops: list) -> int:
    """Whole days a batch takes from start to done: its work-centre minutes plus the waiting hours, rounded up to a day."""
    hours = sum((Decimal(p["minutes"]) / Decimal(60) for p in planned), Decimal("0")) + sum((Decimal(o["wait_hours"]) for o in ops), Decimal("0"))
    return int((hours / Decimal(24)).to_integral_value(rounding=ROUND_CEILING))


__all__ = ["Operation", "check_operations", "valid_version", "plan_operation", "rate_on", "hourly", "activity_cost", "lead_days", "q4"]