dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/replenishment.py
"""Replenishment (INV-44) and the purchase request an accepted proposal becomes (INV-54).
A nightly run reads product's site data (reorder point, max, lot size, rounding profile), projects what will be available
(unrestricted on hand + still to arrive - reserved), and where that is at or below the reorder point writes ONE proposal per
(site, product, business date): to buy up to the max (or a fixed lot), rounded to how we order by product's round-proposal, or to
transfer from the supplying site. A person accepts (optionally changing the quantity), or dismisses with a reason; in auto mode the
system's client sends it. A proposal is a rule, not an AI step; a repeat run writes nothing more."""
import math
from datetime import timedelta
from decimal import Decimal
from fractions import Fraction
from typing import Optional
from erp_platform.bricks import Refused
from brick_inventory.application import docs, planning, products
from brick_inventory.application import unit as U
from brick_inventory.application.brick import INV
from brick_inventory.application.common import need_read, qstr
from brick_inventory.application.inputs import Code, IsoDate, Quantity, Reason, day
from brick_inventory.application.stock import in_transit_in
from brick_inventory.domain.errors import BadInput, GateBroken, RuleBroken
from brick_inventory.domain.values import NO_LOT, UNRESTRICTED, ZERO, fraction
from brick_inventory.infrastructure import seams
KEY = ("warehouse_id", "product_id", "business_date")
def request_ref(wh: str, pid: str, on) -> str:
return f"inv:repl:{wh}:{pid}:{on.isoformat()}"
def _per_unit(unit, pid: str, unit_code: str) -> Fraction:
"""Base units in one `unit_code`."""
c = products.convert(unit, pid, Decimal(1), unit_code)
return Fraction(c.numerator, c.denominator)
def _dec(f: Fraction) -> Decimal:
return Decimal(f.numerator) / Decimal(f.denominator)
def _expected(unit, pid: str, wh: str) -> Decimal:
"""What is still to arrive: purchasing's open orders, transfers on the road, and - until purchasing shows an order - requests
and draft transfers this brick already sent."""
total = in_transit_in(unit, pid, wh)
ordered = ZERO
if seams.purchasing_installed():
try:
for e in seams.expected_receipts(pid, wh):
if e.get("special_stock") in (None, "", "none"):
ordered += Decimal(str(e["quantity_base_outstanding"]))
except seams.Missing:
pass
total += ordered
if not ordered:
for p in unit.repo.find("inv_replenishment_proposal", warehouse_id=wh, product_id=pid, lifecycle="sent"):
if p["procurement"] == "buy" or (p["transfer_document_id"] and (unit.repo.get("inv_document", document_id=p["transfer_document_id"]) or {}).get("lifecycle") == "draft"):
total += _dec(fraction(p["rounded"] or ZERO) * _per_unit(unit, pid, p["unit_code"])) if p["unit_code"] else ZERO
return total
def _dependent(unit, row: dict, wh: str) -> Decimal:
"""INV-61: what planned and released production orders will consume of this product within the horizon (the buy lead time: planned
delivery plus goods-receipt processing days, at least a day), from manufacturing's GET /dependent-requirements. Zero when manufacturing
is not installed or answers nothing; inventory never reads mfg_ tables."""
if not seams.manufacturing_installed():
return ZERO
days = max(1, int(row.get("planned_delivery_days") or 0) + int(row.get("gr_processing_days") or 0))
try:
got = seams.dependent_requirements(row["product_id"], wh, unit.today, unit.today + timedelta(days=days))
except seams.Missing as m:
if m.status == 404:
return ZERO
raise Refused(502, f"manufacturing could not answer the dependent requirements of {row['product_id']} at {wh}: {m.text}") from None
return sum((Decimal(str(r["quantity"])) for r in got), ZERO)
def _proposal_for(unit, row: dict, wh: str) -> Optional[dict]:
pid = row["product_id"]
on_hand = planning.usable_now(unit, pid, wh, unit.today, None) # never quality, blocked or customer-consignment stock
reserved = unit.repo.reserved(wh, pid)
expected = _expected(unit, pid, wh)
projected = on_hand + expected - reserved - _dependent(unit, row, wh)
reorder, max_stock = Decimal(str(row.get("reorder_point") or 0)), Decimal(str(row.get("max_stock") or 0))
if projected > reorder:
return None
transfer = row.get("procurement_type") == "transfer"
if row.get("lot_size_method") == "fixed" and row.get("fixed_lot_size"):
want = Decimal(str(row["fixed_lot_size"]))
else:
want = max_stock - projected
if want <= 0:
return None
figures = {"warehouse_id": wh, "product_id": pid, "business_date": unit.today, "site_row_valid_from": day(row.get("valid_from")),
"on_hand_base": on_hand, "reserved_base": reserved, "expected_base": expected, "projected_base": projected,
"reorder_point": reorder, "max_stock": max_stock, "proposed_base": want, "lifecycle": "proposed",
"procurement": "transfer" if transfer else "buy", "supplying_site_id": row.get("supplying_site_id") or None,
"request_ref": request_ref(wh, pid, unit.today)}
if transfer:
return {**figures, "rounded": want, "unit_code": row.get("base_unit_code") or unit.base_unit(pid)}
pu = row.get("purchase_unit_code") or unit.base_unit(pid)
per = _per_unit(unit, pid, pu)
packs = math.ceil(fraction(want) / per) # whole packs
try:
r = seams.round_proposal(pid, wh, packs, pu, unit.today)
except seams.Missing as m:
raise Refused(502, f"product could not round the proposal for {pid}: {m.text}") from None
return {**figures, "rounded": Decimal(str(r["rounded"])), "unit_code": pu}
def _view(p: dict) -> dict:
return {"warehouse_id": p["warehouse_id"], "product_id": p["product_id"], "business_date": p["business_date"].isoformat(),
"site_row_valid_from": p["site_row_valid_from"].isoformat() if p["site_row_valid_from"] else None,
"procurement": p["procurement"], "supplying_site_id": p["supplying_site_id"], "lifecycle": p["lifecycle"],
"figures": {k: qstr(p[k]) for k in ("on_hand_base", "reserved_base", "expected_base", "projected_base", "reorder_point", "max_stock",
"proposed_base")},
"rounded": qstr(p["rounded"]), "unit_code": p["unit_code"], "request_ref": p["request_ref"],
"purchase_request_id": p.get("purchase_request_id"), "transfer_document_id": p.get("transfer_document_id"),
"decided_by": p.get("decided_by"), "reason": p.get("reason")}
def _send(unit, p: dict, quantity: Optional[Decimal], by: str) -> dict:
wh, pid = p["warehouse_id"], p["product_id"]
qty = quantity if quantity is not None else p["rounded"]
if qty <= 0:
raise BadInput("a quantity to order is above zero")
if p["procurement"] == "buy":
if quantity is not None and quantity != p["rounded"]: # a person's change is checked against the purchase rule only
try:
rule = seams.round_proposal(pid, wh, qty, p["unit_code"], unit.today).get("purchase_rule") or {}
except seams.Missing as m:
raise Refused(502, f"product could not check the purchase rule: {m.text}") from None
if not rule.get("met", True):
raise GateBroken("purchase_rule", rule.get("reason") or "the quantity does not meet the purchase rule", "order a quantity that does", 409)
goods_in = f"{wh}/IN" if unit.repo.get("inv_location", location_id=f"{wh}/IN") else \
next((x["location_id"] for x in unit.repo.locations_in(wh) if x["kind"] == "internal"), f"{wh}/IN")
try:
got = seams.purchase_request({"request_ref": p["request_ref"], "product_id": pid, "quantity": qstr(qty), "unit_code": p["unit_code"],
"warehouse_id": wh, "location_id": goods_in, "needed_by": unit.today.isoformat(),
"reason": f"replenishment: projected {qstr(p['projected_base'])} against reorder point {qstr(p['reorder_point'])}"})
except seams.Missing as m:
raise GateBroken(m.gate or "purchasable", m.text, "make the product purchasable (purchasing)", 409) from None
return {**p, "lifecycle": "sent", "purchase_request_id": got["request_id"], "rounded": qty, "decided_by": by}
src_wh = p["supplying_site_id"]
if not src_wh:
raise RuleBroken(f"no supplying site for {pid} at {wh}")
holder = max((r for r in __import__("brick_inventory.application.ledger", fromlist=["x"]).stock_rows(unit, pid, warehouse_id=src_wh)
if r.usable(unit.today)), key=lambda r: r.quantity_base, default=None)
if holder is None:
raise RuleBroken(f"{src_wh} holds no unrestricted {pid} to send")
dst = f"{wh}/IN" if unit.repo.get("inv_location", location_id=f"{wh}/IN") else next(x["location_id"] for x in unit.repo.locations_in(wh) if x["kind"] == "internal")
doc_id = f"int:{src_wh}:repl-{wh}-{pid}-{p['business_date'].isoformat()}"
docs.stage_document(unit, doc_id, "transfer", src_wh, unit.warehouse(src_wh)["company_code"], lifecycle="draft", source_kind="replenishment",
source_ref=p["request_ref"])
conv = products.convert(unit, pid, qty, p["unit_code"], document_ref=doc_id, what="the quantity")
docs.stage_line(unit, doc_id, 1, line_kind="transfer", product_id=pid, unit_code=p["unit_code"], unit_numerator=conv.numerator,
unit_denominator=conv.denominator, quantity_expected=conv.quantity, quantity_base=conv.base,
from_location_id=holder.location_id, to_location_id=dst, lot_code=NO_LOT, stock_type=UNRESTRICTED)
return {**p, "lifecycle": "sent", "transfer_document_id": doc_id, "rounded": qty, "decided_by": by}
def _get(unit, warehouse_id: str, product_id: str, business_date: str) -> dict:
p = unit.repo.get("inv_replenishment_proposal", warehouse_id=warehouse_id, product_id=product_id, business_date=day(business_date))
if p is None:
raise Refused(404, f"no proposal for {product_id} at {warehouse_id} on {business_date}")
return p
@INV.use_case(kind="command", route="POST /replenishment/run", permission="inv:replenishment:propose", idempotent=True)
def propose_replenishment(caller: dict, warehouse_id: Optional[Code] = None) -> dict:
"""The nightly run: for every product planned at the site whose projected available (unrestricted on hand + still to arrive -
reserved) is at or below its reorder point, one proposal per (site, product, business date) to buy up to the max (rounded to how
we order) or to transfer from the supplying site. Repeating the run writes nothing more. When the owner's replenishment_mode is
auto from a date on or before today, the proposals are sent without a person."""
@U.decide
def body(unit):
if not seams.product_installed():
return {"business_date": unit.today.isoformat(), "proposals": [], "note": "product is not installed: nothing is planned"}
whs = [warehouse_id] if warehouse_id else [w["warehouse_id"] for w in unit.repo.find("inv_warehouse")]
auto = unit.settings.value("replenishment_mode", unit.today) == "auto"
out = []
for wh in whs:
unit.warehouse(wh)
try:
rows = seams.product_sites(wh, unit.today)
except seams.Missing as m:
raise Refused(502, f"product could not answer for site {wh}: {m.text}") from None
for row in rows:
if (row.get("planning_method") or "none") == "none":
continue
if row.get("procurement_type") == "make" or row.get("planning_method") == "mrp":
continue # v6: manufacturing's MRP-lite plans a made product
if unit.repo.get("inv_replenishment_proposal", warehouse_id=wh, product_id=row["product_id"], business_date=unit.today):
continue
p = _proposal_for(unit, row, wh)
if p is None:
continue
if auto:
p = _send(unit, p, None, U.who(caller))
unit.put("inv_replenishment_proposal", p, KEY)
out.append(_view(p))
unit.flush_races()
return {"business_date": unit.today.isoformat(), "proposals": out}
return U.run(caller, body)
@INV.use_case(kind="query", route="GET /replenishment-proposals")
def list_replenishment_proposals(caller: dict, lifecycle: Optional[Code] = None, warehouse_id: Optional[Code] = None) -> dict:
"""Replenishment proposals, newest first: the figures each used, what it became (a purchase request or a draft transfer)."""
need_read(caller)
with U.read() as unit:
where = {k: v for k, v in (("lifecycle", lifecycle), ("warehouse_id", warehouse_id)) if v}
rows = unit.repo.find("inv_replenishment_proposal", order=("business_date", "warehouse_id", "product_id"), **where)
return {"proposals": [_view(p) for p in reversed(rows)]}
@INV.use_case(kind="command", route="POST /replenishment-proposals/{warehouse_id}/{product_id}/{business_date}/accept",
permission="inv:replenishment:write", idempotent=True)
def accept_replenishment(caller: dict, warehouse_id: Code, product_id: Code, business_date: IsoDate, quantity: Optional[Quantity] = None) -> dict:
"""Accept a proposal, optionally with another quantity (in the purchase unit; it must still meet the purchase rule). To buy:
one purchase request in purchasing (idempotent on the proposal's request_ref); to transfer: a draft transfer from the supplying
site. Accepting a sent proposal again returns the same answer."""
@U.decide
def body(unit):
p = _get(unit, warehouse_id, product_id, business_date)
if p["lifecycle"] == "sent":
return {**_view(p), "replay": True}
if p["lifecycle"] != "proposed":
raise RuleBroken(f"the proposal is {p['lifecycle']}")
sent = _send(unit, p, quantity, U.who(caller))
unit.flush_races()
unit.put("inv_replenishment_proposal", sent, KEY)
return {**_view(sent), "replay": False}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /replenishment-proposals/{warehouse_id}/{product_id}/{business_date}/dismiss",
permission="inv:replenishment:write", idempotent=True)
def dismiss_replenishment(caller: dict, warehouse_id: Code, product_id: Code, business_date: IsoDate, reason: Reason) -> dict:
"""Dismiss a proposal with a reason: nothing is bought or moved, and the night's run does not propose it again that day."""
@U.decide
def body(unit):
if not reason.strip():
raise BadInput("a dismissal gives its reason")
p = _get(unit, warehouse_id, product_id, business_date)
if p["lifecycle"] == "dismissed":
return {**_view(p), "replay": True}
if p["lifecycle"] != "proposed":
raise RuleBroken(f"the proposal is {p['lifecycle']}: it was already sent")
out = {**p, "lifecycle": "dismissed", "reason": reason, "decided_by": U.who(caller)}
unit.put("inv_replenishment_proposal", out, KEY)
return {**_view(out), "replay": False}
return U.run(caller, body)