From c2e70f74a95153356265ef522c91a5bb73f49068 Mon Sep 17 00:00:00 2001 From: Mike Arpaia Date: Wed, 30 Sep 2026 22:21:00 -0600 Subject: [PATCH] Export immutable experiment protocols as static LabOP bundles --- scripts/check_install.py | 35 ++- scripts/check_labop.py | 44 +++ src/lab/__init__.py | 7 +- src/lab/compiler.py | 127 ++++++++- src/lab/experiment.py | 227 +++++++++++++++ src/lab/labop/__init__.py | 6 +- src/lab/labop/primitives.py | 76 +++++ src/lab/labop/protocol.py | 540 ++++++++++++++++++++++++++++++++++++ src/lab/methods.py | 82 ++++++ tests/test_labop.py | 141 ++++++++++ 10 files changed, 1278 insertions(+), 7 deletions(-) create mode 100644 scripts/check_labop.py create mode 100644 src/lab/experiment.py create mode 100644 src/lab/labop/primitives.py create mode 100644 src/lab/labop/protocol.py create mode 100644 src/lab/methods.py create mode 100644 tests/test_labop.py diff --git a/scripts/check_install.py b/scripts/check_install.py index 5e2a801..1f531b7 100644 --- a/scripts/check_install.py +++ b/scripts/check_install.py @@ -2,6 +2,7 @@ import json import sys +from decimal import Decimal from importlib import import_module from importlib.metadata import distribution from importlib.resources import files @@ -9,6 +10,9 @@ 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,6 +24,7 @@ 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__ @@ -27,6 +32,8 @@ def main() -> None: 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") diff --git a/scripts/check_labop.py b/scripts/check_labop.py new file mode 100644 index 0000000..d62830a --- /dev/null +++ b/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() diff --git a/src/lab/__init__.py b/src/lab/__init__.py index 938f022..9cbe0d1 100644 --- a/src/lab/__init__.py +++ b/src/lab/__init__.py @@ -1,7 +1,8 @@ """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 @@ -9,6 +10,10 @@ __all__ = [ "Compilation", "CompileError", + "ExperimentCompilation", + "ExperimentPlan", + "MaterialHandoff", + "ProtocolStage", "Plate", "Protocol", "Quantity", diff --git a/src/lab/compiler.py b/src/lab/compiler.py index 62fdb9d..3f9760a 100644 --- a/src/lab/compiler.py +++ b/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: diff --git a/src/lab/experiment.py b/src/lab/experiment.py new file mode 100644 index 0000000..238f1b2 --- /dev/null +++ b/src/lab/experiment.py @@ -0,0 +1,227 @@ +"""Frozen logical experiments, stages, and material handoffs.""" + +import hashlib +from dataclasses import dataclass +from decimal import Decimal + +from lab.artifacts import SourceArtifact, canonical_json +from lab.deck import Deck +from lab.model import RecordedProtocol, semantic +from lab.provenance import DocumentSnapshot, EvidenceState, Implementation, Ref +from lab.provenance.types import require_iri +from lab.samples import Location +from lab.validation import count_trace, logical_bindings, validate + + +@dataclass(frozen=True, kw_only=True) +class MaterialHandoff: + producer: str + implementation: Ref[Implementation] + source: Location + destination: Location + volume_ul: Decimal + count: int | None = None + + def __post_init__(self) -> None: + require_iri(self.producer) + if ( + not isinstance(self.volume_ul, Decimal) + or not self.volume_ul.is_finite() + or self.volume_ul < 0 + or (self.count is None and self.volume_ul <= 0) + or ( + self.count is not None + and (type(self.count) is not int or self.count <= 0 or self.volume_ul != 0) + ) + ): + raise ValueError("A handoff needs either positive volume or a positive count") + + +@dataclass(frozen=True, kw_only=True) +class ProtocolStage: + identity: str + protocol: RecordedProtocol + deck: Deck + depends_on: tuple[str, ...] = () + handoffs: tuple[MaterialHandoff, ...] = () + external: bool = False + + def __post_init__(self) -> None: + require_iri(self.identity) + if type(self.external) is not bool: + raise TypeError("Stage external must be a bool") + if not isinstance(self.protocol, RecordedProtocol) or not isinstance(self.deck, Deck): + raise TypeError("Stages require a frozen protocol and logical deck") + if not isinstance(self.depends_on, tuple) or not isinstance(self.handoffs, tuple): + raise TypeError("Dependencies and handoffs must be immutable tuples") + if len(set(self.depends_on)) != len(self.depends_on): + raise ValueError("Stage dependencies must be unique") + requirements = {item.id: item.labware for item in self.deck.containers} + if set(requirements) != {resource.name for resource in self.protocol.resources}: + raise ValueError("Stage deck requirements must cover exactly the protocol resources") + for resource in self.protocol.resources: + spec = requirements[resource.name] + if (spec.rows, spec.columns) != (resource.rows, resource.columns) or ( + resource.capacity > spec.capacity_ul + ): + raise ValueError("Stage deck geometry and capacity must match the protocol") + + +@dataclass(frozen=True, kw_only=True) +class ExperimentPlan: + identity: str + provenance: DocumentSnapshot + stages: tuple[ProtocolStage, ...] + build_json: str | None = None + inputs: tuple[SourceArtifact, ...] = () + + def __post_init__(self) -> None: + require_iri(self.identity) + if not isinstance(self.inputs, tuple) or not all( + isinstance(item, SourceArtifact) for item in self.inputs + ): + raise TypeError("Experiment inputs must be a tuple of SourceArtifact objects") + if len({item.name for item in self.inputs}) != len(self.inputs): + raise ValueError("Input artifact paths must be unique") + if not isinstance(self.stages, tuple) or not all( + isinstance(stage, ProtocolStage) for stage in self.stages + ): + raise TypeError("Experiment stages must be an immutable tuple") + self.provenance.validate().raise_for_errors() + previous: dict[str, ProtocolStage] = {} + step_ids: set[str | None] = set() + protocol_ids: set[str | None] = set() + # Track carry-over material after every consumer, not the original yield. + remaining: dict[Ref[Implementation], Decimal] = {} + forms: dict[Ref[Implementation], object] = {} + for stage in self.stages: + if stage.identity in previous or not set(stage.depends_on) <= previous.keys(): + raise ValueError("Stages need unique identities and earlier dependencies") + if stage.protocol.identity in protocol_ids: + raise ValueError("Stage protocols need unique identities") + protocol_ids.add(stage.protocol.identity) + for step in stage.protocol.steps: + if step.identity in step_ids: + raise ValueError("Experiment step identities must be unique") + step_ids.add(step.identity) + final = validate(stage.protocol, logical_bindings(stage.protocol)) + final_counts = count_trace(stage.protocol)[-1] + samples = {sample.id: sample for sample in stage.protocol.samples} + places = {place.sample_id: place.location for place in stage.protocol.placements} + at = {location: samples[key] for key, location in places.items()} + initial = { + Location(resource.name, fill.well): fill.volume + for resource in stage.protocol.resources + for fill in resource.fills + } + for key in stage.protocol.input_sample_ids: + if samples[key].count is not None: + initial[places[key]] = Decimal(samples[key].count or 0) + for location, count in final_counts.items(): + final[location] = Decimal(count) + input_refs = [ + samples[key].implementation + for key in stage.protocol.input_sample_ids + if samples[key].implementation is not None + ] + if len(set(input_refs)) != len(input_refs): + raise ValueError("Split input aliquots require distinct implementation identities") + for sample in samples.values(): + if sample.design is not None: + self.provenance.resolve(sample.design) + if sample.implementation is not None: + material = self.provenance.resolve(sample.implementation) + if ( + sample.implementation in forms + and forms[sample.implementation] != sample.form + ): + raise ValueError( + "An implementation cannot change material form across stages" + ) + forms[sample.implementation] = sample.form + if sample.design is None or ( + material.built != sample.design + and sample.design not in material.derived_from + ): + raise ValueError("Sample design conflicts with its implementation") + handoff_locations = {handoff.destination for handoff in stage.handoffs} + if len(handoff_locations) != len(stage.handoffs): + raise ValueError("An input location needs exactly one handoff") + for sample_id in stage.protocol.input_sample_ids: + sample = samples[sample_id] + material_ref = sample.implementation + if material_ref is None or places[sample_id] in handoff_locations: + continue + material = self.provenance.resolve(material_ref) + if material.evidence_state is EvidenceState.PLANNED: + raise ValueError("A planned input requires an upstream material handoff") + location = places[sample_id] + if material_ref in remaining and remaining[material_ref] != initial.get(location): + raise ValueError("Initial volume conflicts with the remaining stock quantity") + remaining[material_ref] = final[location] + for handoff in stage.handoffs: + if handoff.producer not in stage.depends_on: + raise ValueError("Handoff producer must be a stage dependency") + producer = previous[handoff.producer] + outputs = producer.protocol.output_manifest() + output_at = {place.location: place.sample_id for place in outputs.placements} + source = next( + ( + sample + for sample in outputs.samples + if sample.id == output_at.get(handoff.source) + ), + None, + ) + destination = at.get(handoff.destination) + if ( + source is None + or destination is None + or source.implementation != handoff.implementation + or destination.implementation != handoff.implementation + or destination.design != source.design + or destination.form != source.form + or (handoff.count is not None) != (destination.count is not None) + ): + raise ValueError("Handoff must preserve the exact material and design") + amount = Decimal(handoff.count) if handoff.count is not None else handoff.volume_ul + if initial.get(handoff.destination) != amount: + raise ValueError("Handoff volume must match the destination precondition") + if remaining.get(handoff.implementation) != amount: + raise ValueError("Handoff volume conflicts with remaining upstream material") + remaining[handoff.implementation] = final[handoff.destination] + for sample_id in stage.protocol.output_sample_ids: + output_ref = samples[sample_id].implementation + if output_ref is not None: + if output_ref in remaining: + raise ValueError("An output implementation must have one producing stage") + remaining[output_ref] = final[places[sample_id]] + previous[stage.identity] = stage + + @property + def plan_json(self) -> str: + return canonical_json( + { + "format": "lab.experiment.v1", + "identity": self.identity, + "provenance_sha256": self.provenance.digest, + "stages": [ + { + "identity": stage.identity, + "protocol": semantic(stage.protocol), + "depends_on": stage.depends_on, + "handoffs": stage.handoffs, + "external": stage.external, + # Physical layouts do not define the logical experiment. + "containers": stage.deck.containers, + } + for stage in self.stages + ], + "build": self.build_json, + "inputs": {item.name: item.digest for item in self.inputs}, + } + ) + + @property + def digest(self) -> str: + return hashlib.sha256(self.plan_json.encode()).hexdigest() diff --git a/src/lab/labop/__init__.py b/src/lab/labop/__init__.py index 25517db..05f803d 100644 --- a/src/lab/labop/__init__.py +++ b/src/lab/labop/__init__.py @@ -1 +1,5 @@ -"""Pinned schemas and primitive definitions for LabOP interchange.""" +"""Static LabOP interchange from frozen Lab experiments.""" + +from lab.labop.protocol import LabOPDocument, export + +__all__ = ["LabOPDocument", "export"] diff --git a/src/lab/labop/primitives.py b/src/lab/labop/primitives.py new file mode 100644 index 0000000..ce248f3 --- /dev/null +++ b/src/lab/labop/primitives.py @@ -0,0 +1,76 @@ +"""Pinned LabOP primitives and explicit Lab extensions for unmatched semantics.""" + +from dataclasses import dataclass +from importlib.resources import files + +from rdflib import Graph, Namespace, URIRef + +from lab.provenance.vocabulary import LAB as LAB_NAMESPACE + +UML = Namespace("http://bioprotocols.org/uml#") +LABOP = Namespace("http://bioprotocols.org/labop#") +SBOL = Namespace("http://sbols.org/v3#") +OM = Namespace("http://www.ontology-of-units-of-measure.org/resource/om-2/") +LAB = Namespace(LAB_NAMESPACE) +EXT = "https://the-lab-compiler.github.io/lab-py/labop/primitives/" +LIQUID = "https://bioprotocols.org/labop/primitives/liquid_handling/" + + +@dataclass(frozen=True) +class Parameter: + name: str + type: URIRef + required: bool = True + + +EXTENSIONS = { + "TransferAtHeight": ( + Parameter("source", LABOP.SampleCollection), + Parameter("destination", LABOP.SampleCollection), + Parameter("amount", OM.Measure), + Parameter("destinationHeight", OM.Measure), + ), + "Wait": (Parameter("duration", OM.Measure),), + "SetTemperature": ( + Parameter("samples", LABOP.SampleCollection), + Parameter("temperature", OM.Measure), + ), + "Thermocycle": ( + Parameter("samples", LABOP.SampleCollection), + Parameter("profile", LAB.ThermalProfile), + Parameter("cycles", UML.ValueSpecification), + Parameter("lidTemperature", OM.Measure, False), + Parameter("blockVolume", OM.Measure, False), + ), + "Distribute": ( + Parameter("source", LABOP.SampleCollection), + Parameter("destinations", LABOP.SampleCollection), + Parameter("amount", OM.Measure), + Parameter("airGap", OM.Measure, False), + ), + "OperatorPause": (Parameter("instruction", UML.ValueSpecification),), +} + +DESCRIPTIONS = { + "TransferAtHeight": "Transfer liquid, dispensing at the specified height above the " + "destination well bottom. For a plated sample this is a deposition, not colony formation.", + "Wait": "Wait for the specified duration without changing material.", + "SetTemperature": "Set a persistent temperature on the collection's controlling module. " + "The hold continues after the action completes.", + "Thermocycle": "Apply the ordered temperature and duration holds for the specified cycles " + "to the whole resource; then release temperature control. " + "No evaporation or yield change is inferred.", + "Distribute": "Transfer the same liquid amount from one source to each ordered destination, " + "reusing one tip. Air gap is air and does not change the liquid ledger. " + "A target may split aspirations within its capacity.", + "OperatorPause": "Pause for the stated operator instruction; no material change is modeled.", +} + + +def upstream_liquid_primitives() -> Graph: + graph = Graph() + for name in ("liquid_handling", "sample_arrays"): + graph.parse( + data=files("lab.labop").joinpath(f"resources/{name}.ttl").read_text(), format="turtle" + ) + return graph diff --git a/src/lab/labop/protocol.py b/src/lab/labop/protocol.py new file mode 100644 index 0000000..9982eaf --- /dev/null +++ b/src/lab/labop/protocol.py @@ -0,0 +1,540 @@ +"""Static LabOP/UML projection. No execution engine or robot SDK is imported.""" + +import hashlib +import json +from dataclasses import dataclass +from decimal import Decimal +from urllib.parse import quote + +from rdflib import RDF, XSD, Graph, Literal, URIRef +from rdflib.compare import to_canonical_graph + +from lab.experiment import ExperimentPlan, ProtocolStage +from lab.labop.primitives import ( + DESCRIPTIONS, + EXT, + EXTENSIONS, + LAB, + LABOP, + LIQUID, + OM, + SBOL, + UML, + upstream_liquid_primitives, +) +from lab.model import semantic +from lab.operations import ( + Distribute, + ExternalPreparation, + ManualInstruction, + Mix, + SetTemperature, + Thermocycle, + Transfer, + Wait, +) +from lab.samples import Location + + +def turtle(graph: Graph) -> str: + """Sorted N-Triples is also valid Turtle and gives deterministic artifacts.""" + return ( + "\n".join( + sorted( + line + for line in to_canonical_graph(graph).serialize(format="nt").splitlines() + if line + ) + ) + + "\n" + ) + + +@dataclass(frozen=True) +class LabOPDocument: + protocol: str + text: str + + @property + def digest(self) -> str: + return hashlib.sha256(self.text.encode()).hexdigest() + + def graph(self) -> Graph: + return Graph().parse(data=self.text, format="turtle") + + +class _Writer: + def __init__(self) -> None: + self.graph = Graph() + self.upstream = upstream_liquid_primitives() + + def node( + self, identity: str, kind: URIRef, *, name: str | None = None, top: bool = False + ) -> URIRef: + node = URIRef(identity) + self.graph.add((node, RDF.type, kind)) + self.graph.add((node, RDF.type, SBOL.TopLevel if top else SBOL.Identified)) + self.graph.add((node, SBOL.displayId, Literal(identity.rsplit("/", 1)[-1]))) + if top: + self.graph.add((node, SBOL.hasNamespace, URIRef(identity.rsplit("/", 1)[0]))) + if name is not None: + self.graph.add((node, SBOL.name, Literal(name))) + return node + + def add(self, node: URIRef, predicate: URIRef, value: URIRef | Literal) -> None: + self.graph.add((node, predicate, value)) + + def one(self, node: URIRef, predicate: URIRef) -> URIRef: + value = self.graph.value(node, predicate, any=False) + if not isinstance(value, URIRef): + raise ValueError(f"Expected one URI value for {node} {predicate}") + return value + + def literal(self, identity: str, value: int | str) -> URIRef: + node = self.node( + identity, UML.LiteralInteger if isinstance(value, int) else UML.LiteralString + ) + self.add( + node, UML.integerValue if isinstance(value, int) else UML.stringValue, Literal(value) + ) + return node + + def measure(self, identity: str, value: Decimal | int, unit: URIRef) -> URIRef: + node = self.node(identity, OM.Measure) + self.add(node, OM.hasNumericalValue, Literal(str(value), datatype=XSD.decimal)) + self.add(node, OM.hasUnit, unit) + return node + + def parameter( + self, + behavior: URIRef, + name: str, + kind: URIRef, + index: int, + *, + direction: str = "in", + required: bool = True, + ) -> URIRef: + ordered = self.node(f"{behavior}/parameter_{index}", UML.OrderedPropertyValue) + parameter = self.node(f"{ordered}/parameter", UML.Parameter, name=name) + self.add(behavior, UML.ownedParameter, ordered) + self.add(ordered, UML.indexValue, Literal(index)) + self.add(ordered, UML.propertyValue, parameter) + self.add(parameter, UML.direction, UML[direction]) + self.add(parameter, UML.type, kind) + self.add(parameter, UML.isOrdered, Literal(True)) + self.add(parameter, UML.isUnique, Literal(True)) + for field, count in ((UML.lowerValue, int(required)), (UML.upperValue, 1)): + self.add( + parameter, field, self.literal(f"{parameter}/{str(field).split('#')[-1]}", count) + ) + return ordered + + def primitive(self, name: str, *, upstream: bool = False) -> URIRef: + base = ( + "https://bioprotocols.org/labop/primitives/sample_arrays/" + if name == "PlateCoordinates" + else LIQUID + ) + identity = URIRef((base if upstream else EXT) + name) + if (identity, RDF.type, LABOP.Primitive) in self.graph: + return identity + if upstream: + # Include this primitive and all of its owned definitions, unchanged. + for subject, predicate, obj in self.upstream: + if str(subject) == str(identity) or str(subject).startswith(str(identity) + "/"): + self.graph.add((subject, predicate, obj)) + else: + self.node(str(identity), LABOP.Primitive, name=name, top=True) + self.add(identity, SBOL.description, Literal(DESCRIPTIONS[name])) + for index, parameter in enumerate(EXTENSIONS[name]): + self.parameter( + identity, parameter.name, parameter.type, index, required=parameter.required + ) + return identity + + def flow(self, protocol: URIRef, source: URIRef, target: URIRef, *, control: bool) -> None: + index = len(tuple(self.graph.objects(protocol, UML.edge))) + edge = self.node(f"{protocol}/edge_{index}", UML.ControlFlow if control else UML.ObjectFlow) + self.add(protocol, UML.edge, edge) + self.add(edge, UML.source, source) + self.add(edge, UML.target, target) + + def pin( + self, + action: URIRef, + name: str, + value: URIRef | None = None, + *, + owned: bool = False, + output: bool = False, + ) -> URIRef: + identity = f"{action}/{'output' if output else 'input'}_{name}" + pin = self.node( + identity, + UML.OutputPin if output else UML.ValuePin if value is not None else UML.InputPin, + name=name, + ) + self.add(action, UML.output if output else UML.input, pin) + self.add(pin, UML.isOrdered, Literal(True)) + self.add(pin, UML.isUnique, Literal(True)) + if value is not None: + literal = self.node( + f"{pin}/value", UML.LiteralIdentified if owned else UML.LiteralReference + ) + self.add(literal, UML.identifiedValue if owned else UML.referenceValue, value) + self.add(pin, UML.value, literal) + return pin + + def chain(self, protocol: URIRef, actions: list[URIRef]) -> None: + initial = self.node(f"{protocol}/initial", UML.InitialNode) + final = self.node(f"{protocol}/final", UML.FlowFinalNode) + for node in (initial, *actions, final): + self.add(protocol, UML.node, node) + nodes = [initial, *actions, final] + for start, end in zip(nodes, nodes[1:], strict=False): + self.flow(protocol, start, end, control=True) + + def fork(self, protocol: URIRef, source: URIRef) -> URIRef: + node = self.node(f"{source}/fork", UML.ForkNode) + self.add(protocol, UML.node, node) + self.flow(protocol, source, node, control=False) + return node + + def scalar_pin(self, action: URIRef, name: str, value: int | str) -> URIRef: + pin = self.pin(action, name) + self.graph.remove((pin, RDF.type, UML.InputPin)) + self.add(pin, RDF.type, UML.ValuePin) + self.add(pin, UML.value, self.literal(f"{pin}/value", value)) + return pin + + def amount_pin(self, action: URIRef, name: str, value: Decimal | int, unit: URIRef) -> URIRef: + return self.pin( + action, + name, + self.measure( + f"{action}/input_{name}/value/measure", + value, + unit, + ), + owned=True, + ) + + +def export(experiment: ExperimentPlan) -> LabOPDocument: + """Export ordered stage calls and explicit material/control flows.""" + writer = _Writer() + root = writer.node(experiment.identity + "/labop", LABOP.Protocol, top=True) + writer.add(root, LAB.planDigest, Literal(experiment.digest)) + writer.add(root, LAB.evidenceState, LAB.planned) + calls: list[URIRef] = [] + outputs: dict[tuple[str, str], URIRef] = {} + for stage in experiment.stages: + protocol, collections, ports = _stage(writer, stage) + call = writer.node(f"{root}/stage_{len(calls) + 1}", UML.CallBehaviorAction) + writer.add(call, UML.behavior, protocol) + writer.add(call, LAB.plannedActivity, URIRef(stage.identity)) + for resource, (name, collection) in collections.items(): + handoffs = tuple( + item for item in stage.handoffs if item.destination.resource == resource + ) + if not handoffs: + writer.pin(call, name, collection) + else: + sources = {(item.producer, item.source.resource) for item in handoffs} + if len(sources) != 1 or any( + item.source.well != item.destination.well for item in handoffs + ): + raise ValueError("LabOP stage handoffs require whole-plate continuity") + pin = writer.pin(call, name) + writer.flow(root, outputs[next(iter(sources))], pin, control=False) + writer.add( + pin, LAB.handoffs, Literal(json.dumps(semantic(handoffs), sort_keys=True)) + ) + for resource, name in ports.items(): + pin = writer.pin(call, name, output=True) + outputs[(stage.identity, resource)] = writer.fork(root, pin) + calls.append(call) + writer.chain(root, calls) + return LabOPDocument(str(root), turtle(writer.graph)) + + +def _stage( + writer: _Writer, stage: ProtocolStage +) -> tuple[URIRef, dict[str, tuple[str, URIRef]], dict[str, str]]: + recorded = stage.protocol + protocol = writer.node(str(recorded.identity), LABOP.Protocol, name=recorded.name, top=True) + writer.add(protocol, SBOL.description, Literal(recorded.description)) + writer.add(protocol, LAB.planDigest, Literal(recorded.digest)) + writer.add(protocol, LAB.evidenceState, LAB.planned) + writer.add(protocol, LAB.external, Literal(stage.external)) + collections: dict[str, tuple[str, URIRef]] = {} + sources: dict[str, URIRef] = {} + samples = {sample.id: sample for sample in recorded.samples} + for index, resource in enumerate(recorded.resources): + name = f"collection_{index}" + ordered = writer.parameter(protocol, name, LABOP.SampleCollection, index) + parameter = writer.one(ordered, UML.propertyValue) + default = writer.node(f"{parameter}/default", UML.LiteralIdentified) + array = writer.node(f"{default}/samples", LABOP.SampleArray, name=resource.name) + writer.add(parameter, UML.defaultValue, default) + writer.add(default, UML.identifiedValue, array) + container = writer.node( + f"{protocol}/container_{index}", LABOP.ContainerSpec, name=resource.name, top=True + ) + writer.add(container, LAB.rows, Literal(resource.rows)) + writer.add(container, LAB.columns, Literal(resource.columns)) + writer.add( + container, + LAB.capacity, + writer.measure(f"{container}/capacity", resource.capacity, OM.microlitre), + ) + writer.add(array, LABOP.containerType, container) + writer.add(array, LAB.sampleFormat, Literal("json")) + fills = {fill.well: fill for fill in resource.fills} + counted = { + place.location.well: samples[place.sample_id] + for place in recorded.placements + if place.location.resource == resource.name + and place.sample_id in recorded.input_sample_ids + and samples[place.sample_id].count is not None + } + counted_materials = { + well: sample.implementation.identity + if sample.implementation + else sample.material_identity + for well, sample in counted.items() + } + writer.add( + array, + LABOP.initial_contents, + Literal( + quote( + json.dumps( + { + well: ( + fills[well].material + if well in fills + else counted_materials.get(well) + ) + for well in resource.wells + } + ) + ) + ), + ) + writer.add(array, LAB.resource, Literal(resource.name)) + for fill in resource.fills: + node = writer.node(f"{array}/initial_{fill.well}", LAB.InitialMaterial) + writer.add(array, LAB.initialMaterial, node) + writer.add(node, LAB.coordinates, Literal(fill.well)) + writer.add(node, LAB.materialIdentity, Literal(fill.material)) + writer.add( + node, LAB.volume, writer.measure(f"{node}/volume", fill.volume, OM.microlitre) + ) + for well, sample in counted.items(): + node = writer.node(f"{array}/initial_{well}", LAB.InitialMaterial) + writer.add(array, LAB.initialMaterial, node) + writer.add(node, LAB.coordinates, Literal(well)) + writer.add( + node, + LAB.materialIdentity, + Literal( + sample.implementation.identity + if sample.implementation + else sample.material_identity + ), + ) + writer.add(node, LAB.itemCount, Literal(sample.count)) + for place in recorded.placements: + if place.location.resource != resource.name: + continue + sample = samples[place.sample_id] + node = writer.node(f"{array}/sample_{place.location.well}", LAB.SampleAssertion) + writer.add(array, LAB.plannedSample, node) + writer.add(node, LAB.coordinates, Literal(place.location.well)) + writer.add(node, LAB.sampleIdentity, Literal(sample.id)) + if sample.form is not None: + writer.add(node, LAB.materialForm, URIRef(LAB + sample.form.value)) + if sample.count is not None: + writer.add(node, LAB.sampleCount, Literal(sample.count)) + writer.add(node, LAB.sampleRole, Literal(sample.role)) + for predicate, ref in ( + (LAB.design, sample.design), + (LAB.implementation, sample.implementation), + ): + if ref is not None: + writer.add(node, predicate, URIRef(ref.identity)) + input_node = writer.node(f"{protocol}/input_{index}", UML.ActivityParameterNode) + writer.add(protocol, UML.node, input_node) + writer.add(input_node, UML.parameter, ordered) + sources[resource.name] = writer.fork(protocol, input_node) + collections[resource.name] = (name, array) + + actions: list[URIRef] = [] + selections: dict[Location, URIRef] = {} + + def well_source(location: Location) -> URIRef: + if location not in selections: + action = writer.node(f"{protocol}/select_{len(selections)}", UML.CallBehaviorAction) + writer.add(action, UML.behavior, writer.primitive("PlateCoordinates", upstream=True)) + pin = writer.pin(action, "source") + writer.flow(protocol, sources[location.resource], pin, control=False) + writer.scalar_pin(action, "coordinates", location.well) + output = writer.pin(action, "samples", output=True) + selections[location] = writer.fork(protocol, output) + actions.append(action) + return selections[location] + + def collection_pin(action: URIRef, name: str, source: URIRef) -> None: + writer.flow(protocol, source, writer.pin(action, name), control=False) + + for step in recorded.steps: + action = writer.node(str(step.identity), UML.CallBehaviorAction) + writer.add(action, LAB.semanticStep, Literal(json.dumps(semantic(step), sort_keys=True))) + if isinstance(step, Transfer): + primitive = ( + writer.primitive("Transfer", upstream=True) + if step.destination_height_mm is None + else writer.primitive("TransferAtHeight") + ) + collection_pin(action, "source", well_source(step.source)) + collection_pin(action, "destination", well_source(step.destination)) + writer.amount_pin(action, "amount", step.volume, OM.microlitre) + if step.destination_height_mm is not None: + writer.amount_pin( + action, "destinationHeight", step.destination_height_mm, OM.millimetre + ) + elif isinstance(step, Mix): + primitive = writer.primitive("PipetteMix", upstream=True) + collection_pin(action, "samples", well_source(step.location)) + writer.amount_pin(action, "amount", step.volume, OM.microlitre) + writer.amount_pin(action, "cycleCount", step.cycles, OM.one) + elif isinstance(step, Wait): + primitive = writer.primitive("Wait") + writer.amount_pin(action, "duration", step.seconds, OM.second) + elif isinstance(step, ManualInstruction): + primitive = writer.primitive("OperatorPause") + writer.scalar_pin(action, "instruction", step.text) + elif isinstance(step, ExternalPreparation): + primitive = URIRef(EXT + f"ExternalPreparation_{len(step.inputs)}_{len(step.outputs)}") + if (primitive, RDF.type, LABOP.Primitive) not in writer.graph: + writer.node(str(primitive), LABOP.Primitive, top=True) + writer.add( + primitive, + SBOL.description, + Literal( + "Perform the named external procedure using the declared material inputs. " + "Outputs specify expected quantities at distinct destination locations. " + "Counted amounts use OM one; liquid amounts use microlitres. " + "This action specifies an operator-performed preparation." + ), + ) + writer.parameter(primitive, "procedure", UML.ValueSpecification, 0) + writer.parameter(primitive, "instruction", UML.ValueSpecification, 1) + parameter_index = 2 + for prefix, material_ports in ( + ("source", step.inputs), + ("destination", step.outputs), + ): + for index in range(len(material_ports)): + writer.parameter( + primitive, f"{prefix}_{index}", LABOP.SampleCollection, parameter_index + ) + writer.parameter( + primitive, f"{prefix}Amount_{index}", OM.Measure, parameter_index + 1 + ) + parameter_index += 2 + writer.scalar_pin(action, "procedure", step.procedure) + writer.scalar_pin(action, "instruction", step.instructions) + for prefix, material_ports in (("source", step.inputs), ("destination", step.outputs)): + for index, port in enumerate(material_ports): + collection_pin(action, f"{prefix}_{index}", well_source(port.location)) + writer.amount_pin( + action, + f"{prefix}Amount_{index}", + port.count or port.volume_ul, + OM.one if port.count else OM.microlitre, + ) + elif isinstance(step, Distribute): + primitive = writer.primitive("Distribute") + collection_pin(action, "source", well_source(step.source)) + members = tuple(well_source(location) for location in step.destinations) + # A pure structural primitive preserves order and aliases across plates. + group_type = URIRef(EXT + f"OrderedGroup_{len(members)}") + if (group_type, RDF.type, LABOP.Primitive) not in writer.graph: + writer.node(str(group_type), LABOP.Primitive, top=True) + writer.add( + group_type, + SBOL.description, + Literal( + "Return an ordered SampleCollection of the supplied sample aliases. " + "Preserve member order and duplicates; perform no laboratory operation." + ), + ) + for index in range(len(members)): + writer.parameter(group_type, f"member_{index}", LABOP.SampleCollection, index) + writer.parameter( + group_type, "samples", LABOP.SampleCollection, len(members), direction="out" + ) + group = writer.node(f"{protocol}/group_{len(actions)}", UML.CallBehaviorAction) + writer.add(group, UML.behavior, group_type) + for index, source in enumerate(members): + collection_pin(group, f"member_{index}", source) + writer.flow( + protocol, + writer.pin(group, "samples", output=True), + writer.pin(action, "destinations"), + control=False, + ) + actions.append(group) + writer.amount_pin(action, "amount", step.volume, OM.microlitre) + if step.air_gap is not None: + writer.amount_pin(action, "airGap", step.air_gap, OM.microlitre) + elif isinstance(step, (SetTemperature, Thermocycle)): + primitive = writer.primitive(type(step).__name__) + collection_pin(action, "samples", sources[step.resource]) + if isinstance(step, SetTemperature): + writer.amount_pin(action, "temperature", step.celsius, OM.degreeCelsius) + else: + profile = writer.node(f"{action}/input_profile/value/profile", LAB.ThermalProfile) + for index, hold in enumerate(step.profile): + node = writer.node(f"{profile}/hold_{index}", LAB.ThermalHold) + writer.add(profile, LAB.hold, node) + writer.add(node, LAB["index"], Literal(index)) + writer.add( + node, + LAB.temperature, + writer.measure(f"{node}/temperature", hold.celsius, OM.degreeCelsius), + ) + writer.add( + node, + LAB.duration, + writer.measure(f"{node}/duration", hold.seconds, OM.second), + ) + writer.pin(action, "profile", profile, owned=True) + writer.scalar_pin(action, "cycles", step.cycles) + if step.lid_celsius is not None: + writer.amount_pin(action, "lidTemperature", step.lid_celsius, OM.degreeCelsius) + if step.block_volume is not None: + writer.amount_pin(action, "blockVolume", step.block_volume, OM.microlitre) + else: + raise TypeError(f"No LabOP mapping for {type(step).__name__}") + writer.add(action, UML.behavior, primitive) + actions.append(action) + ports: dict[str, str] = {} + for place in recorded.placements: + resource_name = place.location.resource + if place.sample_id not in recorded.output_sample_ids or resource_name in ports: + continue + name = f"result_{len(ports)}" + parameter = writer.parameter( + protocol, name, LABOP.SampleCollection, len(collections) + len(ports), direction="out" + ) + node = writer.node(f"{protocol}/output_{len(ports)}", UML.ActivityParameterNode) + writer.add(protocol, UML.node, node) + writer.add(node, UML.parameter, parameter) + writer.flow(protocol, sources[resource_name], node, control=False) + ports[resource_name] = name + writer.chain(protocol, actions) + return protocol, collections, ports diff --git a/src/lab/methods.py b/src/lab/methods.py new file mode 100644 index 0000000..0347e49 --- /dev/null +++ b/src/lab/methods.py @@ -0,0 +1,82 @@ +"""Digital methods rendered from the complete frozen experiment.""" + +from lab.documents import describe +from lab.experiment import ExperimentPlan +from lab.units import number + + +def render(experiment: ExperimentPlan) -> str: + """Render a planned methods specification without claiming completed work.""" + lines = [ + "# Planned methods", + "", + f"Experiment: <{experiment.identity}>", + "", + f"Semantic plan SHA-256: {experiment.digest}", + "", + "This document specifies planned work. It does not establish that any physical " + "step was performed. The accompanying SBOL document contains design sequences " + "and provenance; the LabOP document specifies control and material flow.", + "", + ] + if not experiment.stages: + lines += [ + "The requested materials are allocated from inventory; no procedure is required.", + "", + ] + for index, stage in enumerate(experiment.stages, 1): + protocol = stage.protocol + lines += [ + f"## {index}. {protocol.name}", + "", + f"Stage: <{stage.identity}>. Protocol: <{protocol.identity}>.", + "", + protocol.description, + "", + "### Initial conditions", + "", + "These loads are preconditions, not inferred dispensing operations.", + "", + ] + for resource in protocol.resources: + lines.append( + f"- {resource.name}: {resource.rows} × {resource.columns} wells, " + f"{number(resource.capacity)} µL capacity and " + f"{number(resource.dead_volume)} µL residual volume per well." + ) + for fill in resource.fills: + lines.append( + f"- {resource.name}:{fill.well}: {number(fill.volume)} µL of {fill.material}." + ) + lines += ["", "### Material identity", ""] + for sample in protocol.samples: + quantity = f"; {sample.count} unit(s)" if sample.count is not None else "" + form = f"; form {sample.form.value}" if sample.form is not None else "" + lines.append( + f"- {sample.id}: design {sample.material_identity}; implementation " + f"{sample.implementation.identity if sample.implementation else 'unspecified'}" + f"{form}{quantity}." + ) + lines += ["", "### Handoffs", ""] + if not stage.handoffs: + lines += ["No upstream stage handoff.", ""] + for handoff in stage.handoffs: + amount = ( + f"{handoff.count} unit(s)" + if handoff.count is not None + else f"{number(handoff.volume_ul)} µL" + ) + lines += [ + f"- Carry {handoff.implementation.identity} from stage <{handoff.producer}> " + f"at {handoff.source} to the logical binding {handoff.destination}, with " + f"{amount} remaining. This identifies the same material " + "and plate; it does not specify an additional liquid transfer.", + ] + lines += ["", "### Procedure", ""] + for step_index, step in enumerate(protocol.steps, 1): + lines += [f"{step_index}. {describe(step)} Step: <{step.identity}>."] + lines += ["", "### Planned outputs", ""] + for sample in protocol.output_manifest().samples: + lines.append(f"- {sample.id}; intended design {sample.material_identity}.") + lines.append("") + return "\n".join(lines) diff --git a/tests/test_labop.py b/tests/test_labop.py new file mode 100644 index 0000000..99b79ec --- /dev/null +++ b/tests/test_labop.py @@ -0,0 +1,141 @@ +import json +from importlib.resources import files +from urllib.parse import unquote + +import pytest +from rdflib import OWL, RDF, RDFS, Graph, URIRef + +from lab import ExperimentPlan, Protocol, ProtocolStage, celsius, seconds, uL +from lab.deck import Container, Deck, DeckSite +from lab.labop import export +from lab.labop.primitives import LAB, LABOP, LIQUID, OM, SBOL, UML +from lab.labware import PCR_PLATE_96 +from lab.provenance import Document + + +def artifact(*, operations=True): + protocol = Protocol("Operation vocabulary", identity="https://example.org/operations") + plate = protocol.plate("plate", capacity=100 * uL) + protocol.load(plate["A1"], "water", volume=50 * uL) + protocol.set_temperature(plate, celsius(25)) + protocol.distribute(plate["A1"], (plate["A2"], plate["A3"]), volume=2 * uL, air_gap=1 * uL) + protocol.wait(1 * seconds) + protocol.manual("Inspect the plate") + protocol.thermocycle( + plate, ((celsius(25), 1 * seconds),), lid_temperature=celsius(40), block_volume=10 * uL + ) + experiment = ExperimentPlan( + identity="https://example.org/operation_experiment", + provenance=Document(namespace="https://example.org/").freeze(), + stages=( + ProtocolStage( + identity="https://example.org/operation_stage", + protocol=protocol.snapshot(), + deck=Deck( + containers=( + Container(id="plate", labware=PCR_PLATE_96, site=DeckSite.THERMOCYCLER), + ) + ), + ), + ), + ) + return experiment, export(experiment) + + +@pytest.mark.parametrize("operations", [True]) +def test_labop_has_protocols_exact_semantic_actions_and_no_execution_claims(operations): + experiment, result = artifact(operations=operations) + graph = result.graph() + assert len(tuple(graph.subjects(RDF.type, LABOP.Protocol))) == 1 + len(experiment.stages) + for kind in ( + LABOP.ProtocolExecution, + LABOP.BehaviorExecution, + LABOP.ActivityNodeExecution, + LABOP.CallBehaviorExecution, + LABOP.ActivityEdgeFlow, + ): + assert not tuple(graph.subjects(RDF.type, kind)) + for predicate in (LABOP.execution, LABOP.completedNormally): + assert not tuple(graph.triples((None, predicate, None))) + assert {str(node) for node in graph.subjects(LAB.semanticStep)} == { + step.identity for stage in experiment.stages for step in stage.protocol.steps + } + for node in graph.subjects(RDF.type, LABOP.SampleArray): + if str(graph.value(node, SBOL.name)) == "products": + assert all( + value is None + for value in json.loads( + unquote(str(graph.value(node, LABOP.initial_contents))) + ).values() + ) + assert not tuple(graph.subjects(UML.behavior, URIRef(LIQUID + "Provision"))) + amounts = tuple(graph.subjects(RDF.type, OM.Measure)) + assert amounts and all(graph.value(node, OM.hasUnit) for node in amounts) + + +@pytest.mark.parametrize("operations", [True]) +def test_calls_follow_upstream_required_parameter_names_and_directions(operations): + _, result = artifact(operations=operations) + graph = result.graph() + for action in graph.subjects(RDF.type, UML.CallBehaviorAction): + behavior = graph.value(action, UML.behavior, any=False) + parameters = [ + graph.value(item, UML.propertyValue) + for item in graph.objects(behavior, UML.ownedParameter) + ] + by_name = {str(graph.value(parameter, SBOL.name)): parameter for parameter in parameters} + supplied = {} + for direction, predicate in (("in", UML.input), ("out", UML.output)): + for pin in graph.objects(action, predicate): + name = str(graph.value(pin, SBOL.name)) + assert name in by_name, (behavior, name) + assert graph.value(by_name[name], UML.direction) == UML[direction] + assert name not in supplied + supplied[name] = pin + if direction == "in": + constant = graph.value(pin, UML.value) + incoming = tuple(graph.subjects(UML.target, pin)) + assert bool(constant) != bool(incoming) + for name, parameter in by_name.items(): + lower = graph.value(parameter, UML.lowerValue) + if graph.value(lower, UML.integerValue).toPython() == 1: + assert name in supplied, (behavior, name) + + +@pytest.mark.parametrize("operations", [True]) +def test_generated_nodes_satisfy_pinned_ontology_cardinalities(operations): + _, result = artifact(operations=operations) + graph = result.graph() + ontology = Graph() + for name in ("uml.ttl", "labop.ttl", "lab.ttl"): + ontology.parse( + data=files("lab.labop").joinpath("resources", name).read_text(), format="turtle" + ) + for subject in set(graph.subjects(RDF.type)): + pending = list(graph.objects(subject, RDF.type)) + seen = set() + while pending: + kind = pending.pop() + if kind in seen: + continue + seen.add(kind) + for parent in ontology.objects(kind, RDFS.subClassOf): + if isinstance(parent, URIRef): + pending.append(parent) + continue + predicate = ontology.value(parent, OWL.onProperty) + count = len(tuple(graph.objects(subject, predicate))) + minimum = ontology.value(parent, OWL.minCardinality) + maximum = ontology.value(parent, OWL.maxCardinality) + exact = ontology.value(parent, OWL.cardinality) + if minimum is not None: + assert count >= int(minimum), (subject, predicate, minimum) + if maximum is not None: + assert count <= int(maximum), (subject, predicate, maximum) + if exact is not None: + assert count == int(exact), (subject, predicate, exact) + # Unknown properties in upstream namespaces generally indicate a spelling or version error. + declared = set(ontology.subjects()) + for predicate in set(graph.predicates()): + if str(predicate).startswith((str(LABOP), str(UML))): + assert predicate in declared, predicate