dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/source_ops.py
"""Stock another module consumes, produces or restocks (INV-58, INV-59): a till sale, a recipe or a production issue takes stock out
(POST /goods-issues/from-source); a production order's output, a by-product and a restocked till refund put it in (POST /receipts
with a source_kind of production_receipt, by_product_receipt or pos_refund_restock).
Both are idempotent on the line's move_ref: a known move_ref answers its move and records nothing. The cost is inventory's: the moves,
their lots and their values are decided and posted here, and the answer lists each move's value so the caller needs no read of inv_ tables."""
import json
from datetime import date
from decimal import Decimal
from typing import Literal, Optional
from erp_platform.bricks import Refused
from brick_inventory.application import docs, ledger, postings, products
from brick_inventory.application import unit as U
from brick_inventory.application.brick import INV
from brick_inventory.application.common import qstr
from brick_inventory.application.inputs import Code, InputLotIn, IsoDate, Rate, Ref, SourceLineIn, day
from brick_inventory.domain import location as L
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, OURS, UNRESTRICTED, ZERO, money
SOURCE_RECEIPTS = ("production_receipt", "by_product_receipt", "pos_refund_restock")
def virtual(unit, company: str, name: str, kind: str) -> str:
"""The company's virtual location, made on first use (PRODUCTION did not exist before v6)."""
lid = L.virtual_id(company, name)
if unit.repo.get("inv_location", location_id=lid) is None:
row = {"location_id": lid, "warehouse_id": "", "company_code": company, "parent_location_id": "", "kind": kind, "name": name.title(),
"stock_status": "unrestricted", "allowed_stock_types": "unrestricted,quality,blocked", "status": "active"}
unit.put("inv_location", row, "location_id")
unit._locations[lid] = row
return lid
def _internal(unit, location_id: str, company: str) -> dict:
from brick_inventory.application.receipts import _internal as internal
loc = internal(unit, location_id)
if loc["company_code"] != company:
raise GateBroken("company", f"{location_id} belongs to company {loc['company_code']}, not {company}", "name a location of the company", 422)
return loc
def _moves_of_ref(unit, move_ref: str) -> list:
"""The moves one move_ref made: the ref itself, then ref#2, ref#3 ... (one per lot)."""
out, n = [], 1
while True:
rows = unit.repo.find("inv_move", move_ref=move_ref if n == 1 else f"{move_ref}#{n}")
if not rows:
return out
out += rows
n += 1
def _answer_move(row: dict) -> dict:
return {"move_ref": row["move_ref"], "move_id": row["move_id"], "quantity_base": qstr(row["quantity_base"]),
"value": f"{abs(row['value_amount'] or ZERO):.2f}", "lot_code": row["lot_code"] or None}
def _fefo(unit, product_id: str, loc: dict, base: Decimal, today: date) -> list:
"""[(lot_code, quantity)] taken first-expiry-first-out from the bin; a shortfall goes on the last lot (a sale that happened)."""
rows = [r for r in ledger.stock_rows(unit, product_id, warehouse_id=loc["warehouse_id"])
if r.location_id == loc["location_id"] and r.stock_type == UNRESTRICTED and r.owner == OURS and r.lot_code and r.quantity_base > 0]
rows.sort(key=lambda r: (r.expires_on or date.max, r.lot_code))
left, out = base, []
for r in rows:
take = min(left, r.quantity_base)
out.append([r.lot_code, take])
left -= take
if left <= 0:
break
if not out:
raise GateBroken("lot_required", f"{product_id} is tracked and no lot is on hand at {loc['location_id']}", "name the lot", 422)
if left > 0:
out[-1][1] += left
return [(lot, q) for lot, q in out]
@INV.use_case(kind="command", route="POST /goods-issues/from-source", permission="inv:issue:intake", idempotent=True, tool=False)
def issue_from_source(caller: dict, source_kind: Literal["pos_sale", "production_issue"], source_id: Ref, company_code: Code, location_id: Code,
issued_on: IsoDate, lines: list[SourceLineIn], account_purpose: Optional[Literal["wip"]] = None) -> dict:
"""INV-58: take out what another module sold or used. One move per line keyed move_ref (a known move_ref answers its move and changes
nothing), from the consumption location of the shop or plant to the company's CONSUMPTION, at the average (or standard) cost, the lot
named else first-expiry-first-out, each move's value rounded once and the remainder on the last move of a product, so the issue totals
round(quantity x cost). A pos_sale is NEVER refused for stock (the sale happened: a short location goes negative); a production_issue
short of stock is refused (gate on_hand) unless the location allows negative stock. The journal is cost_of_sales (pos_sale) or wip
(production_issue) against stock_valuation, one per call."""
if source_kind == "production_issue" and account_purpose not in (None, "wip"):
raise Refused(422, "a production issue posts to wip")
if len({x.move_ref for x in lines}) != len(lines):
raise Refused(422, "a move_ref appears twice in the call")
@U.decide
def body(unit):
on = day(issued_on)
if on > unit.today:
raise BadInput(f"issued_on {on.isoformat()} is in the future: an issue is dated today or earlier")
loc = _internal(unit, location_id, company_code)
doc_id = f"gi:src:{source_kind}:{source_id}"
sink = L.virtual_id(company_code, "CONSUMPTION")
have = unit.repo.moves_of(document_id=doc_id)
next_line = max((m["line_no"] for m in have), default=0) + 1
answers, new_rows, by_class, cum = [], [], {}, {}
first_new = None
for ln in lines:
known = _moves_of_ref(unit, ln.move_ref)
if known:
answers += [_answer_move(m) for m in known]
continue
info = products.stocked(unit, ln.product_id)
conv = products.convert(unit, ln.product_id, ln.quantity, ln.unit_code, document_ref=doc_id, what="the issued quantity")
tracking = unit.tracking(ln.product_id)["tracking"]
if tracking == "none":
splits = [(NO_LOT, conv.base)]
elif ln.lot_code:
splits = [(ln.lot_code, conv.base)]
else:
splits = _fefo(unit, ln.product_id, loc, conv.base, unit.today)
first_new = next_line if first_new is None else first_new
ledger.catch_up(unit, company_code, ln.product_id, on)
method = unit.settings.method(company_code, on)
avg = ledger.position(unit, company_code, ln.product_id).average
for i, (lot, share) in enumerate(splits, start=1):
value_out = None
if method == "average": # each rounded once, the remainder on the last (cumulative rounding)
before = cum.get(ln.product_id, ZERO)
value_out = money((before + share) * avg) - money(before * avg)
cum[ln.product_id] = cum.get(ln.product_id, ZERO) + share
spec = MoveSpec(doc_id, next_line, ln.product_id, share, location_id, sink, on, ln.unit_code,
conv.quantity * share / conv.base if len(splits) > 1 else conv.quantity, conv.numerator, conv.denominator, lot or NO_LOT,
UNRESTRICTED, UNRESTRICTED, OURS, OURS, n=i, move_ref=ln.move_ref if i == 1 else f"{ln.move_ref}#{i}",
allow_negative=source_kind == "pos_sale", value_out=value_out)
try:
pm = ledger.post_move(unit, spec)
except ShortOf as s:
raise GateBroken("on_hand", str(s),
"move stock there first, or the owner allows negative stock at the location", 409) from None
new_rows.append(pm.row)
vc = info.get("valuation_class") or "*"
by_class[vc] = by_class.get(vc, ZERO) + pm.row["value_amount"]
answers.append(_answer_move(pm.row))
docs.stage_line(unit, doc_id, next_line, line_kind="issue", product_id=ln.product_id, unit_code=ln.unit_code, unit_numerator=conv.numerator,
unit_denominator=conv.denominator, quantity_expected=conv.quantity, quantity_done=conv.quantity, quantity_base=conv.base,
from_location_id=location_id, lot_code=splits[0][0] or NO_LOT, stock_type=UNRESTRICTED,
value_amount=sum((r["value_amount"] for r in new_rows if r["line_no"] == next_line), ZERO))
next_line += 1
if first_new is not None:
existing = unit.repo.get("inv_document", document_id=doc_id)
number = existing["number"] if existing else docs.claim_number(unit, loc["warehouse_id"], "issue", doc_id)
docs.stage_document(unit, doc_id, "issue", loc["warehouse_id"], company_code, lifecycle="posted", number=number, done_on=on,
done_by=U.who(caller), purpose=source_kind, source_kind=source_kind, source_ref=source_id)
purpose = "cost_of_sales" if source_kind == "pos_sale" else "wip"
journal = []
for vc, v in by_class.items():
journal += postings.movement_lines(v, purpose, vc, memo=f"{source_kind} {source_id}")
from brick_inventory.application.deliveries import _merge
postings.stage(unit, doc_id, f"issue:{first_new}", company_code, on, f"{source_kind} {source_id}", _merge(journal))
return {"issue_id": doc_id, "moves": answers, "replay": first_new is None}
return U.run(caller, body, lost_message=lambda repo, row=None: "another mover took the stock at the same moment")
def receive_from_source(caller: dict, *, move_ref: str, source_kind: str, source_id: str, company_code: Optional[str], product_id: str, quantity,
unit_code: str, location_id: str, received_on: str, lot_code: Optional[str], unit_cost, reverses_move_ref: Optional[str],
input_lots: Optional[list]) -> dict:
"""INV-59: receive what another module produced or took back. production_receipt and by_product_receipt come from PRODUCTION at the
unit_cost sent (standard, the order's cost, or the by-product's net realisable value), rounded once, crediting wip, with a lot link per
input lot so the trace (INV-40) walks through production. pos_refund_restock comes back from CONSUMPTION at the cost the move it reverses
left at (no unit_cost is sent), crediting cost_of_sales, never more than that move issued. Idempotent on move_ref."""
@U.decide
def body(unit):
known = _moves_of_ref(unit, move_ref)
if known:
return {"move_id": known[0]["move_id"], "quantity_base": qstr(sum((m["quantity_base"] for m in known), ZERO)),
"value": f"{sum((m['value_amount'] or ZERO for m in known), ZERO):.2f}", "replay": True}
on = day(received_on)
if on > unit.today:
raise BadInput(f"received_on {on.isoformat()} is in the future: a receipt is dated today or earlier")
loc = unit.repo.get("inv_location", location_id=location_id)
company = company_code or (loc or {}).get("company_code") or ""
loc = _internal(unit, location_id, company)
info = products.stocked(unit, product_id)
doc_id = f"rcpt:src:{source_kind}:{source_id}"
line_no = max((m["line_no"] for m in unit.repo.moves_of(document_id=doc_id)), default=0) + 1
conv = products.convert(unit, product_id, quantity, unit_code, document_ref=doc_id, what="the received quantity")
tracking = unit.tracking(product_id)
lot = NO_LOT
vclass = info.get("valuation_class") or "*"
if source_kind == "pos_refund_restock":
if not reverses_move_ref:
raise GateBroken("reverses_known", "a restocked refund names the move it reverses (reverses_move_ref)", "send the till sale's move_ref", 422)
origin = [m for m in _moves_of_ref(unit, reverses_move_ref) if m["product_id"] == product_id]
if not origin:
raise GateBroken("reverses_known", f"no issued move {reverses_move_ref} of {product_id}", "send the move_ref the sale was issued under", 422)
issued = sum((m["quantity_base"] for m in origin), ZERO)
back = sum((m["quantity_base"] for o in origin for m in unit.repo.find("inv_move", reverses_move_id=o["move_id"])), ZERO)
if conv.base + back > issued:
raise GateBroken("within_issued", f"{reverses_move_ref} issued {qstr(issued)}, of which {qstr(back)} already came back: {qstr(conv.base)} more is too much",
"restock no more than was issued", 422)
value = money(abs(sum((m["value_amount"] for m in origin), ZERO)) * conv.base / issued)
lot = lot_code or (origin[0]["lot_code"] if len({m["lot_code"] for m in origin}) == 1 else NO_LOT)
src_loc = virtual(unit, company, "CONSUMPTION", "consumption")
spec = MoveSpec(doc_id, line_no, product_id, conv.base, src_loc, location_id, on, unit_code, conv.quantity, conv.numerator, conv.denominator,
lot, UNRESTRICTED, UNRESTRICTED, OURS, OURS, move_ref=move_ref, reverses_move_id=origin[0]["move_id"], value_in=value,
reason=f"restock of {reverses_move_ref}")
credit = "cost_of_sales"
else:
if unit_cost is None:
raise GateBroken("unit_cost", "an output needs its cost per base unit (unit_cost)", "send unit_cost", 422)
if tracking["tracking"] != "none":
if not lot_code:
raise GateBroken("lot_required", f"{product_id} is tracked by {tracking['tracking']}: an output names its lot", "send lot_code", 422)
from brick_inventory.application.receipts import _ensure_lot
_ensure_lot(unit, product_id, lot_code, tracking, on, None, None)
lot = lot_code
value = money(conv.base * Decimal(unit_cost))
src_loc = virtual(unit, company, "PRODUCTION", "production")
spec = MoveSpec(doc_id, line_no, product_id, conv.base, src_loc, location_id, on, unit_code, conv.quantity, conv.numerator, conv.denominator,
lot, UNRESTRICTED, UNRESTRICTED, OURS, OURS, move_ref=move_ref, value_in=value,
reason=f"{source_kind} {source_id}" + (f" reverses {reverses_move_ref}" if reverses_move_ref else ""),
purchase_value=value)
credit = "wip"
for il in input_lots or []:
unit.put("inv_lot_link", {"product_id": product_id, "lot_code": lot or "", "input_product_id": il.product_id, "input_lot_code": il.lot_code,
"move_ref": move_ref, "input_quantity": il.quantity, "source_kind": source_kind, "source_id": source_id},
("product_id", "lot_code", "input_product_id", "input_lot_code", "move_ref"))
existing = unit.repo.get("inv_document", document_id=doc_id)
number = existing["number"] if existing else docs.claim_number(unit, loc["warehouse_id"], "receipt", doc_id)
docs.stage_document(unit, doc_id, "receipt", loc["warehouse_id"], company, source_kind=source_kind, source_ref=source_id, done_on=on,
done_by=U.who(caller), number=number)
docs.stage_line(unit, doc_id, line_no, line_kind="receipt", product_id=product_id, unit_code=unit_code, unit_numerator=conv.numerator,
unit_denominator=conv.denominator, quantity_expected=conv.quantity, quantity_done=conv.quantity, quantity_base=conv.base,
to_location_id=location_id, lot_code=lot, value_amount=value)
pm = ledger.post_move(unit, spec)
got = pm.row["value_amount"]
postings.stage(unit, doc_id, f"receipt:{line_no}", company, on, f"{source_kind} {source_id}",
[postings.entry("stock_valuation", got, valuation_class=vclass), postings.entry(credit, got, natural=postings.CREDIT)])
return {"move_id": pm.row["move_id"], "quantity_base": qstr(conv.base), "value": f"{got:.2f}", "replay": False}
return U.run(caller, body)