Phase 5: event-sourced inventory ledger model

inventory.py — shop-wide, append-only event log; current state derived by
folding (Codex's recommendation). Lean and plumbing-only.

- Events: purchase / consume / create_offcut / discard / adjustment /
  build_recorded. Ledger.load/save to $XDG_DATA_HOME/woodshop/inventory.json.
- Primitives: purchase, adjust, consume_offcut, discard_offcut, record_build
  (deducts consumed stock, keeps/discards each produced offcut).
- Derived: on_hand(), offcut_bin(), available_stock() (one Piece shape for full
  stock + offcuts — the AvailableStock interface), builds(), stats() (spent,
  units built, offcuts kept/burned/trashed, per-project units).
- plan_consumption(plan): derive consumed stock + reusable offcuts from a CutPlan.
- tests: purchase/consume, adjustment, keep/discard offcuts, consume removes
  from bin, plan_consumption, available_stock, save/load, cross-project stats.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/woodshop/inventory.py

@@ -0,0 +1,198 @@
+"""Shop-wide inventory as an append-only event ledger.
+
+Per the plan (and Codex's recommendation), the source of truth is a list of
+immutable events; current state (on-hand stock, the offcut bin, build history,
+stats) is *derived* by folding them. This gives history/audit/stats for free and
+avoids "why is my inventory wrong" drift. The ledger is plumbing — the GUI only
+ever shows the three workflows (purchase / record build / use offcuts) and a
+read-only management view.
+
+Events (each a dict with a "type"):
+  purchase       {stock, qty, is_sheet, price, date}
+  consume        {stock, qty}            OR  {offcut_id}
+  create_offcut  {offcut: {...}}
+  discard        {offcut_id, fate: burned|trashed}
+  adjustment     {stock, delta, reason}
+  build_recorded {build_id, project, units, cost, date}
+
+Shop-wide, stored in $XDG_DATA_HOME/woodshop/inventory.json.
+"""
+from __future__ import annotations
+
+import json
+import os
+from collections import Counter
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+
+from .lumber import is_plywood, normalize_stock
+
+
+@dataclass
+class Piece:
+    """An available piece of stock — a full unit or a reusable offcut. The
+    planner can consume both through this one shape (AvailableStock)."""
+    id: str
+    stock: str
+    length_in: float
+    width_in: float
+    is_sheet: bool
+    material: str = ""
+    is_offcut: bool = False
+    source_project: str = ""
+    bin: str = ""
+
+
+def _data_path() -> Path:
+    base = Path(os.environ.get("XDG_DATA_HOME", "~/.local/share")).expanduser() / "woodshop"
+    return base / "inventory.json"
+
+
+def plan_consumption(plan) -> tuple[dict, list[dict]]:
+    """From a CutPlan, derive (stock consumed as {stock: qty}, reusable offcuts).
+    Offcuts are the reusable waste regions; ids are assigned later by the ledger."""
+    consumed = Counter(sp.stock for sp in plan.stock_pieces)
+    offcuts = []
+    for sp in plan.stock_pieces:
+        for w in sp.waste:
+            if not w.reusable:
+                continue
+            offcuts.append({
+                "stock": sp.stock,
+                "length_in": round(w.length_in, 2),
+                "width_in": round(w.width_in or sp.width_in, 2),
+                "is_sheet": sp.is_sheet,
+            })
+    return dict(consumed), offcuts
+
+
+class Ledger:
+    def __init__(self, events: list | None = None):
+        self.events = list(events or [])
+
+    # ----- persistence --------------------------------------------------
+    @classmethod
+    def load(cls, path: Path | None = None) -> "Ledger":
+        path = path or _data_path()
+        if path.exists():
+            try:
+                return cls(json.loads(path.read_text()).get("events", []))
+            except (ValueError, OSError):
+                pass
+        return cls()
+
+    def save(self, path: Path | None = None) -> None:
+        path = path or _data_path()
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps({"events": self.events}, indent=2))
+
+    def _emit(self, type_: str, **fields) -> dict:
+        e = {"type": type_, **fields}
+        self.events.append(e)
+        return e
+
+    def _next_build_id(self) -> str:
+        n = sum(1 for e in self.events if e["type"] == "build_recorded") + 1
+        return f"b{n}"
+
+    # ----- primitives (each appends events) -----------------------------
+    def purchase(self, stock: str, qty: int, price: float | None = None,
+                 date: str = "", is_sheet: bool | None = None) -> None:
+        stock = normalize_stock(stock)
+        sheet = is_plywood(stock) if is_sheet is None else is_sheet
+        self._emit("purchase", stock=stock, qty=int(qty), is_sheet=sheet,
+                   price=price, date=date)
+
+    def adjust(self, stock: str, delta: int, reason: str = "", date: str = "") -> None:
+        self._emit("adjustment", stock=normalize_stock(stock), delta=int(delta),
+                   reason=reason, date=date)
+
+    def discard_offcut(self, offcut_id: str, fate: str, date: str = "") -> None:
+        self._emit("discard", offcut_id=offcut_id, fate=fate, date=date)
+
+    def consume_offcut(self, offcut_id: str) -> None:
+        self._emit("consume", offcut_id=offcut_id)
+
+    def record_build(self, project: str, units: int, consumed: dict,
+                     offcuts: list[dict], keep_ids=None, fates: dict | None = None,
+                     cost: float | None = None, date: str = "") -> str:
+        """Record a build: deduct consumed full stock, then for each produced
+        offcut either keep it (create_offcut) or discard it (burned/trashed).
+        `keep_ids`/`fates` are keyed by the offcut's assigned id (oc index order).
+        Returns the build id."""
+        build_id = self._next_build_id()
+        for stock, qty in consumed.items():
+            if qty:
+                self._emit("consume", stock=normalize_stock(stock), qty=int(qty),
+                           build_id=build_id)
+        keep = set(keep_ids) if keep_ids is not None else set()
+        fates = fates or {}
+        for i, oc in enumerate(offcuts):
+            oid = f"{build_id}-o{i + 1}"
+            if oid in keep or (keep_ids is None):     # default: keep all offcuts
+                self._emit("create_offcut", offcut={**oc, "id": oid,
+                           "source_project": project, "source_build": build_id})
+            else:
+                self._emit("discard", offcut_id=oid, fate=fates.get(oid, "trashed"))
+        self._emit("build_recorded", build_id=build_id, project=project,
+                   units=int(units), cost=cost, date=date)
+        return build_id
+
+    # ----- derived state (fold the events) ------------------------------
+    def on_hand(self) -> dict:
+        c = Counter()
+        for e in self.events:
+            if e["type"] == "purchase":
+                c[e["stock"]] += e["qty"]
+            elif e["type"] == "consume" and "stock" in e:
+                c[e["stock"]] -= e["qty"]
+            elif e["type"] == "adjustment":
+                c[e["stock"]] += e["delta"]
+        return {k: v for k, v in sorted(c.items()) if v}
+
+    def offcut_bin(self) -> list[Piece]:
+        live: dict[str, dict] = {}
+        for e in self.events:
+            if e["type"] == "create_offcut":
+                oc = e["offcut"]
+                live[oc["id"]] = oc
+            elif e["type"] in ("discard",) and e.get("offcut_id") in live:
+                del live[e["offcut_id"]]
+            elif e["type"] == "consume" and e.get("offcut_id") in live:
+                del live[e["offcut_id"]]
+        return [Piece(id=oc["id"], stock=oc["stock"], length_in=oc["length_in"],
+                      width_in=oc["width_in"], is_sheet=oc["is_sheet"], is_offcut=True,
+                      source_project=oc.get("source_project", ""), bin=oc.get("bin", ""))
+                for oc in live.values()]
+
+    def available_stock(self) -> list[Piece]:
+        """Everything the planner could consume: full on-hand units + offcuts."""
+        pieces = list(self.offcut_bin())
+        for stock, qty in self.on_hand().items():
+            for i in range(qty):
+                pieces.append(Piece(id=f"{stock}#{i + 1}", stock=stock,
+                                    length_in=0.0, width_in=0.0,
+                                    is_sheet=is_plywood(stock)))
+        return pieces
+
+    def builds(self) -> list[dict]:
+        return [e for e in self.events if e["type"] == "build_recorded"]
+
+    def stats(self) -> dict:
+        spent = sum((e.get("price") or 0.0) * e["qty"]
+                    for e in self.events if e["type"] == "purchase")
+        units = sum(e["units"] for e in self.builds())
+        fates = Counter(e["fate"] for e in self.events if e["type"] == "discard")
+        kept = sum(1 for e in self.events if e["type"] == "create_offcut")
+        by_project = Counter()
+        for e in self.builds():
+            by_project[e["project"]] += e["units"]
+        return {
+            "spent": round(spent, 2),
+            "units_built": units,
+            "builds": len(self.builds()),
+            "offcuts_kept": kept,
+            "offcuts_burned": fates.get("burned", 0),
+            "offcuts_trashed": fates.get("trashed", 0),
+            "by_project": dict(by_project),
+        }

