dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/valuation_ops.py
"""The books side of stock: the account map, postings to the ledger and their retry, standard prices and their revaluation, and
the delivery-cost differences purchasing hands over (INV-24, INV-32, INV-48, INV-56, INV-57)."""
import json
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId
from brick_inventory.application import ledger, postings
from brick_inventory.application import unit as U
from brick_inventory.application.brick import INV
from brick_inventory.application.common import need
from brick_inventory.application.inputs import Amount, Code, IsoDate, Rate, Ref, day
from brick_inventory.domain import valuation as val
from brick_inventory.domain.errors import BadInput, GateBroken, RuleBroken
from brick_inventory.domain.values import ZERO, money
PURPOSES = ("stock_valuation", "stock_received_not_billed", "cost_of_goods_sold", "inventory_loss", "consumption", "inventory_scrap",
"free_of_charge", "price_difference", "revaluation", "consignment_liability", "subcontracting", "cost_of_sales", "wip")
@INV.use_case(kind="command", route="POST /account-map", permission="inv:account_map:write", idempotent=True)
def set_inventory_account_map(caller: dict, company_code: Code, purpose: Code, account_id: BigId, valuation_class: Code = "*",
valid_from: Optional[IsoDate] = None) -> dict:
"""Say which ledger account a stock purpose uses for a company (and, optionally, a valuation class: raw materials and
trading goods may post to different stock accounts), from a date. Purposes: stock_valuation, stock_received_not_billed,
cost_of_goods_sold, inventory_loss, consumption, inventory_scrap, free_of_charge, price_difference, revaluation,
consignment_liability, subcontracting. Without an entry the ledger maps the purpose itself."""
@U.decide
def body(unit):
if purpose not in PURPOSES:
raise BadInput(f"purpose is one of {', '.join(PURPOSES)}")
start = day(valid_from) or unit.today
if start < unit.today:
raise RuleBroken("an account map entry starts today or later, never back-dated")
unit.put("inv_account_map", {"company_code": company_code, "purpose": purpose, "valuation_class": valuation_class, "valid_from": start,
"account_id": account_id, "set_by": U.who(caller)},
("company_code", "purpose", "valuation_class", "valid_from"))
return {"company_code": company_code, "purpose": purpose, "valuation_class": valuation_class, "valid_from": start.isoformat(),
"account_id": str(account_id)}
return U.run(caller, body)
@INV.use_case(kind="query", route="GET /account-map")
def get_inventory_account_map(caller: dict, company_code: Optional[Code] = None) -> dict:
"""The account map (the accountant's configuration)."""
need(caller, "inv:valuation:read", as_404=True)
with U.read() as unit:
rows = unit.repo.find("inv_account_map", order=("company_code", "purpose", "valid_from"), **({"company_code": company_code} if company_code else {}))
return {"account_map": [{"company_code": r["company_code"], "purpose": r["purpose"], "valuation_class": r["valuation_class"],
"valid_from": r["valid_from"].isoformat(), "account_id": str(r["account_id"])} for r in rows]}
@INV.use_case(kind="query", route="GET /postings")
def list_postings(caller: dict, status: Optional[Code] = None) -> dict:
"""What was asked of the ledger for each document event and its answer: posted, pending (with the ledger's reason, to fix),
or exported (ledger mode export)."""
need(caller, "inv:valuation:read", as_404=True)
with U.read() as unit:
rows = unit.repo.find("inv_posting", order=("document_id", "event"), **({"status": status} if status else {}))
return {"postings": [{"document_id": r["document_id"], "event": r["event"], "status": r["status"], "journal_id": str(r["journal_id"]),
"reason": r["reason"] or None, "attempts": r["attempts"]} for r in rows]}
@INV.use_case(kind="command", route="POST /postings/retry", permission="inv:valuation:post", idempotent=True, tool=False)
def retry_postings(caller: dict) -> dict:
"""The clock worker (or an external scheduler) retries every posting the ledger refused. Each journal's id is derived from
(document, event), so a retry that lands changes the ledger once."""
return postings.retry_pending()
@INV.use_case(kind="command", route="POST /standard-prices", permission=None)
def schedule_standard_price(caller: dict, company_code: Code, product_id: Code, standard_price: Rate, valid_from: IsoDate,
source_ref: Optional[Ref] = None, released_by: Optional[Ref] = None, released_sub: Optional[Ref] = None) -> dict:
"""Set a product's standard cost for a company from a date (per base unit, never product's own cost_amount). It starts
tomorrow at the earliest, and the company must value at standard on that date (or switch to it that day, from the first of a
later month). On the day it takes effect the stock on hand is revalued once, with its journal. Needs inv:valuation:write.
v6 (INV-60, intake_standard_price): manufacturing's service client (inv:standard_price:intake) sends a released roll-up's price with its
source_ref (mfg:<rollup_id>) and who released it; the releaser's second-person release is the approval, so it is not asked again. The same
(company, product, valid_from, source_ref) answers the same row; a different price already set for that day is 409 standard_price_taken,
never overwritten; a company valued at average cost is refused (values_at_standard)."""
need(caller, "inv:standard_price:intake" if source_ref else "inv:valuation:write")
@U.decide
def body(unit):
start = day(valid_from)
if start <= unit.today:
raise GateBroken("starts_after_today", "a standard price starts tomorrow at the earliest", "pick a later date", 422 if source_ref else 409)
if unit.settings.method(company_code, start) != "standard":
raise GateBroken("values_at_standard" if source_ref else "standard_method",
f"{company_code} values stock at {unit.settings.method(company_code, start)} cost on {start.isoformat()}",
"the owner switches the company to standard from a later month (INV-48), or release for average cost (no standard price)"
if source_ref else "the owner switches the company to standard from the first of a month first", 422 if source_ref else 409)
from brick_inventory.application.products import stocked
stocked(unit, product_id)
taken = next((r for r in unit.repo.find("inv_standard_price", company_code=company_code, product_id=product_id, valid_from=start)), None)
if taken is not None and source_ref:
if taken.get("source_ref") == source_ref and taken["standard_price"] == standard_price:
return {"company_code": company_code, "product_id": product_id, "valid_from": start.isoformat(),
"standard_price": f"{standard_price:f}", "lifecycle": taken["lifecycle"], "replay": True}
raise GateBroken("standard_price_taken", f"a standard price of {taken['standard_price']:f} is already set for {start.isoformat()} by {taken['set_by']}",
"the accountant withdraws theirs, or manufacturing releases for a later date", 409)
by = (released_by or U.who(caller)) if source_ref else U.who(caller)
unit.put("inv_standard_price", {"company_code": company_code, "product_id": product_id, "valid_from": start,
"standard_price": standard_price, "set_by": by, "lifecycle": "scheduled", "source_ref": source_ref or None,
"set_sub": (released_sub or "") or None},
("company_code", "product_id", "valid_from"))
return {"company_code": company_code, "product_id": product_id, "valid_from": start.isoformat(), "standard_price": f"{standard_price:f}",
"lifecycle": "scheduled", "replay": False}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /standard-prices/apply", permission="inv:valuation:post", idempotent=True, tool=False)
def apply_standard_price(caller: dict) -> dict:
"""Start of each business day: the standard prices (and company switches to standard) that take effect today revalue the
quantity on hand once, with their journal. Keyed (company, product, valid_from); a repeat writes nothing."""
@U.decide
def body(unit):
applied = 0
pairs = {(r["company_code"], r["product_id"]) for r in unit.repo.find("inv_standard_price")}
for company, product_id in sorted(pairs):
before = len(unit.races)
ledger.catch_up(unit, company, product_id, unit.today)
applied += 1 if len(unit.races) > before else 0
return {"on": unit.today.isoformat(), "revalued": applied}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /valuation-adjustments", permission="inv:valuation:post", idempotent=True, tool=False)
def post_valuation_adjustment(caller: dict, adjustment_ref: Ref, product_id: Code, amount: Amount, receipt_id: Optional[Ref] = None,
source_kind: Code = "purchase_delivery_cost", purchase_order_id: Optional[Ref] = None,
currency: Optional[Code] = None, on: Optional[IsoDate] = None, production_order_id: Optional[Ref] = None,
company_code: Optional[Code] = None) -> dict:
"""Purchasing hands over a delivery-cost difference found when the bill was approved. It is spread over what is STILL ON HAND
from the named receipt (older stock is assumed to leave first), added to the value of that stock; the rest posts to price
difference, in one journal through the ledger. Under standard cost the whole amount is price difference. Keyed by
adjustment_ref: a repeat changes nothing. Stock from another receipt is never revalued.
v6 (INV-57, source_kind production_settlement, from manufacturing): an average-cost company's production variance (`amount` signed, positive when
the order cost more) is spread over what is STILL ON HAND from the order's output receipts (on_hand_amount, into stock value) and the rest goes to
cost_of_sales (the output already sold), against wip; a standard-cost company is refused (gate values_at_average)."""
@U.decide
def body(unit):
doc_id = f"vadj:{adjustment_ref}"
prior = unit.repo.get("inv_document", document_id=doc_id)
if prior:
return {**json.loads(prior["extra"]), "replay": True}
if source_kind == "production_settlement":
return _settle_production(unit, caller, doc_id, adjustment_ref, product_id, amount, production_order_id, company_code, on)
if not receipt_id:
raise BadInput("receipt_id names the receipt the cost belongs to")
receipt_moves = [m for m in unit.repo.find("inv_move", document_id=f"rcpt:pur:{receipt_id}", product_id=product_id)
if m["from_location_id"].endswith(":SUPPLIER")]
if not receipt_moves:
raise Refused(404, f"no received line of {product_id} on receipt {receipt_id}")
company = receipt_moves[0]["company_code"]
received = sum((m["quantity_base"] for m in receipt_moves), ZERO)
later = sum((m["quantity_base"] for m in unit.repo.find("inv_move", product_id=product_id, company_code=company)
if m["from_location_id"].endswith(":SUPPLIER") and m["moved_at"] > max(r["moved_at"] for r in receipt_moves)), ZERO)
ledger.catch_up(unit, company, product_id, unit.today)
pos = ledger.position(unit, company, product_id)
remaining = min(received, max(pos.quantity - later, ZERO))
method = unit.settings.method(company, unit.today)
if method == "standard":
on_hand = ZERO
else:
on_hand = money(amount * remaining / received)
price_diff = amount - on_hand
on_date = day(on) or unit.today
if on_hand:
after = val.Position(pos.quantity, pos.value + on_hand, val.q4((pos.value + on_hand) / pos.quantity) if pos.quantity > 0 else pos.average, pos.seq + 1)
ledger.append_valuation(unit, company, product_id, val.Step(on_hand, after), kind="adjustment", method=method, move_id="",
quantity_change=ZERO, moved_on=on_date)
vclass = unit.product(product_id).get("valuation_class") or "*"
postings.stage(unit, doc_id, "revaluation", company, on_date, f"delivery cost {adjustment_ref}",
[postings.entry("stock_valuation", on_hand, valuation_class=vclass), postings.entry("price_difference", price_diff),
postings.entry("stock_received_not_billed", amount, natural=postings.CREDIT)])
out = {"adjustment_ref": adjustment_ref, "on_hand_amount": f"{on_hand:.2f}", "price_difference_amount": f"{price_diff:.2f}"}
loc = unit.repo.get("inv_location", location_id=receipt_moves[0]["to_location_id"])
from brick_inventory.application import docs
docs.stage_document(unit, doc_id, "revaluation", loc["warehouse_id"], company, lifecycle="done", done_on=on_date, source_kind=source_kind,
source_ref=receipt_id, extra=docs.dump(out), done_by=U.who(caller))
return {**out, "replay": False}
return U.run(caller, body)
def _settle_production(unit, caller: dict, doc_id: str, adjustment_ref: str, product_id: str, amount, production_order_id: Optional[str],
company_code: Optional[str], on: Optional[str]) -> dict:
"""INV-57: see post_valuation_adjustment. Spread by what is still on hand of the order's output (older stock is assumed to leave first)."""
from brick_inventory.application import docs
if not production_order_id:
raise BadInput("production_order_id names the production order being settled")
outputs = [m for m in unit.repo.find("inv_move", product_id=product_id)
if m["document_id"].startswith("rcpt:src:production_receipt:") and (m["document_id"].endswith(f":{production_order_id}"))]
if not outputs:
raise Refused(404, f"no output of {product_id} received for production order {production_order_id}")
company = company_code or outputs[0]["company_code"]
ledger.catch_up(unit, company, product_id, unit.today)
method = unit.settings.method(company, unit.today)
if method != "average":
raise GateBroken("values_at_average", f"{company} values stock at {method} cost: manufacturing posts its variances to the ledger itself",
"settle through manufacturing's own journal (MFG-15)", 422)
received = sum((m["quantity_base"] for m in outputs), ZERO)
later = sum((m["quantity_base"] for m in unit.repo.find("inv_move", product_id=product_id, company_code=company)
if m["document_id"].startswith("rcpt:src:production_receipt:") and m["moved_at"] > max(r["moved_at"] for r in outputs)), ZERO)
pos = ledger.position(unit, company, product_id)
remaining = min(received, max(pos.quantity - later, ZERO))
on_hand = money(amount * remaining / received)
to_cos = amount - on_hand
on_date = day(on) or unit.today
if on_hand:
after = val.Position(pos.quantity, pos.value + on_hand, val.q4((pos.value + on_hand) / pos.quantity) if pos.quantity > 0 else pos.average, pos.seq + 1)
ledger.append_valuation(unit, company, product_id, val.Step(on_hand, after), kind="adjustment", method=method, move_id="",
quantity_change=ZERO, moved_on=on_date)
vclass = unit.product(product_id).get("valuation_class") or "*"
postings.stage(unit, doc_id, "revaluation", company, on_date, f"production settlement {adjustment_ref}",
[postings.entry("stock_valuation", on_hand, valuation_class=vclass), postings.entry("cost_of_sales", to_cos),
postings.entry("wip", amount, natural=postings.CREDIT)])
out = {"adjustment_ref": adjustment_ref, "on_hand_amount": f"{on_hand:.2f}", "cost_of_sales_amount": f"{to_cos:.2f}"}
loc = unit.repo.get("inv_location", location_id=outputs[0]["to_location_id"])
docs.stage_document(unit, doc_id, "revaluation", loc["warehouse_id"], company, lifecycle="done", done_on=on_date, source_kind="production_settlement",
source_ref=production_order_id, extra=docs.dump(out), done_by=U.who(caller))
return {**out, "replay": False}