Introduce the ability to generate LabOP files for a Lab Compiler protocol

.github/workflows/ci.yml

@@ -39,6 +39,29 @@ jobs:
       - name: Run tests
         run: uv run --no-sync pytest -q
 
+  labop:
+    name: Optional upstream LabOP integration
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
+        with:
+          persist-credentials: false
+      - uses: astral-sh/setup-uv@94527f2e458b27549849d47d273a16bec83a01e9 # v7
+        with:
+          python-version: "3.12"
+      - run: uv sync --locked
+      - name: Generate compiler artifacts
+        run: |
+          uv run --no-sync python -m examples.cloning --target manual --out build/labop_check/cloning
+      - name: Install upstream in a separate environment
+        run: |
+          uv run --no-sync python -m venv integrations/labop/.venv
+          integrations/labop/.venv/bin/python -m pip install -e . -r integrations/labop/requirements.txt
+      - name: Check exported plans with upstream
+        run: |
+          integrations/labop/.venv/bin/python -m examples.labop_export --out build/labop_check/operations
+          integrations/labop/.venv/bin/python integrations/labop/check.py build/labop_check
+
   build:
     name: Build and check distributions
     needs: test
@@ -81,4 +104,3 @@ jobs:
           name: distributions
           path: dist/
           if-no-files-found: error
-

README.md

@@ -57,24 +57,6 @@ compiled = compile(
 )
 ```
 
-`lab.compile()` accepts a `Protocol`. It captures an immutable snapshot, validates resources and volumes, prepares the target, and returns one `Compilation`. Inspect `compiled.protocol` for the captured snapshot. All targets and the document renderer consume those same recorded operations.
-
-`build_assembly()` takes a sequence of `Assembly` recipes; `build_transformation()` takes a sequence of `Transformation` recipes. Their design fields are native `sbol3.Component` objects, including components loaded from an SBOL file with `designs.read()`. Recipe fields specify the materials used by the procedure; SBOL owns their biological descriptions. Builders copy identities into protocol samples so later edits to SBOL objects cannot change the recorded plan.
-
-`transformation_deck(on_module=True)` names the DNA block, cell tubes, and reaction plate. The example writes to `~/.lab/transformation/OT-2/`. Omit `deck` and `liquid_handler` for a manual document. Set `to` to choose an output directory, or `to=None` to compile without writing. The [transformation example](examples/transformation.py) uses native SBOL designs and explicit procedure quantities.
-
-## Samples and protocol outputs
-
-`compiled.manifest` contains declared output samples and logical placements. Pass it as `inputs=` to `build_transformation()`, then pass the resulting transformation manifest to `build_plating()`. The [cloning example](examples/cloning.py) shows assembly, transformation, and plating as separately authored and compiled protocols. Plating processes the manifest's samples in their declared order. Both downstream builders require their source samples to occupy one logical container.
-
-For custom protocols, use `protocol.add_sample(sample, at=well, is_input=True)` or `is_output=True`. `Sample.design` and `Sample.implementation` hold optional SBOL identity strings. Parent IDs identify contributions in the same protocol; imported samples identify their upstream protocol and sample separately. Compilation checks references, locations, and cyclic lineage. Manifests describe planned outputs.
-
-`lab.samples` also defines `Location(resource, well)`, `SamplePlacement`, and `OutputManifest`. Recorded operations, samples, target bindings, and volume accounting share that logical location type. Native SBOL documents remain caller-owned and can be written alongside the compilation artifacts.
-
-## SBOL provenance
-
-Protocol samples can link to native pySBOL3 designs and implementations by identity. The [SBOL provenance guide](docs/sbol-provenance.md) covers authoring, validation, and annotation propagation. Run `uv run python -m examples.sbol_provenance` for an example.
-
 ## Describe a deck
 
 This OT-2 deck places two 96-well plates in slots 1 and 2, a 300 µL tip rack in slot 3, and a P300 pipette on the left mount. It uses the same equipment and placement types as the [deck layouts example](https://github.com/the-lab-compiler/lab-py/blob/master/examples/deck_layouts.py).
@@ -123,10 +105,6 @@ deck = Deck(
 )
 ```
 
-The container ids, `sources` and `assay`, match the names used by the protocol. `DeckLayout` assigns each container a physical model and position, and connects the pipette to its tip rack. A deck can include additional layouts for Flex or STAR using the same container ids.
-
-A deck with explicit layouts must include one for the selected liquid handler. Preset decks, such as `transformation_deck`, instead use `Container` and `DeckSite` from `lab.deck` to assign containers to supported placement groups. A bare `ContainerSpec` requires an explicit layout.
-
 ## Describe physical layouts in Lab
 
 Use `DeckLayout` when equipment or placement needs to be explicit. Each layout names its `LiquidHandler` and places the same shared containers. All objects below belong to Lab:
