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 typing import Dict, List, Mapping, Sequence

from .dates_add_days import epoch_day_from_iso, iso_from_epoch_day  ← from dates.add-days ^1.0.0 · built alongside by fune
from .dates_day_of_week import day_of_week  ← from dates.day-of-week ^1.0.0 · built alongside by fune
from .inventory_cycle_count_schedule_types import CountDay, CountItem


def cycle_count_schedule(
    items: Sequence[CountItem],
    counts_per_period: Mapping[str, int],
    start_date: str,
    end_date: str,
    holidays: Sequence[str],
) -> List[CountDay]:
    """Spread each item's counts evenly over the working days of a period: a
    class counted f times gets f even windows, and its items are spread evenly
    within each window.
    """
    first = epoch_day_from_iso(start_date)
    last = epoch_day_from_iso(end_date)
    if last < first:
        raise ValueError("endDate %s is before startDate %s" % (end_date, start_date))
    closed = {epoch_day_from_iso(h) for h in holidays}
    days: List[str] = []
    for day in range(first, last + 1):
        iso = iso_from_epoch_day(day)
        if day_of_week(iso) <= 5 and day not in closed:
            days.append(iso)

    classes = sorted(counts_per_period.keys())
    for cls in classes:
        f = counts_per_period[cls]
        if isinstance(f, bool) or not isinstance(f, int) or f < 0:
            raise ValueError(
                'countsPerPeriod for class "%s" must be a whole number, not negative, received %r' % (cls, f)
            )
    by_class: Dict[str, List[str]] = {}
    seen = set()
    for item in items:
        if item.sku in seen:
            raise ValueError('duplicate sku "%s"' % item.sku)
        seen.add(item.sku)
        if item.abc_class not in counts_per_period:
            raise ValueError('no count frequency for class "%s" (sku "%s")' % (item.abc_class, item.sku))
        by_class.setdefault(item.abc_class, []).append(item.sku)

    n = len(days)
    schedule: List[List[str]] = [[] for _ in days]
    for cls in classes:
        skus = sorted(by_class.get(cls, []))
        f = counts_per_period[cls]
        if not skus or f == 0:
            continue
        if f > n:
            raise ValueError('class "%s" is counted %d times but the period has only %d working days' % (cls, f, n))
        for k in range(f):
            start = k * n // f
            length = (k + 1) * n // f - start
            for j, sku in enumerate(skus):
                schedule[start + j * length // len(skus)].append(sku)
    return [CountDay(date=date, skus=schedule[i]) for i, date in enumerate(days)]