dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
src/brick_manufacturing/application/engineering.py
"""Work centres, routings and the standard-cost roll-up (MFG-01, MFG-03, MFG-06, MFG-07, MFG-18).
What an hour of a machine costs, how long a step takes, and what a kilo of roasted beans should therefore cost - written down by the head
roaster, rolled up by one cost accountant and released to inventory by another. Everything is dated and versioned, never edited: a new
rate is a row from a later date, a routing is the next version, a roll-up is a new run. A change to any input marks what was derived from it
stale (hub.mark_stale), so nobody releases a cost from yesterday's rates."""
import json
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform import auth, work
from erp_platform.bricks import Refused
from brick_manufacturing.application import dispatch
from brick_manufacturing.application import use_cases as hub
from brick_manufacturing.application.inputs import Code, Day, OperationIn, Qty, Rate0
from brick_manufacturing.domain import costing
from brick_manufacturing.domain import routing as rt
from brick_manufacturing.domain.bom import Bom
from brick_manufacturing.domain.errors import OutOfBounds, RuleBroken
from brick_manufacturing.domain.values import MAX_QTY
from brick_manufacturing.infrastructure import seams
MFG = hub.MFG
_s = hub._s
def _may_see_rates(caller: dict) -> bool:
return hub._costs(caller) or auth.has_permission(caller, "mfg:work_centre:write") or auth.has_permission(caller, "mfg:routing:write")
# ── work centres (MFG-01) ───────────────────────────────────────────────────────────────────────────────────
def _wc_view(row: dict, rates: list, show_rates: bool) -> dict:
out = {k: (_s(v) if isinstance(v, Decimal) else v) for k, v in row.items() if k not in ("created_at", "updated_at")}
if show_rates:
out["rates"] = [{"rate_kind": r["rate_kind"], "valid_from": r["valid_from"].isoformat(), "rate_amount": _s(r["rate_amount"]),
"currency": r["currency"]} for r in rates if r["work_centre_id"] == row["work_centre_id"]]
return out
@MFG.use_case(kind="command", route="POST /work-centres/{work_centre_id}", permission="mfg:work_centre:write", status=201)
def put_work_centre(caller: dict, work_centre_id: Code, company_code: Code, warehouse_id: Code, cost_centre: Code, batch_capacity: Qty,
capacity_unit_code: Code, hours_per_day: Qty, machine_rate: Optional[Rate0] = None,
labour_rate: Optional[Rate0] = None, overhead_rate: Optional[Rate0] = None) -> dict:
"""Write down a machine or a bench: its code (fixed once an order uses it), the warehouse, the gl cost centre its time is posted to
(gl must know it and have it open), the batch it takes and the hours it works a day, and what an hour costs from today (machine,
labour, overhead; NUMERIC(19,4) each). A later rate is a new dated row (set_work_centre_rate), never an edit."""
with hub.repository() as repo, hub._decide():
today = hub._today(repo)
if hours_per_day > 24:
raise OutOfBounds("hours_per_day is at most 24")
if repo.masters.work_centre(work_centre_id):
raise RuleBroken(f"Work centre {work_centre_id} exists: its code is fixed; a rate change is a new dated row", gate="work_centre_exists")
try:
cc = seams.cost_centre(cost_centre)
except seams.Unavailable as e:
if e.http != 404:
raise
raise RuleBroken(f"gl has no cost centre {cost_centre}", gate="cost_centre", fix="create it in gl first") from None
if cc is not None and cc.get("status") != "open":
raise RuleBroken(f"Cost centre {cost_centre} is {cc.get('status')}", gate="cost_centre", fix="use an open cost centre")
cur = repo.settings.all(today)["currency"]
repo.masters.add_work_centre({"work_centre_id": work_centre_id, "company_code": company_code, "warehouse_id": warehouse_id,
"cost_centre": cost_centre, "batch_capacity": batch_capacity,
"capacity_unit_code": capacity_unit_code, "hours_per_day": hours_per_day, "status": "active"}, work.now())
rates = [{"work_centre_id": work_centre_id, "rate_kind": k, "valid_from": today, "rate_amount": v, "currency": cur,
"set_by": hub._email(caller)} for k, v in (("machine", machine_rate), ("labour", labour_rate), ("overhead", overhead_rate))
if v is not None]
repo.masters.add_rates(rates, work.now())
return _wc_view(repo.masters.work_centre(work_centre_id), repo.masters.rates([work_centre_id]), True)
@MFG.use_case(kind="command", route="POST /work-centres/{work_centre_id}/rates", permission="mfg:work_centre:write", status=201)
def set_work_centre_rate(caller: dict, work_centre_id: Code, rate_kind: str, rate_amount: Rate0, valid_from: Optional[Day] = None) -> dict:
"""What an hour of one kind (machine, labour, overhead) costs from a date: a new dated row. The first row of a kind may start today; a
change starts after today, so a rate in force is never rewritten. Every open cost roll-up that used this work centre is marked stale."""
with hub.repository() as repo, hub._decide():
if rate_kind not in rt.RATE_KINDS:
raise Refused(422, f"rate_kind is one of {', '.join(rt.RATE_KINDS)}", gate="bounds")
today = hub._today(repo)
wc = repo.masters.work_centre(work_centre_id)
if wc is None:
raise Refused(404, f"no work centre {work_centre_id}")
start = hub._date(valid_from, "valid_from") or today
have = [r for r in repo.masters.rates([work_centre_id]) if r["rate_kind"] == rate_kind]
if start < today or (have and start <= today):
raise RuleBroken(f"A {rate_kind} rate changes from a day after today ({today}): a rate in force is never rewritten", gate="valid_from")
repo.masters.add_rates([{"work_centre_id": work_centre_id, "rate_kind": rate_kind, "valid_from": start, "rate_amount": rate_amount,
"currency": repo.settings.all(today)["currency"], "set_by": hub._email(caller)}], work.now())
hub.mark_stale(repo, f"{rate_kind} rate of {work_centre_id} changes from {start}", work_centre=work_centre_id)
return _wc_view(wc, repo.masters.rates([work_centre_id]), True)
@MFG.use_case(kind="query", route="GET /work-centres", permission="mfg:bom:read")
def list_work_centres(caller: dict) -> dict:
"""The work centres (rates shown to those who may see costs)."""
with hub.repository() as repo:
rows, rates, show = repo.masters.work_centres(), repo.masters.rates(), _may_see_rates(caller)
return {"work_centres": [_wc_view(r, rates, show) for r in rows]}
@MFG.use_case(kind="query", route="GET /work-centres/{work_centre_id}", permission="mfg:bom:read")
def get_work_centre(caller: dict, work_centre_id: Code) -> dict:
"""One work centre with its dated rates (for those who may see costs)."""
with hub.repository() as repo, hub._decide():
wc = repo.masters.work_centre(work_centre_id)
if wc is None:
raise Refused(404, f"no work centre {work_centre_id}")
return _wc_view(wc, repo.masters.rates([work_centre_id]), _may_see_rates(caller))
# ── routings (MFG-03) ───────────────────────────────────────────────────────────────────────────────────────
def _ops_json(ops: list) -> str:
return json.dumps([{"op_no": o.op_no, "description": o.description, "work_centre_id": o.work_centre_id or "",
"setup_minutes": _s(o.setup_minutes), "run_minutes": _s(o.run_minutes), "per_quantity": _s(o.per_quantity),
"basis": o.basis, "unit_code": o.unit_code or "", "wait_hours": _s(o.wait_hours)}
for o in sorted(ops, key=lambda x: x.op_no)])
def _routing_view(rows: list, today: date) -> dict:
cur = rt.valid_version(rows, today)
return {"product_id": rows[0]["product_id"] if rows else None, "valid_today": cur["routing_version"] if cur else None,
"versions": [{"routing_version": r["routing_version"], "valid_from": r["valid_from"].isoformat(),
"operations": json.loads(r["operations_json"])} for r in rows]}
@MFG.use_case(kind="command", route="POST /routings/{product_id}/versions", permission="mfg:routing:write", status=201)
def put_routing(caller: dict, product_id: Code, routing_version: int, valid_from: Day, operations: list[OperationIn]) -> dict:
"""Write a version of a product's routing: its steps on work centres (setup once, run per batch; a step with no work centre is
waiting time). Dated like a BOM: it must start after today and after every earlier version, and it is never edited - the next
version is the change. Open cost roll-ups and the last MRP run are marked stale."""
with hub.repository() as repo, hub._decide():
today = hub._today(repo)
start = hub._date(valid_from, "valid_from")
ops = [rt.Operation(o.op_no, o.description, o.work_centre_id or "", o.setup_minutes, o.run_minutes, o.per_quantity, o.basis,
o.unit_code or "", o.wait_hours) for o in operations]
rt.check_operations(ops)
for o in ops:
if o.work_centre_id and repo.masters.work_centre(o.work_centre_id) is None:
raise RuleBroken(f"Operation {o.op_no} names work centre {o.work_centre_id}, which does not exist", gate="work_centre")
have = repo.masters.routings(product_id)
if start <= today:
raise RuleBroken(f"A routing starts after today ({today}): orders already running keep the copy they froze", gate="valid_from")
if have and (start <= max(r["valid_from"] for r in have) or routing_version <= max(r["routing_version"] for r in have)):
raise RuleBroken("A new routing version starts after, and numbers above, every earlier one: versions never overlap",
gate="valid_from")
repo.masters.add_routing({"product_id": product_id, "routing_version": routing_version, "valid_from": start,
"operations_json": _ops_json(ops), "written_by": hub._email(caller)}, work.now())
hub.mark_stale(repo, f"routing of {product_id} version {routing_version} written, valid from {start}", routing_product=product_id)
return _routing_view(repo.masters.routings(product_id), today)
@MFG.use_case(kind="query", route="GET /routings/{product_id}", permission="mfg:bom:read")
def get_routing(caller: dict, product_id: Code) -> dict:
"""A product's routing versions with their operations, and which is in force today."""
with hub.repository() as repo, hub._decide():
rows = repo.masters.routings(product_id)
if not rows:
raise Refused(404, f"{product_id} has no routing")
return _routing_view(rows, hub._today(repo))
def plan_for(repo, order_like_lines: list, output_base: Decimal, scale: Decimal, ops: list, day: date, rates: Optional[list] = None) -> list:
"""Each operation planned for `scale` runs of a BOM: [{op_no, work_centre_id, batches, minutes, wait_hours, cost}]."""
rates = repo.masters.rates() if rates is None else rates # one read for a whole roll-up, not one per product
out = []
for op in ops:
basis = rt.basis_quantity(op, scale, order_like_lines, output_base)
out.append(rt.plan_operation(op, basis, rt.hourly(rates, op["work_centre_id"], day) if op["work_centre_id"] else Decimal("0")))
return out
@MFG.use_case(kind="query", route="GET /routings/{product_id}/plan", permission="mfg:bom:read")
def plan_routing(caller: dict, product_id: Code, planned_quantity: Qty, on: Optional[Day] = None) -> dict:
"""What making `planned_quantity` of a product takes on the routing and BOM in force on a date: batches, minutes and (for those who may
see costs) the cost per operation. For 100 kg of green beans into a 15 kg roaster: ceil(100/15) = 7 batches, 10 + 7 x 20 = 150
minutes; waiting time plans the date but costs nothing."""
with hub.repository() as repo, hub._decide():
day = hub._date(on, "on") or hub._today(repo)
bom, _, _ = hub._need_bom(repo, Bom.make_id(product_id, "production"))
v = bom.valid_on(day)
row = rt.valid_version(repo.masters.routings(product_id), day)
if v is None or row is None:
raise RuleBroken(f"{product_id} has no BOM and routing both valid on {day}", gate="routing_valid")
ops = json.loads(row["operations_json"])
lines = [x.__dict__ for x in bom.lines_of(v.version)]
planned = plan_for(repo, lines, v.output_quantity_base, planned_quantity / v.output_quantity, ops, day)
show = _may_see_rates(caller)
total = sum((p["cost"] for p in planned), Decimal("0"))
return {"product_id": product_id, "routing_version": row["routing_version"], "bom_version": v.version, "on": day.isoformat(),
"planned_quantity": _s(planned_quantity), "lead_days": rt.lead_days(planned, ops),
"operations": [{**{k: x for k, x in p.items() if k != "cost"}, **({"cost": _s(hub.q2(p["cost"]))} if show else {})} for p in planned],
**({"total_cost": _s(hub.q2(total))} if show else {})}
# ── the cost roll-up (MFG-06, MFG-07) ───────────────────────────────────────────────────────────────────────
def _rollup_view(repo, row: dict) -> dict:
out = {k: (v.isoformat() if isinstance(v, date) else v) for k, v in row.items() if k not in ("created_at", "updated_at", "used_json")}
out["used"] = json.loads(row["used_json"] or "{}")
out["lines"] = [{k: (_s(v) if isinstance(v, Decimal) else v) for k, v in ln.items() if k not in ("created_at", "updated_at", "rollup_id")}
for ln in repo.costs.lines(row["rollup_id"])]
return out
@MFG.use_case(kind="command", route="POST /cost-rollups", permission="mfg:cost:run", status=201)
def run_cost_rollup(caller: dict, company_code: Code, effective_on: Day) -> dict:
"""Roll up the standard cost of every production BOM of the company, as it stands on the effective date: (components at their standard
price + the routing's activity at the work centre rates - the by-products' value) / the output, divided once at the end; a made
component costs what its own roll-up says. A component with no price stops the run with the product named. The result is a PROPOSAL
(computed): a second cost accountant approves it and it is released to inventory (approve_cost_rollup)."""
with hub.repository() as repo, hub._decide():
eff = hub._date(effective_on, "effective_on")
info: dict = {}
for bid in repo.boms.ids():
parts = bid.split(":")
if len(parts) != 2 or parts[1] != "production":
continue
bom, _, _ = repo.boms.get(bid)
v = bom.valid_on(eff)
if bom.company_code == company_code and v is not None:
info[bom.product_id] = (bom, v, [x.__dict__ for x in bom.lines_of(v.version)])
if not info:
raise RuleBroken(f"No production BOM of {company_code} is valid on {eff}", gate="bom_valid", fix="release a BOM version first")
rates, results, prices, used = repo.masters.rates(), {}, {}, {"boms": [], "work_centres": set(), "routings": set()}
def stocked(pid: str) -> Decimal:
if pid not in prices:
price = seams.unit_cost(pid, company_code, eff.isoformat())
if price is None:
raise RuleBroken(f"{pid} has no standard price in inventory: the roll-up cannot cost what uses it", gate="standard_price",
fix="release a standard price for it, or give it a BOM")
prices[pid] = price
return prices[pid]
def cost(pid: str, path: tuple = ()) -> Decimal:
if pid not in info:
return stocked(pid)
if pid in path:
raise RuleBroken(f"{pid} is its own component", gate="cycle")
if pid not in results:
bom, v, lines = info[pid]
for x in lines:
if x["kind"] == "component":
cost(x["product_id"], path + (pid,))
routing = rt.valid_version(repo.masters.routings(pid), eff)
conversion = Decimal("0")
if routing:
used["routings"].add(pid)
for op in json.loads(routing["operations_json"]):
if op["work_centre_id"]:
used["work_centres"].add(op["work_centre_id"])
conversion = sum((p["cost"] for p in plan_for(repo, lines, v.output_quantity_base, Decimal(1),
json.loads(routing["operations_json"]), eff, rates)), Decimal("0"))
got = costing.cost_of(lines, v.output_quantity_base, lambda p: cost(p, path + (pid,)), conversion)
if got["total"] > MAX_QTY or got["total"] < 0:
raise OutOfBounds(f"The cost of {pid} ({got['total']}) is not a cost a price column can hold")
results[pid] = {**got, "bom_id": bom.bom_id, "bom_version": v.version, "base_unit_code": v.output_base_unit_code}
used["boms"].append(bom.bom_id)
return results[pid]["total"]
for pid in sorted(info):
cost(pid)
base = f"{company_code}-CR-{eff.year}-{eff.month:02d}"
n = repo.claims.counts("rollup", [base])[base] + 1
rid = base if n == 1 else f"{base}-{n}"
repo.claims.take([("rollup", base, n, rid)], hub._email(caller), work.now())
repo.costs.add({"rollup_id": rid, "company_code": company_code, "effective_on": eff, "status": "computed",
"ran_by": hub._email(caller), "ran_sub": hub._sub(caller), "approved_by": "", "approved_sub": "",
"stale_cause": "", "used_json": hub.dump({"boms": sorted(set(used["boms"])), "work_centres": sorted(used["work_centres"]),
"routings": sorted(used["routings"])})},
[{"rollup_id": rid, "product_id": pid, "bom_id": r["bom_id"], "bom_version": r["bom_version"],
"base_unit_code": r["base_unit_code"], "material": r["material"], "conversion": r["conversion"],
"by_product_credit": r["by_product_credit"], "total": r["total"]} for pid, r in sorted(results.items())], work.now())
return _rollup_view(repo, repo.costs.get(rid))
@MFG.use_case(kind="query", route="GET /cost-rollups", permission="mfg:report:read")
def list_cost_rollups(caller: dict) -> dict:
"""The cost roll-ups by id."""
with hub.repository() as repo:
return {"cost_rollups": repo.costs.ids()}
@MFG.use_case(kind="query", route="GET /cost-rollups/{rollup_id}", permission="mfg:report:read")
def get_cost_rollup(caller: dict, rollup_id: Code) -> dict:
"""One roll-up: status (computed, stale, released), who ran and approved it, and each product's material, conversion and total per base unit."""
with hub.repository() as repo, hub._decide():
row = repo.costs.get(rollup_id)
if row is None:
raise Refused(404, f"no cost roll-up {rollup_id}")
return _rollup_view(repo, row)
@MFG.use_case(kind="command", route="POST /cost-rollups/{rollup_id}/approve", permission="mfg:cost:approve")
def approve_cost_rollup(caller: dict, rollup_id: Code) -> dict:
"""A SECOND cost accountant approves a roll-up and releases it: each product's standard goes to inventory as a dated standard price
from the effective date (sent after the commit from a pending row; inventory revalues stock on hand that day). Refused for whoever
ran it (email or sub), for a stale roll-up, and once released it never changes - the next is a new run."""
with hub.repository() as repo, hub._decide():
row = repo.costs.get(rollup_id)
if row is None:
raise Refused(404, f"no cost roll-up {rollup_id}")
today = hub._today(repo)
if row["status"] == "stale":
raise RuleBroken(f"{rollup_id} is stale: {row['stale_cause']}", gate="stale", fix="run a new roll-up")
if row["status"] != "computed":
raise RuleBroken(f"{rollup_id} is {row['status']}: only a computed roll-up is approved", gate="state")
if hub._same_person(caller, row["ran_by"], row["ran_sub"]):
raise RuleBroken("You ran this roll-up yourself: a second person must approve what you computed", gate="second_person",
fix="ask another cost accountant")
if row["effective_on"] <= today:
raise RuleBroken(f"A standard price starts after today ({today}); this roll-up is effective {row['effective_on']}",
gate="starts_after_today", fix="run a new roll-up for a later date")
repo.costs.release(rollup_id, hub._email(caller), hub._sub(caller), today)
doc = f"rollup:{rollup_id}"
repo.outbox.add([{"target": "inventory.standard_price", "ref": f"mfg:{rollup_id}:{ln['product_id']}", "production_order_id": doc,
"payload": hub.dump({"company_code": row["company_code"], "product_id": ln["product_id"],
"valid_from": row["effective_on"].isoformat(), "standard_price": _s(ln["total"]),
"source_ref": f"mfg:{rollup_id}", "released_by": hub._email(caller),
"released_sub": hub._sub(caller)})} for ln in repo.costs.lines(rollup_id)], work.now())
sent = dispatch.send(hub.repository, doc)
with hub.repository() as repo, hub._decide():
return {**_rollup_view(repo, repo.costs.get(rollup_id)), "sent": sent,
"outbox": [{"ref": r["ref"], "status": r["status"], "last_error": r["last_error"]} for r in repo.outbox.for_order(doc)]}