dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/warehouse_ops.py
"""On the warehouse floor: putaway rules and bin capacity (INV-50), pick batches in walking order (INV-51), packing into
SSCC-numbered packages (INV-52)."""
import json
import uuid
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from brick_inventory.application import deliveries as D
from brick_inventory.application import docs, ledger, 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, PackIn, Quantity, Ref, day
from brick_inventory.domain.errors import BadInput, GateBroken, RuleBroken, ShortOf
from brick_inventory.domain.moves import MoveSpec
from brick_inventory.domain.values import NO_LOT, UNRESTRICTED, ZERO
STRATEGIES = ("fixed", "next_with_room", "addition")
# ── INV-50: putaway rules and bin capacity ───────────────────────────────────────────────────
@INV.use_case(kind="command", route="POST /putaway-rules", permission="inv:warehouse:write", idempotent=True)
def set_putaway_rule(caller: dict, warehouse_id: Code, seq: int, applies_to: Code, strategy: Code, target: Code,
valid_from: Optional[IsoDate] = None) -> dict:
"""Where goods-in puts stock away. Rules are tried in `seq` order; `applies_to` is product:<id>, category:<code> or *;
`strategy` fixed (the bin in `target`), next_with_room (the next bin of the zone `target` with room) or addition (the bin that
already holds the product, else the zone `target`). A rule is dated: the one in force today is used."""
@U.decide
def body(unit):
unit.warehouse(warehouse_id)
if strategy not in STRATEGIES:
raise BadInput(f"strategy is one of {', '.join(STRATEGIES)}")
if not (applies_to == "*" or applies_to.startswith(("product:", "category:"))):
raise BadInput("applies_to is product:<id>, category:<code> or *")
unit.location(target)
on = day(valid_from) or unit.today
unit.put("inv_putaway_rule", {"warehouse_id": warehouse_id, "seq": seq, "valid_from": on, "applies_to": applies_to, "target": target,
"strategy": strategy, "target_location_id": target if strategy == "fixed" else None},
("warehouse_id", "seq", "valid_from"))
return {"warehouse_id": warehouse_id, "seq": seq, "valid_from": on.isoformat(), "applies_to": applies_to, "strategy": strategy, "target": target}
return U.run(caller, body)
def _rules(unit, wh: str) -> list:
latest: dict = {}
for r in unit.repo.find("inv_putaway_rule", order=("seq", "valid_from"), warehouse_id=wh):
if r["valid_from"] <= unit.today:
latest[r["seq"]] = r
return [latest[k] for k in sorted(latest)]
def _applies(unit, rule: dict, pid: str) -> bool:
a = rule["applies_to"]
if a == "*":
return True
info = unit.product(pid)
return a == f"product:{pid}" or a == f"category:{info.get('category_code') or info.get('category')}"
def _per_base(unit, pid: str, field: str, loc_id: str, capacity: str) -> Decimal:
v = products.measures(unit.product(pid), unit.base_unit(pid)).get(field)
if v is None:
raise RuleBroken(f"{field.replace('gross_', '')} unknown for {pid}: {loc_id} has a {capacity} capacity that cannot be checked")
return v
def room(unit, loc: dict, pid: str) -> tuple:
"""(base quantity of `pid` that still fits in the bin, the capacity that limits it), from the bin's capacity (quantity, weight,
volume) and the units' copied measures; (None, '') when the bin is unlimited. Never counts an unknown measure as zero."""
best = (None, "")
held = unit.repo.stock(locations={loc["location_id"]})
for cap_field, name, measure in (("max_quantity_base", "quantity", None), ("max_weight", "weight", "gross_weight"), ("max_volume", "volume", "volume")):
cap = loc.get(cap_field)
if cap is None:
continue
used = ZERO
for (p, _l, _lot, _t, _o), q in held.items():
used += q * (Decimal(1) if measure is None else _per_base(unit, p, measure, loc["location_id"], name))
per = Decimal(1) if measure is None else _per_base(unit, pid, measure, loc["location_id"], name)
fits = max((Decimal(str(cap)) - used) / per, ZERO) if per else Decimal("1e12")
if best[0] is None or fits < best[0]:
best = (fits.quantize(Decimal("0.0001"), rounding="ROUND_DOWN"), name)
return best
def _zone_bins(unit, zone: str, exclude: str) -> list:
rows = [r for r in unit.repo.find("inv_location", kind="internal") if r["location_id"].startswith(zone + "/") and r["location_id"] != exclude
and r["stock_status"] == UNRESTRICTED and r["status"] == "active"]
return sorted(rows, key=lambda r: (r["pick_sequence"] if r["pick_sequence"] is not None else 10 ** 9, r["location_id"]))
def propose_putaway(unit, wh: str, pid: str, base: Decimal, source: str) -> list:
"""[(bin, base quantity)] by the rules in force: each rule fills what room it has, the next takes the rest."""
rem, out = base, []
for rule in _rules(unit, wh):
if rem <= 0:
break
if not _applies(unit, rule, pid):
continue
if rule["strategy"] == "fixed":
bins = [unit.location(rule["target_location_id"])]
elif rule["strategy"] == "addition":
holding = {k[1] for k, q in unit.repo.stock(product_id=pid).items() if q > 0 and k[3] == UNRESTRICTED and not k[4]}
bins = [unit.location(b) for b in sorted(holding) if b != source and unit.location(b)["kind"] == "internal"
and unit.location(b)["warehouse_id"] == wh] or _zone_bins(unit, rule["target"], source)
else:
bins = _zone_bins(unit, rule["target"], source)
for b in bins:
r, _name = room(unit, b, pid)
take = rem if r is None else min(rem, r)
if take > 0:
out.append((b["location_id"], take))
rem -= take
if rem <= 0:
break
if rem > 0:
raise GateBroken("bin_capacity", f"no bin has room for the last {qstr(rem)} {unit.base_unit(pid)} of {pid}",
"add a bin or raise a capacity, or choose a bin yourself", 409)
return out
@INV.use_case(kind="query", route="GET /putaway/proposal")
def get_putaway_proposal(caller: dict, warehouse_id: Code, product_id: Code, quantity: Quantity, unit_code: Code,
from_location_id: Optional[Code] = None) -> dict:
"""Which bins the putaway rules send this stock to, and how much to each (the bin's room by quantity, weight and volume)."""
need_read(caller)
with U.read() as unit:
src = from_location_id or f"{warehouse_id}/IN"
conv = products.convert(unit, product_id, quantity, unit_code, what="the quantity")
try:
splits = propose_putaway(unit, warehouse_id, product_id, conv.base, src)
except GateBroken as g:
raise Refused(g.status, g.as_dict()) from None
return {"from_location_id": src, "splits": [{"to_location_id": b, "quantity_base": qstr(q)} for b, q in splits]}
@INV.use_case(kind="command", route="POST /putaway", permission="inv:transfer:write", idempotent=True)
def put_away_stock(caller: dict, warehouse_id: Code, product_id: Code, quantity: Quantity, unit_code: Code, from_location_id: Optional[Code] = None,
to_location_id: Optional[Code] = None, lot_code: Optional[Code] = None, request_id: Optional[Code] = None) -> dict:
"""Put goods away from goods-in: one transfer per bin the rules propose (no value change), or into the bin the operator chose,
recorded with who. A bin over its capacity (quantity, weight or volume) is refused naming the capacity."""
@U.decide
def body(unit):
src = from_location_id or f"{warehouse_id}/IN"
rid = request_id or uuid.uuid4().hex[:12]
first = f"put:{warehouse_id}:{rid}:1"
if unit.repo.get("inv_document", document_id=first):
return {"replay": True, "putaways": [{"document_id": first}]}
conv = products.convert(unit, product_id, quantity, unit_code, document_ref=first, what="the quantity")
proposed = propose_putaway(unit, warehouse_id, product_id, conv.base, src) if not to_location_id else None
if to_location_id:
dst = unit.location(to_location_id)
r, name = room(unit, dst, product_id)
if r is not None and conv.base > r:
raise GateBroken("bin_capacity", f"{to_location_id} would exceed its {name} capacity: room for {qstr(r)} {unit.base_unit(product_id)}, "
f"{qstr(conv.base)} {unit.base_unit(product_id)} put away", "choose another bin", 409)
splits = [(to_location_id, conv.base)]
try:
suggested = propose_putaway(unit, warehouse_id, product_id, conv.base, src)
except GateBroken:
suggested = []
chosen_note = f"chosen by {U.who(caller)}" + (f" instead of {suggested[0][0]}" if suggested and suggested[0][0] != to_location_id else "")
else:
splits, chosen_note = proposed, ""
out = []
for k, (b, q) in enumerate(splits, start=1):
doc_id = f"put:{warehouse_id}:{rid}:{k}"
company = unit.warehouse(warehouse_id)["company_code"]
docs.stage_document(unit, doc_id, "putaway", warehouse_id, company, lifecycle="done", done_on=unit.today, done_by=U.who(caller),
source_kind="putaway", source_ref=src, reason=chosen_note or None,
number=docs.claim_number(unit, warehouse_id, "putaway", doc_id))
try:
pm = ledger.post_move(unit, MoveSpec(doc_id, 1, product_id, q, src, b, unit.today, conv.unit_code, None, 1, 1, lot_code or NO_LOT,
reason=chosen_note))
except ShortOf as s:
raise GateBroken("on_hand", str(s), "receive the goods first", 409) from None
out.append({"document_id": doc_id, "to_location_id": b, "quantity_base": qstr(q), "move_id": pm.row["move_id"]})
unit.flush_races()
return {"replay": False, "putaways": out, "note": chosen_note}
return U.run(caller, body, lost_message=lambda repo, row=None: "another mover took the stock at the same moment")
# ── INV-51: pick batches ─────────────────────────────────────────────────────────────────────
def _batch(unit, batch_id: str) -> dict:
b = unit.repo.get("inv_pick_batch", pick_batch_id=batch_id)
if b is None:
raise Refused(404, f"no pick batch {batch_id}")
return b
def _batch_list(unit, b: dict) -> list:
lines = []
for did in json.loads(b["delivery_document_ids"]):
d = D._delivery(unit, did)
ful = D.fulfilment_of(d["source_ref"]) if d["source_kind"] == "sales_order" else dict(D.DEFAULT_FULFILMENT)
for ln in D.pick_proposal(unit, d, ful):
for p in ln["proposals"]:
lines.append({"delivery_document_id": did, "line_no": ln["line_no"], "product_id": ln["product_id"], "location_id": p["location_id"],
"lot_code": p["lot_code"], "quantity_base": p["quantity_base"], "pick_sequence": p["pick_sequence"]})
return sorted(lines, key=lambda x: (x["pick_sequence"] if x["pick_sequence"] is not None else 10 ** 9, x["location_id"], x["delivery_document_id"]))
@INV.use_case(kind="command", route="POST /pick-batches", permission="inv:delivery:validate")
def create_pick_batch(caller: dict, delivery_document_ids: list[Ref]) -> dict:
"""Batch ready deliveries of one warehouse so they are picked together, in walking order. A delivery joins whole, once."""
@U.decide
def body(unit):
ds = [D._delivery(unit, i) for i in dict.fromkeys(delivery_document_ids)]
if not ds:
raise BadInput("name the deliveries to batch")
wh = ds[0]["warehouse_id"]
for d in ds:
if d["warehouse_id"] != wh:
raise RuleBroken("a pick batch is of one warehouse")
if d["lifecycle"] != "ready":
raise RuleBroken(f"{d['document_id']} is {d['lifecycle']}: only a ready delivery is batched")
if d["pick_batch_id"]:
raise RuleBroken(f"{d['document_id']} is already in pick batch {d['pick_batch_id']}")
batch_id = docs.claim_number(unit, wh, "pick_batch", f"pb:{uuid.uuid4().hex[:12]}")
for d in ds:
unit.put("inv_document", {**d, "pick_batch_id": batch_id}, "document_id")
unit.put("inv_pick_batch", {"pick_batch_id": batch_id, "warehouse_id": wh, "status": "open", "created_by": U.who(caller),
"delivery_document_ids": json.dumps([d["document_id"] for d in ds])}, "pick_batch_id")
return {"pick_batch_id": batch_id, "warehouse_id": wh, "status": "open", "delivery_document_ids": [d["document_id"] for d in ds]}
return U.run(caller, body, lost_message=lambda repo, row=None: "another writer took the same batch number")
@INV.use_case(kind="query", route="GET /pick-batches")
def get_pick_batch(caller: dict, pick_batch_id: Optional[Ref] = None) -> dict:
"""One pick list over all deliveries of a batch, sorted by the bins' pick sequence, each line naming the lot (by the removal
strategy) and the delivery it is for; with no id, the open batches."""
need_read(caller)
with U.read() as unit:
if not pick_batch_id:
return {"batches": [{"pick_batch_id": b["pick_batch_id"], "status": b["status"]} for b in unit.repo.find("inv_pick_batch", order=("pick_batch_id",))]}
b = _batch(unit, pick_batch_id)
return {"pick_batch_id": b["pick_batch_id"], "status": b["status"], "lines": _batch_list(unit, b) if b["status"] == "open" else []}
@INV.use_case(kind="command", route="POST /pick-batches/pick", permission="inv:delivery:validate", idempotent=True)
def pick_batch(caller: dict, pick_batch_id: Ref) -> dict:
"""Pick the batch: goods move bin -> <warehouse>/OUT-STAGE (a transfer, no value change) and are sorted to their deliveries as
picks. Each delivery's validation still claims stock per delivery."""
@U.decide
def body(unit):
b = _batch(unit, pick_batch_id)
if b["status"] == "picked":
return {"pick_batch_id": pick_batch_id, "status": "picked", "replay": True}
stage = f"{b['warehouse_id']}/OUT-STAGE"
if unit.repo.get("inv_location", location_id=stage) is None:
raise GateBroken("staging_location", f"{stage} does not exist", "add the staging bin for two-step picking", 409)
lines = _batch_list(unit, b)
if not lines:
raise RuleBroken("nothing to pick")
doc_id = f"pbpick:{b['warehouse_id']}:{uuid.uuid4().hex[:10]}"
wh = unit.warehouse(b["warehouse_id"])
docs.stage_document(unit, doc_id, "transfer", b["warehouse_id"], wh["company_code"], lifecycle="done", done_on=unit.today,
done_by=U.who(caller), source_kind="pick_batch", source_ref=pick_batch_id)
per_delivery: dict = {}
for n, ln in enumerate(lines, start=1):
q = Decimal(ln["quantity_base"])
ledger.post_move(unit, MoveSpec(doc_id, n, ln["product_id"], q, ln["location_id"], stage, unit.today, unit.base_unit(ln["product_id"]),
q, 1, 1, ln["lot_code"] or NO_LOT, reason=f"picked for {ln['delivery_document_id']}"))
per_delivery.setdefault(ln["delivery_document_id"], []).append((ln, q))
for did, items in per_delivery.items():
D._delivery(unit, did)
demand = {l["line_no"]: l for l in D._demand_lines(unit, did)}
for ln_no in {i[0]["line_no"] for i in items}:
unit.repo.delete_where("inv_document_line", document_id=did, line_kind="pick", parent_line_no=ln_no)
unit.memo.pop(("inv_document_line", did), None)
nxt = max([l["line_no"] for l in D._pick_lines(unit, did)] + [1000]) + 1
for ln, q in items:
dl = demand[ln["line_no"]]
docs.stage_line(unit, did, nxt, line_kind="pick", parent_line_no=ln["line_no"], product_id=ln["product_id"], unit_code=dl["unit_code"],
unit_numerator=dl["unit_numerator"], unit_denominator=dl["unit_denominator"],
quantity_done=D._in_unit(q, dl["unit_numerator"], dl["unit_denominator"]), quantity_base=q, from_location_id=stage,
lot_code=ln["lot_code"] or NO_LOT, stock_type=UNRESTRICTED,
**products.measures(unit.product(ln["product_id"]), dl["unit_code"]))
nxt += 1
unit.flush_races()
unit.put("inv_pick_batch", {**b, "status": "picked"}, "pick_batch_id")
return {"pick_batch_id": pick_batch_id, "status": "picked", "replay": False, "moved": len(lines), "to_location_id": stage}
return U.run(caller, body, lost_message=lambda repo, row=None: "another mover took the stock at the same moment")
@INV.use_case(kind="command", route="POST /pick-batches/remove", permission="inv:delivery:validate", idempotent=True)
def remove_from_pick_batch(caller: dict, pick_batch_id: Ref, delivery_document_id: Ref) -> dict:
"""Take a whole delivery out of a batch that is not picked yet (so it can be cancelled or picked alone)."""
@U.decide
def body(unit):
b = _batch(unit, pick_batch_id)
ids = json.loads(b["delivery_document_ids"])
if delivery_document_id not in ids:
return {"pick_batch_id": pick_batch_id, "delivery_document_id": delivery_document_id, "replay": True}
if b["status"] == "picked":
raise RuleBroken(f"{pick_batch_id} is picked: its goods are already in staging")
d = D._delivery(unit, delivery_document_id)
unit.put("inv_document", {**d, "pick_batch_id": None}, "document_id")
rest = [i for i in ids if i != delivery_document_id]
unit.put("inv_pick_batch", {**b, "delivery_document_ids": json.dumps(rest), "status": "open" if rest else "empty"}, "pick_batch_id")
return {"pick_batch_id": pick_batch_id, "delivery_document_id": delivery_document_id, "remaining": rest, "replay": False}
return U.run(caller, body)
# ── INV-52: packing and SSCC ─────────────────────────────────────────────────────────────────
def sscc_check_digit(digits17: str) -> int:
"""GS1 mod-10: from the right, weights 3, 1, 3, ..."""
total = sum(int(c) * (3 if i % 2 == 0 else 1) for i, c in enumerate(reversed(digits17)))
return (10 - total % 10) % 10
def make_sscc(prefix: str, serial: int) -> str:
"""18 digits: extension 0, the GS1 company prefix, the serial reference padded to fill 17 digits, then the check digit."""
width = 16 - len(prefix)
if width < 1 or serial >= 10 ** width:
raise RuleBroken("the SSCC series for this company prefix is used up")
body = "0" + prefix + str(serial).rjust(width, "0")
return body + str(sscc_check_digit(body))
@INV.use_case(kind="command", route="POST /package-types", permission="inv:warehouse:write", idempotent=True)
def set_package_type(caller: dict, package_type_code: Code, tare_weight: Quantity, weight_unit_code: Code, length: Optional[Quantity] = None,
width: Optional[Quantity] = None, height: Optional[Quantity] = None, dimension_unit_code: Optional[Code] = None,
max_weight: Optional[Quantity] = None) -> dict:
"""A kind of box or pallet: its own weight (tare) and size. Packages of this type weigh the tare plus their contents."""
@U.decide
def body(unit):
unit.put("inv_package_type", {"package_type_code": package_type_code, "tare_weight": tare_weight, "weight_unit_code": weight_unit_code,
"length": length, "width": width, "height": height, "dimension_unit_code": dimension_unit_code,
"max_weight": max_weight}, "package_type_code")
return {"package_type_code": package_type_code, "tare_weight": qstr(tare_weight), "weight_unit_code": weight_unit_code}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /deliveries/{document_id}/pack", permission="inv:delivery:validate")
def pack_delivery(caller: dict, document_id: Ref, packages: list[PackIn]) -> dict:
"""Pack what was picked into packages, each with its own SSCC (18 digits, check digit computed, from a series that never
reuses a number) and a gross weight = the package type's tare + the contents' copied weights. A package's contents never
exceed what was picked (less what is packed already)."""
@U.decide
def body(unit):
d = D._delivery(unit, document_id)
if d["lifecycle"] not in ("ready", "waiting"):
raise RuleBroken(f"{document_id} is {d['lifecycle']}: nothing more is packed on it")
prefix = unit.settings.value("gs1_company_prefix", unit.today)
if not prefix.isdigit() or not 6 <= len(prefix) <= 11:
raise GateBroken("gs1_company_prefix", "no GS1 company prefix is set", "the owner sets gs1_company_prefix in the settings", 409)
picks = D._pick_lines(unit, document_id)
left: dict = {}
for p in picks:
k = (p["product_id"], p["lot_code"] or "")
left[k] = left.get(k, ZERO) + p["quantity_base"]
for k, q in D._packed_base(unit, document_id).items():
left[k] = left.get(k, ZERO) - q
out = []
for pk in packages:
ptype = unit.repo.get("inv_package_type", package_type_code=pk.package_type_code)
if ptype is None:
raise RuleBroken(f"there is no package type {pk.package_type_code}")
weight, contents = ptype["tare_weight"], []
for c in pk.contents:
conv = products.convert(unit, c.product_id, c.quantity, c.unit_code, document_ref=document_id, what="a packed quantity")
k = (c.product_id, c.lot_code or "")
if left.get(k, ZERO) < conv.base:
raise RuleBroken(f"{qstr(conv.base)} {unit.base_unit(c.product_id)} of {c.product_id} packed, {qstr(max(left.get(k, ZERO), ZERO))} picked and not packed yet: "
"a package never holds more than was picked")
left[k] -= conv.base
m = products.measures(unit.product(c.product_id), c.unit_code)
if m["gross_weight"] is None:
raise RuleBroken(f"weight unknown for {c.product_id} ({c.unit_code}): give the unit its weight in the product catalogue")
if m["weight_unit_code"] != ptype["weight_unit_code"]:
raise RuleBroken(f"{c.product_id} is weighed in {m['weight_unit_code']}, the package type in {ptype['weight_unit_code']}")
weight += conv.quantity * m["gross_weight"]
contents.append({"product_id": c.product_id, "lot_code": c.lot_code or "", "quantity_base": qstr(conv.base)})
sid = f"SSCC/{prefix}"
n = unit.next_number(sid)
unit.race("inv_number_series", {"series_id": sid, "number": n, "document_id": document_id, "claimed_at": unit.now_utc.replace(tzinfo=None)})
sscc = make_sscc(prefix, n)
unit.put("inv_package", {"sscc": sscc, "document_id": document_id, "package_type_code": pk.package_type_code, "contents": json.dumps(contents),
"gross_weight": weight, "weight_unit_code": ptype["weight_unit_code"], "packed_by": U.who(caller),
"packed_at": unit.now_utc.replace(tzinfo=None)}, "sscc")
out.append({"sscc": sscc, "package_type_code": pk.package_type_code, "gross_weight": qstr(weight),
"weight_unit_code": ptype["weight_unit_code"], "contents": contents})
unit.flush_races()
return {"document_id": document_id, "packages": out}
return U.run(caller, body, lost_message=lambda repo, row=None: "another packer took the same SSCC number")