impl/python.py

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

from functools import cmp_to_key
from typing import List, Sequence

from .inventory_abc_classification_types import AbcItem, ConsumptionItem
from .math_round_div import round_div  ← from math.round-div ^1.0.0 · built alongside by fune
from .money_amount import assert_same_currency, money  ← from money.amount ^1.0.0 · built alongside by fune

LETTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
MAX_SAFE = 2**53 - 1


def _by_value_then_sku(a: tuple, b: tuple) -> int:
    if a[1] != b[1]:
        return -1 if a[1] > b[1] else 1
    return -1 if a[0] < b[0] else 1 if a[0] > b[0] else 0


def abc_classification(items: Sequence[ConsumptionItem], cutoff_basis_points: Sequence[int]) -> List[AbcItem]:
    """Rank items by annual consumption value and class them by the band of the
    cumulative share each one starts in, so the item that crosses a cut-off
    stays in the higher class.
    """
    if len(cutoff_basis_points) == 0 or len(cutoff_basis_points) > 25:
        raise ValueError("cutoffBasisPoints needs 1 to 25 cut-offs, received %d" % len(cutoff_basis_points))
    previous = 0
    for cutoff in cutoff_basis_points:
        if isinstance(cutoff, bool) or not isinstance(cutoff, int) or cutoff <= previous or cutoff >= 10000:
            raise ValueError(
                "cutoffBasisPoints must ascend strictly within 1..9999: received %s after %s" % (cutoff, previous)
            )
        previous = cutoff
    seen = set()
    valued = []
    for item in items:
        if item.sku in seen:
            raise ValueError('duplicate sku "%s"' % item.sku)
        seen.add(item.sku)
        q = item.annual_quantity
        if isinstance(q, bool) or not isinstance(q, int) or q < 0:
            raise ValueError(
                'annualQuantity must be a whole number, not negative, received %r for "%s"' % (q, item.sku)
            )
        if item.unit_cost.minor < 0:
            raise ValueError('unitCost must not be negative, received %d for "%s"' % (item.unit_cost.minor, item.sku))
        assert_same_currency(items[0].unit_cost, item.unit_cost)
        valued.append((item.sku, q * item.unit_cost.minor, item.unit_cost.currency))
    total = sum(v[1] for v in valued)
    if total * 10000 > MAX_SAFE:
        raise ValueError("the total annual value is too large: it must stay within (2^53 - 1) / 10000 minor units")
    valued.sort(key=cmp_to_key(_by_value_then_sku))
    result: List[AbcItem] = []
    before = 0
    for index, (sku, value, currency) in enumerate(valued):
        band = len(cutoff_basis_points)
        for i, cutoff in enumerate(cutoff_basis_points):
            if before * 10000 < cutoff * total:
                band = i
                break
        before += value
        result.append(
            AbcItem(
                sku=sku,
                annual_value=money(value, currency),
                rank=index + 1,
                cumulative_basis_points=0 if total == 0 else round_div(before * 10000, total, "half-up"),
                abc_class=LETTERS[band],
            )
        )
    return result