[5/11] Export immutable experiment protocols as static LabOP bundles

scripts/check_install.py

@@ -2,13 +2,17 @@
 
 import json
 import sys
+from decimal import Decimal
 from importlib import import_module
 from importlib.metadata import distribution
 from importlib.resources import files
 from pathlib import Path
 from tempfile import TemporaryDirectory
 
 import lab
+from lab.deck import ContainerSpec, Deck
+from lab.experiment import ExperimentPlan, ProtocolStage
+from lab.labware import LabwareKind, LabwareSpec
 from lab.provenance import Activity, Document
 from lab.targets import Manual
 
@@ -20,13 +24,16 @@ def main() -> None:
     import_module("lab.samples")
     import_module("lab.provenance")
     import_module("lab.inventory")
+    import_module("lab.labop")
 
     package = distribution("lab-compiler")
     assert package.version == lab.__version__
     assert package.metadata["Name"] == "lab-compiler"
     assert set(package.metadata.get_all("Provides-Extra", [])) == {"opentrons", "star"}
     assert files("lab").joinpath("py.typed").is_file()
     assert files("lab.provenance").joinpath("resources/lab.ttl").is_file()
+    assert files("lab.labop").joinpath("resources/upstream.json").is_file()
+    assert files("lab.labop").joinpath("resources/LICENSE.txt").is_file()
     assert any(str(path).endswith("licenses/LICENSE") for path in package.files or ())
     assert lab.__file__ is not None
     assert not Path(lab.__file__).resolve().is_relative_to(Path(__file__).resolve().parents[1])
@@ -43,13 +50,39 @@ def main() -> None:
         plan = json.loads((output / "plan.json").read_text())
         assert plan["compiler_version"] == package.version
         assert (output / "protocol.html").stat().st_size > 0
-
         provenance = Document(namespace="https://example.org/install_check")
         provenance.add(Activity(identity=provenance.iri("activity")))
         snapshot = provenance.freeze()
         snapshot.write(output / "provenance.ttl")
         assert Document.read(output / "provenance.ttl").freeze().digest == snapshot.digest
         assert not snapshot.to_sbol3().validate().errors
+        experiment = ExperimentPlan(
+            identity="https://example.org/install_check/experiment",
+            provenance=snapshot,
+            stages=(
+                ProtocolStage(
+                    identity="https://example.org/install_check/stage",
+                    protocol=protocol.snapshot(),
+                    deck=Deck(
+                        containers=tuple(
+                            ContainerSpec(
+                                id=name,
+                                labware=LabwareSpec(
+                                    kind=LabwareKind.PLATE,
+                                    rows=1,
+                                    columns=1,
+                                    capacity_ul=Decimal(200),
+                                ),
+                            )
+                            for name in ("source", "destination")
+                        )
+                    ),
+                ),
+            ),
+        )
+        bundle = lab.compile(experiment, Manual())
+        bundle.write(output / "experiment")
+        assert "protocol.labop.ttl" in bundle.files
 
     assert not any(name.split(".")[0] in {"opentrons", "pylabrobot"} for name in sys.modules)
     print(f"lab-compiler {package.version}: installed package check passed")

scripts/check_labop.py

@@ -0,0 +1,44 @@
+"""Validate a generated protocol with isolated SBOLFactory 1.1.2 / pySBOL3.
+
+Run this script in a separate interpreter: SBOLFactory registers global builders.
+It is a validation tool, never a compiler dependency or execution engine.
+"""
+
+import argparse
+from pathlib import Path
+
+import sbol3
+from rdflib import RDF, Graph, Namespace
+from sbol_factory import SBOLFactory
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("protocol", type=Path)
+    parser.add_argument("--schemas", type=Path, required=True)
+    args = parser.parse_args()
+    lab_namespace = "https://the-lab-compiler.github.io/lab-py/ns#"
+    SBOLFactory("lab", str(args.schemas / "lab.ttl"), lab_namespace)
+    SBOLFactory("uml", str(args.schemas / "uml.ttl"), "http://bioprotocols.org/uml#")
+    SBOLFactory("labop", str(args.schemas / "labop.ttl"), "http://bioprotocols.org/labop#")
+    document = sbol3.Document()
+    document.read(str(args.protocol))
+    report = document.validate()
+    assert not report.errors, tuple(str(error) for error in report.errors)
+    graph = Graph().parse(args.protocol, format="turtle")
+    namespace = Namespace(lab_namespace)
+    profiles = tuple(graph.subjects(RDF.type, namespace.ThermalProfile))
+    for identity in profiles:
+        profile = document.find(str(identity))
+        assert profile is not None and profile.holds
+        for hold in profile.holds:
+            assert hold.temperature.unit and hold.duration.unit
+            assert hold.index >= 0
+    print(
+        f"{len(document.objects)} top-level objects, {len(profiles)} typed thermal profiles; "
+        f"{len(report.errors)} errors, {len(report.warnings)} warnings"
+    )
+
+
+if __name__ == "__main__":
+    main()

