dodil/erp-brick-sales-order
DODIL ERP brick: sales order
src/brick_sales_order/domain/promise.py
"""Promised dates (SO-35, SO-36): what inventory's availability says, moved by the days goods need to leave and to arrive.
Pure: no I/O. The order asks inventory (GET /atp) for each stocked line and these functions turn the answer into schedule lines. A shortfall
never blocks anything: it is a schedule line with nothing confirmed. A re-promise (SO-36) never moves a date the customer was told earlier on
its own: an earlier date is offered, and applied only to quantity not yet promised to the customer (or when reschedule_mode is `both`).
"""
from dataclasses import dataclass
from datetime import date, timedelta
from decimal import ROUND_DOWN, Decimal
from typing import Optional
ZERO = Decimal(0)
@dataclass(frozen=True)
class Chunk:
"""A quantity of the line (in the line's unit) promised for a date; on None: nothing confirms it yet."""
on: Optional[date]
quantity: Decimal
base: Decimal = ZERO
material_on: Optional[date] = None
def to_base(quantity: Decimal, numerator: int, denominator: int) -> Decimal:
return Decimal(quantity) * numerator / denominator
def places(quantity: Decimal) -> int:
return max(0, -Decimal(quantity).as_tuple().exponent)
def promise_line(quantity: Decimal, numerator: int, denominator: int, confirmations: list, *, pick_pack_days: int, transit_days: int,
requested_on: Optional[date], complete: bool = False) -> list:
"""Chunks for `quantity` of a line from inventory's confirmations [{available_on, quantity_base}]: each takes what it can, in date order,
and its delivery date is the day the goods are available plus pick-and-pack plus the ship-to's transit, never before the requested date.
complete: one chunk for the whole line, on the date the last of it is available. What no confirmation covers is a chunk with no date."""
need = to_base(quantity, numerator, denominator)
q = Decimal(quantity)
exp = Decimal(1).scaleb(-places(q))
left_base, left_q, out = need, q, []
for c in sorted(confirmations, key=lambda c: str(c["available_on"])):
if left_base <= 0:
break
take = min(left_base, Decimal(str(c["quantity_base"])))
if take <= 0:
continue
avail = date.fromisoformat(str(c["available_on"])) if not isinstance(c["available_on"], date) else c["available_on"]
on = avail + timedelta(days=pick_pack_days + transit_days)
if requested_on and on < requested_on:
on = requested_on
part = left_q if take >= left_base else (take * denominator / numerator).quantize(exp, rounding=ROUND_DOWN)
out.append(Chunk(on, part, take, avail))
left_base, left_q = left_base - take, left_q - part
if left_base > 0 or left_q > 0:
out.append(Chunk(None, left_q, left_base))
if complete:
dated = [c.on for c in out if c.on]
if len(dated) == len(out) and out:
return [Chunk(max(dated), q, need, max(c.material_on for c in out))]
return [Chunk(None, q, need)]
return out
def lift_to_latest(per_line: dict) -> dict:
"""complete_order: every line waits for the last one — each line's dates become the latest date any line is promised."""
dates = [c.on for chunks in per_line.values() for c in chunks if c.on]
if not dates or any(c.on is None for chunks in per_line.values() for c in chunks):
return per_line
latest = max(dates)
return {k: [Chunk(latest, c.quantity, c.base, c.material_on) for c in chunks] for k, chunks in per_line.items()}
def open_chunks(promised: list, delivered: Decimal) -> list:
"""What is promised for the quantity still to deliver: what has been delivered comes off the earliest promises first."""
left, out = Decimal(delivered), []
for c in sorted(promised, key=lambda c: (c.on is None, c.on or date.max)):
if left >= c.quantity:
left -= c.quantity
continue
out.append(Chunk(c.on, c.quantity - left, c.base, c.material_on))
left = ZERO
return out
def merge(old: list, new: list, *, keep_told: bool) -> tuple:
"""(chunks to promise, offers): `new` is what inventory now says for the open quantity, `old` what is promised for it. With keep_told,
where `new` is EARLIER than `old` the old date stays and the earlier one is an offer; everywhere else `new` is applied."""
if not keep_told:
return new, []
old = sorted(old, key=lambda c: (c.on is None, c.on or date.max))
cuts = sorted({*_cum(old), *_cum(new)})
out, offers, lo = [], [], ZERO
for hi in cuts:
o, n = _at(old, lo), _at(new, lo)
qty = hi - lo
if n.on is not None and o.on is not None and n.on < o.on:
on = o.on
offers.append(Chunk(n.on, qty))
else:
on = n.on
out.append(Chunk(on, qty, n.base * qty / n.quantity if n.quantity else ZERO, n.material_on))
lo = hi
return _coalesce(out), _coalesce(offers)
def _cum(chunks: list) -> list:
total, out = ZERO, []
for c in chunks:
total += c.quantity
out.append(total)
return out
def _at(chunks: list, pos: Decimal) -> Chunk:
total = ZERO
for c in chunks:
total += c.quantity
if pos < total:
return c
return chunks[-1] if chunks else Chunk(None, ZERO)
def _coalesce(chunks: list) -> list:
out = []
for c in chunks:
if out and out[-1].on == c.on and out[-1].material_on == c.material_on:
p = out[-1]
out[-1] = Chunk(p.on, p.quantity + c.quantity, p.base + c.base, p.material_on)
else:
out.append(c)
return out
def moved_later(old: list, new: list) -> list:
"""The quantities whose promised date moved LATER (or lost its date): what the owner must tell the customer."""
old = sorted(old, key=lambda c: (c.on is None, c.on or date.max))
cuts = sorted({*_cum(old), *_cum(new)})
out, lo = [], ZERO
for hi in cuts:
o, n = _at(old, lo), _at(new, lo)
if o.on is not None and (n.on is None or n.on > o.on):
out.append(Chunk(n.on, hi - lo))
lo = hi
return _coalesce(out)
def same(a: list, b: list) -> bool:
return [(c.on, c.quantity) for c in a] == [(c.on, c.quantity) for c in b]
def say(chunks: list, unit: str = "") -> str:
"""'partly confirmed: 240 on 4 Nov, 240 on 12 Nov' for a line (what the order shows and the confirmation prints)."""
def n(q):
q = Decimal(q).normalize()
return format(q, "f")
bits = [f"{n(c.quantity)} on {c.on.day} {c.on:%b}" if c.on else f"{n(c.quantity)} not yet confirmed" for c in chunks]
if not chunks:
return ""
if len(chunks) == 1 and chunks[0].on:
return "confirmed: " + bits[0]
return ("partly confirmed: " if any(c.on for c in chunks) else "not yet confirmed: ") + ", ".join(bits)