dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
src/brick_manufacturing/domain/production_order.py
"""The ProductionOrder aggregate: an instruction to make a quantity of a product by a date, and what happened to it.
The state machine only moves forward: created -> released -> in_progress -> completed -> closed, with cancelled reachable
from created and released only. Stock and money never move here: the order decides, the use case records the movement and a
pending outbox row, and a neighbour (inventory, gl) is told only after the local commit.
"""
import json
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Optional
from brick_manufacturing.domain.errors import RuleBroken
from brick_manufacturing.domain.values import OrderState, q4, quantity
@dataclass
class ProductionOrder:
production_order_id: str
company_code: str = ""
product_id: str = ""
warehouse_id: str = ""
planned_quantity: Decimal = Decimal("0")
unit_code: str = ""
due_on: Optional[date] = None
bom_id: str = ""
bom_version: int = 0
frozen_bom_json: str = "" # the BOM version copied at release; a later BOM change never reaches it
status: str = OrderState.CREATED.value
created_by: str = ""
released_on: Optional[date] = None
released_by: str = ""
shortage_warning: str = ""
completed_on: Optional[date] = None
closed_on: Optional[date] = None
cancel_reason: str = ""
routing_version: int = 0
frozen_routing_json: str = "" # the routing version copied at release (optional: a product may have none)
planned_order_ref: str = "" # the MRP-lite proposal this order was converted from
stale_note: str = "" # an unreleased order only: a newer BOM version is now valid (MFG-18)
# ── creation ────────────────────────────────────────────────────────────────────────────────────────────
@classmethod
def create_order(cls, production_order_id: str, *, company_code: str, product_id: str, warehouse_id: str,
planned_quantity: Decimal, unit_code: str, due_on: date, bom_id: str, bom_version: int, by: str):
quantity(planned_quantity, "the planned quantity")
return cls(production_order_id, company_code, product_id, warehouse_id, planned_quantity, unit_code, due_on,
bom_id, bom_version, created_by=by)
# ── the frozen copy ─────────────────────────────────────────────────────────────────────────────────────
@property
def frozen(self) -> dict:
return json.loads(self.frozen_bom_json) if self.frozen_bom_json else {}
@property
def frozen_routing(self) -> dict:
return json.loads(self.frozen_routing_json) if self.frozen_routing_json else {}
def scale(self) -> Decimal:
"""planned quantity / the BOM's output quantity: what each BOM line is multiplied by for THIS order."""
return self.planned_quantity / Decimal(self.frozen["output_quantity"])
def planned_components(self) -> list:
"""[{line_no, product_id, planned_base, base_unit_code, line_quantity, unit_code}] for the frozen version."""
k = self.scale()
return [{"line_no": x["line_no"], "product_id": x["product_id"], "planned_base": q4(Decimal(x["quantity_base"]) * k),
"base_unit_code": x["base_unit_code"], "unit_code": x["unit_code"],
"base_per_unit": Decimal(x["quantity_base"]) / Decimal(x["quantity"])}
for x in self.frozen["lines"] if x["kind"] == "component"]
def planned_output_base(self) -> Decimal:
return q4(Decimal(self.frozen["output_quantity_base"]) * self.scale())
# ── the lifecycle (module.yaml workflows: production_order) ────────────────────────────────────────────
def release_order(self, *, today: date, frozen: dict, shortages: list, allow_shortage: bool, by: str,
routing: Optional[dict] = None) -> None:
if self.status != OrderState.CREATED.value:
raise RuleBroken(f"Order {self.production_order_id} is {self.status}: only a created order is released "
"(a released order never goes back)", gate="state")
if shortages and not allow_shortage:
raise RuleBroken("Components are short: " + "; ".join(shortages) + ". Release is refused", gate="components_available",
fix="wait for the receipt, or the owner allows release with shortages")
self.frozen_bom_json = json.dumps(frozen, sort_keys=True)
self.bom_version = frozen["version"]
self.status, self.released_on, self.released_by = OrderState.RELEASED.value, today, by
self.shortage_warning = "; ".join(shortages)
self.stale_note = ""
if routing:
self.frozen_routing_json, self.routing_version = json.dumps(routing, sort_keys=True), routing["routing_version"]
def start_work(self) -> None:
"""An issue or an output on a released order puts it in progress; a completed, closed or cancelled one takes none."""
if self.status not in (OrderState.RELEASED.value, OrderState.IN_PROGRESS.value):
raise RuleBroken(f"Order {self.production_order_id} is {self.status}: nothing more is issued or reported on it",
gate="state")
self.status = OrderState.IN_PROGRESS.value
def complete_order(self, *, today: date) -> None:
if self.status != OrderState.IN_PROGRESS.value:
raise RuleBroken(f"Order {self.production_order_id} is {self.status}: only an order in progress is completed", gate="state")
self.status, self.completed_on = OrderState.COMPLETED.value, today
def close_order(self, *, today: date) -> None:
if self.status != OrderState.COMPLETED.value:
raise RuleBroken(f"Order {self.production_order_id} is {self.status}: only a completed order is closed", gate="state")
self.status, self.closed_on = OrderState.CLOSED.value, today
def cancel_order(self, reason: str, *, has_movements: bool) -> None:
if self.status not in (OrderState.CREATED.value, OrderState.RELEASED.value):
raise RuleBroken(f"Order {self.production_order_id} is {self.status}: only a created or released order is cancelled", gate="state")
if has_movements:
raise RuleBroken("A movement or confirmation is already recorded on this order: it cannot be cancelled", gate="no_movement")
if not reason.strip():
raise RuleBroken("A reason is required")
self.status, self.cancel_reason = OrderState.CANCELLED.value, reason.strip()
# ── issue (MFG-11): the tolerance ───────────────────────────────────────────────────────────────────────
def check_issue(self, line: dict, add_base: Decimal, issued_base: Decimal, tolerance_pct: Decimal) -> bool:
"""True when this issue takes the component above its plan (an over-issue); refuses above the tolerance."""
allowed = q4(line["planned_base"] * (Decimal(100) + tolerance_pct) / Decimal(100))
total = issued_base + add_base
if total > allowed:
raise RuleBroken(f"Issuing {add_base} {line['base_unit_code']} of {line['product_id']} takes {total} against a plan of "
f"{line['planned_base']} and a tolerance of {tolerance_pct}% ({allowed} at most)",
gate="over_issue_tolerance", fix="the owner raises the tolerance, or the planner re-plans the order")
return total > line["planned_base"]
# ── output (MFG-13): the plausibility gate ──────────────────────────────────────────────────────────────
def check_output(self, add_base: Decimal, reported_base: Decimal, plausible_pct: Decimal, planner_confirmed: bool) -> None:
cap = q4(self.planned_output_base() * plausible_pct / Decimal(100))
if reported_base + add_base > cap and not planner_confirmed:
raise RuleBroken(f"{reported_base + add_base} reported is above {plausible_pct}% of the plan "
f"({self.planned_output_base()}): that looks like a typo", gate="output_plausibility",
fix="the planner confirms it")