src/lab/__init__.py

@@ -1,14 +1,19 @@
 """Build and compile laboratory protocols using ordinary Python."""
 
 from lab._version import __version__
-from lab.compiler import Compilation, compile
+from lab.compiler import Compilation, ExperimentCompilation, compile
+from lab.experiment import ExperimentPlan, MaterialHandoff, ProtocolStage
 from lab.protocol import Plate, Protocol, Well
 from lab.units import Quantity, celsius, minutes, mL, seconds, uL
 from lab.validation import CompileError
 
 __all__ = [
     "Compilation",
     "CompileError",
+    "ExperimentCompilation",
+    "ExperimentPlan",
+    "MaterialHandoff",
+    "ProtocolStage",
     "Plate",
     "Protocol",
     "Quantity",

src/lab/compiler.py

@@ -3,16 +3,20 @@
 import hashlib
 import json
 import os
+from collections.abc import Mapping
 from dataclasses import dataclass, replace
 from decimal import Decimal
+from importlib.resources import files as resource_files
 from pathlib import Path
 from typing import Any, overload
 from typing import Protocol as Interface
 
 import lab.documents as documents
+import lab.methods as methods
 from lab._version import __version__
 from lab.artifacts import write_bundle
 from lab.deck import Deck
+from lab.experiment import ExperimentPlan
 from lab.experiments.cloning.stages.assembly import build_assembly
 from lab.experiments.cloning.stages.plating import build_plating
 from lab.experiments.cloning.stages.transformation import build_transformation
@@ -23,6 +27,7 @@
     Transformation,
     TransformationRequest,
 )
+from lab.labop import export as export_labop
 from lab.model import RecordedProtocol, encode
 from lab.operations import Distribute, Mix, Transfer
 from lab.protocol import Protocol
@@ -190,6 +195,73 @@ def _compile_protocol(
     return Compilation(recorded, prepared, tuple(volumes.items()))
 
 
+@dataclass(frozen=True, kw_only=True)
+class ExperimentCompilation:
+    experiment: ExperimentPlan
+    stages: tuple[Compilation, ...]
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.stages, tuple) or len(self.stages) != len(self.experiment.stages):
+            raise ValueError("Compile every experiment stage into an immutable tuple")
+        if any(
+            compilation.protocol != stage.protocol
+            for compilation, stage in zip(self.stages, self.experiment.stages, strict=True)
+        ):
+            raise ValueError("Stage compilations must use the frozen experiment's protocols")
+
+    @property
+    def files(self) -> dict[str, str]:
+        files = {
+            "experiment.json": self.experiment.plan_json,
+            "provenance.ttl": self.experiment.provenance.to_turtle(),
+            "protocol.labop.ttl": export_labop(self.experiment).text,
+            "methods.md": methods.render(self.experiment),
+        }
+        files.update({f"inputs/{item.name}": item.text for item in self.experiment.inputs})
+        for name in ("lab.ttl", "labop.ttl", "uml.ttl", "upstream.json", "LICENSE.txt"):
+            files[f"schemas/{name}"] = (
+                resource_files("lab.labop")
+                .joinpath(
+                    "resources",
+                    name,
+                )
+                .read_text(encoding="utf-8")
+            )
+        for index, compilation in enumerate(self.stages, 1):
+            files.update(
+                {f"stages/{index}/{name}": text for name, text in compilation.files.items()}
+            )
+        if self.experiment.build_json is not None:
+            files["build.json"] = self.experiment.build_json
+        checksums = {
+            name: hashlib.sha256(text.encode()).hexdigest() for name, text in sorted(files.items())
+        }
+        files["bundle.json"] = canonical_json(
+            {
+                "format": "lab.bundle.v1",
+                "experiment": self.experiment.identity,
+                "semantic_sha256": self.experiment.digest,
+                "sha256": checksums,
+                "stages": [
+                    {
+                        "protocol": compilation.protocol.identity,
+                        "target": compilation.target.name,
+                        "plan_sha256": compilation.digest,
+                    }
+                    for compilation in self.stages
+                ],
+            }
+        )
+        return files
+
+    @property
+    def digest(self) -> str:
+        return hashlib.sha256(self.files["bundle.json"].encode()).hexdigest()
+
+    def write(self, directory: str | Path) -> Path:
+        return write_bundle(directory, self.files)
+
+
 class _DefaultOutput:
     """Marks compile's default output directory."""
 
