dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
src/brick_manufacturing/application/planning.py
"""Planning (MFG-08, MFG-09, MFG-17, MFG-23, MFG-27): what to make from what is ordered, what the shops will use and what is in stock.
MRP-lite proposes planned production orders only. It reads open sales-order demand (never so_ tables), the planner's forecasts, stock and
site data, explodes level by level from the BOM graph, nets week by week, and rounds the shortfall UP through product (never itself).
Each planned order publishes its components as dependent requirements for inventory's and purchasing's replenishment - this brick never
raises a purchase request. A run is a pure function of its inputs: the same inputs give the same planned orders, the run takes the next
run_no with a plain INSERT (a second starter loses), unfirmed orders of the earlier run are superseded and firmed ones are kept."""
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional
from erp_platform import work
from erp_platform.bricks import Refused
from brick_manufacturing.application import use_cases as hub
from brick_manufacturing.application.engineering import plan_for
from brick_manufacturing.application.inputs import Code, Day, ForecastIn
from brick_manufacturing.domain import mrp
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, q4
from brick_manufacturing.infrastructure import seams
MFG = hub.MFG
_s = hub._s
LIVE = ("proposed", "firmed")
def _view(p: dict) -> dict:
return {"planned_order_ref": p["planned_order_ref"], "warehouse_id": p["warehouse_id"], "product_id": p["product_id"],
"due_on": p["need_date"].isoformat(), "run_no": p["run_no"], "quantity": _s(p["quantity"]), "unit_code": p["unit_code"],
"status": p["status"], "production_order_id": p["production_order_id"] or None,
"cause": __import__("json").loads(p["cause_json"] or "{}")}
# ── forecasts ───────────────────────────────────────────────────────────────────────────────────────────────
@MFG.use_case(kind="command", route="POST /forecasts/{warehouse_id}", permission="mfg:forecast:write")
def set_forecasts(caller: dict, warehouse_id: Code, forecasts: list[ForecastIn]) -> dict:
"""Enter what the shops will use, per product and week (the week is its Monday): demand for MRP-lite until pos publishes its sales
history. Entering a week again replaces it. The last MRP run is marked stale."""
with hub.repository() as repo, hub._decide():
rows = {}
for f in forecasts:
wk = mrp.monday(hub._date(f.week_start, "week_start"))
rows[(f.product_id, wk)] = {"warehouse_id": warehouse_id, "product_id": f.product_id, "week_start": wk, "quantity": f.quantity,
"unit_code": f.unit_code or "", "entered_by": hub._email(caller)}
repo.plan.set_forecasts(list(rows.values()), work.now())
hub.mark_stale(repo, f"forecast changed for {len(rows)} product-week(s)", plan=True, bom_id="", from_day=None)
return {"warehouse_id": warehouse_id, "weeks": sorted(f"{p}:{w.isoformat()}" for p, w in rows)}
@MFG.use_case(kind="query", route="GET /forecasts/{warehouse_id}", permission="mfg:plan:run")
def list_forecasts(caller: dict, warehouse_id: Code, start: Day, end: Day) -> dict:
"""The forecasts entered for a warehouse between two dates (by week)."""
with hub.repository() as repo, hub._decide():
rows = repo.plan.forecasts(warehouse_id, mrp.monday(hub._date(start, "start")), hub._date(end, "end"))
return {"forecasts": [{"product_id": r["product_id"], "week_start": r["week_start"].isoformat(), "quantity": _s(r["quantity"]),
"entered_by": r["entered_by"]} for r in rows]}
# ── MRP-lite (MFG-08, MFG-17) ───────────────────────────────────────────────────────────────────────────────
def _components(bom: Bom, version, qty_units: Decimal) -> list:
"""[(component_product, quantity_base, base_unit)] one order of `qty_units` (in the BOM's output unit) needs."""
k = qty_units / version.output_quantity
return [(x.product_id, q4(x.quantity_base * k), x.base_unit_code) for x in bom.lines_of(version.version) if x.kind == "component"]
def _run(repo, caller: dict, warehouse_id: str, weeks: Optional[int]) -> dict:
today = hub._today(repo)
horizon = weeks or int(repo.settings.all(today)["mrp_horizon_weeks"])
end = today + timedelta(weeks=horizon)
n = repo.claims.counts("mrp_run", ["run"])["run"] + 1
try:
repo.claims.take([("mrp_run", "run", n, f"run {n}")], hub._email(caller), work.now())
except hub.IntegrityError:
raise Refused(409, "A run is in progress: someone started run %d a moment ago - read the plan, or start again" % n) from None
sites = seams.product_sites(warehouse_id, today.isoformat())
edges = repo.boms.edges("")
levels = mrp.low_level_codes(edges)
boms: dict = {}
for pid in sorted(edges):
got = repo.boms.get(Bom.make_id(pid, "production"))
if got:
boms[pid] = got[0]
forecasts: dict = {}
for f in repo.plan.forecasts(warehouse_id, mrp.monday(today), end):
forecasts.setdefault(f["product_id"], []).append((f["week_start"], f["quantity"]))
firmed = [p for p in repo.plan.planned(warehouse_id, ("firmed",))]
open_orders = repo.orders.open_for(warehouse_id, ("created", "released"))
dependents: dict = {} # component product -> [(date, qty)] from parents planned so far (and firmed ones)
for p in firmed:
bom = boms.get(p["product_id"])
v = bom.valid_on(p["need_date"]) if bom else None
for comp, qb, _u in (_components(bom, v, p["quantity"]) if v else []):
dependents.setdefault(comp, []).append((p["need_date"], qb))
rows, reqs, made = [], [], []
for pid in sorted(boms, key=lambda x: (levels.get(x, 0), x)):
site = sites.get(pid)
if site is None or site.get("procurement_type") != "make":
continue # bought, or not planned here: not MRP-lite's to make
bom = boms[pid]
v_first = bom.valid_on(today) or bom.valid_on(end)
if v_first is None:
continue
needs = [(date.fromisoformat(str(d["needed_on"])[:10]), d["open_base"]) for d in seams.sales_demand(pid, warehouse_id, today.isoformat(), end.isoformat())]
needs += forecasts.get(pid, []) + dependents.get(pid, [])
if not needs:
continue
ratio = v_first.output_quantity_base / v_first.output_quantity
supply = [(p["need_date"], p["quantity"] * ratio) for p in firmed if p["product_id"] == pid]
supply += [(o["due_on"], o["planned_quantity"] * ratio) for o in open_orders if o["product_id"] == pid]
safety = Decimal(str(site.get("safety_stock") or 0))
on_hand = seams.available(pid, warehouse_id)
routing = rt.valid_version(repo.masters.routings(pid), today) or rt.valid_version(repo.masters.routings(pid), end)
ops = __import__("json").loads(routing["operations_json"]) if routing else []
lines = [x.__dict__ for x in bom.lines_of(v_first.version)]
def lead(qty_base: Decimal, _day: date, pid=pid, v=v_first, lines=lines, ops=ops) -> int:
return rt.lead_days(plan_for(repo, lines, v.output_quantity_base, qty_base / v.output_quantity_base, ops, today, []), ops) if ops else 0
def round_up(short: Decimal, pid=pid, v=v_first) -> Decimal:
rounded = seams.round_up(pid, warehouse_id, short, v.output_base_unit_code, today.isoformat())
if rounded < short or rounded > MAX_QTY:
raise OutOfBounds(f"{pid}: product rounded {short} to {rounded}, which is not a lot that covers it")
return rounded
for o in mrp.plan_product(needs, supply, on_hand, safety, today, lead, round_up):
due = o["due"]
version = bom.valid_on(due) or v_first
units = q4(o["quantity"] / (version.output_quantity_base / version.output_quantity))
ref = f"{warehouse_id}:{pid}:{due.isoformat()}:run {n}"
rows.append({"warehouse_id": warehouse_id, "product_id": pid, "need_date": due, "run_no": n, "quantity": units,
"unit_code": version.output_unit_code, "planned_order_ref": ref, "status": "proposed", "production_order_id": "",
"firmed_by": "", "firmed_sub": "",
"cause_json": hub.dump({"week": o["week"].isoformat(), "need": _s(o["need"]), "first_need": o["first_need"].isoformat(),
"short": _s(o["short"]), "on_hand": _s(on_hand), "safety_stock": _s(safety),
"bom_version": version.version})})
for comp, qb, unit in _components(bom, version, units):
dependents.setdefault(comp, []).append((due, qb))
reqs.append({"warehouse_id": warehouse_id, "product_id": comp, "need_date": due, "source_ref": ref, "quantity": qb,
"unit_code": unit, "source_kind": "planned_order", "run_no": n})
made.append(ref)
repo.plan.supersede(warehouse_id)
repo.plan.add_planned(rows, work.now())
repo.plan.add_requirements(reqs, work.now())
repo.plan.add_run({"run_no": n, "warehouse_id": warehouse_id, "from_date": today, "horizon_weeks": horizon, "planned_count": len(rows),
"stale": False, "stale_cause": "", "started_by": hub._email(caller)}, work.now())
return {"run_no": n, "warehouse_id": warehouse_id, "from": today.isoformat(), "to": end.isoformat(), "horizon_weeks": horizon,
"planned_orders": [_view(r) for r in rows], "stale": False}
@MFG.use_case(kind="command", route="POST /mrp/runs", permission="mfg:plan:run", status=201)
def run_mrp(caller: dict, warehouse_id: Code, weeks: Optional[int] = None) -> dict:
"""Run MRP-lite for a warehouse: the open sales-order demand, the forecasts entered and the components of the orders planned above,
netted against stock (less the safety stock) and what is already planned or ordered, week by week; each shortfall is rounded UP to a lot
through product and becomes one PLANNED order, due a lead time before its first need. It also publishes each order's components as
dependent requirements for replenishment. A re-run replaces the unfirmed planned orders and keeps the firmed ones."""
if weeks is not None and not 1 <= weeks <= 26:
raise Refused(422, "weeks is 1 to 26", gate="bounds")
with hub.repository() as repo, hub._decide():
return _run(repo, caller, warehouse_id, weeks)
@MFG.use_case(kind="command", route="POST /mrp/derive", permission=hub.PLAN_DERIVE, status=201, tool=False)
def derive_mrp(caller: dict, warehouse_id: Code, weeks: Optional[int] = None) -> dict:
"""The same run, for the composition's worker (a machine grant: no server-side scheduler exists, so something calls this)."""
if weeks is not None and not 1 <= weeks <= 26:
raise Refused(422, "weeks is 1 to 26", gate="bounds")
with hub.repository() as repo, hub._decide():
return _run(repo, caller, warehouse_id, weeks)
@MFG.use_case(kind="query", route="GET /planned-orders", permission="mfg:plan:run")
def list_planned_orders(caller: dict, warehouse_id: Code, status: Optional[str] = None) -> dict:
"""The planned orders of a warehouse (optionally of one status: proposed, firmed, converted, superseded), the cause of each, and
whether the last run is stale and why."""
with hub.repository() as repo, hub._decide():
run = repo.plan.last_run()
rows = repo.plan.planned(warehouse_id, (status,) if status else None)
return {"run": {"run_no": run["run_no"], "stale": bool(run["stale"]), "stale_cause": run["stale_cause"] or None} if run else None,
"planned_orders": [_view(r) for r in rows]}
def _need_planned(repo, ref: str) -> dict:
p = repo.plan.planned_by_ref(ref)
if p is None:
raise Refused(404, f"no planned order {ref}")
return p
@MFG.use_case(kind="command", route="POST /planned-orders/{planned_order_ref}/firm", permission="mfg:plan:firm")
def firm_planned_order(caller: dict, planned_order_ref: Code) -> dict:
"""Firm a proposed planned order: a later MRP run keeps it (and counts it as supply) instead of replacing it."""
with hub.repository() as repo, hub._decide():
p = _need_planned(repo, planned_order_ref)
if p["status"] != "proposed":
why = {"superseded": "a later run replaced it", "firmed": "it is already firmed", "converted": "it is already an order"}[p["status"]]
raise RuleBroken(f"{planned_order_ref} is {p['status']}: {why}", gate="state")
repo.plan.move_planned(p, "proposed", "firmed", firmed_by=hub._email(caller), firmed_sub=hub._sub(caller))
return _view({**p, "status": "firmed"})
@MFG.use_case(kind="command", route="POST /planned-orders/{planned_order_ref}/convert", permission="mfg:order:create", status=201)
def convert_planned_order(caller: dict, planned_order_ref: Code) -> dict:
"""Turn a FIRMED planned order into a production order (created: its BOM and routing are named now and copied when it is released).
The planned order is then converted and never proposed again; a second convert is refused and names the order."""
with hub.repository() as repo, hub._decide():
p = _need_planned(repo, planned_order_ref)
if p["status"] == "converted":
raise RuleBroken(f"{planned_order_ref} is already converted: order {p['production_order_id']}", gate="already_converted",
fix=f"open {p['production_order_id']}")
if p["status"] != "firmed":
raise RuleBroken(f"{planned_order_ref} is {p['status']}: firm it first", gate="planned_firmed", fix="firm the planned order")
got = repo.boms.get(Bom.make_id(p["product_id"], "production"))
company = got[0].company_code if got else ""
order = hub._new_order(repo, caller, company, p["product_id"], p["warehouse_id"], p["quantity"], p["need_date"], p["unit_code"],
planned_order_ref=planned_order_ref)
repo.plan.move_planned(p, "firmed", "converted", production_order_id=order.production_order_id)
return {"planned_order": _view({**p, "status": "converted", "production_order_id": order.production_order_id}),
"production_order": hub._order_view(repo, caller, order, 1)}
# ── what replenishment asks (MFG-27) ────────────────────────────────────────────────────────────────────────
@MFG.use_case(kind="query", route="GET /dependent-requirements", permission=hub.READ_REQUIREMENTS, tool=False,
aliases={"date_from": "from", "date_to": "to"})
def dependent_requirements(caller: dict, product_id: Code, warehouse_id: Code, date_from: Day, date_to: Day) -> dict:
"""What planned and released production orders will take of a component, per date, in its base unit - for inventory's replenishment
to add to its demand. It never raises a purchase request. (A route for inventory's service client, never a tool; the dates are the
query's `from` and `to`.)"""
with hub.repository() as repo, hub._decide():
a, b = hub._date(date_from, "from"), hub._date(date_to, "to")
if a > b:
raise Refused(422, "from is not after to", gate="bounds")
out = [{"required_on": r["need_date"].isoformat(), "quantity": _s(r["quantity"]), "source": "planned", "production_order_id": None}
for r in repo.plan.requirements(product_id, warehouse_id, a, b, LIVE)]
for o in repo.orders.open_for(warehouse_id, ("released", "in_progress")):
if f'"product_id": "{product_id}"' not in (o["frozen_bom_json"] or "") or not a <= o["due_on"] <= b:
continue
order, _, _ = repo.orders.get(o["production_order_id"])
issued = hub._net_issued(repo.movements.for_order(order.production_order_id))
left = sum((max(c["planned_base"] - issued.get(c["line_no"], Decimal("0")), Decimal("0")) for c in order.planned_components()
if c["product_id"] == product_id), Decimal("0"))
if left > 0:
out.append({"required_on": order.due_on.isoformat(), "quantity": _s(left), "source": "released",
"production_order_id": order.production_order_id})
out.sort(key=lambda r: (r["required_on"], r["source"], r["production_order_id"] or ""))
return {"product_id": product_id, "warehouse_id": warehouse_id, "requirements": out}