dodil/erp-brick-product

src/brick_product/domain/site.py

"""Site data and rounding profiles (PRD-37, PRD-38): the rules, and nothing else.

Product STORES how a product is procured, planned and allowed at a site, and ANSWERS a proposed quantity rounded to pallets or
layers; inventory computes proposals, sales-order and purchasing enforce the site's status. All arithmetic is exact (Fractions)."""
from decimal import Decimal
from fractions import Fraction
from typing import Optional

from brick_product.domain.errors import RuleBroken

SITE_STATUS = ("active", "no_purchase", "no_sale", "blocked")
PROCUREMENT = ("buy", "transfer", "make")
PLANNING = ("none", "reorder_point", "min_max", "manual", "mrp")
LOT_SIZE = ("exact", "fixed", "up_to_max")
DEFAULTS = {"site_status": "active", "procurement_type": "buy", "planning_method": "none", "lot_size_method": "exact"}
RESERVED = "procurement make and planning mrp need the production module, which this system does not have: install manufacturing, or use buy or transfer, and reorder_point, min_max or manual"


def check_row(site_id: str, row: dict, manufacturing: bool = False) -> None:
    """The site row's own rules: make and mrp only when manufacturing is installed (v6, gate manufacturing_installed), quantities consistent (safety stock <= reorder point <= max),
    a transfer names another site, a planning method has what it plans from."""
    for key, allowed in (("site_status", SITE_STATUS), ("procurement_type", PROCUREMENT), ("planning_method", PLANNING),
                         ("lot_size_method", LOT_SIZE)):
        if row.get(key) not in allowed:
            raise RuleBroken(f"{key} is one of {', '.join(allowed)}, not '{row.get(key)}'")
    if (row["procurement_type"] == "make" or row["planning_method"] == "mrp") and not manufacturing:
        raise RuleBroken(RESERVED)
    rp, ss, mx = row.get("reorder_point"), row.get("safety_stock"), row.get("max_stock")
    if ss is not None and rp is not None and ss > rp:
        raise RuleBroken(f"the safety stock ({ss}) is above the reorder point ({rp}): correct the quantities")
    if rp is not None and mx is not None and rp > mx:
        raise RuleBroken(f"the reorder point ({rp}) is above the maximum ({mx}): correct the quantities")
    if ss is not None and mx is not None and ss > mx:
        raise RuleBroken(f"the safety stock ({ss}) is above the maximum ({mx}): correct the quantities")
    if row["procurement_type"] == "transfer":
        if not row.get("supplying_site_id"):
            raise RuleBroken("a transfer names the site that supplies it (supplying_site_id)")
        if row["supplying_site_id"] == site_id:
            raise RuleBroken(f"{site_id} cannot be supplied by itself: a transfer names another site of the same company")
    if row["planning_method"] in ("reorder_point", "min_max") and rp is None:
        raise RuleBroken(f"planning {row['planning_method']} needs a reorder point")
    if row["planning_method"] == "min_max" and mx is None:
        raise RuleBroken("planning min_max needs a maximum")
    if row["lot_size_method"] == "fixed" and not row.get("fixed_lot_size"):
        raise RuleBroken("lot size fixed needs the fixed lot size")


def check_steps(steps: list) -> list:
    """A rounding profile's steps as exact [{multiple, round_up_pct}] strings, largest first: each rounds UP from its percentage."""
    if not steps:
        raise RuleBroken("a rounding profile has at least one step (a pallet, a layer)")
    out, last = [], None
    for st in steps:
        m, pct = Decimal(str(st["multiple"])), Decimal(str(st["round_up_pct"]))
        if m <= 0 or m >= Decimal("1e12") or m.as_tuple().exponent < -6:
            raise RuleBroken("a step is more than zero and below 1,000,000,000,000, with at most 6 decimal places")
        if not (Decimal(0) < pct <= Decimal(100)):
            raise RuleBroken("round_up_pct is above 0 and at most 100")
        if last is not None and m >= last:
            raise RuleBroken("steps go largest first: each step is smaller than the one before")
        last = m
        out.append({"multiple": format(m.normalize(), "f"), "round_up_pct": format(pct.normalize(), "f")})
    return out


def apply_steps(q: Fraction, steps: list) -> tuple:
    """(rounded, applied) for a proposal `q` in the profile's unit. From the largest step down: whole steps are taken, and what is left
    over rounds UP to one more of that step when it reaches the step's percentage (75 of an 80 pallet is 94%: 80), else it falls to the
    next smaller step; under the smallest step's percentage it is dropped (163 = 2 pallets of 80, the 3 left are under both: 160).
    `applied`: [(multiple, round_up_pct, quantity this step contributed)]."""
    left, total, applied = q, Fraction(0), []
    for st in steps:
        m, pct = Fraction(Decimal(st["multiple"])), Decimal(st["round_up_pct"])
        n = left // m
        rest = left - n * m
        up = rest > 0 and rest / m * 100 >= Fraction(pct)
        if up:
            n, rest = n + 1, Fraction(0)
        if n:
            applied.append((m, pct, n * m))
            total += n * m
        left = rest
        if left == 0:
            break
    return total, applied


def quantity_text(v: Optional[Fraction]) -> Optional[str]:
    from brick_product.domain import rounding
    return None if v is None else rounding.fmt(v)