diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2662206..b803f45 100644 --- a/.github/workflows/ci.yml +++ b/.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 - diff --git a/README.md b/README.md index 56bb138..c40f1ee 100644 --- a/README.md +++ b/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. diff --git a/examples/labop_export.py b/examples/labop_export.py new file mode 100644 index 0000000..ffa38bf --- /dev/null +++ b/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() diff --git a/integrations/labop/README.md b/integrations/labop/README.md new file mode 100644 index 0000000..4fc46fe --- /dev/null +++ b/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 +``` diff --git a/integrations/labop/check.py b/integrations/labop/check.py new file mode 100644 index 0000000..aefa0ba --- /dev/null +++ b/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() diff --git a/integrations/labop/export.py b/integrations/labop/export.py new file mode 100644 index 0000000..d253a32 --- /dev/null +++ b/integrations/labop/export.py @@ -0,0 +1,241 @@ +"""Export lab.plan.v1 with upstream LabOP in a separate Python environment.""" + +import argparse +import hashlib +import json +import math +from pathlib import Path +from typing import Any +from urllib.parse import quote + +import labop +import sbol3 +import uml +from rdflib import Namespace +from rdflib.compare import to_canonical_graph + +LAB = Namespace("https://the-lab-compiler.github.io/lab-py/ns#") +LABOP = Namespace("http://bioprotocols.org/labop#") +UML = Namespace("http://bioprotocols.org/uml#") +OM = Namespace("http://www.ontology-of-units-of-measure.org/resource/om-2/") +EXT = "https://the-lab-compiler.github.io/lab-py/labop/primitives/" + +# These describe Lab operations whose semantics differ from the upstream library. +EXTENSIONS = { + "Wait": (("duration", OM.Measure),), + "OperatorPause": (("instruction", UML.ValueSpecification),), + "SetTemperature": (("samples", LABOP.SampleCollection), ("temperature", OM.Measure)), + "Thermocycle": ( + ("samples", LABOP.SampleCollection), + ("profile", UML.ValueSpecification), + ("cycles", UML.ValueSpecification), + ), + "Distribute": ( + ("source", LABOP.SampleCollection), + ("destinations", LABOP.SampleCollection), + ("amount", OM.Measure), + ), +} +DESCRIPTIONS = { + "Wait": "Wait for the duration without changing material.", + "OperatorPause": "Pause for the operator instruction; no material change is modeled.", + "SetTemperature": "Set a persistent temperature; the hold continues after this action.", + "Thermocycle": "Repeat the ordered JSON profile of {celsius, seconds} holds, " + "then release temperature control.", + "Distribute": "Transfer equal amounts to ordered destinations, reusing one tip. " + "Air gap is air; a target may split aspirations within its capacity.", +} + + +def semantic(value: Any) -> Any: + if isinstance(value, dict): + return {key: semantic(item) for key, item in value.items() if key != "origin"} + if isinstance(value, list): + return [semantic(item) for item in value] + return value + + +def text(obj: sbol3.Identified, name: str, value: str) -> None: + setattr(obj, "lab_" + name, sbol3.TextProperty(obj, str(LAB[name]), 0, 1, initial_value=value)) + + +def measure(value: str | int | float, unit: str) -> sbol3.Measure: + return sbol3.Measure(float(value), str(OM[unit])) + + +def export(plan: dict[str, Any]) -> sbol3.Document: + """Map a compiler artifact to an upstream SBOL document; never execute the plan.""" + if plan.get("format") != "lab.plan.v1": + raise ValueError("Expected a lab.plan.v1 compilation artifact") + if plan["units"] != {"volume": "microliter", "duration": "second", "temperature": "celsius"}: + raise ValueError("Unsupported compilation units") + recorded = semantic(plan["protocol"]) + digest = hashlib.sha256(json.dumps(recorded, sort_keys=True).encode()).hexdigest() + identity = "https://the-lab-compiler.github.io/lab-py/protocols/p_" + digest + document = sbol3.Document() + protocol = labop.Protocol(identity, name=recorded["name"], description=recorded["description"]) + document.add(protocol) + # Upstream starts with an initial-to-final edge; replace it with the actual sequence. + protocol.edges.clear() + text(protocol, "planDigest", digest) + protocol.evidence = sbol3.URIProperty( + protocol, str(LAB.evidenceState), 1, 1, initial_value=str(LAB.planned) + ) + labop.import_library("liquid_handling") + labop.import_library("sample_arrays") + sources: dict[str, uml.ActivityParameterNode] = {} + selections: dict[tuple[str, str], uml.OutputPin] = {} + samples = {sample["id"]: sample for sample in recorded["samples"]} + for index, resource in enumerate(recorded["resources"]): + spec = labop.ContainerSpec(f"{identity}/container_{index}", name=resource["name"]) + document.add(spec) + text(spec, "resource", json.dumps(resource, sort_keys=True)) + array = labop.SampleArray(name=resource["name"], container_type=spec) + fills = {fill["well"]: fill["material"] for fill in resource["fills"]} + wells = { + f"{chr(65 + row)}{column}": fills.get(f"{chr(65 + row)}{column}") + for column in range(1, resource["columns"] + 1) + for row in range(resource["rows"]) + } + array.initial_contents = quote(json.dumps(wells, sort_keys=True)) + text(array, "sampleFormat", "json") + annotations = [ + {"well": place["location"]["well"], "sample": samples[place["sample_id"]]} + for place in recorded["placements"] + if place["location"]["resource"] == resource["name"] + ] + text(array, "samples", json.dumps(annotations, sort_keys=True)) + for name in ("design", "implementation"): + refs = sorted({item["sample"][name] for item in annotations if item["sample"][name]}) + setattr( + array, + name, + sbol3.URIProperty(array, str(LAB[name]), 0, math.inf, initial_value=refs), + ) + sources[resource["name"]] = protocol.input_value( + f"collection_{index}", + str(LABOP.SampleCollection), + default_value=uml.LiteralIdentified(value=array), + ) + + def well(location: dict[str, str]) -> uml.OutputPin: + key = location["resource"], location["well"] + if key not in selections: + selections[key] = protocol.primitive_step( + "PlateCoordinates", source=sources[key[0]], coordinates=key[1] + ).output_pin("samples") + return selections[key] + + def extension(name: str) -> labop.Primitive: + primitive = document.find(EXT + name) + if primitive is None: + primitive = labop.Primitive(EXT + name, description=DESCRIPTIONS[name]) + document.add(primitive) + for parameter, kind in EXTENSIONS[name]: + primitive.add_input(parameter, str(kind)) + optional = {"Thermocycle": ("lidTemperature", "blockVolume"), "Distribute": ("airGap",)} + for parameter in optional.get(name, ()): + primitive.add_input(parameter, str(OM.Measure), optional=True) + return primitive + + for index, step in enumerate(recorded["steps"]): + kind = step["kind"] + primitive: str | labop.Primitive + arguments: dict[str, object] + if kind == "Transfer": + primitive = "Transfer" + arguments = { + "source": well(step["source"]), + "destination": well(step["destination"]), + "amount": measure(step["volume"], "microlitre"), + } + elif kind == "Mix": + primitive = "PipetteMix" + arguments = { + "samples": well(step["location"]), + "amount": measure(step["volume"], "microlitre"), + "cycleCount": measure(step["cycles"], "one"), + } + elif kind == "Wait": + primitive, arguments = extension(kind), {"duration": measure(step["seconds"], "second")} + elif kind == "ManualInstruction": + primitive, arguments = extension("OperatorPause"), {"instruction": step["text"]} + elif kind == "Distribute": + members = [well(location) for location in step["destinations"]] + group = document.find(EXT + f"OrderedGroup_{len(members)}") + if group is None: + group = labop.Primitive( + EXT + f"OrderedGroup_{len(members)}", + description="Return the ordered sample aliases, including duplicates; " + "no laboratory operation.", + ) + document.add(group) + for member in range(len(members)): + group.add_input(f"member_{member}", str(LABOP.SampleCollection)) + group.add_output("samples", str(LABOP.SampleCollection)) + selected = protocol.primitive_step( + group, **{f"member_{i}": value for i, value in enumerate(members)} + ) + primitive = extension(kind) + arguments = { + "source": well(step["source"]), + "destinations": selected.output_pin("samples"), + "amount": measure(step["volume"], "microlitre"), + } + if step["air_gap"] is not None: + arguments["airGap"] = measure(step["air_gap"], "microlitre") + elif kind in ("SetTemperature", "Thermocycle"): + primitive, arguments = extension(kind), {"samples": sources[step["resource"]]} + if kind == "SetTemperature": + arguments["temperature"] = measure(step["celsius"], "degreeCelsius") + else: + arguments.update( + profile=json.dumps(step["profile"], sort_keys=True), cycles=step["cycles"] + ) + for field, name, unit in ( + ("lid_celsius", "lidTemperature", "degreeCelsius"), + ("block_volume", "blockVolume", "microlitre"), + ): + if step[field] is not None: + arguments[name] = measure(step[field], unit) + else: + raise ValueError(f"No LabOP mapping for {kind}") + action = protocol.primitive_step(primitive, **arguments) + text(action, "semanticStep", json.dumps(step, sort_keys=True)) + text(action, "stepIndex", str(index)) + outputs = { + place["location"]["resource"] + for place in recorded["placements"] + if place["sample_id"] in recorded["output_sample_ids"] + } + for index, name in enumerate(sorted(outputs)): + protocol.designate_output(f"result_{index}", str(LABOP.SampleCollection), sources[name]) + protocol.order(protocol.get_last_step(), protocol.final()) + report = document.validate() + if report.errors: + raise ValueError("\n".join(str(error) for error in report.errors)) + return document + + +def export_file(plan: Path, output: Path | None = None) -> Path: + """Write LabOP beside a saved plan, refusing to replace a different artifact.""" + document = export(json.loads(plan.read_text(encoding="utf-8"))) + graph = to_canonical_graph(document.graph()) + content = "\n".join(sorted(graph.serialize(format="nt").splitlines())) + "\n" + path = output or plan.with_name("protocol.labop.ttl") + if path.exists() and path.read_text(encoding="utf-8") != content: + raise FileExistsError(f"{path} already contains a different artifact") + path.write_text(content, encoding="utf-8") + return path + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("plan", type=Path) + parser.add_argument("--out", type=Path) + args = parser.parse_args() + print(export_file(args.plan, args.out)) + + +if __name__ == "__main__": + main() diff --git a/integrations/labop/requirements.txt b/integrations/labop/requirements.txt new file mode 100644 index 0000000..7418027 --- /dev/null +++ b/integrations/labop/requirements.txt @@ -0,0 +1,6 @@ +# Installed only in this integration's separate virtual environment. +labop @ git+https://github.com/Bioprotocols/labop@2e2bd88150c71a440771fd369f40295dfd622324 +sbol3==1.2.0.post0 +sbol-factory==1.1.2 +# Upstream imports IPython from its Markdown converter during package import. +ipython==9.17.1 diff --git a/pyproject.toml b/pyproject.toml index 0d888ac..efef5ab 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,6 +56,7 @@ include = [ "/tests", "/examples", "/scripts", + "/integrations", "/docs", "/README.md", "/LICENSE", @@ -85,10 +86,10 @@ asyncio_mode = "auto" markers = ["integration: requires the optional robot SDKs; never connects to hardware"] [tool.mypy] -files = ["src/lab"] +files = ["src/lab", "integrations/labop"] check_untyped_defs = true disallow_untyped_defs = true [[tool.mypy.overrides]] -module = ["opentrons.*", "pylabrobot.*", "sbol3.*"] +module = ["opentrons.*", "pylabrobot.*", "sbol3.*", "labop.*", "uml.*"] ignore_missing_imports = true diff --git a/scripts/check_install.py b/scripts/check_install.py index fd1bc79..2ce53e1 100644 --- a/scripts/check_install.py +++ b/scripts/check_install.py @@ -45,7 +45,9 @@ def main() -> None: assert plan["compiler_version"] == package.version assert (output / "protocol.html").stat().st_size > 0 - assert not any(name.split(".")[0] in {"opentrons", "pylabrobot"} for name in sys.modules) + isolated = {"opentrons", "pylabrobot", "labop", "uml", "sbol_factory"} + assert not any(name.split(".")[0] in isolated for name in sys.modules) + assert not files("lab").joinpath("labop").is_dir() print(f"lab-compiler {package.version}: installed package check passed")