dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
src/brick_manufacturing/domain/bom.py
"""The Bom aggregate: what goes into one product (usage production) or what one sold product consumes (usage sale).
Rules here are true whoever calls and however it is stored. A version is written as a draft, released once (never edited
again: a change is the next version), and made obsolete with a last valid day. Validity is COMPUTED, never stored twice: a
released version is valid from its valid_from until the day before the next released version starts (or its own valid_to),
so two versions of one BOM can never overlap. Quantities arrive already converted EXACTLY to the product's base unit by
product (the use case asks it); this module never converts.
"""
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import Optional
from brick_manufacturing.domain.errors import OutOfBounds, RuleBroken
from brick_manufacturing.domain.values import BomState, BomUsage, LineKind, q4, quantity
@dataclass
class BomLine:
version: int
line_no: int
kind: str
product_id: str
quantity: Decimal # in unit_code (the product's production unit)
unit_code: str
quantity_base: Decimal # the same, converted exactly by product
base_unit_code: str
replaces_product_id: str = "" # a modifier line: the component it replaces
modifier_id: str = "" # a modifier line: what the till names (MOD-OAT)
nrv_unit_cost: Decimal = Decimal("0") # a by-product: net realisable value per base unit (IAS 2.14)
@dataclass
class BomVersion:
version: int
valid_from: date
output_quantity: Decimal
output_unit_code: str
output_quantity_base: Decimal
output_base_unit_code: str
planned_yield_pct: Optional[Decimal] = None
valid_to: Optional[date] = None
status: str = BomState.DRAFT.value
released_by: str = ""
released_sub: str = ""
@dataclass
class Bom:
bom_id: str # "<product_id>:<usage>[:<alternative>]"
product_id: str = ""
usage: str = BomUsage.PRODUCTION.value
alternative: str = ""
company_code: str = ""
versions: list = field(default_factory=list)
lines: list = field(default_factory=list)
# ── identity ────────────────────────────────────────────────────────────────────────────────────────────
@staticmethod
def make_id(product_id: str, usage: str, alternative: str = "") -> str:
return f"{product_id}:{usage}" + (f":{alternative}" if alternative else "")
# ── lookups ─────────────────────────────────────────────────────────────────────────────────────────────
def version_of(self, n: int) -> Optional[BomVersion]:
return next((v for v in self.versions if v.version == n), None)
def lines_of(self, n: int) -> list:
return sorted((x for x in self.lines if x.version == n), key=lambda x: x.line_no)
def valid_on(self, day: date) -> Optional[BomVersion]:
"""The released version in force on `day`: the latest released one that has started, unless it was made
obsolete before `day`. None when nothing is."""
started = [v for v in self.versions
if v.status in (BomState.RELEASED.value, BomState.OBSOLETE.value) and v.valid_from <= day]
if not started:
return None
cur = max(started, key=lambda v: v.valid_from)
if cur.valid_to is not None and day > cur.valid_to:
return None
return cur
def components_of(self, n: int) -> list:
return [x.product_id for x in self.lines_of(n) if x.kind == LineKind.COMPONENT.value]
# ── write a version (MFG-02) ────────────────────────────────────────────────────────────────────────────
def write_version(self, n: int, *, valid_from: date, output_quantity: Decimal, output_unit_code: str,
output_quantity_base: Decimal, output_base_unit_code: str, lines: list, edges: dict) -> BomVersion:
"""Write (or rewrite, while it is still a draft) version `n`. `lines` are BomLine drafts; `edges` is every OTHER
production BOM's product -> components walk, for the cycle check."""
have = self.version_of(n)
if have is not None and have.status != BomState.DRAFT.value:
raise RuleBroken(f"Version {n} of {self.bom_id} is {have.status}: a released version is never edited - write version "
f"{max(v.version for v in self.versions) + 1}", gate="version_frozen")
if have is None and n != max((v.version for v in self.versions), default=0) + 1:
raise RuleBroken(f"The next version of {self.bom_id} is {max((v.version for v in self.versions), default=0) + 1}, not {n}",
gate="version_number")
quantity(output_quantity, "the output quantity")
comps = [x for x in lines if x.kind == LineKind.COMPONENT.value]
if not comps:
raise RuleBroken("A bill of materials needs at least one component", gate="components")
nos = [x.line_no for x in lines]
if len(set(nos)) != len(nos):
raise RuleBroken("Two lines share a line number")
for x in lines:
quantity(x.quantity, f"the quantity of line {x.line_no}")
if x.kind == LineKind.BY_PRODUCT.value and self.usage == BomUsage.SALE.value:
raise RuleBroken("A usage-sale BOM has no by-products: nothing is produced", gate="usage")
if x.kind == LineKind.MODIFIER.value:
if self.usage != BomUsage.SALE.value:
raise RuleBroken("A modifier belongs to a usage-sale BOM", gate="usage")
if not x.modifier_id or x.replaces_product_id not in {c.product_id for c in comps}:
raise RuleBroken(f"Modifier line {x.line_no} must name its modifier and a component of this BOM it replaces",
gate="modifier")
if x.product_id == self.product_id:
raise RuleBroken(f"{self.product_id} cannot be a component or by-product of itself", gate="cycle")
if self.usage == BomUsage.PRODUCTION.value:
self._no_cycle([x.product_id for x in comps], edges)
pct = None
if len(comps) == 1 and comps[0].base_unit_code == output_base_unit_code:
pct = q4(output_quantity_base / comps[0].quantity_base * 100)
v = BomVersion(n, valid_from, output_quantity, output_unit_code, output_quantity_base, output_base_unit_code, pct)
self.versions = [x for x in self.versions if x.version != n] + [v]
self.lines = [x for x in self.lines if x.version != n] + [BomLine(**{**x.__dict__, "version": n}) for x in lines]
return v
def _no_cycle(self, components: list, edges: dict) -> None:
"""A walk down from each new component: reaching this BOM's own product is a cycle (direct or through another BOM)."""
seen, stack = set(), list(components)
while stack:
p = stack.pop()
if p == self.product_id:
raise RuleBroken(f"{self.product_id} would contain itself, through another bill of materials", gate="cycle")
if p in seen:
continue
seen.add(p)
stack.extend(edges.get(p, ()))
# ── the lifecycle (module.yaml workflows: bom_version) ─────────────────────────────────────────────────
def release_bom(self, n: int, *, today: date, by: str, sub: str) -> BomVersion:
v = self._need(n)
if v.status != BomState.DRAFT.value:
raise RuleBroken(f"Version {n} of {self.bom_id} is {v.status}; only a draft is released")
if v.valid_from <= today:
raise RuleBroken(f"Version {n} is valid from {v.valid_from}, which is not after today ({today}): a release is for a "
"later day, so no order already planned changes under anyone", gate="valid_from")
latest = max((x.valid_from for x in self.versions if x.status != BomState.DRAFT.value), default=None)
if latest is not None and v.valid_from <= latest:
raise RuleBroken(f"Version {n} would overlap an earlier released version (valid from {latest}): start it later",
gate="overlap")
v.status, v.released_by, v.released_sub = BomState.RELEASED.value, by, sub
return v
def obsolete_bom(self, n: int, *, valid_to: date) -> BomVersion:
v = self._need(n)
if v.status != BomState.RELEASED.value:
raise RuleBroken(f"Version {n} of {self.bom_id} is {v.status}; only a released version is made obsolete")
if valid_to < v.valid_from:
raise RuleBroken("valid_to is before valid_from", gate="valid_to")
v.status, v.valid_to = BomState.OBSOLETE.value, valid_to
return v
def _need(self, n: int) -> BomVersion:
v = self.version_of(n)
if v is None:
raise RuleBroken(f"{self.bom_id} has no version {n}")
return v
# ── explode (MFG-30): deterministic for the same query ─────────────────────────────────────────────────
def explode(self, qty: Decimal, on: date, modifiers: tuple = ()) -> tuple:
"""(version, [(product_id, base quantity, base unit)]) for `qty` of the output (in its output unit), on `on`."""
v = self.valid_on(on)
if v is None:
raise LookupError(self.bom_id)
ratio = quantity(qty) / v.output_quantity
mods = {x.modifier_id: x for x in self.lines_of(v.version) if x.kind == LineKind.MODIFIER.value}
for m in modifiers:
if m not in mods:
raise OutOfBounds(f"{self.bom_id} has no modifier {m}")
replaced = {mods[m].replaces_product_id: mods[m] for m in modifiers}
out = []
for x in self.lines_of(v.version):
if x.kind != LineKind.COMPONENT.value:
continue
use = replaced.get(x.product_id, x)
out.append((use.product_id, q4(use.quantity_base * ratio), use.base_unit_code))
return v, out