dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
src/brick_manufacturing/application/use_cases.py
"""The use cases - what a person (or their agent) can do here. Slice A: bills of materials and the production order workflow.
Each is declared ONCE and served as an API route and as an MCP tool. Two things are decided here and nowhere else: WHO MAY
(the `permission` on the declaration) and WHICH ROWS (`_see_order`, from the caller: an order the caller may not see is a 404,
never a 403). Stock and money never move in a use case: a movement and a PENDING outbox row commit together, and
application/dispatch.py tells inventory and gl afterwards.
Row visibility (module.yaml): operators see released orders (of their work centres); everyone else with mfg:order:read sees every
order; costs need mfg:report:read or the cost accountant's role, and the shop-floor view never returns cost fields.
"""
import json
from contextlib import contextmanager
from datetime import date
from decimal import Decimal
from typing import Literal, Optional
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from erp_platform import auth, work
from erp_platform.bricks import Brick, Refused
from erp_platform.seam import derived_id
from erp_platform.types import text
from sqlalchemy.exc import IntegrityError
from brick_manufacturing.application import dispatch
from brick_manufacturing.application.inputs import BomLineIn, ByProductIn, Code, Day, IssueLineIn, Qty, Reason
from brick_manufacturing.domain.bom import Bom, BomLine
from brick_manufacturing.domain.errors import OutOfBounds, RuleBroken
from brick_manufacturing.domain.production_order import ProductionOrder
from brick_manufacturing.domain.settlement import settle
from brick_manufacturing.domain.values import BomUsage, LineKind, OrderState, q2, q4
from brick_manufacturing.infrastructure import seams
from brick_manufacturing.infrastructure.tables import SETTING_DEFS, dump
EXPLODE, READ_REQUIREMENTS, PLAN_DERIVE, OUTBOX_DISPATCH, GENEALOGY, NUMBER_SERIES, MIG_LOAD, MIG_RECONCILE = (
"mfg:bom:explode", "mfg:plan:read_requirements", "mfg:plan:derive", "mfg:outbox:dispatch", "mfg:genealogy:derive",
"mfg:number_series:claim", "mfg:migration:load", "mfg:migration:reconcile")
CONTROLS = "mfg:settings:controls"
MIG_READ, MIG_SIGN = "mfg:migration:read", "mfg:migration:sign"
# Whoever holds one of these sees every order; an operator (none of them) sees only released and in-progress ones.
SEES_ALL = ("mfg:order:create", "mfg:order:release", "mfg:order:close", "mfg:order:complete", "mfg:report:read", "mfg:bom:write",
"mfg:trace:read", CONTROLS)
COSTS = ("mfg:report:read", "mfg:order:close")
MFG = Brick("mfg", version="0.2.0",
machine_permissions=(EXPLODE, READ_REQUIREMENTS, PLAN_DERIVE, OUTBOX_DISPATCH, GENEALOGY, NUMBER_SERIES, MIG_LOAD,
MIG_RECONCILE),
instructions=(
"Make your own products. A bill of materials says what goes into one product (and what comes out beside it); a "
"released version is never edited - the next one has its own start date. A production order copies its BOM "
"version when it is released. Issue components and report output on the order: stock and its value are "
"inventory's, so every movement is sent to inventory after it is recorded, and an order closes only when all of "
"them were taken. Weight lost in roasting is the BOM's planned yield (normal, in the cost); only output below plan "
"is a variance. A closing variance above the threshold needs a second cost accountant. Work centres and routings cost the time "
"(confirmations post activity to gl); scrap is expensed, never received; a wrong entry is reversed the same day by a "
"mirror. A cost roll-up is approved and released to inventory by a second person. MRP-lite proposes planned orders; a "
"planner firms and converts them. A model may suggest the cause of a variance (explain_variance): a person decides."))
# The process view (BRICKS.md): the signed workflows as BPMN, GET /process/bpmn.
MFG.declare_process("process.yaml")
# ── the unit of work ────────────────────────────────────────────────────────────────────────────────────────
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
def _lost_a_unique_race(exc: IntegrityError) -> bool:
text = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text or "duplicate key" in text or "UNIQUE constraint" in text
@contextmanager
def _on_engine(engine):
"""One transaction. On DataK3 a racer's duplicate claim can surface at COMMIT, after the use case has returned: it is caught
here and answered as the retry-safe 409 a conflict seen mid-use-case gets, never a 500."""
from brick_manufacturing.infrastructure.tables import Repository
try:
with engine.begin() as conn:
yield Repository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else wrote the same record a moment ago - load it again and retry") from None
repository = _repository_default # tests swap in one on SQLite
@contextmanager
def _decide():
"""A broken rule is a 409 with the rule's own words; a bound is a 422; a neighbour that could not answer is its own status."""
try:
yield
except OutOfBounds as bad:
raise Refused(422, str(bad), gate="bounds", fix=bad.fix or None) from None
except RuleBroken as broken:
raise Refused(409, str(broken), gate=broken.gate or None, fix=broken.fix or None) from None
except seams.Unavailable as down:
raise Refused(down.http, down.reason) from None
except IntegrityError:
raise Refused(409, "Someone else wrote the same record a moment ago - load it again and retry") from None
def _email(caller: dict) -> str:
return (caller.get("email") or caller.get("sub") or "").strip().lower()
def _sub(caller: dict) -> str:
return (caller.get("sub") or caller.get("email") or "").strip().lower()
def _same_person(caller: dict, email: str, sub: str) -> bool:
"""The same person under EITHER identity: a changed email or a second account does not make a second approver."""
return bool(email and _email(caller) == (email or "").lower()) or bool(sub and _sub(caller) == (sub or "").lower())
def _date(v: Optional[str], what: str) -> Optional[date]:
if v in (None, ""):
return None
try:
d = date.fromisoformat(str(v).strip()[:10])
except ValueError:
raise Refused(422, f"{what} is a date like 2026-11-01") from None
if not date(1900, 1, 1) <= d <= date(2199, 12, 31):
raise Refused(422, f"{what} is out of range")
return d
def _today(repo) -> date:
"""Business today: the date NOW (work.now()) in business_timezone, read from ONE place (SettingsRepository.timezone)."""
return work.now().astimezone(ZoneInfo(repo.settings.timezone())).date()
def _s(v) -> str:
return f"{Decimal(v):f}"
# ── settings (MFG-31) ───────────────────────────────────────────────────────────────────────────────────────
def _check_setting(key: str, value: str) -> str:
kind = SETTING_DEFS[key][1]
v = str(value).strip()
try:
if kind == "decimal":
d = Decimal(v)
if not d.is_finite() or d < 0 or d > Decimal("100000") or d.as_tuple().exponent < -2:
raise ValueError
elif kind == "bool" and v.lower() not in ("true", "false"):
raise ValueError
elif kind == "tz":
ZoneInfo(v)
elif kind == "method" and v not in ("standard", "average"):
raise ValueError
elif kind == "currency" and not (len(v) == 3 and v.isalpha() and v.isascii()):
raise ValueError
elif kind == "aimode" and v not in ("off", "suggest"):
raise ValueError
elif kind == "weeks" and not (v.isdigit() and 1 <= int(v) <= 26):
raise ValueError
except (ValueError, ZoneInfoNotFoundError, ArithmeticError):
raise Refused(422, f"{value!r} is not a valid {key} ({kind})", gate="bounds") from None
return v.lower() if kind == "bool" else v.upper() if kind == "currency" else v
@MFG.use_case(kind="query", route="GET /settings", permission="mfg:order:read")
def list_mfg_settings(caller: dict) -> dict:
"""The shop-floor controls in force today (over_issue_tolerance_pct, release_with_shortage, output_plausibility_pct,
variance_approval_threshold, business_timezone, valuation_method, currency), and every dated row behind them."""
with repository() as repo:
today = _today(repo)
return {"today": today.isoformat(), "settings": repo.settings.all(today),
"rows": [{**r, "valid_from": r["valid_from"].isoformat(), "created_at": None, "updated_at": None}
for r in repo.settings.rows()]}
@MFG.use_case(kind="command", route="POST /settings/{setting_key}", permission=CONTROLS)
def set_mfg_setting(caller: dict, setting_key: Literal[tuple(SETTING_DEFS)], value: Code, valid_from: Optional[Day] = None) -> dict:
"""Change a control from a date: a dated row, never an edit. The first row of a key may start today; after that a change
starts after today. Only the owner holds this (mfg:settings:controls): the planner it steers cannot widen its own limits."""
with repository() as repo, _decide():
today = _today(repo)
start = _date(valid_from, "valid_from") or today
v = _check_setting(setting_key, value)
have = [r for r in repo.settings.rows() if r["setting_key"] == setting_key]
if start < today or (have and start <= today):
raise RuleBroken(f"{setting_key} changes from a day after today ({today}): a row in force is never rewritten",
gate="valid_from")
repo.settings.set(setting_key, v, start, _email(caller), work.now())
return {"setting_key": setting_key, "value": v, "valid_from": start.isoformat(), "settings": repo.settings.all(today)}
def mark_stale(repo, cause: str, *, bom_id: str = "", from_day: Optional[date] = None, product_id: str = "", work_centre: str = "",
routing_product: str = "", plan: bool = True, unreleased_note: str = "") -> None:
"""MFG-18: an input changed, so what was derived from it says so (a flag, never a rewrite): the open cost roll-ups that used it (they
can no longer be released), the last MRP run, and - for a new BOM version - the unreleased orders due on or after its start.
A released roll-up and a released order keep what they froze."""
hit = []
for r in repo.costs.open_ones():
used = json.loads(r["used_json"] or "{}")
if (bom_id and bom_id in used.get("boms", []) and (from_day is None or r["effective_on"] >= from_day)) \
or (work_centre and work_centre in used.get("work_centres", [])) \
or (routing_product and routing_product in used.get("routings", [])):
hit.append(r["rollup_id"])
repo.costs.mark_stale(hit, cause)
if plan:
repo.plan.mark_stale(cause)
if unreleased_note and product_id and from_day:
for oid in repo.orders.unreleased_due_from(product_id, from_day):
got = repo.orders.get(oid)
if got and not got[0].stale_note:
o, ver, loaded = got
o.stale_note = unreleased_note
repo.orders.save(o, ver, loaded)
# ── bills of materials (MFG-02, 04, 05, 30) ─────────────────────────────────────────────────────────────────
def _bom_view(bom: Bom, version: int, today: date) -> dict:
cur = bom.valid_on(today)
return {"bom_id": bom.bom_id, "product_id": bom.product_id, "usage": bom.usage, "alternative": bom.alternative,
"company_code": bom.company_code, "version": version, "valid_today": cur.version if cur else None,
"versions": [{**{k: (_s(x) if isinstance(x, Decimal) else x.isoformat() if isinstance(x, date) else x)
for k, x in v.__dict__.items()},
"lines": [{k: (_s(x) if isinstance(x, Decimal) else x) for k, x in ln.__dict__.items() if k != "version"}
for ln in bom.lines_of(v.version)]}
for v in sorted(bom.versions, key=lambda v: v.version)]}
def _need_bom(repo, bom_id: str) -> tuple:
got = repo.boms.get(bom_id)
if got is None:
raise Refused(404, f"no bill of materials {bom_id}")
return got
@MFG.use_case(kind="query", route="GET /boms", permission="mfg:bom:read")
def list_boms(caller: dict) -> dict:
"""The bills of materials, by id (<product>:<usage>, e.g. ROAST-ETH:production). Get one for its versions and lines."""
with repository() as repo:
return {"boms": repo.boms.ids()}
@MFG.use_case(kind="query", route="GET /boms/{bom_id}", permission="mfg:bom:read")
def get_bom(caller: dict, bom_id: Code) -> dict:
"""One bill of materials with every version and line, the planned yield, and which version is in force today."""
with repository() as repo, _decide():
bom, ver, _ = _need_bom(repo, bom_id)
return _bom_view(bom, ver, _today(repo))
@MFG.use_case(kind="command", route="POST /boms/{bom_id}/versions", permission="mfg:bom:write", idempotent=True)
def write_bom(caller: dict, bom_id: Code, version: int, valid_from: Day, output_quantity: Qty, output_unit_code: Code,
lines: list[BomLineIn], company_code: Optional[Code] = None) -> dict:
"""Write a version of a bill of materials as a DRAFT (the BOM is created with its first version). `bom_id` is
<product>:<production|sale>[:<alternative>]. Every quantity must convert EXACTLY to the product's base unit (product answers;
it never rounds). The planned yield (output per input) is shown, never typed: weight lost in roasting is planned here, so it is
part of the cost. A component that leads back to the output, directly or through another BOM, is refused. A draft can be
written again; release it with release_bom."""
with repository() as repo, _decide():
today = _today(repo)
parts = bom_id.split(":")
if len(parts) not in (2, 3) or parts[1] not in (u.value for u in BomUsage):
raise Refused(422, f"{bom_id!r} is not a BOM id: use <product>:<production|sale>[:<alternative>]", gate="bounds")
got = repo.boms.get(bom_id)
if got is None:
if not company_code:
raise Refused(422, "company_code is needed to create a bill of materials", gate="bounds")
bom, ver, loaded = Bom(bom_id, parts[0], parts[1], parts[2] if len(parts) == 3 else "", company_code), 0, None
else:
bom, ver, loaded = got
start = _date(valid_from, "valid_from")
base_unit: dict = {}
def base(product_id: str, qty: Decimal, unit: str) -> tuple:
if product_id not in base_unit:
base_unit[product_id] = seams.product(product_id, today.isoformat())["base_unit_code"]
b = base_unit[product_id]
try:
return seams.to_base(product_id, qty, unit, b), b
except seams.Unavailable as e:
if e.http == 422:
raise OutOfBounds(f"{qty} {unit} of {product_id} does not convert exactly to its base unit {b}") from None
raise
out_base, out_unit = base(bom.product_id, output_quantity, output_unit_code)
drafts = []
for n, ln in enumerate(lines, 1):
qb, bu = base(ln.product_id, ln.quantity, ln.unit_code)
drafts.append(BomLine(0, n, ln.kind, ln.product_id, ln.quantity, ln.unit_code, qb, bu, ln.replaces_product_id or "",
ln.modifier_id or "", ln.nrv_unit_cost or Decimal("0")))
bom.write_version(version, valid_from=start, output_quantity=output_quantity, output_unit_code=output_unit_code,
output_quantity_base=out_base, output_base_unit_code=out_unit, lines=drafts,
edges=repo.boms.edges(bom_id))
ver = repo.boms.save(bom, ver, loaded)
return _bom_view(bom, ver, today)
@MFG.use_case(kind="command", route="POST /boms/{bom_id}/versions/{version}/release", permission="mfg:bom:release")
def release_bom(caller: dict, bom_id: Code, version: int) -> dict:
"""Release a draft version. It must start after today and after every released version (so two versions never overlap),
and from then on it is never edited: a change is the next version. Orders already released keep the copy they froze."""
with repository() as repo, _decide():
bom, ver, loaded = _need_bom(repo, bom_id)
bom.release_bom(version, today=_today(repo), by=_email(caller), sub=_sub(caller))
ver = repo.boms.save(bom, ver, loaded)
released = next(x for x in bom.versions if x.version == version)
mark_stale(repo, f"{bom_id} version {version} released, valid from {released.valid_from}", bom_id=bom_id,
from_day=released.valid_from, product_id=bom.product_id,
unreleased_note=f"BOM v{version} now valid" if bom.usage == "production" else "")
return _bom_view(bom, ver, _today(repo))
@MFG.use_case(kind="command", route="POST /boms/{bom_id}/versions/{version}/obsolete", permission="mfg:bom:release")
def obsolete_bom(caller: dict, bom_id: Code, version: int, valid_to: Day) -> dict:
"""Retire a released version after a last valid day. Refused while a created order that names it falls due after that day."""
with repository() as repo, _decide():
bom, ver, loaded = _need_bom(repo, bom_id)
end = _date(valid_to, "valid_to")
bom.obsolete_bom(version, valid_to=end)
named = repo.orders.naming_bom(bom_id, version, end)
if named:
raise RuleBroken(f"Orders {', '.join(named)} are not released yet and fall due after {end}: they name version {version}",
gate="orders_named", fix="re-plan or cancel them first")
ver = repo.boms.save(bom, ver, loaded)
return _bom_view(bom, ver, _today(repo))
@MFG.use_case(kind="query", route="GET /boms/{product_id}/explode", permission=EXPLODE, tool=False)
def explode_bom(caller: dict, product_id: Code, quantity: Qty, usage: Literal["sale", "production"] = "sale",
unit_code: Optional[Code] = None, modifiers: Optional[text(500)] = None, on: Optional[Day] = None) -> dict:
"""What `quantity` of a product consumes on a date, from the released BOM version in force: components in their base unit,
with the BOM id and version (pos freezes both on its receipt line). Modifiers (comma-separated ids) replace the component they
name. The same question gets the same answer. A route the tills call, never a tool."""
with repository() as repo, _decide():
bom_id = Bom.make_id(product_id, usage)
day = _date(on, "on") or _today(repo)
bom, _, _ = _need_bom(repo, bom_id)
mods = tuple(m.strip() for m in (modifiers or "").split(",") if m.strip())
try:
v, comps = bom.explode(quantity, day, mods)
except LookupError:
raise Refused(404, f"{bom_id} has no released version valid on {day}") from None
if unit_code and unit_code != v.output_unit_code:
raise OutOfBounds(f"{bom_id} is written per {v.output_unit_code}: ask in that unit")
return {"bom_id": bom_id, "version": v.version, "valid_from": v.valid_from.isoformat(),
"components": [{"product_id": p, "quantity": _s(q), "unit_code": u} for p, q, u in comps]}
# ── production orders ───────────────────────────────────────────────────────────────────────────────────────
def _sees_all(caller: dict) -> bool:
return any(auth.has_permission(caller, p) for p in SEES_ALL)
def _costs(caller: dict) -> bool:
return any(auth.has_permission(caller, p) for p in COSTS)
def _see_order(repo, caller: dict, order_id: str) -> tuple:
"""(order, version, loaded) if the caller may see it; otherwise 404, never 403. An operator sees released orders."""
got = repo.orders.get(order_id)
if got is None or (not _sees_all(caller) and got[0].status not in (OrderState.RELEASED.value, OrderState.IN_PROGRESS.value)):
raise Refused(404, f"no production order {order_id}")
return got
def _live(movements: list, kind: str) -> list:
return [m for m in movements if m["kind"] == kind and m["status"] != "rejected"]
def _net_issued(movements: list) -> dict:
"""line_no -> base quantity issued, NET of reversals (a reversal that inventory did not refuse counts as returned)."""
out: dict = {}
for m in _live(movements, "issue"):
out[m["line_no"]] = out.get(m["line_no"], Decimal("0")) + m["quantity_base"]
for m in _live(movements, "issue_reverse"):
out[m["line_no"]] = out.get(m["line_no"], Decimal("0")) - m["quantity_base"]
return out
def _net_lots(movements: list) -> dict:
"""(product_id, lot_code) -> base quantity issued net of reversals: the ingredients of the output lot."""
out: dict = {}
for m in _live(movements, "issue"):
if m["lot_code"]:
out[(m["product_id"], m["lot_code"])] = out.get((m["product_id"], m["lot_code"]), Decimal("0")) + m["quantity_base"]
for m in _live(movements, "issue_reverse"):
if m["lot_code"]:
out[(m["product_id"], m["lot_code"])] = out.get((m["product_id"], m["lot_code"]), Decimal("0")) - m["quantity_base"]
return {k: v for k, v in out.items() if v > 0}
def _figures(order: ProductionOrder, movements: list, activity_cost: Decimal = Decimal("0"), scraps: Optional[list] = None):
"""Settlement figures from the SENT movements (those inventory valued), the confirmations' activity cost and the approved scrap
(expensed at its standard value, never received)."""
comps = []
for c in order.planned_components():
mine = [m for m in movements if m["kind"] == "issue" and m["line_no"] == c["line_no"] and m["status"] == "sent"]
back = [m for m in movements if m["kind"] == "issue_reverse" and m["line_no"] == c["line_no"] and m["status"] == "sent"]
comps.append({"planned_base": c["planned_base"],
"issued_base": sum((m["quantity_base"] for m in mine), Decimal("0")) - sum((m["quantity_base"] for m in back), Decimal("0")),
"issued_value": sum((m["value"] or Decimal("0") for m in mine), Decimal("0"))
- sum((m["value"] or Decimal("0") for m in back), Decimal("0"))})
scraps = [x for x in (scraps or []) if x["status"] == "approved"]
sent = [m for m in movements if m["kind"] in ("output", "byproduct") and m["status"] == "sent"]
out_base = sum((m["quantity_base"] for m in sent if m["kind"] == "output"), Decimal("0"))
return settle(comps, outputs_value=sum((m["value"] or Decimal("0") for m in sent), Decimal("0")), activity_cost=activity_cost,
output_base=out_base, planned_output_base=order.planned_output_base(),
abnormal_scrap=sum((x["standard_value"] for x in scraps), Decimal("0")),
scrap_base=sum((x["quantity_base"] for x in scraps), Decimal("0")))
def _plain_rows(rows: list, costs: bool, cost_fields: tuple) -> list:
"""Fact rows as plain JSON; the shop-floor view (no cost permission) never carries the cost fields."""
return [{k: (_s(v) if isinstance(v, Decimal) else v.isoformat() if isinstance(v, date) else v) for k, v in r.items()
if k not in ("created_at", "updated_at", "decided_at") and (costs or k not in cost_fields)} for r in rows]
def _order_view(repo, caller: dict, order: ProductionOrder, version: int) -> dict:
movements = repo.movements.for_order(order.production_order_id)
costs = _costs(caller)
frozen = order.frozen
out_base = sum((m["quantity_base"] for m in _live(movements, "output")), Decimal("0"))
comps_issued = _net_issued(movements)
only = None # a yield is output per input, so it is shown for a one-component BOM (the roast)
if frozen and sum(1 for x in frozen["lines"] if x["kind"] == "component") == 1:
only = next(x for x in frozen["lines"] if x["kind"] == "component")
actual_in = comps_issued.get(only["line_no"], Decimal("0")) if only else Decimal("0")
pct = frozen.get("planned_yield_pct") if frozen else None
out = {"production_order_id": order.production_order_id, "status": order.status, "company_code": order.company_code,
"product_id": order.product_id, "warehouse_id": order.warehouse_id, "planned_quantity": _s(order.planned_quantity),
"unit_code": order.unit_code, "due_on": order.due_on.isoformat() if order.due_on else None, "bom_id": order.bom_id,
"bom_version": order.bom_version or None, "released_on": order.released_on.isoformat() if order.released_on else None,
"shortage_warning": order.shortage_warning or None, "version": version, "routing_version": order.routing_version or None,
"stale_note": order.stale_note or None, "planned_order_ref": order.planned_order_ref or None,
"planned_yield_pct": pct, "yield_pct": _s(q2(out_base / actual_in * 100)) if (only and actual_in and out_base) else None,
"planned_components": [{**c, "planned_base": _s(c["planned_base"]), "base_per_unit": None,
"issued_base": _s(q4(comps_issued.get(c["line_no"], Decimal("0"))))}
for c in (order.planned_components() if frozen else [])],
"movements": [{k: (_s(v) if isinstance(v, Decimal) else v.isoformat() if isinstance(v, date) else v)
for k, v in m.items() if k not in ("created_at", "updated_at")
and (costs or k not in ("value", "unit_cost"))} for m in movements]}
confirmations, scraps = repo.floor.confirmations(order.production_order_id), repo.floor.scraps(order.production_order_id)
out["confirmations"] = _plain_rows(confirmations, costs, ("cost",))
out["scraps"] = _plain_rows(scraps, costs, ("standard_value",))
if costs:
activity = sum((c["cost"] or Decimal("0") for c in confirmations if c["status"] != "void"), Decimal("0"))
out["wip_balance"] = _s(_figures(order, movements, activity, scraps).balance) if frozen else "0.00"
out["settlements"] = [{k: (_s(v) if isinstance(v, Decimal) else v.isoformat() if isinstance(v, date) else v)
for k, v in s.items() if k not in ("created_at", "updated_at")}
for s in repo.facts.settlements(order.production_order_id)]
out["outbox"] = [{"target": r["target"], "ref": r["ref"], "status": r["status"], "attempts": r["attempts"],
"last_error": r["last_error"]} for r in repo.outbox.for_order(order.production_order_id)]
return out
@MFG.use_case(kind="query", route="GET /production-orders", permission="mfg:order:read")
def list_production_orders(caller: dict, status: Optional[Literal[tuple(s.value for s in OrderState)]] = None) -> dict:
"""The production orders the caller may see, by id, optionally of one status. An operator sees released orders only."""
with repository() as repo:
if not _sees_all(caller):
wanted = [s for s in ("released", "in_progress") if status in (None, s)]
return {"production_orders": sorted(i for s in wanted for i in repo.orders.ids(s))}
return {"production_orders": repo.orders.ids(status)}
@MFG.use_case(kind="query", route="GET /production-orders/{production_order_id}", permission="mfg:order:read")
def get_production_order(caller: dict, production_order_id: Code) -> dict:
"""One order: its frozen BOM, what each component is planned and has been issued, every movement and whether inventory took
it, the yield against plan, and (for those who may see costs) its WIP balance, settlement and pending sends."""
with repository() as repo, _decide():
order, ver, _ = _see_order(repo, caller, production_order_id)
return _order_view(repo, caller, order, ver)
@MFG.use_case(kind="command", route="POST /production-orders", permission="mfg:order:create")
def create_order(caller: dict, company_code: Code, product_id: Code, warehouse_id: Code, planned_quantity: Qty, due_on: Day,
unit_code: Optional[Code] = None, alternative: Optional[Code] = None) -> dict:
"""Create a production order: a quantity of a product by a date, against the released BOM and version valid on the due date.
The id is claimed from the company's yearly series (UK01-MO-2026-00118). Nothing moves until it is released."""
with repository() as repo, _decide():
order = _new_order(repo, caller, company_code, product_id, warehouse_id, planned_quantity, _date(due_on, "due_on"), unit_code,
alternative or "")
return _order_view(repo, caller, order, 1)
def _new_order(repo, caller: dict, company_code: str, product_id: str, warehouse_id: str, planned_quantity: Decimal, due: date,
unit_code: Optional[str], alternative: str = "", planned_order_ref: str = "") -> ProductionOrder:
"""The one way an order is born (by hand or from a planned order): a released BOM version, and a routing if the product has one, both
valid on the due date; its id claimed from the company's yearly series."""
from brick_manufacturing.domain import routing as rt
today = _today(repo)
bom, _, _ = repo.boms.get(Bom.make_id(product_id, "production", alternative)) or (None, 0, None)
if bom is None:
raise RuleBroken(f"{product_id} has no production bill of materials", gate="bom_valid", fix="write and release one")
v = bom.valid_on(due)
if v is None:
raise RuleBroken(f"No released version of {bom.bom_id} is valid on {due}", gate="bom_valid", fix="release a version")
if unit_code and unit_code != v.output_unit_code:
raise RuleBroken(f"{bom.bom_id} is written per {v.output_unit_code}: order in that unit", gate="units_exact")
routings = repo.masters.routings(product_id)
if routings and rt.valid_version(routings, due) is None:
raise RuleBroken(f"No routing of {product_id} is valid on {due}", gate="routing_valid", fix="write a routing that starts earlier")
n = repo.facts.next_order_no(company_code, today.year)
oid = f"{company_code}-MO-{today.year}-{n:05d}"
order = ProductionOrder.create_order(oid, company_code=company_code, product_id=product_id, warehouse_id=warehouse_id,
planned_quantity=planned_quantity, unit_code=v.output_unit_code, due_on=due,
bom_id=bom.bom_id, bom_version=v.version, by=_email(caller))
order.planned_order_ref = planned_order_ref
repo.facts.take_order_no(company_code, today.year, n, oid, work.now())
repo.orders.save(order, 0)
return order
def _freeze(bom: Bom, v) -> dict:
return {"bom_id": bom.bom_id, "version": v.version, "valid_from": v.valid_from.isoformat(),
"output_quantity": _s(v.output_quantity), "output_unit_code": v.output_unit_code,
"output_quantity_base": _s(v.output_quantity_base), "output_base_unit_code": v.output_base_unit_code,
"planned_yield_pct": _s(v.planned_yield_pct) if v.planned_yield_pct is not None else None,
"lines": [{k: (_s(x) if isinstance(x, Decimal) else x) for k, x in ln.__dict__.items() if k != "version"}
for ln in bom.lines_of(v.version)]}
@MFG.use_case(kind="command", route="POST /production-orders/{production_order_id}/release", permission="mfg:order:release")
def release_order(caller: dict, production_order_id: Code) -> dict:
"""Release a created order: the BOM version valid today is COPIED onto it and frozen (a later BOM change never reaches it),
and every component must be available in the warehouse unless the owner allows release with shortages (then a warning is
kept). A released order never goes back to created."""
with repository() as repo, _decide():
order, ver, loaded = _see_order(repo, caller, production_order_id)
today = _today(repo)
bom, _, _ = _need_bom(repo, order.bom_id)
v = bom.valid_on(today)
if v is None:
raise RuleBroken(f"No released version of {bom.bom_id} is valid on {today}", gate="bom_valid", fix="release a version")
frozen = _freeze(bom, v)
k = order.planned_quantity / v.output_quantity
need: dict = {}
for x in bom.lines_of(v.version):
if x.kind == LineKind.COMPONENT.value:
need[x.product_id] = need.get(x.product_id, Decimal("0")) + q4(x.quantity_base * k)
need[("unit", x.product_id)] = x.base_unit_code
shortages = []
for pid in [p for p in need if isinstance(p, str)]:
have = seams.available(pid, order.warehouse_id)
if have < need[pid]:
shortages.append(f"{pid} needs {_s(need[pid])} {need[('unit', pid)]}, {_s(have)} available")
allow = repo.settings.all(today)["release_with_shortage"] == "true"
from brick_manufacturing.domain import routing as rt
routings = repo.masters.routings(order.product_id)
frozen_routing = None
if routings:
row = rt.valid_version(routings, today)
if row is None:
raise RuleBroken(f"No routing of {order.product_id} is valid on {today}", gate="routing_valid", fix="write a routing that starts earlier")
frozen_routing = {"routing_version": row["routing_version"], "valid_from": row["valid_from"].isoformat(),
"operations": json.loads(row["operations_json"])}
order.release_order(today=today, frozen=frozen, shortages=shortages, allow_shortage=allow, by=_email(caller),
routing=frozen_routing)
ver = repo.orders.save(order, ver, loaded)
return _order_view(repo, caller, order, ver)
def _after_commit(order_id: str) -> dict:
return dispatch.send(repository, order_id)
def _refuse_if_rejected(sent: dict) -> None:
if sent["rejected"]:
r = sent["rejected"][0]
raise Refused(409, f"Inventory refused it: {r['reason']}", gate=r["gate"] or "inventory_refused",
fix="the movement is recorded as rejected and counts for nothing; correct it and report again")
@MFG.use_case(kind="command", route="POST /production-orders/{production_order_id}/issue", permission="mfg:order:issue")
def issue_components(caller: dict, production_order_id: Code, location_id: Code, lines: list[IssueLineIn]) -> dict:
"""Issue components from a stock location into the order (scan the lot). Each line is recorded with its own move_ref
(mfg:<order>:issue:<line>:<seq>) and sent to inventory after the commit, which moves the stock to the order's WIP and answers
its value. Up to over_issue_tolerance_pct above plan is allowed and flagged as over-issue; beyond it is refused. Quantities
must be above zero and convert exactly."""
with repository() as repo, _decide():
order, ver, loaded = _see_order(repo, caller, production_order_id)
today = _today(repo)
order.start_work()
tol = Decimal(repo.settings.all(today)["over_issue_tolerance_pct"])
planned = {c["product_id"]: c for c in order.planned_components()}
movements = repo.movements.for_order(production_order_id)
issued = _net_issued(movements)
counts = repo.claims.counts("issue", [f"{production_order_id}:{c['line_no']}" for c in planned.values()])
rows, claims, payload, seqs = [], [], [], {}
for ln in lines:
c = planned.get(ln.product_id)
if c is None:
raise RuleBroken(f"{ln.product_id} is not a component of order {production_order_id}", gate="component")
unit = ln.unit_code or c["unit_code"]
if unit == c["unit_code"]:
base = ln.quantity * c["base_per_unit"]
elif unit == c["base_unit_code"]:
base = ln.quantity
else:
raise OutOfBounds(f"{unit} is neither {c['unit_code']} (the BOM's unit) nor {c['base_unit_code']} (the base unit) "
f"of {ln.product_id}")
if base != q4(base):
raise OutOfBounds(f"{ln.quantity} {unit} of {ln.product_id} does not convert exactly to {c['base_unit_code']}")
over = order.check_issue(c, base, issued.get(c["line_no"], Decimal("0")), tol)
issued[c["line_no"]] = issued.get(c["line_no"], Decimal("0")) + base
key = f"{production_order_id}:{c['line_no']}"
seqs[key] = seqs.get(key, counts[key]) + 1
ref = f"mfg:{production_order_id}:issue:{c['line_no']}:{seqs[key]}"
claims.append(("issue", key, seqs[key], ref))
rows.append({"move_ref": ref, "production_order_id": production_order_id, "kind": "issue", "line_no": c["line_no"],
"seq": seqs[key], "product_id": ln.product_id, "lot_code": ln.lot_code or "", "quantity": ln.quantity,
"unit_code": unit, "quantity_base": base, "base_unit_code": c["base_unit_code"],
"location_id": location_id, "over_issue": over, "status": "pending", "recorded_by": _email(caller),
"recorded_sub": _sub(caller), "business_date": today})
payload.append({"move_ref": ref, "product_id": ln.product_id, "quantity": _s(ln.quantity), "unit_code": unit,
**({"lot_code": ln.lot_code} if ln.lot_code else {})})
body = {"source_kind": "production_issue", "source_id": production_order_id, "company_code": order.company_code,
"location_id": location_id, "issued_on": today.isoformat(), "account_purpose": "wip", "lines": payload}
repo.orders.save(order, ver, loaded)
repo.claims.take(claims, _email(caller), work.now())
repo.movements.add(rows, work.now())
repo.outbox.add([{"target": "inventory.issue", "ref": rows[0]["move_ref"], "production_order_id": production_order_id,
"payload": dump(body)}], work.now())
sent = _after_commit(production_order_id)
_refuse_if_rejected(sent)
with repository() as repo, _decide():
order, ver, _ = _see_order(repo, caller, production_order_id)
return {**_order_view(repo, caller, order, ver), "sent": sent}
@MFG.use_case(kind="command", route="POST /production-orders/{production_order_id}/output", permission="mfg:order:report_output")
def report_output(caller: dict, production_order_id: Code, location_id: Code, quantity: Qty, lot_code: Optional[Code] = None,
by_products: Optional[list[ByProductIn]] = None, unit_code: Optional[Code] = None,
confirm_above_plan: bool = False) -> dict:
"""Report what came out: the main product (with its NEW lot code) and any by-products. Each is received through inventory at
the cost inventory holds for the product (the standard) - a by-product at its net realisable value - and the lots issued to
the order are recorded as the output lot's ingredients (genealogy). Weight lost in the roast is not a loss here: it is the
BOM's planned yield. Above output_plausibility_pct of plan is refused as a typo unless a planner confirms
(confirm_above_plan, honoured only from someone who may release orders)."""
with repository() as repo, _decide():
order, ver, loaded = _see_order(repo, caller, production_order_id)
today = _today(repo)
order.start_work()
st = repo.settings.all(today)
frozen = order.frozen
unit = unit_code or frozen["output_unit_code"]
if unit != frozen["output_unit_code"]:
raise OutOfBounds(f"{order.product_id} is made per {frozen['output_unit_code']}: report in that unit")
ratio = Decimal(frozen["output_quantity_base"]) / Decimal(frozen["output_quantity"])
base = quantity * ratio
if base != q4(base):
raise OutOfBounds(f"{quantity} {unit} does not convert exactly to {frozen['output_base_unit_code']}")
movements = repo.movements.for_order(production_order_id)
reported = sum((m["quantity_base"] for m in _live(movements, "output")), Decimal("0"))
confirmed = confirm_above_plan and auth.has_permission(caller, "mfg:order:release")
if confirm_above_plan and not confirmed:
raise RuleBroken("Only a planner confirms output above the plausible limit", gate="output_plausibility")
order.check_output(base, reported, Decimal(st["output_plausibility_pct"]), confirmed)
cost = seams.unit_cost(order.product_id, order.company_code, today.isoformat())
if cost is None:
raise RuleBroken(f"Inventory holds no cost for {order.product_id}: the output cannot be valued", gate="standard_price",
fix="release a standard cost for it (a cost roll-up) first")
by_line = {x["product_id"]: x for x in frozen["lines"] if x["kind"] == "by_product"}
out_keys = [production_order_id] + [f"{production_order_id}:{x['line_no']}" for x in by_line.values()]
n_out = repo.claims.counts("output", [production_order_id])[production_order_id] + 1
n_by = repo.claims.counts("byproduct", out_keys[1:]) if by_line else {}
input_lots = _net_lots(movements)
ref = f"mfg:{production_order_id}:output:{n_out}"
claims = [("output", production_order_id, n_out, ref)]
rows = [{"move_ref": ref, "production_order_id": production_order_id, "kind": "output", "line_no": 0, "seq": n_out,
"product_id": order.product_id, "lot_code": lot_code or "", "quantity": quantity, "unit_code": unit,
"quantity_base": base, "base_unit_code": frozen["output_base_unit_code"], "location_id": location_id,
"over_issue": False, "unit_cost": cost, "status": "pending", "recorded_by": _email(caller),
"recorded_sub": _sub(caller), "business_date": today}]
currency = st["currency"]
bodies = [(ref, {"move_ref": ref, "source_kind": "production_receipt", "source_id": production_order_id,
"company_code": order.company_code, "product_id": order.product_id, "quantity": _s(base),
"unit_code": frozen["output_base_unit_code"], "location_id": location_id,
"received_on": today.isoformat(), "unit_cost": _s(cost), "currency": currency,
**({"lot_code": lot_code} if lot_code else {}),
"input_lots": [{"product_id": p, "lot_code": lc, "quantity": _s(q)} for (p, lc), q in sorted(input_lots.items())]})]
seq_by: dict = {}
for bp in by_products or []:
line = by_line.get(bp.product_id)
if line is None:
raise RuleBroken(f"{bp.product_id} is not a by-product of {order.bom_id}", gate="by_product")
bb = bp.quantity * (Decimal(line["quantity_base"]) / Decimal(line["quantity"]))
if bb != q4(bb):
raise OutOfBounds(f"{bp.quantity} {line['unit_code']} of {bp.product_id} does not convert exactly")
key = f"{production_order_id}:{line['line_no']}"
seq_by[key] = seq_by.get(key, n_by[key]) + 1
bref = f"mfg:{production_order_id}:byproduct:{line['line_no']}:{seq_by[key]}"
nrv = Decimal(line["nrv_unit_cost"])
claims.append(("byproduct", key, seq_by[key], bref))
rows.append({"move_ref": bref, "production_order_id": production_order_id, "kind": "byproduct", "line_no": line["line_no"],
"seq": seq_by[key], "product_id": bp.product_id, "lot_code": bp.lot_code or "", "quantity": bp.quantity,
"unit_code": line["unit_code"], "quantity_base": bb, "base_unit_code": line["base_unit_code"],
"location_id": location_id, "over_issue": False, "unit_cost": nrv, "status": "pending",
"recorded_by": _email(caller), "recorded_sub": _sub(caller), "business_date": today})
bodies.append((bref, {"move_ref": bref, "source_kind": "by_product_receipt", "source_id": production_order_id,
"company_code": order.company_code, "product_id": bp.product_id, "quantity": _s(bb),
"unit_code": line["base_unit_code"], "location_id": location_id, "received_on": today.isoformat(),
"unit_cost": _s(nrv), "currency": currency, **({"lot_code": bp.lot_code} if bp.lot_code else {})}))
repo.orders.save(order, ver, loaded)
repo.claims.take(claims, _email(caller), work.now())
repo.movements.add(rows, work.now())
repo.outbox.add([{"target": "inventory.receipt", "ref": r, "production_order_id": production_order_id, "payload": dump(b)}
for r, b in bodies], work.now())
if lot_code and input_lots:
repo.facts.genealogy([{"output_product_id": order.product_id, "output_lot_code": lot_code, "input_product_id": p,
"input_lot_code": lc, "production_order_id": production_order_id, "input_quantity": q,
"output_quantity": base} for (p, lc), q in input_lots.items()])
sent = _after_commit(production_order_id)
_refuse_if_rejected(sent)
with repository() as repo, _decide():
order, ver, _ = _see_order(repo, caller, production_order_id)
return {**_order_view(repo, caller, order, ver), "sent": sent}
@MFG.use_case(kind="command", route="POST /production-orders/{production_order_id}/complete", permission="mfg:order:complete")
def complete_order(caller: dict, production_order_id: Code) -> dict:
"""Technically complete an order in progress: no further issues or outputs. Costs may still arrive until it is closed. A
completed order never returns to in progress."""
with repository() as repo, _decide():
order, ver, loaded = _see_order(repo, caller, production_order_id)
order.complete_order(today=_today(repo))
ver = repo.orders.save(order, ver, loaded)
return _order_view(repo, caller, order, ver)
def _settled(repo, order: ProductionOrder) -> tuple:
"""(figures, movements) for a completed order; refuses while any outbox row of the order is still waiting for a neighbour."""
waiting = [r for r in repo.outbox.for_order(order.production_order_id) if r["status"] == "pending"]
if waiting:
raise RuleBroken("Not every movement was taken yet: " + "; ".join(f"{r['ref']} ({r['last_error'] or 'waiting'})" for r in waiting),
gate="outbox_empty", fix="wait, or retry the failed ones (dispatch_outbox)")
scraps = repo.floor.scraps(order.production_order_id)
pending = [x for x in scraps if x["status"] == "pending_approval"]
if pending:
raise RuleBroken(f"Scrap {', '.join(str(x['scrap_no']) for x in pending)} above the threshold is waiting for a planner's approval",
gate="scrap_approved", fix="a planner approves it (approve_scrap)")
movements = repo.movements.for_order(order.production_order_id)
return _figures(order, movements, repo.floor.activity_cost(order.production_order_id), scraps), movements
@MFG.use_case(kind="command", route="POST /production-orders/{production_order_id}/approve-variance", permission="mfg:order:close")
def approve_variance(caller: dict, production_order_id: Code) -> dict:
"""A SECOND cost accountant approves a closing variance above variance_approval_threshold. The approval records who (email and
sub) and the variance it covers; whoever then closes the order must be someone else, and an approval below the current
variance is not enough."""
with repository() as repo, _decide():
order, _, _ = _see_order(repo, caller, production_order_id)
if order.status != OrderState.COMPLETED.value:
raise RuleBroken(f"Order {production_order_id} is {order.status}: only a completed order has a variance to approve", gate="state")
figures, _ = _settled(repo, order)
threshold = Decimal(repo.settings.all(_today(repo))["variance_approval_threshold"])
if figures.variance <= threshold:
raise RuleBroken(f"The variance {_s(figures.variance)} is within {_s(threshold)}: no approval is needed", gate="variance_approved")
n = repo.claims.counts("approval", [production_order_id])[production_order_id] + 1
repo.claims.take([("approval", production_order_id, n, f"{production_order_id}:approval:{n}")], _email(caller), work.now())
repo.facts.add_approval({"production_order_id": production_order_id, "approval_no": n,
"variance": figures.variance, "approved_by": _email(caller), "approved_sub": _sub(caller),
"approved_on": _today(repo)}, work.now())
return {"production_order_id": production_order_id, "variance": _s(figures.variance), "approved_by": _email(caller)}
@MFG.use_case(kind="command", route="POST /production-orders/{production_order_id}/close", permission="mfg:order:close")
def close_order(caller: dict, production_order_id: Code) -> dict:
"""Close and settle a completed order. Every movement must have been taken by inventory first. The order's WIP balance is
split exactly into material usage variance (including output below the planned yield) and activity variance; the split always
sums to the balance, so WIP is 0.00 afterwards. A variance above variance_approval_threshold needs a second cost accountant's
approval (approve_variance; never the person closing, by email or sub). On a standard-cost company the variances post to gl; on an
average-cost company the balance goes to inventory as a valuation adjustment. Both are sent after the commit."""
with repository() as repo, _decide():
order, ver, loaded = _see_order(repo, caller, production_order_id)
if order.status != OrderState.COMPLETED.value:
raise RuleBroken(f"Order {production_order_id} is {order.status}: only a completed order is closed", gate="state")
today = _today(repo)
st = repo.settings.all(today)
figures, _ = _settled(repo, order)
threshold, approver = Decimal(st["variance_approval_threshold"]), None
if figures.variance > threshold:
ok = [a for a in repo.facts.approvals(production_order_id) if a["variance"] >= figures.variance]
if not ok:
raise RuleBroken(f"The variance {_s(figures.variance)} is above {_s(threshold)}", gate="variance_approved",
fix="another cost accountant approves it (approve_variance)")
if all(_same_person(caller, a["approved_by"], a["approved_sub"]) for a in ok):
raise RuleBroken("You approved this variance yourself: a second person must approve what you settle", gate="second_person",
fix="ask another cost accountant")
approver = next(a for a in ok if not _same_person(caller, a["approved_by"], a["approved_sub"]))
n = repo.claims.counts("settle", [production_order_id])[production_order_id] + 1
method = st["valuation_method"]
outbox, journal_id = [], None
if figures.balance != 0 or figures.variance != 0:
if method == "average":
body = {"adjustment_ref": f"mfg:{production_order_id}:settle:{n}", "source_kind": "production_settlement",
"production_order_id": production_order_id, "company_code": order.company_code,
"product_id": order.product_id, "amount": _s(figures.balance), "currency": st["currency"],
"on": today.isoformat()}
outbox.append({"target": "inventory.adjustment", "ref": body["adjustment_ref"], "payload": dump(body)})
else:
journal_id = derived_id(production_order_id, "settle")
lines = [(p, v) for p, v in (("material_usage_variance", figures.material_usage_variance),
("activity_variance", figures.activity_variance),
("abnormal_scrap", figures.abnormal_scrap), ("wip", -figures.balance)) if v != 0]
body = {"journal_id": str(journal_id), "source_module": "manufacturing", "source_document": production_order_id,
"event": "settle", "company_code": order.company_code, "journal_date": today.isoformat(),
"currency": st["currency"], "memo": f"Settlement of {production_order_id}",
"lines": [{"purpose": p, **({"debit": _s(v)} if v > 0 else {"credit": _s(-v)})} for p, v in lines]}
outbox.append({"target": "gl.journal", "ref": f"mfg:{production_order_id}:settle:{n}", "payload": dump(body)})
order.close_order(today=today)
repo.claims.take([("settle", production_order_id, n, f"mfg:{production_order_id}:settle:{n}")], _email(caller), work.now())
repo.facts.add_settlement({"production_order_id": production_order_id, "settlement_no": n, "wip_debit": figures.wip_debit,
"wip_credit": figures.wip_credit, "standard_material": figures.standard_material,
"abnormal_scrap": figures.abnormal_scrap, "material_usage_variance": figures.material_usage_variance,
"activity_variance": figures.activity_variance, "journal_id": journal_id,
"valuation_method": method, "settled_by": _email(caller), "settled_sub": _sub(caller),
"approved_by": approver["approved_by"] if approver else "",
"approved_sub": approver["approved_sub"] if approver else "", "business_date": today}, work.now())
repo.orders.save(order, ver, loaded)
repo.outbox.add([{**o, "production_order_id": production_order_id} for o in outbox], work.now())
sent = _after_commit(production_order_id) if outbox else {"sent": 0, "rejected": [], "pending": 0}
with repository() as repo, _decide():
order, ver, _ = _see_order(repo, caller, production_order_id)
return {**_order_view(repo, caller, order, ver), "sent": sent}
@MFG.use_case(kind="command", route="POST /production-orders/{production_order_id}/cancel", permission="mfg:order:cancel")
def cancel_order(caller: dict, production_order_id: Code, reason: Reason) -> dict:
"""Cancel an order that is created or released and has had no movement. A reason is required; a cancelled order is final."""
with repository() as repo, _decide():
order, ver, loaded = _see_order(repo, caller, production_order_id)
order.cancel_order(reason, has_movements=bool(repo.movements.for_order(production_order_id)))
ver = repo.orders.save(order, ver, loaded)
return _order_view(repo, caller, order, ver)
@MFG.use_case(kind="command", route="POST /outbox/dispatch", permission=OUTBOX_DISPATCH, idempotent=True, tool=False)
def dispatch_outbox(caller: dict, production_order_id: Optional[Code] = None) -> dict:
"""Send every pending outbox row (of one order, or all) to inventory and gl: the composition's worker calls this. A repeat is
the neighbour's replay on the same reference; nothing is sent twice."""
return dispatch.send(repository, production_order_id)
# Slice B declares its use cases on the same brick (each module is a use-case family; none holds a rule the domain owns).
from brick_manufacturing.application import ai_step, engineering, floor, migration, planning, trace # noqa: E402,F401