dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/counts.py
"""Physical inventory: count what is on the shelves and make the system match (INV-10 .. INV-12, INV-33, INV-34, INV-42, INV-43).
A count never sets a quantity: it appends the DIFFERENCE as moves between the bin and LOSS. The difference is taken against what
the moves said at the moment each line was counted, so stock keeps moving during a count and nothing need stop. Small
differences are applied at once; a large one waits for a second person who is neither a counter nor the submitter, matched by
email AND by sign-in subject, so a second account of the same person is refused.
"""
from datetime import date, timedelta
from decimal import Decimal
from typing import 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 need_read, qstr, shaped
from brick_inventory.application.inputs import Code, CountEntryIn, Reason, Ref
from brick_inventory.domain import location as L
from brick_inventory.domain.errors import BadInput, RuleBroken, ShortOf
from brick_inventory.domain.moves import MoveSpec
from brick_inventory.domain.values import NO_LOT, OURS, UNRESTRICTED, ZERO, money
from brick_inventory.infrastructure import ai_steps
OPEN = ("counting", "submitted", "approved", "rejected")
def _ids_of(email, sub) -> set:
return {x.strip().lower() for x in (email, sub) if x and x.strip()}
def _count(unit, count_id: str) -> dict:
c = unit.repo.get("inv_count", count_id=count_id)
if c is None:
raise Refused(404, f"no count {count_id}")
return c
def _lines(unit, count_id: str) -> list:
return unit.repo.find("inv_count_line", order=("location_id", "product_id", "lot_code"), count_id=count_id)
def _scope_locations(unit, count: dict) -> set:
rows = unit.repo.locations_in(count["warehouse_id"])
root = count["location_id"]
return {r["location_id"] for r in rows if r["kind"] == "internal" and r["status"] != "archived" and L.in_subtree(r["location_id"], root)} \
| {r["location_id"] for r in rows if r["kind"] == "internal" and L.in_subtree(r["location_id"], root)}
def unit_cost(unit, company: str, product_id: str) -> Decimal:
"""What a unit costs now: the average, or the standard price under standard cost (zero when nothing was ever valued)."""
ledger.catch_up(unit, company, product_id, unit.today)
pos = ledger.position(unit, company, product_id)
return pos.average if pos.quantity != 0 or pos.average else ZERO
@INV.use_case(kind="command", route="POST /counts", permission="inv:count:write")
def start_count(caller: dict, location_id: Code, product_ids: Optional[list[Code]] = None, source: Code = "manual") -> dict:
"""Start a count of a warehouse zone or bin (everything under it), or of listed products over EVERY bin that holds them
(the cycle count: pass the warehouse as location_id and the products). Every product, lot and stock type with stock in scope
is listed WITHOUT an expected quantity, so the counter counts blind. No other count may be open on the same ground."""
@U.decide
def body(unit):
loc = unit.repo.get("inv_location", location_id=location_id)
if loc is None:
raise Refused(404, f"no location {location_id}")
if loc["kind"] not in ("view", "internal") or loc["status"] == "archived":
raise RuleBroken(f"{location_id} cannot be renamed, archived or counted" if loc["kind"] not in ("view", "internal") else f"{location_id} is archived")
if source not in ("manual", "cycle_plan"):
raise BadInput("source is manual or cycle_plan")
scope = "products" if product_ids else "subtree"
wid = loc["warehouse_id"]
for c in unit.repo.find("inv_count", warehouse_id=wid):
if c["lifecycle"] in OPEN:
clash = (c["scope"] == "subtree" and scope == "subtree" and (L.in_subtree(location_id, c["location_id"]) or L.in_subtree(c["location_id"], location_id))) \
or (c["scope"] == "products" and scope == "products" and set((c["product_ids"] or "").split(",")) & set(product_ids or []))
if clash:
raise RuleBroken(f"{c['count_id']} is already open on it: apply or reject it first")
n = unit.next_number(docs.series_id(wid, "C"))
unit.race("inv_number_series", {"series_id": docs.series_id(wid, "C"), "number": n, "document_id": f"{wid}:C-{n}",
"claimed_at": unit.now_utc.replace(tzinfo=None)})
count_id = f"{wid}:C-{n}"
count = {"count_id": count_id, "warehouse_id": wid, "location_id": location_id, "scope": scope,
"product_ids": ",".join(product_ids or []), "source": source, "lifecycle": "counting", "started_by": U.who(caller),
"started_by_sub": (caller.get("sub") or "").lower() or None}
unit.put("inv_count", count, "count_id")
ids = _scope_locations(unit, {**count, "location_id": location_id if scope == "subtree" else wid})
stock = unit.repo.stock(locations=ids, product_ids=set(product_ids) if product_ids else None)
listed = 0
for (p, l, lot, t, o), q in sorted(stock.items()):
if q <= 0 or o != OURS:
continue
unit.put("inv_count_line", {"count_id": count_id, "location_id": l, "product_id": p, "lot_code": lot, "stock_type": t}, ("count_id", "location_id", "product_id", "lot_code", "stock_type"))
listed += 1
return {"count_id": count_id, "location_id": location_id, "scope": scope, "lifecycle": "counting", "lines_listed": listed}
return U.run(caller, body)
def _line_key(e: CountEntryIn) -> tuple:
return (e.location_id, e.product_id, e.lot_code or NO_LOT, e.stock_type or UNRESTRICTED)
@INV.use_case(kind="command", route="POST /counts/{count_id}/record", permission="inv:count:write")
def record_counted(caller: dict, count_id: Ref, entries: list[CountEntryIn]) -> dict:
"""Record what is on the shelf, in any unit of the product: 5 PK24 and 6 C62 are two entries of one line, summed exactly
into the counted quantity in the base unit. The difference is taken against what the moves said at the moment the line was
counted, not at the start; nothing changes on hand until the count is applied. Counting a product into a bin it was never
moved to is allowed (a found item). A wrong entry is corrected by a negative one. A quantity outside its bounds, in a unit the
product lacks, or with more decimals than the unit allows is refused with the reason."""
@U.decide
def body(unit):
c = _count(unit, count_id)
if c["lifecycle"] != "counting":
raise RuleBroken(f"{count_id} is {c['lifecycle']}: it is not being counted")
scope_ids = _scope_locations(unit, {**c, "location_id": c["location_id"] if c["scope"] == "subtree" else c["warehouse_id"]})
existing = {(l["location_id"], l["product_id"], l["lot_code"], l["stock_type"]): l for l in _lines(unit, count_id)}
by_line: dict = {}
out = []
for e in entries:
if e.location_id not in scope_ids:
raise RuleBroken(f"{e.location_id} is not part of count {count_id}")
products.stocked(unit, e.product_id)
tr = unit.tracking(e.product_id)["tracking"]
if tr != "none" and not e.lot_code:
raise RuleBroken(f"{e.product_id} is tracked by {tr}: count it by {'serial number' if tr == 'serial' else 'lot'}")
conv = products.convert(unit, e.product_id, abs(e.quantity), e.unit_code, document_ref=count_id, what="a counted quantity", positive=False)
base = conv.base if e.quantity >= 0 else -conv.base
qty = e.quantity
k = _line_key(e)
if tr == "serial" and k not in existing:
where = ledger.serial_where(unit, e.product_id, e.lot_code)
if where and where != e.location_id:
raise RuleBroken(f"{e.lot_code} is on hand at {where}: record a transfer or a return first")
if not where:
left = _last_move_to_customer(unit, e.product_id, e.lot_code)
if left:
raise RuleBroken(f"{e.lot_code} left on {left}: record a return first")
by_line.setdefault(k, []).append((qty, e.unit_code, conv, base))
now = unit.now_utc
for k, items in by_line.items():
line = existing.get(k) or {"count_id": count_id, "location_id": k[0], "product_id": k[1], "lot_code": k[2], "stock_type": k[3],
"counted_base": None}
entry_rows = unit.repo.find("inv_count_entry", count_id=count_id, location_id=k[0], product_id=k[1], lot_code=k[2], stock_type=k[3])
no = max([r["entry_no"] for r in entry_rows] + [0])
total = line.get("counted_base") or ZERO
for qty, ucode, conv, base in items:
no += 1
total += base
unit.put("inv_count_entry", {"count_id": count_id, "location_id": k[0], "product_id": k[1], "lot_code": k[2], "stock_type": k[3],
"entry_no": no, "quantity": qty, "unit_code": ucode, "unit_numerator": conv.numerator,
"unit_denominator": conv.denominator, "quantity_base": base,
"counted_by": U.who(caller), "counted_sub": (caller.get("sub") or "").lower() or None,
"counted_at": now.replace(tzinfo=None)}, ("count_id", "location_id", "product_id", "lot_code", "stock_type", "entry_no"))
if total < 0:
raise RuleBroken(f"{k[1]} at {k[0]}: the entries add up to {qstr(total)}: a count is never below zero")
expected = unit.repo.stock(product_id=k[1], locations={k[0]}, lot_code=k[2], as_of=now).get((k[1], k[0], k[2], k[3], OURS), ZERO)
unit.put("inv_count_line", {"count_id": count_id, "location_id": k[0], "product_id": k[1], "lot_code": k[2], "stock_type": k[3],
"counted_base": total, "counted_at": now.replace(tzinfo=None), "counted_by": U.who(caller),
"counted_sub": (caller.get("sub") or "").lower() or None, "expected_base": expected,
"difference_base": None, "value_amount": None},
("count_id", "location_id", "product_id", "lot_code", "stock_type"))
out.append({"location_id": k[0], "product_id": k[1], "lot_code": k[2] or None, "counted_base": qstr(total),
"entries": len(entry_rows) + len(items)})
return {"count_id": count_id, "lifecycle": "counting", "lines": out}
return U.run(caller, body)
def _last_move_to_customer(unit, product_id: str, serial: str) -> str:
moves = [m for m in unit.repo.moves_of(product_id=product_id, lot_code=serial) if m["to_location_id"].endswith(":CUSTOMER")]
return moves[-1]["document_id"] if moves else ""
def _differences(unit, count: dict, lines: list) -> tuple:
company = unit.warehouse(count["warehouse_id"])["company_code"]
rows, total_abs, net = [], ZERO, ZERO
for l in lines:
diff = (l["counted_base"] or ZERO) - (l["expected_base"] or ZERO)
cost = unit_cost(unit, company, l["product_id"])
value = money(diff * cost)
rows.append({**l, "difference_base": diff, "value_amount": value})
total_abs += abs(value)
net += value
return rows, total_abs, net
def line_view(l: dict, show: bool) -> dict:
out = {"location_id": l["location_id"], "product_id": l["product_id"], "lot_code": l["lot_code"] or None, "stock_type": l["stock_type"],
"counted_base": qstr(l["counted_base"]), "counted_at": l["counted_at"].isoformat() if l["counted_at"] else None,
"counted_by": l["counted_by"]}
if show:
out.update(expected_base=qstr(l["expected_base"]), difference_base=qstr(l["difference_base"]),
value_amount=f"{l['value_amount']:.2f}" if l["value_amount"] is not None else None, suggestion=l["suggestion"])
return out
@INV.use_case(kind="command", route="POST /counts/{count_id}/submit", permission="inv:count:write", idempotent=True)
def submit_count(caller: dict, count_id: Ref) -> dict:
"""Submit the count: each line's difference (counted less what the moves said at its counted moment) and its value at the
average cost are fixed on the count. A line not counted yet is refused (record 0 for an empty shelf). With the AI step on
(the owner's setting), each line with a difference gets a SUGGESTED likely cause, marked as a suggestion; nothing is adjusted
or approved by it."""
@U.decide
def body(unit):
c = _count(unit, count_id)
if c["lifecycle"] == "submitted":
return _submitted_view(unit, c, caller, replay=True)
if c["lifecycle"] != "counting":
raise RuleBroken(f"{count_id} is {c['lifecycle']}: only a count being counted is submitted")
lines = _lines(unit, count_id)
missing = [l for l in lines if l["counted_base"] is None]
if missing:
raise RuleBroken(f"{len(missing)} line(s) are not counted yet, for example {missing[0]['product_id']} at {missing[0]['location_id']}: "
"record 0 for an empty shelf")
rows, total_abs, net = _differences(unit, c, lines)
mode = unit.settings.value("ai_explain_variance", unit.today)
suggestions = {}
if mode != "off":
for r in rows:
if r["difference_base"] != 0:
s = _suggest(unit, c, r, mode)
if s:
suggestions[(r["location_id"], r["product_id"], r["lot_code"])] = s
for r in rows:
sug = suggestions.get((r["location_id"], r["product_id"], r["lot_code"]))
unit.put("inv_count_line", {"count_id": count_id, "location_id": r["location_id"], "product_id": r["product_id"], "lot_code": r["lot_code"],
"stock_type": r["stock_type"], "difference_base": r["difference_base"], "value_amount": r["value_amount"],
"suggestion": sug}, ("count_id", "location_id", "product_id", "lot_code", "stock_type"))
unit.put("inv_count", {"count_id": count_id, "lifecycle": "submitted", "submitted_by": U.who(caller),
"submitted_by_sub": (caller.get("sub") or "").lower() or None, "value_amount": total_abs}, "count_id")
needs = _needs_second_person(unit, total_abs)
return {"count_id": count_id, "lifecycle": "submitted", "value_amount": f"{net:.2f}", "absolute_value": f"{total_abs:.2f}",
"needs_second_person": needs, "lines": [{**line_view(r, True), "suggestion": suggestions.get((r["location_id"], r["product_id"], r["lot_code"]))}
for r in rows]} if U.can(caller, "inv:valuation:read") else \
{"count_id": count_id, "lifecycle": "submitted", "needs_second_person": needs,
"lines": [line_view({**r, "value_amount": None}, True) | {"value_amount": None} for r in rows]}
return U.run(caller, body)
def _needs_second_person(unit, total_abs: Decimal) -> bool:
return total_abs >= unit.settings.number("count_approval_threshold", unit.today)
def _suggest(unit, c: dict, r: dict, mode: str) -> Optional[str]:
"""What an AI step proposes for one line (never applied). The provider reads JSON scalars only: the line and the moves
since the count started, with the pack conversions."""
info = unit.product(r["product_id"])
moves = [m for m in unit.repo.moves_of(product_id=r["product_id"]) if m["moved_at"] >= (c["created_at"] or m["moved_at"])]
ctx = {"count_id": c["count_id"], "location_id": r["location_id"], "product_id": r["product_id"], "lot_code": r["lot_code"] or "",
"counted_base": str(r["counted_base"]), "expected_base": str(r["expected_base"]), "difference_base": str(r["difference_base"]),
"base_unit": info.get("base_unit_code") or "",
"recent_moves": "; ".join(f"{m['moved_at']:%H:%M} {m['quantity_base']} {m['from_location_id']}->{m['to_location_id']} ({m['document_id']})" for m in moves[-10:]),
"pack_conversions": "; ".join(f"1 {u['unit_code']} = {u['numerator']}/{u['denominator']} {info.get('base_unit_code')}" for u in info.get("units") or [])}
answer = ai_steps.ask("ai_explain_variance", mode, f"{c['count_id']}|{r['location_id']}|{r['product_id']}|{r['lot_code']}", 1, ["suggestion"], ctx)
if answer.kind != "done":
return None
text = (answer.suggestion or {}).get("text") or answer.rationale
if not text:
return None
unit.put("inv_suggestion", {"count_id": c["count_id"], "location_id": r["location_id"], "product_id": r["product_id"], "lot_code": r["lot_code"],
"step": "ai_explain_variance", "text": text}, ("count_id", "location_id", "product_id", "lot_code", "step"))
return text
def _submitted_view(unit, c: dict, caller: dict, replay: bool = False) -> dict:
lines = _lines(unit, c["count_id"])
show = True
return shaped(caller, {"count_id": c["count_id"], "lifecycle": c["lifecycle"], "value_amount": f"{c['value_amount'] or 0:.2f}", "replay": replay,
"needs_second_person": _needs_second_person(unit, c["value_amount"] or ZERO),
"lines": [line_view(l, show) for l in lines]})
@INV.use_case(kind="command", route="POST /counts/{count_id}/explain", permission="inv:count:suggest", idempotent=True, tool=False)
def ai_explain_variance(caller: dict, count_id: Ref) -> dict:
"""The system (a machine) re-runs the AI step for a submitted count: a suggested likely cause per line with a difference,
marked as a suggestion. It writes inv_suggestion and changes no quantity, no approval and no decision a person made."""
@U.decide
def body(unit):
c = _count(unit, count_id)
if c["lifecycle"] != "submitted":
raise RuleBroken(f"{count_id} is {c['lifecycle']}: suggestions are made for a submitted count")
mode = unit.settings.value("ai_explain_variance", unit.today)
if mode == "off":
return {"count_id": count_id, "mode": "off", "suggestions": 0}
n = 0
for l in _lines(unit, count_id):
if l["difference_base"]:
got = _suggest(unit, c, l, mode)
if got:
unit.put("inv_count_line", {"count_id": count_id, "location_id": l["location_id"], "product_id": l["product_id"],
"lot_code": l["lot_code"], "stock_type": l["stock_type"], "suggestion": got},
("count_id", "location_id", "product_id", "lot_code", "stock_type"))
n += 1
return {"count_id": count_id, "mode": mode, "suggestions": n}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /counts/{count_id}/suggestions/decide", permission="inv:adjustment:approve", idempotent=True)
def decide_suggestion(caller: dict, count_id: Ref, location_id: Code, product_id: Code, decision: Code, lot_code: Optional[Code] = None) -> dict:
"""The approver accepts or dismisses a suggestion (the decision is logged with who). Accepting changes nothing by itself."""
@U.decide
def body(unit):
if decision not in ("accepted", "dismissed"):
raise BadInput("decision is accepted or dismissed")
row = unit.repo.get("inv_suggestion", count_id=count_id, location_id=location_id, product_id=product_id, lot_code=lot_code or "",
step="ai_explain_variance")
if row is None:
raise Refused(404, "no such suggestion")
unit.put("inv_suggestion", {**{k: row[k] for k in ("count_id", "location_id", "product_id", "lot_code", "step")}, "decided": decision,
"decided_by": U.who(caller), "decided_at": unit.now_utc.replace(tzinfo=None)},
("count_id", "location_id", "product_id", "lot_code", "step"))
return {"count_id": count_id, "decision": decision, "decided_by": U.who(caller)}
return U.run(caller, body)
def _apply(unit, c: dict, caller: dict) -> dict:
"""Append the difference of every line as moves between its bin and LOSS, and ask the ledger to post the loss or gain."""
lines = [l for l in _lines(unit, c["count_id"]) if l["difference_base"]]
wh = unit.warehouse(c["warehouse_id"])
loss = L.virtual_id(wh["company_code"], "LOSS")
doc_id = f"adj:{c['count_id']}"
docs.stage_document(unit, doc_id, "adjustment", c["warehouse_id"], wh["company_code"], lifecycle="done", done_on=unit.today,
done_by=U.who(caller), source_kind="count", source_ref=c["count_id"],
number=c["count_id"])
total = ZERO
posted = []
by_class: dict = {}
for i, l in enumerate(lines, start=1):
diff = l["difference_base"]
src, dst = (l["location_id"], loss) if diff < 0 else (loss, l["location_id"])
spec = MoveSpec(doc_id, i, l["product_id"], abs(diff), src, dst, unit.today, "", None, 1, 1, l["lot_code"] or NO_LOT,
l["stock_type"], l["stock_type"], OURS, OURS, reason=f"count {c['count_id']}")
try:
pm = ledger.post_move(unit, spec)
except ShortOf as s:
raise RuleBroken(f"{s}: the stock moved after it was counted - recount") from None
posted.append(pm)
v = pm.row["value_amount"]
total += v
vc = unit.product(l["product_id"]).get("valuation_class") or "*"
by_class[vc] = by_class.get(vc, ZERO) + v
docs.stage_line(unit, doc_id, i, line_kind="adjustment", product_id=l["product_id"], quantity_base=abs(diff), from_location_id=src,
to_location_id=dst, lot_code=l["lot_code"] or NO_LOT, stock_type=l["stock_type"], value_amount=v)
journal = []
for vc, v in by_class.items():
journal += postings.movement_lines(v, "inventory_loss", vc, memo=f"count {c['count_id']}")
from brick_inventory.application.deliveries import _merge
postings.stage(unit, doc_id, "adjustment", wh["company_code"], unit.today, f"stock count {c['count_id']}", _merge(journal))
unit.put("inv_count", {"count_id": c["count_id"], "lifecycle": "applied", "applied_on": unit.today}, "count_id")
return {"count_id": c["count_id"], "lifecycle": "applied", "document_id": doc_id, "adjustments": len(posted), "value_amount": f"{total:.2f}"}
@INV.use_case(kind="command", route="POST /counts/{count_id}/apply", permission="inv:adjustment:apply", idempotent=True)
def apply_small(caller: dict, count_id: Ref) -> dict:
"""Apply a count whose differences are under the approval threshold (the absolute value of its lines, in the company's
currency): one adjustment move per line between the bin and LOSS, and the loss or gain posted to the ledger. A count at or
above the threshold waits for a second person (approve_count)."""
@U.decide
def body(unit):
c = _count(unit, count_id)
if c["lifecycle"] == "applied":
return {"count_id": count_id, "lifecycle": "applied", "replay": True}
if c["lifecycle"] != "submitted":
raise RuleBroken(f"{count_id} is {c['lifecycle']}: submit it first")
if _needs_second_person(unit, c["value_amount"] or ZERO):
raise RuleBroken(f"{count_id} is worth {c['value_amount']:.2f}, at or above the {unit.settings.value('count_approval_threshold', unit.today)} "
"threshold: a second person approves it")
return {**_apply(unit, c, caller), "replay": False}
return U.run(caller, body, lost_message=lambda repo, row=None: "stock moved while the count was applied: try again")
def _approver_clash(unit, c: dict, caller: dict) -> Optional[str]:
me = set(U.identities(caller))
counters: set = set()
for l in _lines(unit, c["count_id"]):
counters |= _ids_of(l["counted_by"], l["counted_sub"])
for e in unit.repo.find("inv_count_entry", count_id=c["count_id"]):
counters |= _ids_of(e["counted_by"], e["counted_sub"])
submitter = _ids_of(c["submitted_by"], c["submitted_by_sub"])
if me & submitter:
return f"you submitted {c['count_id']}: a second person approves"
if me & counters:
return f"you counted lines of {c['count_id']}: a second person approves"
return None
@INV.use_case(kind="command", route="POST /counts/{count_id}/approve", permission="inv:adjustment:approve", idempotent=True)
def approve_count(caller: dict, count_id: Ref) -> dict:
"""A second person approves a submitted count, and it applies at once: neither a counter nor the submitter, compared by
email AND by sign-in subject (a second account of the same person is refused). The decision is recorded."""
@U.decide
def body(unit):
c = _count(unit, count_id)
if c["lifecycle"] == "applied":
return {"count_id": count_id, "lifecycle": "applied", "replay": True}
if c["lifecycle"] not in ("submitted", "approved"):
raise RuleBroken(f"{count_id} is {c['lifecycle']}: only a submitted count is approved")
clash = _approver_clash(unit, c, caller)
if clash:
from brick_inventory.domain.errors import GateBroken
raise GateBroken("approver_not_counter", clash, "ask another stock controller", 409)
unit.put("inv_count", {"count_id": count_id, "lifecycle": "approved", "decided_by": U.who(caller),
"decided_sub": (caller.get("sub") or "").lower() or None}, "count_id")
done = _apply(unit, {**c, "lifecycle": "approved"}, caller)
return {**done, "decided_by": U.who(caller), "replay": False}
return U.run(caller, body, lost_message=lambda repo, row=None: "stock moved while the count was applied: try again")
@INV.use_case(kind="command", route="POST /counts/{count_id}/apply-approved", permission="inv:adjustment:approve", idempotent=True)
def apply_approved(caller: dict, count_id: Ref) -> dict:
"""Apply a count that was approved but not yet applied (a retry): the adjustment moves and the posting."""
@U.decide
def body(unit):
c = _count(unit, count_id)
if c["lifecycle"] == "applied":
return {"count_id": count_id, "lifecycle": "applied", "replay": True}
if c["lifecycle"] != "approved":
raise RuleBroken(f"{count_id} is {c['lifecycle']}: only an approved count is applied here")
return {**_apply(unit, c, caller), "replay": False}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /counts/{count_id}/reject", permission="inv:adjustment:approve", idempotent=True)
def reject_count(caller: dict, count_id: Ref, reason: Reason) -> dict:
"""Reject a submitted count with a reason: it returns to the warehouse manager to recount; nothing is applied. Not by a counter
or the submitter."""
@U.decide
def body(unit):
if not reason.strip():
raise BadInput("a reason is required to reject a count")
c = _count(unit, count_id)
if c["lifecycle"] == "rejected":
return {"count_id": count_id, "lifecycle": "rejected", "replay": True}
if c["lifecycle"] != "submitted":
raise RuleBroken(f"{count_id} is {c['lifecycle']}: only a submitted count is rejected")
clash = _approver_clash(unit, c, caller)
if clash:
raise RuleBroken(clash)
unit.put("inv_count", {"count_id": count_id, "lifecycle": "rejected", "reason": reason, "decided_by": U.who(caller),
"decided_sub": (caller.get("sub") or "").lower() or None}, "count_id")
return {"count_id": count_id, "lifecycle": "rejected", "reason": reason, "replay": False}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /counts/{count_id}/recount", permission="inv:count:write", idempotent=True)
def recount(caller: dict, count_id: Ref) -> dict:
"""A rejected count goes back to counting: the lines are recorded again from nothing."""
@U.decide
def body(unit):
c = _count(unit, count_id)
if c["lifecycle"] == "counting":
return {"count_id": count_id, "lifecycle": "counting", "replay": True}
if c["lifecycle"] != "rejected":
raise RuleBroken(f"{count_id} is {c['lifecycle']}: only a rejected count is recounted")
for l in _lines(unit, count_id):
unit.put("inv_count_line", {"count_id": count_id, "location_id": l["location_id"], "product_id": l["product_id"], "lot_code": l["lot_code"],
"stock_type": l["stock_type"], "counted_base": None, "counted_at": None, "counted_by": None,
"counted_sub": None, "expected_base": None, "difference_base": None, "value_amount": None,
"suggestion": None}, ("count_id", "location_id", "product_id", "lot_code", "stock_type"))
unit.repo.delete_where("inv_count_entry", count_id=count_id)
unit.put("inv_count", {"count_id": count_id, "lifecycle": "counting", "submitted_by": None, "submitted_by_sub": None,
"decided_by": None, "decided_sub": None, "value_amount": None}, "count_id")
return {"count_id": count_id, "lifecycle": "counting", "replay": False}
return U.run(caller, body)
@INV.use_case(kind="query", route="GET /counts/{count_id}")
def get_count(caller: dict, count_id: Ref) -> dict:
"""A count with its lines. While it is being counted the counter sees no expected quantity; once submitted each line shows
expected (at its counted moment), counted, difference and value (value only to inv:valuation:read), and any suggestion marked
as a suggestion. The approver cannot change a counted quantity."""
need_read(caller)
with U.read() as unit:
c = _count(unit, count_id)
show = c["lifecycle"] != "counting"
lines = _lines(unit, count_id)
sug = {(s["location_id"], s["product_id"], s["lot_code"]): s for s in unit.repo.find("inv_suggestion", count_id=count_id)}
out = []
for l in lines:
v = line_view(l, show)
s = sug.get((l["location_id"], l["product_id"], l["lot_code"]))
if show and s:
v["suggestion"] = {"text": s["text"], "marked": "suggestion", "decided": s["decided"], "decided_by": s["decided_by"]}
out.append(v)
return shaped(caller, {"count_id": count_id, "warehouse_id": c["warehouse_id"], "location_id": c["location_id"], "scope": c["scope"],
"source": c["source"], "lifecycle": c["lifecycle"], "started_by": c["started_by"], "submitted_by": c["submitted_by"],
"decided_by": c["decided_by"], "reason": c["reason"],
"value_amount": f"{c['value_amount']:.2f}" if c["value_amount"] is not None and show else None, "lines": out})
@INV.use_case(kind="query", route="GET /counts")
def list_counts(caller: dict, lifecycle: Optional[Code] = None, warehouse_id: Optional[Code] = None) -> dict:
"""Counts, newest first."""
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_count", order=("-count_id",), limit=200, **where)
return {"counts": [{"count_id": c["count_id"], "lifecycle": c["lifecycle"], "location_id": c["location_id"], "source": c["source"]} for c in rows]}
# ── ABC classes and the cycle count plan (INV-42, INV-43) ─────────────────────────────────────────────────
def _add_months(d: date, n: int) -> date:
y, m = divmod(d.month - 1 + n, 12)
year, month = d.year + y, m + 1
import calendar
return date(year, month, min(d.day, calendar.monthrange(year, month)[1]))
def due_date(last: Optional[date], per_year: int, today: date) -> date:
"""When a product is next due: 12/frequency months after its last count (A 12 a year: monthly; B 4: quarterly; C 1: yearly);
a frequency that does not divide 12 uses 365/frequency days. Never counted: due now."""
if last is None:
return today
return _add_months(last, 12 // per_year) if 12 % per_year == 0 else last + timedelta(days=round(365 / per_year))
def class_on(unit, warehouse_id: str, product_id: str, on: date) -> Optional[dict]:
rows = [r for r in unit.repo.find("inv_abc_class", order=("valid_from",), warehouse_id=warehouse_id, product_id=product_id) if r["valid_from"] <= on]
return rows[-1] if rows else None
def last_counted(unit, warehouse_id: str, product_id: str) -> Optional[date]:
best = None
for c in unit.repo.find("inv_count", warehouse_id=warehouse_id, lifecycle="applied"):
if c["applied_on"] and (best is None or c["applied_on"] > best):
if any(l["product_id"] == product_id for l in _lines(unit, c["count_id"])) or (c["scope"] == "products" and product_id in (c["product_ids"] or "").split(",")):
best = c["applied_on"]
return best
@INV.use_case(kind="command", route="POST /abc/classify", permission="inv:abc:classify", idempotent=True, tool=False)
def classify_abc(caller: dict, warehouse_id: Optional[Code] = None) -> dict:
"""The monthly classification: each product is classed A, B or C at each warehouse by the cumulative share of the value that
moved out in the last twelve months (deliveries and issues), counting from the products above it - so counting effort
follows value. Dated rows from today; a re-run the same day writes nothing; a class a manager overrode is kept and reported."""
@U.decide
def body(unit):
a, b = (int(x) for x in unit.settings.value("abc_thresholds", unit.today).split(","))
whs = [warehouse_id] if warehouse_id else [w["warehouse_id"] for w in unit.repo.find("inv_warehouse", order=("warehouse_id",))]
since = date(unit.today.year - 1, unit.today.month, min(unit.today.day, 28))
result = []
for wid in whs:
unit.warehouse(wid)
locs = {r["location_id"] for r in unit.repo.locations_in(wid) if r["kind"] == "internal"}
sinks = {r["location_id"] for r in unit.repo.find("inv_location") if r["kind"] in ("customer", "consumption")}
usage: dict = {}
for m in unit.repo.find("inv_move"):
if m["from_location_id"] in locs and m["to_location_id"] in sinks and m["moved_on"] > since and m["value_amount"]:
usage[m["product_id"]] = usage.get(m["product_id"], ZERO) + abs(m["value_amount"])
total = sum(usage.values(), ZERO)
running = ZERO
written = changed = 0
overridden = []
for pid, val in sorted(usage.items(), key=lambda kv: (-kv[1], kv[0])):
share = (running / total * 100) if total else ZERO
cls = "A" if share < a else ("B" if share < b else "C")
running += val
cur = unit.repo.get("inv_abc_class", warehouse_id=wid, product_id=pid, valid_from=unit.today)
latest = class_on(unit, wid, pid, unit.today)
if latest and latest["source"] == "override":
if latest["abc_class"] != cls:
overridden.append({"product_id": pid, "computed": cls, "override": latest["abc_class"], "reason": latest["reason"]})
continue
if cur and cur["abc_class"] == cls:
continue
unit.put("inv_abc_class", {"warehouse_id": wid, "product_id": pid, "valid_from": unit.today, "abc_class": cls, "consumption_value": val,
"cumulative_share": (share).quantize(Decimal("0.0001")), "source": "computed", "set_by": "system"},
("warehouse_id", "product_id", "valid_from"))
written += 1
changed += 1 if (latest is None or latest["abc_class"] != cls) else 0
result.append({"warehouse_id": wid, "classified": len(usage), "written": written, "overridden": overridden})
return {"valid_from": unit.today.isoformat(), "warehouses": result}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /abc/override", permission="inv:count:write", idempotent=True)
def override_abc_class(caller: dict, warehouse_id: Code, product_id: Code, abc_class: Code, reason: Reason) -> dict:
"""The warehouse manager overrides a product's class from today, with a reason: kept until a run that would change the class
names it as overridden. It changes nothing about stock or value."""
@U.decide
def body(unit):
if abc_class not in ("A", "B", "C") or not reason.strip():
raise BadInput("abc_class is A, B or C, with a reason")
unit.warehouse(warehouse_id)
cur = unit.repo.get("inv_abc_class", warehouse_id=warehouse_id, product_id=product_id, valid_from=unit.today) or {}
unit.put("inv_abc_class", {"warehouse_id": warehouse_id, "product_id": product_id, "valid_from": unit.today, "abc_class": abc_class,
"consumption_value": cur.get("consumption_value"), "cumulative_share": cur.get("cumulative_share"),
"source": "override", "reason": reason, "set_by": U.who(caller)}, ("warehouse_id", "product_id", "valid_from"))
return {"warehouse_id": warehouse_id, "product_id": product_id, "abc_class": abc_class, "source": "override", "reason": reason}
return U.run(caller, body)
@INV.use_case(kind="query", route="GET /cycle-count-plan")
def cycle_count_plan(caller: dict, warehouse_id: Code) -> dict:
"""What to count this week: every classed product whose next count is due (more often for what matters: A monthly, B
quarterly, C yearly, by the setting count_frequency), overdue ones marked. Start the counts from it with start_count
(source cycle_plan) over the listed products."""
need_read(caller)
with U.read() as unit:
unit.warehouse(warehouse_id)
freq = {k: int(v) for k, v in (p.split(":") for p in unit.settings.value("count_frequency", unit.today).split(","))}
due, later = [], []
for r in unit.repo.find("inv_abc_class", warehouse_id=warehouse_id):
cur = class_on(unit, warehouse_id, r["product_id"], unit.today)
if cur is None or cur["valid_from"] != r["valid_from"]:
continue
last = last_counted(unit, warehouse_id, r["product_id"])
when = due_date(last, freq[cur["abc_class"]], unit.today)
row = {"product_id": r["product_id"], "abc_class": cur["abc_class"], "last_counted_on": last.isoformat() if last else None,
"due_on": when.isoformat(), "overdue": when < unit.today}
(due if when <= unit.today else later).append(row)
return {"warehouse_id": warehouse_id, "on": unit.today.isoformat(), "due": sorted(due, key=lambda x: x["due_on"]),
"not_yet_due": sorted(later, key=lambda x: x["due_on"])}