From 30a10adabc878675b66206fb585ee7c5c3ceed2c Mon Sep 17 00:00:00 2001 From: rob Date: Sat, 30 May 2026 19:16:28 -0300 Subject: [PATCH] Phase 5: event-sourced inventory ledger model MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- src/woodshop/inventory.py | 198 ++++++++++++++++++++++++++++++++++++++ tests/test_inventory.py | 79 +++++++++++++++ 2 files changed, 277 insertions(+) create mode 100644 src/woodshop/inventory.py create mode 100644 tests/test_inventory.py diff --git a/src/woodshop/inventory.py b/src/woodshop/inventory.py new file mode 100644 index 0000000..ec88440 --- /dev/null +++ b/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), + } diff --git a/tests/test_inventory.py b/tests/test_inventory.py new file mode 100644 index 0000000..f063227 --- /dev/null +++ b/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}