from typing import Dict, List, Mapping, Sequence from .dates_add_days import epoch_day_from_iso, iso_from_epoch_day from .dates_day_of_week import day_of_week 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)]