tests/test_inventory.py

@@ -0,0 +1,79 @@
+"""Phase 5: event-sourced inventory ledger (no GUI)."""
+from woodshop.cutplan import build_cut_plan
+from woodshop.inventory import Ledger, plan_consumption
+from woodshop.scene import Scene
+
+
+def test_purchase_then_consume_on_hand():
+    led = Ledger()
+    led.purchase("2x4", 5)
+    led.purchase("ply-3/4", 2)
+    assert led.on_hand() == {"2x4": 5, "ply-3/4": 2}
+    led.record_build("table", 1, consumed={"2x4": 2}, offcuts=[])
+    assert led.on_hand()["2x4"] == 3
+
+
+def test_adjustment_corrects_count():
+    led = Ledger()
+    led.purchase("2x4", 5)
+    led.adjust("2x4", -1, reason="broke one")
+    assert led.on_hand()["2x4"] == 4
+
+
+def test_record_build_keeps_and_discards_offcuts():
+    led = Ledger()
+    offcuts = [{"stock": "2x4", "length_in": 20, "width_in": 3.5, "is_sheet": False},
+               {"stock": "2x4", "length_in": 8, "width_in": 3.5, "is_sheet": False}]
+    bid = led.record_build("shelf", 1, consumed={"2x4": 1}, offcuts=offcuts,
+                           keep_ids={f"{ 'b1'}-o1"}, fates={"b1-o2": "burned"})
+    bin_ = led.offcut_bin()
+    assert len(bin_) == 1 and bin_[0].id == "b1-o1"          # kept the 20"
+    stats = led.stats()
+    assert stats["offcuts_kept"] == 1 and stats["offcuts_burned"] == 1
+
+
+def test_consume_offcut_removes_it_from_bin():
+    led = Ledger()
+    led.record_build("a", 1, consumed={}, offcuts=[
+        {"stock": "2x4", "length_in": 20, "width_in": 3.5, "is_sheet": False}])
+    oid = led.offcut_bin()[0].id
+    led.consume_offcut(oid)
+    assert led.offcut_bin() == []
+
+
+def test_plan_consumption_from_cutplan():
+    s = Scene()
+    s.place("2x4", 60)            # leaves a reusable ~35" offcut on a 96" stick
+    plan = build_cut_plan(s)
+    consumed, offcuts = plan_consumption(plan)
+    assert consumed.get("2x4") == 1
+    assert offcuts and offcuts[0]["stock"] == "2x4" and offcuts[0]["length_in"] > 12
+
+
+def test_available_stock_combines_full_and_offcuts():
+    led = Ledger()
+    led.purchase("2x4", 2)
+    led.record_build("a", 1, consumed={}, offcuts=[
+        {"stock": "2x4", "length_in": 20, "width_in": 3.5, "is_sheet": False}])
+    avail = led.available_stock()
+    assert sum(1 for p in avail if p.is_offcut) == 1
+    assert sum(1 for p in avail if not p.is_offcut and p.stock == "2x4") == 2
+
+
+def test_ledger_save_load_roundtrip(tmp_path, monkeypatch):
+    monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path))
+    led = Ledger()
+    led.purchase("2x4", 3, price=3.98)
+    led.save()
+    again = Ledger.load()
+    assert again.on_hand()["2x4"] == 3
+    assert again.stats()["spent"] == round(3 * 3.98, 2)
+
+
+def test_stats_aggregate_across_projects():
+    led = Ledger()
+    led.record_build("table", 2, consumed={}, offcuts=[])
+    led.record_build("shelf", 1, consumed={}, offcuts=[])
+    st = led.stats()
+    assert st["units_built"] == 3 and st["builds"] == 2
+    assert st["by_project"] == {"table": 2, "shelf": 1}