@@ -272,24 +344,35 @@ def compile(
 ) -> Compilation: ...
 
 
+@overload
+def compile(
+    protocol: ExperimentPlan,
+    target: Target | LiquidHandler | Mapping[str, Target | Deck],
+    *,
+    liquid_handler: LiquidHandler | None = None,
+    to: str | Path | None = ...,
+) -> ExperimentCompilation: ...
+
+
 def compile(
     protocol: (
         Protocol
         | RecordedProtocol
+        | ExperimentPlan
         | Assembly
         | AssemblyRequest
         | Transformation
         | TransformationRequest
         | PlatingRequest
     ),
-    target: Target | None = None,
+    target: Target | LiquidHandler | Mapping[str, Target | Deck] | None = None,
     *,
     deck: Deck | None = None,
     liquid_handler: LiquidHandler | None = None,
     inputs: OutputManifest | None = None,
     to: str | Path | None | _DefaultOutput = _DEFAULT_OUTPUT,
-) -> Compilation:
-    """Compile a protocol or a cloning request.
+) -> Compilation | ExperimentCompilation:
+    """Compile a protocol, an experiment, or a cloning request.
 
     One assembly uses its id as the protocol name. An assembly request names a
     protocol that holds every assembly. One transformation uses its id as the
@@ -300,12 +383,48 @@ def compile(
     ``LAB_HOME`` replaces ``~/.lab``. ``to=None`` skips the write. A deck
     contains shared requirements and optional Lab-owned layouts. The selected
     backend validates and translates its layout or supported preset. Concrete
-    backend targets are also accepted for low-level integrations.
+    backend targets are also accepted for low-level integrations. An experiment
+    accepts a handler, one document target, or an exact stage-to-target/deck
+    mapping; use its compilation's write() method to save the complete bundle.
     """
     if inputs is not None and not isinstance(
         protocol, (Transformation, TransformationRequest, PlatingRequest)
     ):
         raise TypeError("Pass inputs with a transformation or plating request.")
+    if isinstance(protocol, ExperimentPlan):
+        if deck is not None or isinstance(target, Deck):
+            raise TypeError("Map stage identities to decks when compiling an experiment")
+        experiment_target = target if target is not None else Manual()
+        if isinstance(experiment_target, Mapping) and set(experiment_target) != {
+            stage.identity for stage in protocol.stages
+        }:
+            raise ValueError("Hardware mapping must cover exactly the experiment's stages")
+        stages = []
+        for stage in protocol.stages:
+            stage_target = (
+                stage.deck
+                if isinstance(experiment_target, LiquidHandler)
+                else experiment_target[stage.identity]
+                if isinstance(experiment_target, Mapping)
+                else experiment_target
+            )
+            handler = (
+                experiment_target
+                if isinstance(experiment_target, LiquidHandler)
+                else liquid_handler
+            )
+            if stage.external:
+                if isinstance(experiment_target, Mapping) and not isinstance(stage_target, Manual):
+                    raise ValueError("Map external preparation stages to Manual targets")
+                stage_target = Manual()
+                handler = None
+            stages.append(_compile_protocol(stage.protocol, stage_target, liquid_handler=handler))
+        experiment_compilation = ExperimentCompilation(experiment=protocol, stages=tuple(stages))
+        if to is not None and not isinstance(to, _DefaultOutput):
+            experiment_compilation.write(to)
+        return experiment_compilation
+    if isinstance(target, (LiquidHandler, Mapping)):
+        raise TypeError("A single protocol requires a target or a Deck")
     if isinstance(target, Deck):
         raise TypeError("Pass a deck with deck=.")
     if deck is not None and target is not None: