dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/planning.py
"""What can be promised by date (ATP), what we hold at customers, and what to buy or move when stock runs low (INV-44, INV-45,
INV-49, INV-54). All derived on read from the moves, the reservations, purchasing's expected receipts and transfers on their way;
nothing is reserved by asking."""
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional
from erp_platform import auth
from erp_platform.bricks import Refused
from brick_inventory.application import ledger, products
from brick_inventory.application import unit as U
from brick_inventory.application.brick import INV
from brick_inventory.application.common import AVAILABILITY, qstr
from brick_inventory.application.inputs import Code, IsoDate, Quantity, day
from brick_inventory.domain.values import ZERO
from brick_inventory.infrastructure import seams
def _can_ask(caller: dict) -> None:
if not (auth.has_permission(caller, AVAILABILITY) or auth.has_permission(caller, "inv:stock:read")):
raise Refused(404, "not found")
def goods_in_days(unit, product_id: str, warehouse_id: str) -> int:
if not seams.product_installed():
return 0
try:
for r in seams.product_sites(warehouse_id, unit.today):
if r.get("product_id") == product_id:
return int(r.get("gr_processing_days") or 0)
except seams.Missing:
pass
return 0
def usable_now(unit, product_id: str, warehouse_id: str, on: date, min_shelf_life_days: Optional[int]) -> Decimal:
"""Unrestricted stock of ours or a consigning supplier's at the warehouse, not expired by `on` and with at least the minimum
shelf life left at `on`: never quality, blocked, expired, customer-consignment or subcontractor stock."""
total = ZERO
for r in ledger.stock_rows(unit, product_id, warehouse_id=warehouse_id):
if not r.usable(unit.today):
continue
if r.expires_on is not None:
if r.expires_on <= on:
continue
if min_shelf_life_days and (r.expires_on - on).days < min_shelf_life_days:
continue
total += r.quantity_base
return total
@INV.use_case(kind="query", route="GET /atp/{product_id}")
def get_atp(caller: dict, product_id: Code, warehouse_id: Code, quantity: Quantity, unit_code: Code, requested_on: Optional[IsoDate] = None,
min_shelf_life_days: Optional[int] = None, exclude_sales_order_id: Optional[Code] = None) -> dict:
"""What can be promised, by date: unrestricted stock now less what is reserved, plus purchasing's expected receipts after the
site's goods-in days and transfers by their expected arrival, less reservations dated later. Stock expiring within the
customer's minimum shelf life of its date is not counted; quality, blocked and customer-consignment stock never is. The
quantity asked is confirmed in steps (now, then by date), with the shortfall. Derived on read: nothing is reserved by asking."""
_can_ask(caller)
with U.read() as unit:
info = unit.product(product_id)
conv = products.convert(unit, product_id, quantity, unit_code, what="the quantity asked")
unit.warehouse(warehouse_id)
want_on = day(requested_on) or unit.today
shelf = min_shelf_life_days
reservations = [r for r in unit.repo.find("inv_reservation", warehouse_id=warehouse_id, product_id=product_id) if r["kind"] != "marker"
and r["sales_order_id"] != exclude_sales_order_id]
reserved_now = sum((r["quantity_base"] for r in reservations if not r["confirmed_on"] or r["confirmed_on"] <= unit.today), ZERO)
stock_now = usable_now(unit, product_id, warehouse_id, max(want_on, unit.today), shelf)
available_now = max(stock_now - reserved_now, ZERO)
events, sources = [], [{"kind": "on_hand", "ref": warehouse_id, "on": unit.today.isoformat(), "quantity_base": qstr(available_now)}]
gr = goods_in_days(unit, product_id, warehouse_id)
try:
expected = seams.expected_receipts(product_id, warehouse_id) if seams.purchasing_installed() else []
except seams.Missing:
expected = []
for e in expected:
if e.get("special_stock") not in (None, "", "none", "supplier_consignment"):
continue
on = day(e["expected_on"]) + timedelta(days=gr)
q = Decimal(str(e["quantity_base_outstanding"]))
events.append((on, q))
sources.append({"kind": "purchase", "ref": f"{e['purchase_order_id']}:{e['line_no']}", "on": on.isoformat(), "quantity_base": qstr(q)})
for d in unit.repo.find("inv_document", kind="transfer", lifecycle="in_transit"):
for ln in unit.repo.find("inv_document_line", document_id=d["document_id"]):
if ln["product_id"] == product_id and ln["line_kind"] == "transit" and (ln["to_location_id"] or "").split("/")[0] == warehouse_id:
left = (ln["quantity_base"] or ZERO) - (ln["quantity_done"] or ZERO)
on = d["expected_arrival_on"] or unit.today
events.append((on, left))
sources.append({"kind": "transit", "ref": d["document_id"], "on": on.isoformat(), "quantity_base": qstr(left)})
for r in reservations:
if r["confirmed_on"] and r["confirmed_on"] > unit.today:
events.append((r["confirmed_on"], -r["quantity_base"]))
confirmations, left, running = [], conv.base, available_now
if available_now > 0 and left > 0:
take = min(left, available_now)
confirmations.append({"available_on": unit.today.isoformat(), "quantity_base": qstr(take)})
left -= take
running -= take
by_day: dict = {}
for on, q in events:
by_day[on] = by_day.get(on, ZERO) + q
free = ZERO
for on in sorted(by_day):
free += by_day[on]
if left > 0 and free > 0:
take = min(left, free)
confirmations.append({"available_on": on.isoformat(), "quantity_base": qstr(take)})
left -= take
free -= take
days = int(unit.settings.value(f"pick_pack_days:{warehouse_id}", unit.today) or unit.settings.value("pick_pack_days", unit.today))
return {"product_id": product_id, "base_unit_code": info.get("base_unit_code"), "warehouse_id": warehouse_id,
"requested_base": qstr(conv.base), "requested_on": want_on.isoformat(), "available_now": qstr(available_now),
"confirmations": confirmations, "shortfall_base": qstr(max(left, ZERO)), "sources": sources, "pick_pack_days": days}
@INV.use_case(kind="query", route="GET /consignment-stock/customers/{bp_id}")
def get_consignment_stock(caller: dict, bp_id: str, product_id: Optional[Code] = None) -> dict:
"""What we hold at a customer's site: ours, valued, never available to anyone else. sales-order gates its consignment
issue, pick-up and return on it."""
_can_ask(caller)
with U.read() as unit:
rows = []
for loc in unit.repo.find("inv_location", kind="consignment_customer"):
if str(loc["bp_id"]) != str(bp_id):
continue
for (p, l, lot, t, o), q in sorted(unit.repo.stock(locations={loc["location_id"]}, product_id=product_id).items()):
if q > 0 and not o:
rows.append({"product_id": p, "base_unit_code": unit.base_unit(p), "on_hand_base": qstr(q), "location_id": l})
merged: dict = {}
for r in rows:
m = merged.setdefault(r["product_id"], {**r, "on_hand_base": ZERO})
m["on_hand_base"] += Decimal(r["on_hand_base"])
return {"bp_id": str(bp_id), "products": [{**m, "on_hand_base": qstr(m["on_hand_base"])} for m in merged.values()]}