dodil/erp-brick-manufacturing

DODIL ERP brick: manufacturing

src/brick_manufacturing/application/trace.py

"""Traceability (MFG-16): a batch forward to everything made from it, a bag back to its beans.

The genealogy is DERIVED, never typed: every output reported on an order writes an edge input lot -> order -> output lot (Regulation (EC)
No 178/2002 art. 18, one step back and one step forward). A trace walks those edges one level at a time - one statement per level
whatever the number of lots - and asks inventory where each lot went (customers, shops, on hand) or came from (the supplier's receipt);
a lot inventory cannot answer is listed as unaccounted, so a recall sees the gap instead of a quiet 'nothing'."""
from typing import Literal

from erp_platform.bricks import Refused

from brick_manufacturing.application import use_cases as hub
from brick_manufacturing.application.inputs import Code
from brick_manufacturing.infrastructure import seams

MFG = hub.MFG
MAX_DEPTH = 25


def _walk(repo, root: tuple, forward: bool) -> tuple:
    """([{product_id, lot_code, depth, order}], [edges]) from the root lot, one query per level."""
    seen, found, edges, frontier = {root}, [], [], [root]
    for depth in range(1, MAX_DEPTH + 1):
        rows = repo.trace.outputs_of(frontier) if forward else repo.trace.inputs_of(frontier)
        frontier = []
        for r in sorted(rows, key=lambda x: (x["output_product_id"], x["output_lot_code"], x["input_product_id"], x["input_lot_code"])):
            src, dst = ((r["input_product_id"], r["input_lot_code"]), (r["output_product_id"], r["output_lot_code"]))
            edges.append({"input": {"product_id": src[0], "lot_code": src[1]}, "order": r["production_order_id"],
                          "output": {"product_id": dst[0], "lot_code": dst[1]}, "input_quantity": hub._s(r["input_quantity"]),
                          "output_quantity": hub._s(r["output_quantity"])})
            nxt = dst if forward else src
            if nxt not in seen:
                seen.add(nxt)
                found.append({"product_id": nxt[0], "lot_code": nxt[1], "depth": depth, "order": r["production_order_id"]})
                frontier.append(nxt)
        if not frontier:
            break
    return found, edges


@MFG.use_case(kind="query", route="GET /trace/{product_id}/{lot_code}", permission="mfg:trace:read")
def trace_lot(caller: dict, product_id: Code, lot_code: Code, direction: Literal["forward", "back"] = "forward") -> dict:
    """Trace a lot FORWARD (every roasted and bagged lot made from it, and for each lot inventory's answer of where it went: customers,
    shops, on hand) or BACK (every lot that went into it, and the supplier's receipt inventory knows). Derived from the orders' issues and
    outputs. A lot inventory cannot answer is listed as unaccounted."""
    with hub.repository() as repo, hub._decide():
        root = (product_id, lot_code)
        found, edges = _walk(repo, root, direction == "forward")
        if not found:
            asked = seams.lot_trace(product_id, lot_code)
            if not asked:
                raise Refused(404, f"no trace for {product_id} lot {lot_code}: no order used or made it, and inventory does not know it")
    lots, unaccounted = [], []
    with hub._decide():
        for item in [{"product_id": product_id, "lot_code": lot_code, "depth": 0, "order": None}] + found:
            inv = seams.lot_trace(item["product_id"], item["lot_code"])
            if not inv:
                unaccounted.append({"product_id": item["product_id"], "lot_code": item["lot_code"]})
            lots.append({**item, "inventory": {k: inv.get(k, []) for k in ("delivered_to", "on_hand", "history")} if inv else None})
    return {"product_id": product_id, "lot_code": lot_code, "direction": direction, "lots": lots, "edges": edges, "unaccounted": unaccounted}