dodil/erp-brick-sales-order

DODIL ERP brick: sales order

src/brick_sales_order/application/promising.py

"""Promised dates (SO-35): ask inventory what can be delivered when, and append the answer as the order's schedule lines.

Availability never blocks anything. With inventory installed and a delivering warehouse known (the `default_warehouse_id` setting), each STOCK
line is asked (GET /atp, nothing reserved) for its quantity and the line's schedule lines are inventory's confirmations moved by its pick-and-pack
days and the ship-to's transit days; without either, the requested date is kept and the schedule line says nothing was checked. Where the lines
land: `so_schedule_line` rows under one order-wide schedule version (above every version so far), written with the order in the caller's save.
"""
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional

from brick_sales_order.domain import promise as pm
from brick_sales_order.infrastructure.seams import Missing, ports


def line_is_stocked(repo, ln) -> bool:
    """A line inventory is asked about: its item category is delivered from stock (a service or a drop-ship line is not)."""
    cat = next((c for c in repo.config.categories() if c.item_category_code == ln.item_category_code), None)
    return (cat.fulfilment if cat else "stock") == "stock" and ln.usage != "drop_ship"


def warehouse(repo, today: date) -> str:
    return (repo.settings.value_on("default_warehouse_id", today) or "").strip()


def transit_days(repo, o, wh: str, today: date) -> int:
    """Days from the warehouse to the order's ship-to, from the customer_partner row of that ship-to in force today (0: none recorded)."""
    ship_to = o.ship_to_bp_id or o.bp_id
    rows = [r for r in repo.config.partners() if r.sold_to_bp_id == o.bp_id and r.function == "ship_to" and r.partner_bp_id == ship_to
            and r.valid_from <= today and r.status == "active"]
    return max(rows, key=lambda r: r.valid_from).transit_days(wh) if rows else 0


def _material_date(requested: Optional[date], today: date, lead: int) -> date:
    return max(today, requested - timedelta(days=lead)) if requested else today


def propose(repo, o, ln, today: date, *, quantity: Decimal, requested_on: Optional[date], exclude_own: bool, wh: str) -> tuple:
    """(chunks, why): for `quantity` of line `ln`. why is "" when inventory answered, else the reason nothing was checked."""
    if not ports.inventory_installed():
        return [pm.Chunk(requested_on, quantity, pm.to_base(quantity, ln.unit_numerator, ln.unit_denominator))], "availability not checked: inventory is not installed"
    if not wh:
        return [pm.Chunk(requested_on, quantity, pm.to_base(quantity, ln.unit_numerator, ln.unit_denominator))], "availability not checked: no delivering warehouse is set"
    if not line_is_stocked(repo, ln):
        return [pm.Chunk(requested_on, quantity, pm.to_base(quantity, ln.unit_numerator, ln.unit_denominator))], "availability not checked: not a stocked line"
    transit = transit_days(repo, o, wh, today)
    pick = int(repo.settings.value_on("pick_pack_days", today) or 1)
    try:
        ans = ports.inventory.atp(product_id=ln.product_id, warehouse_id=wh, quantity=pm_num(quantity), unit_code=ln.unit_code,
                                  requested_on=_material_date(requested_on, today, pick + transit).isoformat(),
                                  min_shelf_life_days=o.min_shelf_life_days or 0, exclude_sales_order_id=o.sales_order_id if exclude_own else "")
    except Missing as m:
        return [pm.Chunk(requested_on, quantity, pm.to_base(quantity, ln.unit_numerator, ln.unit_denominator))], f"availability not checked: {m.reason}"
    pick = int(ans.get("pick_pack_days") if ans.get("pick_pack_days") is not None else pick)
    return pm.promise_line(quantity, ln.unit_numerator, ln.unit_denominator, ans.get("confirmations") or [], pick_pack_days=pick,
                           transit_days=transit, requested_on=requested_on, complete=o.partial_delivery in ("complete_line", "complete_order")), ""


def pm_num(q: Decimal) -> str:
    """A quantity as text without a trailing exponent: 480, 2.5."""
    return format(Decimal(q).normalize(), "f")


def promise_order(repo, o, today: date, *, at, reason: str, exclude_own: bool, trigger_key: str = "") -> int:
    """Promise every line of the order afresh and append the result under the next schedule version (nothing is appended when it equals what is
    in force). Returns the version in force. Called at confirm and at amend (the reservation follows, with this version)."""
    wh = warehouse(repo, today)
    per_line, whys = {}, {}
    for ln in o.lines:
        per_line[ln.line_no], whys[ln.line_no] = propose(repo, o, ln, today, quantity=ln.quantity, requested_on=o.requested_delivery_on,
                                                         exclude_own=exclude_own, wh=wh)
    if o.partial_delivery == "complete_order":
        per_line = pm.lift_to_latest(per_line)
    current = {ln.line_no: [pm.Chunk(r.confirmed_on, r.requested_quantity, r.confirmed_base) for r in o.in_force(ln.line_no)] for ln in o.lines}
    if all(current[k] and pm.same(current[k], per_line[k]) for k in per_line):
        return o.schedule_version()
    version = o.schedule_version() + 1
    for ln in o.lines:
        o.append_schedule(version, ln.line_no, per_line[ln.line_no], requested_on=o.requested_delivery_on,
                          reason=whys[ln.line_no] or reason, at=at, trigger_key=trigger_key)
    return version


def line_summary(o, ln) -> str:
    rows = o.in_force(ln.line_no)
    if not rows:
        return ""
    if any("not checked" in (r.reason or "") for r in rows):
        return f"{rows[0].reason}; requested date kept" if rows[0].confirmed_on else rows[0].reason
    return pm.say([pm.Chunk(r.confirmed_on, r.requested_quantity) for r in rows])