[2/11] Model material inventory with explicit quantities and provenance

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__

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

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

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())