dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
src/brick_manufacturing/application/ai_step.py
"""explain_variance (ai_steps.explain_variance): the likely cause of an order's variance, suggested to the cost accountant, who decides.
Consent and control: the owner holds the mode (off | suggest). A suggestion is only ever a row in mfg_ai_finding - the model, a digest of what it was
shown, the causes with their evidence, and then who accepted or dismissed it and when. Nothing is posted or changed by the step; closing an order
never waits for it. With the step off, no provider, or no usable answer, the response is the variance split alone."""
import json
from decimal import Decimal
from typing import Literal
from erp_platform import work
from erp_platform.bricks import Refused
from brick_manufacturing.application import use_cases as hub
from brick_manufacturing.application.inputs import Code
from brick_manufacturing.domain import ai
from brick_manufacturing.domain.errors import RuleBroken
from brick_manufacturing.infrastructure import ai_steps, seams
MFG = hub.MFG
_s = hub._s
REPORT = "mfg:report:read"
def _split(f) -> dict:
return {"wip_debit": _s(f.wip_debit), "wip_credit": _s(f.wip_credit), "balance": _s(f.balance), "standard_material": _s(f.standard_material),
"abnormal_scrap": _s(f.abnormal_scrap), "material_usage_variance": _s(f.material_usage_variance), "activity_variance": _s(f.activity_variance)}
def _context(repo, order, movements: list, confirmations: list, scraps: list, figures) -> tuple:
"""The step's inputs, as JSON scalars (the provider reads nothing itself): this order, and the same product's last 20 closed orders."""
planned = {c["line_no"]: c for c in order.planned_components()}
issued = hub._net_issued(movements)
out_base = sum((m["quantity_base"] for m in hub._live(movements, "output")), Decimal("0"))
one = next(iter(issued.values()), Decimal("0")) if len(planned) == 1 else None
lots = hub._net_lots(movements)
suppliers = {}
for (pid, lot) in sorted(lots):
history = (seams.lot_trace(pid, lot) or {}).get("history", [])
suppliers[f"{pid}/{lot}"] = sorted({h["bp_id"] for h in history if h.get("bp_id")})
recent = repo.ai.recent_settlements(order.product_id, order.production_order_id)
ctx = {"production_order_id": order.production_order_id, "product_id": order.product_id, "planned_quantity": _s(order.planned_quantity),
"planned_yield_pct": order.frozen.get("planned_yield_pct"),
"actual_yield_pct": _s(hub.q2(out_base / one * 100)) if one and out_base else None, "variance": _split(figures),
"movements": [{"kind": m["kind"], "product_id": m["product_id"], "lot_code": m["lot_code"], "quantity_base": _s(m["quantity_base"]),
"over_issue": bool(m["over_issue"]), "status": m["status"], "business_date": m["business_date"].isoformat()} for m in movements],
"planned_components": [{"product_id": c["product_id"], "planned_base": _s(c["planned_base"]), "issued_base": _s(issued.get(c["line_no"], 0))}
for c in planned.values()],
"confirmations": [{"op_no": c["op_no"], "machine_minutes": _s(c["machine_minutes"]), "labour_minutes": _s(c["labour_minutes"]),
"reverses": c["reverses_confirmation_no"] or None, "status": c["status"]} for c in confirmations],
"planned_operations": [{"op_no": o["op_no"], "run_minutes": o["run_minutes"], "per_quantity": o["per_quantity"]}
for o in order.frozen_routing.get("operations", [])],
"scrap": [{"quantity_base": _s(x["quantity_base"]), "reason_code": x["reason_code"], "status": x["status"]} for x in scraps],
"input_lot_suppliers": suppliers,
"last_orders": [{"production_order_id": r["production_order_id"], "planned_quantity": _s(r["planned_quantity"]),
"material_usage_variance": _s(r["material_usage_variance"]), "activity_variance": _s(r["activity_variance"]),
"abnormal_scrap": _s(r["abnormal_scrap"])} for r in recent]}
return ctx, ai.digest(ctx)
@MFG.use_case(kind="command", route="POST /production-orders/{production_order_id}/explain-variance", permission=REPORT, status=201)
def explain_variance(caller: dict, production_order_id: Code) -> dict:
"""Ask for the likely cause of an order's variance (a lower yield on one lot, a longer roast, an over-issue not reversed): up to three
causes with their evidence, suggested to you - nothing is posted or changed, and you accept or dismiss each finding. With the step off (the
owner's setting), no explanation service wired, or no usable answer, you get the variance split alone. Every finding is audited: the model, a
digest of what it was shown, the finding, your decision, who and when."""
with hub.repository() as repo, hub._decide():
order, _, _ = hub._see_order(repo, caller, production_order_id)
if order.status not in ("completed", "closed"):
raise RuleBroken(f"Order {production_order_id} is {order.status}: a variance exists once it is completed", gate="state")
mode = repo.settings.all(hub._today(repo))["ai_explain_variance_mode"]
movements = repo.movements.for_order(production_order_id)
confirmations, scraps = repo.floor.confirmations(production_order_id), repo.floor.scraps(production_order_id)
figures = hub._figures(order, movements, repo.floor.activity_cost(production_order_id), scraps)
base = {"production_order_id": production_order_id, "mode": mode, "variance": _split(figures), "explained": False, "causes": []}
if mode != "suggest":
return {**base, "reason": "the step is off: the variance split is all there is"}
ctx, digest = _context(repo, order, movements, confirmations, scraps, figures)
got, why = ai_steps.explain(production_order_id, digest, ctx)
if got is None:
return {**base, "reason": why, "inputs_digest": digest}
model, suggestion, rationale, confidence, sources = got
causes = ai.causes(suggestion)
if not causes:
return {**base, "reason": "the explanation service gave no usable cause", "inputs_digest": digest}
with hub.repository() as repo, hub._decide():
n = len(repo.ai.findings(production_order_id)) + 1
repo.ai.add({"production_order_id": production_order_id, "finding_no": n, "step_key": ai.STEP, "model": model, "inputs_digest": digest,
"finding_json": hub.dump({"causes": causes, "rationale": rationale[:1000], "confidence": confidence, "sources": sources[:20]}),
"decision": "pending", "requested_by": hub._email(caller), "requested_sub": hub._sub(caller), "decided_by": "", "decided_sub": ""},
work.now())
return {**base, "explained": True, "causes": causes, "finding_no": n, "model": model, "confidence": confidence, "rationale": rationale,
"inputs_digest": digest, "decision": "pending"}
def _finding_view(r: dict) -> dict:
f = json.loads(r["finding_json"] or "{}")
return {"finding_no": r["finding_no"], "step": r["step_key"], "model": r["model"], "inputs_digest": r["inputs_digest"], "causes": f.get("causes", []),
"rationale": f.get("rationale", ""), "confidence": f.get("confidence"), "decision": r["decision"], "requested_by": r["requested_by"],
"decided_by": r["decided_by"] or None, "decided_at": r["decided_at"].isoformat() if r["decided_at"] else None}
@MFG.use_case(kind="query", route="GET /production-orders/{production_order_id}/variance-findings", permission=REPORT)
def list_variance_findings(caller: dict, production_order_id: Code) -> dict:
"""The audit of the explain_variance step on an order: each finding with its model, the digest of what it saw, its causes and the decision."""
with hub.repository() as repo, hub._decide():
hub._see_order(repo, caller, production_order_id)
return {"production_order_id": production_order_id, "findings": [_finding_view(r) for r in repo.ai.findings(production_order_id)]}
@MFG.use_case(kind="command", route="POST /production-orders/{production_order_id}/variance-findings/{finding_no}/decide", permission=REPORT)
def decide_variance_finding(caller: dict, production_order_id: Code, finding_no: int, decision: Literal["accepted", "dismissed"]) -> dict:
"""Accept or dismiss a suggested explanation, once: the decision, who and when are recorded. Accepting changes nothing about the order or its
variance - it records that a person agrees with the cause."""
with hub.repository() as repo, hub._decide():
hub._see_order(repo, caller, production_order_id)
row = next((r for r in repo.ai.findings(production_order_id) if r["finding_no"] == finding_no), None)
if row is None:
raise Refused(404, f"no finding {finding_no} on {production_order_id}")
if row["decision"] != "pending":
raise RuleBroken(f"Finding {finding_no} is already {row['decision']}: a decision is not changed", gate="state")
repo.ai.decide(production_order_id, finding_no, decision, hub._email(caller), hub._sub(caller), work.now())
return _finding_view(next(r for r in repo.ai.findings(production_order_id) if r["finding_no"] == finding_no))