dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
src/brick_manufacturing/application/migration.py
"""The Odoo slice (MFG-32): Odoo 17 mrp - work centres, bills of materials, routings and the done and open production orders - carried read-only,
reconciled to Odoo per product and per month, and signed by the owner (workflow mfg_migration_run: load -> reconcile -> sign | discard).
The load is idempotent: every carried row is recorded in mfg_migration_map on (odoo_db, odoo_model, odoo_id), so a second load writes nothing
new. History is never re-priced: a done order is a read-only CLOSED order with its output, its lots and its minutes - no value, no journal (gl's
migration carries Odoo's entries). A BOM becomes released version 1 valid from the cut-over; normal -> usage production, phantom -> usage sale,
subcontract stays purchasing's. Open orders are finished in Odoo (the default: the run stops and lists them) or re-created here as `created`."""
import json
from datetime import timedelta
from decimal import Decimal, InvalidOperation
from typing import Annotated, Literal
from erp_platform import work
from erp_platform.bricks import Refused
from pydantic import Field
from brick_manufacturing.application import use_cases as hub
from brick_manufacturing.application.inputs import Code, Day, 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.infrastructure import odoo as odoo_port
from brick_manufacturing.infrastructure import seams
MFG = hub.MFG
_s = hub._s
CHECKS = (("boms_carried", "equal"), ("output_per_product_and_month", "equal"), ("genealogy_of_output_lots", "equal"),
("phantom_boms_mapped_to_usage_sale", "explained"))
def _dec(v, what: str) -> Decimal:
try:
d = Decimal(str(v))
except (InvalidOperation, ValueError):
raise OutOfBounds(f"Odoo's {what} {v!r} is not a number") from None
if not d.is_finite() or d < 0 or d.as_tuple().exponent < -4 or d > Decimal("999999999999999"):
raise OutOfBounds(f"Odoo's {what} {v!r} is not a quantity this brick can hold (zero or more, four places)")
return d
def _reader():
try:
return odoo_port.ports.get()
except odoo_port.OdooUnavailable as down:
raise Refused(down.http, down.reason) from None
def _view(run: dict) -> dict:
return {"run_id": run["run_id"], "odoo_db": run["odoo_db"], "company_code": run["company_code"], "cutover_date": run["cutover_date"].isoformat(),
"status": run["status"], "doors": json.loads(run["doors_json"] or "{}"), "summary": json.loads(run["summary_json"] or "{}"),
"checks": json.loads(run["checks_json"] or "[]"), "loaded_by": run["loaded_by"], "signed_by": run["signed_by"] or None,
"signed_on": run["signed_on"].isoformat() if run["signed_on"] else None, "discard_reason": run["discard_reason"] or None}
def _need_run(repo, run_id: str) -> dict:
run = repo.migration.run(run_id)
if run is None:
raise Refused(404, f"no migration run {run_id}")
return run
@MFG.use_case(kind="command", route="POST /migration/odoo/load", permission=hub.MIG_LOAD, status=201)
def load_mfg_from_odoo(caller: dict, odoo_db: Code, cutover_date: Day, company_code: Code, warehouse_id: Code, cost_centre: Code,
byproduct_nrv: Annotated[dict[Annotated[str, Field(max_length=64)], Annotated[str, Field(max_length=24)]], Field(max_length=500)],
years: Annotated[list[Annotated[int, Field(ge=2000, le=2100)]], Field(min_length=1, max_length=10)],
open_orders: Literal["finish_in_odoo", "recreate"] = "finish_in_odoo") -> dict:
"""Load Odoo's manufacturing, read-only, in one transaction. The doors are the owner's: `cost_centre` (gl's, for every work centre: Odoo has
none), `byproduct_nrv` (each by-product's net realisable value per base unit - Odoo's cost_share is not one), `years` of done orders to carry
for traceability, and what to do with open orders (finish them in Odoo, the default, or re-create them here as created orders). Idempotent
on the map: a second load writes nothing new. Reads nothing in Odoo but mrp models."""
rd = _reader()
with hub.repository() as repo, hub._decide():
today = hub._today(repo)
cutover = hub._date(cutover_date, "cutover_date")
try:
cc = seams.cost_centre(cost_centre)
except seams.Unavailable as e:
if e.http != 404:
raise
raise RuleBroken(f"gl has no cost centre {cost_centre}", gate="cost_centre") from None
if cc is not None and cc.get("status") != "open":
raise RuleBroken(f"Cost centre {cost_centre} is {cc.get('status')}", gate="cost_centre")
nrv = {k: _dec(v, f"net realisable value of {k}") for k, v in byproduct_nrv.items()}
try:
wcs, boms, prods = rd.workcentres(odoo_db), rd.boms(odoo_db), rd.productions(odoo_db, years)
except odoo_port.OdooUnavailable as down:
raise Refused(down.http, down.reason) from None
open_ = [p for p in prods if p["state"] in ("draft", "confirmed", "progress", "to_close")]
if open_ and open_orders == "finish_in_odoo":
raise RuleBroken("Open orders in Odoo: " + ", ".join(f"{p['name']} ({p['state']})" for p in open_) + " - finish them in Odoo, or load again with "
"open_orders=recreate", gate="open_orders", fix="finish them in Odoo (recommended) or choose recreate")
db = odoo_db
n = repo.claims.counts("migration_run", [db])[db] + 1
run_id = f"mfg-odoo-{db}-{n}"
repo.claims.take([("migration_run", db, n, run_id)], hub._email(caller), work.now())
mapped = {m: repo.migration.mapped(db, m) for m in ("mrp.workcenter", "mrp.bom", "mrp.production")}
maps, summary = [], {"work_centres": 0, "boms": 0, "sale_boms": 0, "routings": 0, "subcontract_skipped": 0, "closed_orders": 0, "open_orders": 0,
"confirmations": 0}
cur = repo.settings.all(today)["currency"]
# work centres: costs_hour is the machine rate from the cut-over
for w in wcs:
wid = str(w["id"])
if wid in mapped["mrp.workcenter"]:
continue
code = str(w.get("code") or w["name"]).strip()[:64] or f"WC-{wid}"
if repo.masters.work_centre(code):
raise RuleBroken(f"Work centre {code} exists already and is not from Odoo's {wid}: name it at the door", gate="work_centre_exists")
repo.masters.add_work_centre({"work_centre_id": code, "company_code": company_code, "warehouse_id": warehouse_id, "cost_centre": cost_centre,
"batch_capacity": _dec(w.get("capacity") or 1, "capacity"), "capacity_unit_code": "", "hours_per_day": Decimal(8),
"status": "active"}, work.now())
repo.masters.add_rates([{"work_centre_id": code, "rate_kind": "machine", "valid_from": cutover,
"rate_amount": _dec(w.get("costs_hour") or 0, "costs_hour"), "currency": cur, "set_by": "migration"}], work.now())
maps.append({"odoo_db": db, "odoo_model": "mrp.workcenter", "odoo_id": wid, "our_key": code, "run_id": run_id})
mapped["mrp.workcenter"][wid] = code
summary["work_centres"] += 1
wc_code = mapped["mrp.workcenter"]
# bills of materials -> released version 1 valid from the cut-over, and a routing from the operations
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
for b in boms:
bid_odoo = str(b["id"])
if bid_odoo in mapped["mrp.bom"]:
continue
if b["type"] == "subcontract":
summary["subcontract_skipped"] += 1 # stays purchasing's
continue
usage = "sale" if b["type"] == "phantom" else "production"
bom_id = Bom.make_id(b["product"], usage)
if repo.boms.get(bom_id):
raise RuleBroken(f"{bom_id} exists already: a second Odoo BOM for one product and usage is an alternative - name it at the door",
gate="bom_exists")
bom = Bom(bom_id, b["product"], usage, "", company_code)
out_q = _dec(b["qty"], "product_qty")
out_base, out_unit = base(b["product"], out_q, b["uom"])
drafts, n_line = [], 0
for ln in b.get("lines", []):
n_line += 1
q = _dec(ln["qty"], "line quantity")
qb, bu = base(ln["product"], q, ln["uom"])
drafts.append(BomLine(0, n_line, "component", ln["product"], q, ln["uom"], qb, bu))
for bp in b.get("byproducts", []):
n_line += 1
if bp["product"] not in nrv:
raise RuleBroken(f"By-product {bp['product']} of Odoo BOM {bid_odoo}: say its net realisable value (byproduct_nrv) - Odoo's cost_share "
"is not one", gate="byproduct_nrv", fix="add it to byproduct_nrv")
q = _dec(bp["qty"], "by-product quantity")
qb, bu = base(bp["product"], q, bp["uom"])
drafts.append(BomLine(0, n_line, "by_product", bp["product"], q, bp["uom"], qb, bu, nrv_unit_cost=nrv[bp["product"]]))
bom.write_version(1, valid_from=cutover, output_quantity=out_q, output_unit_code=b["uom"], output_quantity_base=out_base,
output_base_unit_code=out_unit, lines=drafts, edges=repo.boms.edges(bom_id))
bom.release_bom(1, today=cutover - timedelta(days=1), by="migration", sub="migration") # released as of the day before the cut-over
repo.boms.save(bom, 0)
maps.append({"odoo_db": db, "odoo_model": "mrp.bom", "odoo_id": bid_odoo, "our_key": bom_id, "run_id": run_id})
summary["sale_boms" if usage == "sale" else "boms"] += 1
ops = [o for o in b.get("operations", []) if str(o.get("workcenter_id")) in wc_code]
if usage == "production" and ops and not repo.masters.routings(b["product"]):
repo.masters.add_routing({"product_id": b["product"], "routing_version": 1, "valid_from": cutover, "written_by": "migration",
"operations_json": json.dumps([{"op_no": int(o["sequence"]), "description": str(o["name"])[:200],
"work_centre_id": wc_code[str(o["workcenter_id"])], "setup_minutes": "0",
"run_minutes": _s(_dec(o.get("time_cycle_manual") or 0, "time_cycle_manual")),
"per_quantity": "1", "basis": "output", "unit_code": "", "wait_hours": "0"}
for o in sorted(ops, key=lambda x: x["sequence"])])}, work.now())
summary["routings"] += 1
# production orders: done -> closed history (output, lots, minutes); open -> created, if the door says so
rows, genealogy, confirmations = [], [], []
for p in prods:
pid = str(p["id"])
if pid in mapped["mrp.production"] or p["state"] == "cancel":
continue
oid = f"odoo:{db}:mrp.production:{pid}"
done = p["state"] == "done"
finished = hub._date(p.get("date_finished"), "date_finished") if done else None
qty = _dec(p["product_qty"], "product_qty")
if done:
produced = _dec(p["qty_produced"], "qty_produced")
order = ProductionOrder(oid, company_code, p["product"], warehouse_id, qty, p["uom"], finished, f"{p['product']}:production", 0,
json.dumps({"output_quantity": _s(qty), "output_quantity_base": _s(qty), "output_unit_code": p["uom"],
"output_base_unit_code": p["uom"], "lines": []}), "closed", "migration", finished, "migration", "",
finished, finished, "")
repo.orders.save(order, 0)
ref = f"{oid}:output"
rows.append({"move_ref": ref, "production_order_id": oid, "kind": "output", "line_no": 0, "seq": 1, "product_id": p["product"],
"lot_code": p.get("lot_producing") or "", "quantity": produced, "unit_code": p["uom"], "quantity_base": produced,
"base_unit_code": p["uom"], "location_id": "", "over_issue": False, "status": "sent", "recorded_by": "migration",
"recorded_sub": "migration", "business_date": finished})
if p.get("lot_producing"):
for c in p.get("consumed", []):
if c.get("lot"):
genealogy.append({"output_product_id": p["product"], "output_lot_code": p["lot_producing"], "input_product_id": c["product"],
"input_lot_code": c["lot"], "production_order_id": oid, "input_quantity": _dec(c["qty"], "consumed quantity"),
"output_quantity": produced})
for i, wo in enumerate(sorted(p.get("workorders", []), key=lambda w: w.get("sequence", 0)), 1):
mins = _dec(wo.get("duration") or 0, "duration")
confirmations.append({"production_order_id": oid, "op_no": i, "confirmation_no": 1,
"work_centre_id": wc_code.get(str(wo.get("workcenter_id")), ""), "machine_minutes": mins, "labour_minutes": mins,
"is_final": True, "cost": Decimal("0"), "journal_id": "", "reverses_confirmation_no": 0,
"recorded_by": "migration", "recorded_sub": "migration", "business_date": finished, "status": "recorded",
"outbox_ref": ""})
summary["closed_orders"] += 1
else:
bom_key = f"{p['product']}:production"
got = repo.boms.get(bom_key)
v = got[0].valid_on(max(cutover, today)) if got else None
if v is None:
raise RuleBroken(f"Open Odoo order {p['name']} makes {p['product']}, which has no released BOM here", gate="bom_valid")
order = ProductionOrder(oid, company_code, p["product"], warehouse_id, qty, v.output_unit_code, max(cutover, today), bom_key,
v.version, created_by="migration")
repo.orders.save(order, 0)
summary["open_orders"] += 1
maps.append({"odoo_db": db, "odoo_model": "mrp.production", "odoo_id": pid, "our_key": oid, "run_id": run_id})
repo.movements.add(rows, work.now())
if genealogy:
repo.facts.genealogy(genealogy)
for c in confirmations:
repo.floor.add_confirmation(c, work.now())
summary["confirmations"] += 1
repo.migration.add_map(maps, work.now())
repo.migration.add_run({"run_id": run_id, "odoo_db": db, "company_code": company_code, "cutover_date": cutover, "status": "loaded",
"doors_json": hub.dump({"cost_centre": cost_centre, "warehouse_id": warehouse_id, "open_orders": open_orders,
"years": years, "byproduct_nrv": {k: _s(v) for k, v in nrv.items()}}),
"checks_json": "[]", "summary_json": hub.dump(summary), "loaded_by": hub._email(caller), "signed_by": "", "signed_sub": "",
"discard_reason": ""}, work.now())
return _view(repo.migration.run(run_id))
@MFG.use_case(kind="command", route="POST /migration/odoo/{run_id}/reconcile", permission=hub.MIG_RECONCILE)
def reconcile_mfg_migration(caller: dict, run_id: Code) -> dict:
"""Compare what was carried with Odoo, check by check: BOMs carried (normal + phantom), output quantity per product and month, the genealogy
of every output lot, and phantom BOMs mapped to usage sale. Each check says Odoo's figure, ours, and whether they are equal; the owner signs when
every one is equal or explained."""
rd = _reader()
with hub.repository() as repo, hub._decide():
run = _need_run(repo, run_id)
if run["status"] != "loaded":
raise RuleBroken(f"Run {run_id} is {run['status']}: only a loaded run is reconciled", gate="state")
db, doors = run["odoo_db"], json.loads(run["doors_json"])
try:
boms, prods = rd.boms(db), rd.productions(db, doors["years"])
except odoo_port.OdooUnavailable as down:
raise Refused(down.http, down.reason) from None
carried = [b for b in boms if b["type"] != "subcontract"]
phantom = [b for b in boms if b["type"] == "phantom"]
ours_boms = repo.migration.mapped(db, "mrp.bom")
sale = [k for k in ours_boms.values() if k.endswith(":sale")]
theirs_out: dict = {}
for p in prods:
if p["state"] == "done":
k = f"{p['product']}|{str(p['date_finished'])[:7]}"
theirs_out[k] = _s(hub.q4(Decimal(theirs_out.get(k, "0")) + _dec(p["qty_produced"], "qty_produced")))
ours_out: dict = {}
for m in repo.migration.migrated_outputs(db):
k = f"{m['product_id']}|{m['business_date'].isoformat()[:7]}"
ours_out[k] = _s(hub.q4(Decimal(ours_out.get(k, "0")) + m["quantity_base"]))
theirs_gen = sorted(f"{p['product']}|{p['lot_producing']}|{c['product']}|{c['lot']}" for p in prods if p["state"] == "done" and p.get("lot_producing")
for c in p.get("consumed", []) if c.get("lot"))
ours_gen = sorted(f"{g['output_product_id']}|{g['output_lot_code']}|{g['input_product_id']}|{g['input_lot_code']}"
for g in repo.migration.migrated_genealogy(db))
checks = [{"check": "boms_carried", "must": "equal", "odoo": len(carried), "ours": len(ours_boms), "equal": len(carried) == len(ours_boms)},
{"check": "output_per_product_and_month", "must": "equal", "odoo": theirs_out, "ours": ours_out, "equal": theirs_out == ours_out},
{"check": "genealogy_of_output_lots", "must": "equal", "odoo": len(theirs_gen), "ours": len(ours_gen), "equal": theirs_gen == ours_gen},
{"check": "phantom_boms_mapped_to_usage_sale", "must": "explained", "odoo": len(phantom), "ours": len(sale), "equal": len(phantom) == len(sale)}]
repo.migration.update_run(run_id, status="reconciled", checks_json=hub.dump(checks))
return _view(repo.migration.run(run_id))
def sign_gates(checks: list, explanations: dict) -> list:
"""The gate of sign_mfg_reconciliation as {gate, met, reason, fix}: every check equal, or explained by the owner."""
open_ = [c["check"] for c in checks if not c["equal"] and not (explanations.get(c["check"]) or "").strip()]
return [{"gate": "checks_reconciled", "met": bool(checks) and not open_,
"reason": ("checks neither equal nor explained: " + "; ".join(open_)) if open_ else ("the run has not been reconciled" if not checks else ""),
"fix": "fix the mapping and load again, or explain each difference"}]
@MFG.use_case(kind="command", route="POST /migration/odoo/{run_id}/sign", permission=hub.MIG_SIGN)
def sign_mfg_reconciliation(caller: dict, run_id: Code, explanations: Annotated[dict[Annotated[str, Field(max_length=64)], Annotated[str, Field(max_length=500)]],
Field(max_length=20)] = {}) -> dict: # noqa: B006 (never mutated)
"""The owner signs that nothing was lost: every check equal, or explained in the owner's words. A signed run is frozen - nothing loads into it
and it is never reopened."""
with hub.repository() as repo, hub._decide():
run = _need_run(repo, run_id)
if run["status"] != "reconciled":
raise RuleBroken(f"Run {run_id} is {run['status']}: only a reconciled run is signed", gate="state")
gates = sign_gates(json.loads(run["checks_json"]), explanations)
if not all(g["met"] for g in gates):
g = next(g for g in gates if not g["met"])
raise RuleBroken(g["reason"], gate=g["gate"], fix=g["fix"])
checks = [{**c, "explained": explanations.get(c["check"], "") if not c["equal"] else ""} for c in json.loads(run["checks_json"])]
repo.migration.update_run(run_id, status="signed", checks_json=hub.dump(checks), signed_by=hub._email(caller), signed_sub=hub._sub(caller),
signed_on=hub._today(repo))
return _view(repo.migration.run(run_id))
@MFG.use_case(kind="command", route="POST /migration/odoo/{run_id}/discard", permission=hub.MIG_SIGN)
def discard_mfg_migration(caller: dict, run_id: Code, reason: Reason) -> dict:
"""Discard a loaded or reconciled run with a reason. It removes nothing already carried (history is never deleted, and a second load is
idempotent on the map); it only says this run was not accepted."""
with hub.repository() as repo, hub._decide():
run = _need_run(repo, run_id)
if run["status"] not in ("loaded", "reconciled"):
raise RuleBroken(f"Run {run_id} is {run['status']}: only a loaded or reconciled run is discarded", gate="state")
if not reason.strip():
raise OutOfBounds("a discard needs a reason")
repo.migration.update_run(run_id, status="discarded", discard_reason=reason.strip())
return _view(repo.migration.run(run_id))
@MFG.use_case(kind="query", route="GET /migration/odoo/{run_id}", permission=hub.MIG_READ)
def get_mfg_migration_run(caller: dict, run_id: Code) -> dict:
"""One migration run: its doors, what it carried, its checks and its status."""
with hub.repository() as repo, hub._decide():
return _view(_need_run(repo, run_id))
@MFG.use_case(kind="query", route="GET /migration-map/{odoo_db}/{odoo_model}", permission=hub.MIG_READ)
def get_mfg_migration_map(caller: dict, odoo_db: Code, odoo_model: Code) -> dict:
"""Which of our rows each Odoo record became (mrp.workcenter, mrp.bom, mrp.production): {odoo_id: our key}."""
with hub.repository() as repo:
return {"odoo_db": odoo_db, "odoo_model": odoo_model, "map": repo.migration.mapped(odoo_db, odoo_model)}