From 3fefa148b13ea2e1bf05759b13e5cd9f58d8742f Mon Sep 17 00:00:00 2001 From: Mike Arpaia Date: Wed, 30 Sep 2026 22:15:15 -0600 Subject: [PATCH] Model material inventory with explicit quantities and provenance --- scripts/check_install.py | 1 + src/lab/artifacts.py | 76 ++++++++++++++ src/lab/inventory.py | 213 +++++++++++++++++++++++++++++++++++++++ tests/test_inventory.py | 50 +++++++++ 4 files changed, 340 insertions(+) create mode 100644 src/lab/artifacts.py create mode 100644 src/lab/inventory.py create mode 100644 tests/test_inventory.py diff --git a/scripts/check_install.py b/scripts/check_install.py index 72d9d2b..5e2a801 100644 --- a/scripts/check_install.py +++ b/scripts/check_install.py @@ -19,6 +19,7 @@ def main() -> None: import_module("lab.part") import_module("lab.samples") import_module("lab.provenance") + import_module("lab.inventory") package = distribution("lab-compiler") assert package.version == lab.__version__ diff --git a/src/lab/artifacts.py b/src/lab/artifacts.py new file mode 100644 index 0000000..287a986 --- /dev/null +++ b/src/lab/artifacts.py @@ -0,0 +1,76 @@ +"""Deterministic artifact encoding and non-destructive bundle writes.""" + +import hashlib +import json +from dataclasses import dataclass, fields, is_dataclass +from datetime import datetime +from decimal import Decimal +from enum import Enum +from pathlib import Path, PurePosixPath +from typing import Any + +from lab.units import number + + +@dataclass(frozen=True, kw_only=True) +class SourceArtifact: + name: str + text: str + + def __post_init__(self) -> None: + relative = PurePosixPath(self.name) + if relative.is_absolute() or ".." in relative.parts or not relative.parts: + raise ValueError("Source artifacts require a relative bundle path") + if not isinstance(self.text, str): + raise TypeError("Source artifact content must be text") + + @property + def digest(self) -> str: + return hashlib.sha256(self.text.encode()).hexdigest() + + +def encode(value: Any) -> Any: + if isinstance(value, Decimal): + return number(value) + if isinstance(value, datetime): + return value.isoformat() + if isinstance(value, Enum): + return value.value + if is_dataclass(value) and not isinstance(value, type): + return { + field.name: encode(getattr(value, field.name)) + for field in fields(value) + if not field.name.startswith("_") + } + if isinstance(value, (tuple, list)): + return [encode(item) for item in value] + if isinstance(value, dict): + return {key: encode(item) for key, item in value.items()} + return value + + +def canonical_json(value: Any) -> str: + return ( + json.dumps(encode(value), ensure_ascii=False, sort_keys=True, indent=2, allow_nan=False) + + "\n" + ) + + +def digest(value: Any) -> str: + return hashlib.sha256(canonical_json(value).encode()).hexdigest() + + +def write_bundle(directory: str | Path, files: dict[str, str]) -> Path: + directory = Path(directory) + for name, content in files.items(): + relative = PurePosixPath(name) + if relative.is_absolute() or ".." in relative.parts or not relative.parts: + raise ValueError(f"Invalid bundle path {name!r}") + path = directory / name + if path.exists() and path.read_text(encoding="utf-8") != content: + raise FileExistsError(f"{path} already contains a different artifact") + for name, content in files.items(): + path = directory / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + return directory diff --git a/src/lab/inventory.py b/src/lab/inventory.py new file mode 100644 index 0000000..143963e --- /dev/null +++ b/src/lab/inventory.py @@ -0,0 +1,213 @@ +"""Explicit material assertions and immutable inventory snapshots.""" + +import json +from dataclasses import dataclass +from decimal import Decimal +from enum import StrEnum +from pathlib import Path + +from lab.artifacts import canonical_json, digest, write_bundle +from lab.provenance import Component, DocumentSnapshot, EvidenceState, Implementation, Ref +from lab.provenance.types import require_iri +from lab.units import magnitude, uL, units + + +class MaterialForm(StrEnum): + DNA = "dna" + COMPETENT_CELLS = "competent_cells" + CULTURE = "culture" + BACTERIAL_STAB = "bacterial_stab" + PLATED_SAMPLE = "plated_sample" + REAGENT = "reagent" + + @property + def counted(self) -> bool: + return self in (MaterialForm.BACTERIAL_STAB, MaterialForm.PLATED_SAMPLE) + + +@dataclass(frozen=True, slots=True) +class StockLocation: + container: str + position: str + + def __post_init__(self) -> None: + if not self.container.strip() or not self.position.strip(): + raise ValueError("Stock locations need a container and position") + + +@dataclass(frozen=True, kw_only=True, init=False) +class Stock: + identity: str + implementation: Ref[Implementation] + design: Ref[Component] + form: MaterialForm + quantity_ul: Decimal + concentration_ng_ul: Decimal | None + location: StockLocation | None + supplier_item: str | None + + def __init__( + self, + *, + identity: str, + implementation: Ref[Implementation], + design: Ref[Component], + form: MaterialForm, + quantity: object, + concentration: object = None, + location: StockLocation | None = None, + supplier_item: str | None = None, + ) -> None: + require_iri(identity) + if not isinstance(implementation, Ref) or not isinstance(design, Ref): + raise TypeError("Stock implementation and design must be references") + if not isinstance(form, MaterialForm): + raise TypeError("Pass a MaterialForm") + if form.counted: + raise ValueError("Use CountedStock or counted Receipts, not liquid-volume stocks") + if location is not None and not isinstance(location, StockLocation): + raise TypeError("Pass a StockLocation") + if supplier_item is not None: + require_iri(supplier_item) + values = { + "identity": identity, + "implementation": implementation, + "design": design, + "form": form, + "quantity_ul": magnitude(quantity, "microliter", positive=False), + "concentration_ng_ul": None + if concentration is None + else magnitude(concentration, "nanogram/microliter"), + "location": location, + "supplier_item": supplier_item, + } + for name, value in values.items(): + object.__setattr__(self, name, value) + + @property + def amount(self) -> Decimal: + return self.quantity_ul + + +@dataclass(frozen=True, kw_only=True) +class CountedStock: + """Available whole material units, without an inferred liquid volume.""" + + identity: str + implementation: Ref[Implementation] + design: Ref[Component] + form: MaterialForm + count: int + location: StockLocation | None = None + supplier_item: str | None = None + + def __post_init__(self) -> None: + require_iri(self.identity) + if not isinstance(self.implementation, Ref) or not isinstance(self.design, Ref): + raise TypeError("Stock implementation and design must be references") + if not isinstance(self.form, MaterialForm) or not self.form.counted: + raise ValueError("Counted stocks require a counted material form") + if type(self.count) is not int or self.count < 0: + raise ValueError("Available count must be a nonnegative integer") + if self.location is not None and not isinstance(self.location, StockLocation): + raise TypeError("Pass a StockLocation") + if self.supplier_item is not None: + require_iri(self.supplier_item) + + @property + def amount(self) -> Decimal: + return Decimal(self.count) + + +@dataclass(frozen=True, kw_only=True) +class Inventory: + """Available amounts, never a live database or an automatically depleted ledger. + + ``design`` is an explicit inventory assertion. It does not mean the stock's + sequence was verified. Liquid volumes and whole-unit counts are distinct. + """ + + identity: str + stocks: tuple[Stock | CountedStock, ...] = () + + def __post_init__(self) -> None: + require_iri(self.identity) + if not isinstance(self.stocks, tuple) or not all( + isinstance(stock, (Stock, CountedStock)) for stock in self.stocks + ): + raise TypeError("Inventory stocks must be a tuple of Stock objects") + if len({stock.identity for stock in self.stocks}) != len(self.stocks): + raise ValueError("Stock identities must be unique") + if len({stock.implementation.identity for stock in self.stocks}) != len(self.stocks): + raise ValueError( + "Each stock needs its own implementation; " + "duplicate lots would double-count material" + ) + occupied = [stock.location for stock in self.stocks if stock.location is not None] + if len(set(occupied)) != len(occupied): + raise ValueError("Inventory locations must be unique") + object.__setattr__( + self, "stocks", tuple(sorted(self.stocks, key=lambda stock: stock.identity)) + ) + + def validate(self, document: DocumentSnapshot) -> None: + for stock in self.stocks: + implementation = document.get(stock.implementation.identity, Implementation) + document.get(stock.design.identity, Component) + if implementation.evidence_state is not EvidenceState.RECORDED: + raise ValueError( + f"Inventory stock {stock.identity} must reference a recorded implementation" + ) + if implementation.built is not None and implementation.built != stock.design: + raise ValueError(f"Stock {stock.identity} conflicts with its realized design") + if implementation.built is None and stock.design not in implementation.derived_from: + raise ValueError( + f"Stock {stock.identity} must identify its intended design in provenance" + ) + + @property + def digest(self) -> str: + return digest(self) + + def write(self, path: str | Path) -> Path: + path = Path(path) + write_bundle( + path.parent, + {path.name: canonical_json({"format": "lab.inventory.v1", "inventory": self})}, + ) + return path + + @classmethod + def read(cls, path: str | Path, *, document: DocumentSnapshot) -> "Inventory": + data = json.loads(Path(path).read_text(encoding="utf-8")) + if data.get("format") != "lab.inventory.v1": + raise ValueError("Expected lab.inventory.v1") + item = data["inventory"] + stocks = tuple( + CountedStock( + identity=row["identity"], + implementation=Ref(row["implementation"]["identity"]), + design=Ref(row["design"]["identity"]), + form=MaterialForm(row["form"]), + count=row["count"], + location=None if row["location"] is None else StockLocation(**row["location"]), + supplier_item=row["supplier_item"], + ) + if MaterialForm(row["form"]).counted + else Stock( + identity=row["identity"], + implementation=Ref(row["implementation"]["identity"]), + design=Ref(row["design"]["identity"]), + form=MaterialForm(row["form"]), + quantity=Decimal(row["quantity_ul"]) * uL, + concentration=None + if row["concentration_ng_ul"] is None + else Decimal(row["concentration_ng_ul"]) * units.nanogram / uL, + location=None if row["location"] is None else StockLocation(**row["location"]), + supplier_item=row["supplier_item"], + ) + for row in item["stocks"] + ) + result = cls(identity=item["identity"], stocks=stocks) + result.validate(document) + return result diff --git a/tests/test_inventory.py b/tests/test_inventory.py new file mode 100644 index 0000000..c3b256a --- /dev/null +++ b/tests/test_inventory.py @@ -0,0 +1,50 @@ +from dataclasses import replace + +import pytest + +from lab import uL +from lab.inventory import CountedStock, Inventory, MaterialForm, Stock +from lab.provenance import Component, Document, EvidenceState, Implementation +from lab.provenance.vocabulary import DNA + + +def test_inventory_round_trip_preserves_quantities_and_rejects_planned_stock(tmp_path): + document = Document(namespace="https://example.org/inventory") + design = Component(identity=document.iri("design"), types=(DNA,)) + liquid = Implementation( + identity=document.iri("liquid"), + derived_from=(design.ref,), + evidence_state=EvidenceState.RECORDED, + ) + counted = replace(liquid, identity=document.iri("counted")) + document.add(design, liquid, counted) + stocks = ( + Stock( + identity=document.iri("aliquot"), + design=design.ref, + implementation=liquid.ref, + form=MaterialForm.DNA, + quantity=10 * uL, + ), + CountedStock( + identity=document.iri("stab"), + design=design.ref, + implementation=counted.ref, + form=MaterialForm.BACTERIAL_STAB, + count=2, + ), + ) + inventory = Inventory(identity=document.iri("inventory"), stocks=stocks) + snapshot = document.freeze() + inventory.validate(snapshot) + assert ( + Inventory.read(inventory.write(tmp_path / "inventory.json"), document=snapshot) == inventory + ) + assert inventory.stocks[0].amount == 10 and inventory.stocks[1].amount == 2 + with pytest.raises(ValueError, match="double-count"): + replace( + inventory, stocks=(stocks[1], replace(stocks[1], identity=document.iri("duplicate"))) + ) + document.replace(replace(liquid, evidence_state=EvidenceState.PLANNED)) + with pytest.raises(ValueError, match="recorded"): + inventory.validate(document.freeze())