@@ -187,10 +165,6 @@ The cloning example writes separate `assembly`, `transformation`, and `plating`
 
 The manifest uses logical container and well names; physical bindings remain in `plan.json`. Rewriting an identical bundle succeeds. If any generated file would replace different contents, `write()` rejects the write before changing any bundle files.
 
-## Status
-
-Lab Compiler is an early prototype. It checks a plan and emits a document or device program for the OT-2, Flex, and STAR. Software checks are not calibration, collision safety, or qualification of a physical run. Generated instructions need a person and a facility before anyone uses them at the bench.
-
 ## Development
 
 From the repository root:
@@ -202,25 +176,4 @@ uv run --no-sync mypy
 uv run --no-sync pytest
 ```
 
-Shared types and compiler code live directly under `src/lab`; experiment families and target implementations have their own packages:
-
-```text
-src/lab/
-    protocol.py       # Protocol builder, Plate, Well
-    model.py          # Recorded operations, protocol snapshots, target plans
-    part.py           # SBOL part identity
-    samples.py        # Sample, Location, SamplePlacement, OutputManifest
-    labware.py        # Logical labware specifications
-    equipment.py      # Liquid handlers and equipment identifiers
-    deck.py           # Container requirements and physical layouts
-    units.py          # Quantities and unit conversion
-    compiler.py       # Compilation and output bundles
-    validation.py     # Volume, binding, and sample validation
-    documents.py      # Printable protocol rendering
-    experiments/
-        cloning/      # Cloning types, stage builders, decks, and workflow composition
-            types.py  # Assembly and transformation designs, stage inputs
-    targets/          # Manual, Opentrons, and STAR backends
-```
-
 See the [release guide](https://github.com/the-lab-compiler/lab-py/blob/master/docs/releasing.md) for package validation and PyPI publishing.

examples/labop_export.py

@@ -0,0 +1,36 @@
+"""Compile a protocol covering the operations supported by the optional LabOP exporter."""
+
+import argparse
+from pathlib import Path
+
+import lab
+from integrations.labop.export import export_file
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--out", type=Path, default=Path("build/labop"))
+    args = parser.parse_args()
+    protocol = lab.Protocol("LabOP operations")
+    plate = protocol.plate("operations", capacity=100 * lab.uL)
+    protocol.load(plate["A1"], "water", volume=50 * lab.uL)
+    protocol.set_temperature(plate, lab.celsius(25))
+    protocol.distribute(
+        plate["A1"], (plate["A2"], plate["A3"]), volume=2 * lab.uL, air_gap=1 * lab.uL
+    )
+    protocol.transfer(plate["A1"], plate["A4"], volume=2 * lab.uL)
+    protocol.mix(plate["A1"], volume=2 * lab.uL, cycles=2)
+    protocol.wait(1 * lab.seconds)
+    protocol.manual("Inspect the plate")
+    protocol.thermocycle(
+        plate,
+        ((lab.celsius(25), 1 * lab.seconds),),
+        lid_temperature=lab.celsius(40),
+        block_volume=10 * lab.uL,
+    )
+    output = lab.compile(protocol, to=None).write(args.out)
+    print(export_file(output / "plan.json"))
+
+
+if __name__ == "__main__":
+    main()

integrations/labop/README.md

@@ -0,0 +1,38 @@
+# Export a compilation as LabOP
+
+Because LabOP is not currently on PyPi, we have made the integration with tthe Lab Compiler optional. The integration reads the compiler's `lab.plan.v1` JSON artifact and writes a prospective LabOP protocol. It uses upstream LabOP's classes, primitive libraries, and SBOL serialization.
+
+Install the integration from a checkout, using a separate Python 3.11 or 3.12 environment:
+
+```sh
+python3.12 -m venv integrations/labop/.venv
+integrations/labop/.venv/bin/python -m pip install -e . -r integrations/labop/requirements.txt
+```
+
+The upstream revision and direct compatibility dependencies are pinned in `requirements.txt`. This environment uses `pip` because upstream's container-ontology URL is not accepted by `uv`. Upstream's transitive dependencies are not fully locked. Installation needs network access; export uses the installed libraries. This environment is separate because upstream import modifies pySBOL3's registrations, classes, and process logging.
+
+Run the example to compile with Lab and write LabOP:
+
+```sh
+integrations/labop/.venv/bin/python -m examples.labop_export
+```
+
+The example writes `build/labop/protocol.labop.ttl` beside `plan.json` and `protocol.html`. `--out directory` selects another output directory. It refuses to replace a different existing artifact. Normal compilation emits `plan.json`, HTML, and target artifacts; LabOP is an explicit export step.
+
+The integration consumes only the protocol portion of `plan.json`. Target configuration and authoring file locations do not affect the exported identity. Samples retain SBOL design and implementation references. The RDF records the intended operations, initial resource metadata, and declared outputs; it records no execution or observed products.
+
+Upstream owns the LabOP/UML object model, RDF structure, and the `Transfer`, `PipetteMix`, and `PlateCoordinates` definitions. The adapter owns the mapping from Lab's operations. Small Lab extension declarations cover waits, persistent temperatures, thermal profiles, ordered distribution, and operator instructions. Consumers need support for those semantics; graph validation does not establish executable compatibility with every LabOP interpreter.
+
+Thermocycle's `profile` argument is an ordered JSON list of `{celsius, seconds}` holds from the compiler artifact. Resource metadata and sample placements are JSON annotations on upstream container and sample objects; design and implementation identities are also RDF URI annotations. This preserves Lab data without defining another object model or extension ontology.
+
+Checks validate the constructed upstream document, operation parameters, control-flow order, and RDF serialization round-trips. Upstream's behavior-enabled `Protocol` constructor adds an initial-to-final edge when reading a document; object-level readback can therefore change its graph. The adapter does not patch upstream or run its execution engine.
+
+`plan.json` is the integration boundary. The adapter imports no Lab Python modules, and Lab imports no adapter. A future supported PyPI release of LabOP can replace the Git requirement without changing compiler inputs, the artifact format, or adding a second protocol model. There is no bundled fallback implementation.
+
+To check the integration against fresh compiler output:
+
+```sh
+integrations/labop/.venv/bin/python -m examples.labop_export --out build/labop_check/operations
+uv run python -m examples.cloning --target manual --out build/labop_check/cloning
+integrations/labop/.venv/bin/python integrations/labop/check.py build/labop_check
+```

integrations/labop/check.py

@@ -0,0 +1,174 @@
+"""Check upstream export against compiler-generated fixtures, in the integration environment."""
+
+import argparse
+import copy
+import json
+from pathlib import Path
+from typing import Any
+
+import sbol3
+from export import LAB, LABOP, OM, UML, export, semantic
+from rdflib import RDF, Graph, Literal, Namespace, URIRef
+from rdflib.compare import isomorphic
+from rdflib.term import Node
+
+SBOL = Namespace("http://sbols.org/v3#")
+
+
+def literal(graph: Graph, subject: Node | None, predicate: URIRef) -> Literal:
+    assert subject is not None
+    value = graph.value(subject, predicate, any=False)
+    assert isinstance(value, Literal), (subject, predicate, value)
+    return value
+
+
+def check(plan: dict[str, Any]) -> Graph:
+    document = export(plan)
+    graph: Graph = document.graph()
+    recorded = semantic(plan["protocol"])
+    protocols = list(graph.subjects(RDF.type, LABOP.Protocol))
+    assert len(protocols) == 1
+    assert graph.value(protocols[0], LAB.evidenceState) == LAB.planned
+    assert not any("Execution" in str(kind) for kind in graph.objects(None, RDF.type))
+    actions = sorted(
+        graph.subjects(LAB.semanticStep), key=lambda node: int(literal(graph, node, LAB.stepIndex))
+    )
+    assert [json.loads(str(graph.value(node, LAB.semanticStep))) for node in actions] == recorded[
+        "steps"
+    ]
+    for action, step in zip(actions, recorded["steps"], strict=True):
+        pins = {str(graph.value(pin, SBOL.name)): pin for pin in graph.objects(action, UML.input)}
+        if "volume" in step:
+            argument = graph.value(pins["amount"], UML.value)
+            amount = graph.value(argument, UML.identifiedValue)
+            assert graph.value(amount, OM.hasUnit) == OM.microlitre
+            assert float(literal(graph, amount, OM.hasNumericalValue)) == float(step["volume"])
+        if step["kind"] == "Thermocycle":
+            argument = graph.value(pins["profile"], UML.value)
+            assert json.loads(str(literal(graph, argument, UML.stringValue))) == step["profile"]
+            argument = graph.value(pins["cycles"], UML.value)
+            assert int(literal(graph, argument, UML.integerValue)) == step["cycles"]
+    samples = {sample["id"]: sample for sample in recorded["samples"]}
+    for resource in recorded["resources"]:
+        container = next(
+            node
+            for node in graph.subjects(RDF.type, LABOP.ContainerSpec)
+            if str(graph.value(node, SBOL.name)) == resource["name"]
+        )
+        assert json.loads(str(graph.value(container, LAB.resource))) == resource
+        array = next(graph.subjects(LABOP.containerType, container))
+        expected = [
+            {"well": place["location"]["well"], "sample": samples[place["sample_id"]]}
+            for place in recorded["placements"]
+            if place["location"]["resource"] == resource["name"]
+        ]
+        assert json.loads(str(graph.value(array, LAB.samples))) == expected
+    for node in graph.subjects(RDF.type, OM.Measure):
+        assert graph.value(node, OM.hasUnit)
+        assert graph.value(node, OM.hasNumericalValue) is not None
+    for name in ("design", "implementation"):
+        expected_refs = {
+            sample[name]
+            for sample in recorded["samples"]
+            if sample[name] is not None
+            and any(place["sample_id"] == sample["id"] for place in recorded["placements"])
+        }
+        assert {str(value) for value in graph.objects(None, LAB[name])} == expected_refs
+
+    for action in graph.subjects(RDF.type, UML.CallBehaviorAction):
+        behavior = graph.value(action, UML.behavior)
+        for item in graph.objects(behavior, UML.ownedParameter):
+            parameter = graph.value(item, UML.propertyValue)
+            name = str(graph.value(parameter, SBOL.name))
+            direction = graph.value(parameter, UML.direction)
+            predicate = UML.output if direction == UML["out"] else UML.input
+            parameter_pins = [
+                pin
+                for pin in graph.objects(action, predicate)
+                if str(graph.value(pin, SBOL.name)) == name
+            ]
+            assert len(parameter_pins) == 1, (behavior, name)
+            if direction == UML["in"]:
+                lower = graph.value(parameter, UML.lowerValue)
+                required = int(literal(graph, lower, UML.integerValue)) == 1
+                if required:
+                    assert graph.value(parameter_pins[0], UML.value) is not None or list(
+                        graph.subjects(UML.target, parameter_pins[0])
+                    )
+
+    # There must be one sequential control path, including all selection/group actions.
+    first = next(graph.subjects(RDF.type, UML.InitialNode))
+    last = next(graph.subjects(RDF.type, UML.FinalNode))
+    successors: dict[Node, Node] = {}
+    for edge in graph.subjects(RDF.type, UML.ControlFlow):
+        source = graph.value(edge, UML.source)
+        target = graph.value(edge, UML.target)
+        assert source is not None and target is not None
+        assert source not in successors
+        successors[source] = target
+    visited: list[Node] = []
+    node = first
+    while node != last:
+        assert node not in visited
+        visited.append(node)
+        node = successors[node]
+    assert set(visited[1:]) == set(graph.subjects(RDF.type, UML.CallBehaviorAction))
+    assert [item for item in visited if item in actions] == actions
+    assert len(visited) == len(successors)
+
+    # Check RDF serialization. Upstream Protocol's constructor mutates control flow on read.
+    restored = Graph().parse(data=document.write_string(sbol3.TURTLE), format="turtle")
+    assert isomorphic(graph, restored)
+    return graph
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("fixtures", type=Path)
+    args = parser.parse_args()
+    paths = sorted(args.fixtures.rglob("plan.json"))
+    if not paths:
+        raise ValueError("Generate the example compilation fixtures first")
+    kinds: set[str] = set()
+    for path in paths:
+        plan = json.loads(path.read_text())
+        graph = check(plan)
+        kinds.update(step["kind"] for step in plan["protocol"]["steps"])
+        print(f"{path}: validated and RDF round-tripped")
+    assert kinds == {
+        "Transfer",
+        "Mix",
+        "Distribute",
+        "Wait",
+        "SetTemperature",
+        "Thermocycle",
+        "ManualInstruction",
+    }
+    moved = copy.deepcopy(plan)
+    moved["target"] = {"name": "Different robot"}
+    for step in moved["protocol"]["steps"]:
+        step["origin"] = {"file": "/another/computer.py", "line": 999}
+    assert isomorphic(graph, export(moved).graph())
+    moved["protocol"]["name"] += " changed"
+    assert set(graph.subjects(RDF.type, LABOP.Protocol)) != set(
+        export(moved).graph().subjects(RDF.type, LABOP.Protocol)
+    )
+    for invalid, expected_error in (
+        ({"format": "unknown"}, "Expected a lab.plan.v1"),
+        ({**plan, "units": {}}, "Unsupported compilation units"),
+        (
+            {**plan, "protocol": {**plan["protocol"], "steps": [{"kind": "unknown"}]}},
+            "No LabOP mapping for unknown",
+        ),
+    ):
+        try:
+            export(invalid)
+        except ValueError as error:
+            assert expected_error in str(error)
+        else:
+            raise AssertionError("Unsupported input must fail")
+    print("Operation mappings, control flow, SBOL links, round-trips, and identities passed")
+
+
+if __name__ == "__main__":
+    main()