From 6c3e306c8b3874bbc461b8a0d1bdd04b16fe159f Mon Sep 17 00:00:00 2001 From: Mike Arpaia Date: Wed, 30 Sep 2026 22:23:11 -0600 Subject: [PATCH] Generate staged cloning protocols from resolved build plans --- examples/cloning_workflow.py | 168 ++++++++++ examples/provenance_build.py | 120 +++++++ src/lab/experiments/cloning/__init__.py | 2 + src/lab/experiments/cloning/build.py | 398 ++++++++++++++++++++++++ tests/cloning_integration_fixture.py | 139 +++++++++ tests/test_cloning_integration.py | 298 ++++++++++++++++++ tests/test_experiments.py | 121 +++++++ tests/test_labop.py | 81 +++-- 8 files changed, 1300 insertions(+), 27 deletions(-) create mode 100644 examples/cloning_workflow.py create mode 100644 examples/provenance_build.py create mode 100644 src/lab/experiments/cloning/build.py create mode 100644 tests/cloning_integration_fixture.py create mode 100644 tests/test_cloning_integration.py create mode 100644 tests/test_experiments.py diff --git a/examples/cloning_workflow.py b/examples/cloning_workflow.py new file mode 100644 index 0000000..ea5991f --- /dev/null +++ b/examples/cloning_workflow.py @@ -0,0 +1,168 @@ +"""Compile all cloning stages from explicit synthetic inputs. + +The arbitrary sequences, quantities, and parameters exercise the compiler; +this example does not supply validated laboratory methods. +""" + +import argparse +from dataclasses import replace +from decimal import Decimal + +import lab +from examples.provenance_build import NS +from examples.provenance_build import inputs as assembly_inputs +from lab import uL +from lab.experiments import cloning +from lab.experiments.cloning.methods import ( + ExternalPreparationMethod, + PlatingMethod, + Reagent, + TransformationMethod, +) +from lab.experiments.cloning.planning import BuildRequest, CountTarget +from lab.experiments.cloning.systems import ( + CloningSystem, + ExternalPreparationRecipe, + PlatingRecipe, + TransformationRecipe, +) +from lab.inventory import CountedStock, Inventory, MaterialForm, Stock +from lab.operations import Hold +from lab.provenance import Component, Document, EvidenceState, Implementation +from lab.provenance.vocabulary import SMALL_MOLECULE +from lab.targets import LiquidHandler, Manual + + +def inputs(count=2): + _, inputs = assembly_inputs() + doc = Document.from_snapshot(inputs["document"]) + vector = doc.get(NS + "/vector", Component) + target = doc.get(NS + "/target", Component) + cells = Component(identity=NS + "/cells", types=("https://example.org/cell",)) + strain = Component(identity=NS + "/strain", types=("https://example.org/cell",)) + broth = Component(identity=NS + "/broth", types=(SMALL_MOLECULE,)) + substrate = Component(identity=NS + "/substrate", types=(SMALL_MOLECULE,)) + doc.add(cells, strain, broth, substrate) + stocks = [s for s in inputs["inventory"].stocks if s.design != vector.ref] + for design, form in ((cells, MaterialForm.COMPETENT_CELLS), (broth, MaterialForm.REAGENT)): + material = Implementation( + identity=design.identity + "/material", + derived_from=(design.ref,), + evidence_state=EvidenceState.RECORDED, + ) + doc.add(material) + stocks.append( + Stock( + identity=design.identity + "/stock", + design=design.ref, + implementation=material.ref, + form=form, + quantity=20 * uL, + ) + ) + material = Implementation( + identity=NS + "/received_stab", + derived_from=(vector.ref,), + evidence_state=EvidenceState.RECORDED, + ) + doc.add(material) + stocks.append( + CountedStock( + identity=NS + "/stab_stock", + design=vector.ref, + implementation=material.ref, + form=MaterialForm.BACTERIAL_STAB, + count=1, + ) + ) + preparation = ExternalPreparationMethod( + identity=NS + "/preparation_method", + procedure=NS + "/sop", + instructions=( + "Apply the supplied test procedure and record the resulting material and quantity." + ), + source_count=1, + output_volume_ul=Decimal(10), + ) + transformation = TransformationMethod( + identity=NS + "/transformation_method", + cell_volume_ul=Decimal(2), + dna_volume_ul=Decimal(1), + recovery=Reagent(component=broth.ref, volume_ul=Decimal(2)), + profile=(Hold(Decimal(25), Decimal(1)),), + recovery_profile=(Hold(Decimal(25), Decimal(1)),), + output_volume_ul=Decimal(5), + cell_mix_volume_ul=Decimal(1), + cell_mix_cycles=1, + dna_mix_cycles=1, + initial_celsius=Decimal(25), + ) + plating = PlatingMethod( + identity=NS + "/plating_method", + substrate=substrate.ref, + diluent=broth.ref, + transfer_volume_ul=Decimal(1), + dilution_factors=(Decimal(2), Decimal(3)), + spot_volume_ul=Decimal(1), + spot_height_mm=Decimal(2), + mix_volume_ul=Decimal(1), + mix_cycles=1, + ) + system = CloningSystem( + identity=NS + "/complete_system", + recipes=( + *inputs["system"].recipes, + ExternalPreparationRecipe( + identity=NS + "/preparation_recipe", + product=vector.ref, + source=vector.ref, + source_form=MaterialForm.BACTERIAL_STAB, + output_form=MaterialForm.DNA, + method=preparation.identity, + ), + TransformationRecipe( + identity=NS + "/transformation_recipe", + product=strain.ref, + chassis=cells.ref, + plasmids=(target.ref,), + method=transformation.identity, + ), + PlatingRecipe( + identity=NS + "/plating_recipe", product=strain.ref, method=plating.identity + ), + ), + ) + return BuildRequest( + identity=NS + "/integrated_request", targets=(CountTarget(design=strain.ref, count=count),) + ), { + **inputs, + "document": doc.freeze(), + "inventory": Inventory(identity=NS + "/integrated_inventory", stocks=tuple(stocks)), + "system": system, + "methods": replace( + inputs["methods"], + transformations=(transformation,), + platings=(plating,), + preparations=(preparation,), + ), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--target", choices=("manual", "ot2", "flex", "star"), default="manual") + parser.add_argument("--out") + args = parser.parse_args() + request, supplied = inputs() + plan = cloning.plan(request, **supplied) + plan.require_ready() + experiment = cloning.build(plan) + target = Manual() if args.target == "manual" else LiquidHandler(args.target) + compilation = lab.compile(experiment, target) + print(compilation.write(args.out or f"build/cloning/{args.target}")) + for stage in compilation.stages: + print(f"{stage.protocol.name}: {stage.target.name}") + + +if __name__ == "__main__": + main() diff --git a/examples/provenance_build.py b/examples/provenance_build.py new file mode 100644 index 0000000..294b3ac --- /dev/null +++ b/examples/provenance_build.py @@ -0,0 +1,120 @@ +"""Compile a synthetic software fixture from SBOL designs through robot artifacts. + +The short DNA strings and arbitrary method numbers demonstrate the compiler. +They are not a laboratory cloning recipe. +""" + +import argparse +from decimal import Decimal + +import lab +from lab import uL +from lab.experiments import cloning +from lab.experiments.cloning.methods import AssemblyMethod, CloningMethods, Reagent +from lab.experiments.cloning.planning import BuildRequest, BuildTarget +from lab.experiments.cloning.sequences import CIRCULAR, LINEAR +from lab.experiments.cloning.systems import AssemblyRecipe, CloningSystem, FragmentSelection +from lab.inventory import Inventory, MaterialForm, Stock +from lab.operations import Hold +from lab.provenance import Component, Document, EvidenceState, Implementation, Sequence +from lab.provenance.vocabulary import DNA, IUPAC_DNA, SMALL_MOLECULE +from lab.targets import LiquidHandler, Manual + +NS = "https://example.org/lab_demo" + + +def inputs(): + volume, stock_volume = "3", "10" + doc = Document(namespace=NS) + vector_sequence = Sequence( + identity=doc.iri("vector_sequence"), elements="AAAAGAATTCTTTT", encoding=IUPAC_DNA + ) + + insert_sequence = Sequence( + identity=doc.iri("insert_sequence"), elements="GGGAATTCCCCGAATTCGG", encoding=IUPAC_DNA + ) + vector = Component( + identity=doc.iri("vector"), types=(DNA, CIRCULAR), sequences=(vector_sequence.ref,) + ) + insert = Component( + identity=doc.iri("insert"), types=(DNA, LINEAR), sequences=(insert_sequence.ref,) + ) + target = Component(identity=doc.iri("target"), types=(DNA, CIRCULAR)) + buffer = Component(identity=doc.iri("buffer"), types=(SMALL_MOLECULE,)) + water = Component(identity=doc.iri("water"), types=(SMALL_MOLECULE,)) + doc.add(vector_sequence, insert_sequence, vector, insert, target, buffer, water) + stocks = [] + for design, form in ( + (vector, MaterialForm.DNA), + (insert, MaterialForm.DNA), + (buffer, MaterialForm.REAGENT), + (water, MaterialForm.REAGENT), + ): + implementation = Implementation( + identity=design.identity + "/implementation", + derived_from=(design.ref,), + evidence_state=EvidenceState.RECORDED, + ) + doc.add(implementation) + stocks.append( + Stock( + identity=design.identity + "/stock", + design=design.ref, + implementation=implementation.ref, + form=form, + quantity=Decimal(stock_volume) * uL, + ) + ) + recipe = AssemblyRecipe( + identity=doc.iri("recipe"), + product=target.ref, + enzyme="EcoRI", + fragments=( + FragmentSelection(component=vector.ref, left_cut=5, right_cut=5), + FragmentSelection(component=insert.ref, left_cut=3, right_cut=12), + ), + method=doc.iri("method"), + ) + method = AssemblyMethod( + identity=recipe.method, + enzyme="EcoRI", + dna_volume_ul=Decimal(1), + reaction_volume_ul=Decimal(5), + output_volume_ul=Decimal(5), + reagents=(Reagent(component=buffer.ref, volume_ul=Decimal(1)),), + diluent=water.ref, + profile=(Hold(Decimal(25), Decimal(1)),), + cycles=1, + mix_volume_ul=Decimal(2), + mix_cycles=1, + ) + return ( + BuildRequest( + identity=doc.iri("request"), + targets=(BuildTarget(design=target.ref, volume_ul=Decimal(volume)),), + ), + dict( + document=doc.freeze(), + inventory=Inventory(identity=doc.iri("inventory"), stocks=tuple(stocks)), + system=CloningSystem(identity=doc.iri("system"), recipes=(recipe,)), + methods=CloningMethods(assemblies=(method,)), + ), + ) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--target", choices=("manual", "ot2", "flex", "star"), default="manual") + parser.add_argument("--out", default=None) + args = parser.parse_args() + request, supplied = inputs() + planned = cloning.plan(request, **supplied) + planned.require_ready() + experiment = cloning.build(planned) + target = Manual() if args.target == "manual" else LiquidHandler(args.target) + compilation = lab.compile(experiment, target) + print(compilation.write(args.out or f"build/provenance/{args.target}")) + + +if __name__ == "__main__": + main() diff --git a/src/lab/experiments/cloning/__init__.py b/src/lab/experiments/cloning/__init__.py index fb3c438..c926be7 100644 --- a/src/lab/experiments/cloning/__init__.py +++ b/src/lab/experiments/cloning/__init__.py @@ -4,6 +4,7 @@ containers and sites; compilation lowers it for one handler. """ +from lab.experiments.cloning.build import build from lab.experiments.cloning.decks import assembly_deck, plating_deck, transformation_deck from lab.experiments.cloning.methods import ( AssemblyMethod, @@ -74,6 +75,7 @@ "PlatingMethod", "PlatingRecipe", "Reagent", + "build", "plan", "AssemblyLayout", "AssemblyReaction", diff --git a/src/lab/experiments/cloning/build.py b/src/lab/experiments/cloning/build.py new file mode 100644 index 0000000..6da2c9c --- /dev/null +++ b/src/lab/experiments/cloning/build.py @@ -0,0 +1,398 @@ +"""Freeze planned cloning routes as ordinary protocols and explicit operator stages.""" + +from collections import defaultdict +from dataclasses import replace +from decimal import Decimal + +from lab.artifacts import SourceArtifact, canonical_json +from lab.deck import Container, Deck, DeckSite +from lab.experiment import ExperimentPlan, MaterialHandoff, ProtocolStage +from lab.experiments.cloning.methods import ( + AssemblyMethod, + ExternalPreparationMethod, + PlatingMethod, + TransformationMethod, +) +from lab.experiments.cloning.planning import Allocation, BuildPlan +from lab.experiments.cloning.systems import TransformationRecipe +from lab.labware import PCR_PLATE_96, ContainerSpec, LabwareKind, LabwareSpec +from lab.operations import MaterialPort +from lab.protocol import Protocol, Well +from lab.provenance import Activity, Document, EvidenceState, Implementation, Plan +from lab.samples import Location, Sample +from lab.units import celsius, seconds, uL, units + + +def _transfers( + protocol: Protocol, + wells: dict[str, Well], + allocations: tuple[Allocation, ...], + destination: Well, +) -> None: + for allocation in allocations: + protocol.transfer( + wells[allocation.implementation.identity], destination, volume=allocation.volume_ul * uL + ) + + +def _role(inputs: tuple[Allocation, ...], name: str) -> tuple[Allocation, ...]: + return tuple(allocation for allocation in inputs if allocation.role == name) + + +def build( + plan: BuildPlan, + *, + reagent_container: Container | None = None, + preparation_container: ContainerSpec | None = None, +) -> ExperimentPlan: + """Generate one stage per selected reaction, preserving material quantities. + + External procedures compile as operator stages. Counted stocks occupy logical + positions there; those positions do not prescribe robot labware. Every input + load and substrate is an explicit precondition, not an inferred preparation. + """ + plan.require_ready() + reagent_container = reagent_container or Container( + id="reagents", + labware=PCR_PLATE_96, + site=DeckSite.PLATES, + ) + preparation_container = preparation_container or ContainerSpec( + id="products", labware=PCR_PLATE_96 + ) + if reagent_container.id != "reagents" or preparation_container.id != "products": + raise ValueError("Container ids must be 'reagents' and 'products', respectively") + document = Document.from_snapshot(plan.document) + stages: list[ProtocolStage] = [] + remaining = {stock.implementation.identity: stock.amount for stock in plan.inventory.stocks} + produced_at: dict[str, tuple[str, Location, ContainerSpec]] = {} + for task in plan.tasks: + method = task.method + if method is None or task.recipe is None: + raise ValueError("The builder requires resolved typed recipes and methods") + external = isinstance(method, ExternalPreparationMethod) + plating = isinstance(method, PlatingMethod) + product_container = ( + preparation_container + if external + else Container( + id="agar" if plating else "products", + labware=PCR_PLATE_96, + site=DeckSite.MORE_PLATES if plating else DeckSite.THERMOCYCLER, + ) + ) + protocol_id = task.identity + "/protocol" + description = f"Planned {task.kind.value} using {method.identity}." + if isinstance(method, PlatingMethod): + description += ( + f" Supplied substrate: {method.substrate.identity}. " + "Output is a deposited sample, not a confirmed colony." + ) + protocol = Protocol( + name=f"{task.kind.value.capitalize()} {len(stages) + 1}", + identity=protocol_id, + description=description, + ) + containers: list[ContainerSpec] = [reagent_container, product_container] + plates = { + container.id: protocol.plate( + container.id, + shape=(container.labware.rows, container.labware.columns), + capacity=container.labware.capacity_ul * uL, + ) + for container in containers + } + container: ContainerSpec + if plating: + container = Container(id="dilutions", labware=PCR_PLATE_96, site=DeckSite.PLATES) + containers.append(container) + plates[container.id] = protocol.plate( + container.id, capacity=container.labware.capacity_ul * uL + ) + counted_stocks = { + a.implementation for a in task.inputs if a.form.counted and a.producer is None + } + if counted_stocks: + container = ContainerSpec( + id="external_inputs", + labware=LabwareSpec( + kind=LabwareKind.PLATE, + rows=1, + columns=len(counted_stocks), + capacity_ul=Decimal(1), + ), + ) + containers.append(container) + plates[container.id] = protocol.plate( + container.id, shape=(1, len(counted_stocks)), capacity=1 * uL + ) + wells: dict[str, Well] = {} + handoffs: list[MaterialHandoff] = [] + stock_index = count_index = 0 + samples: dict[str, Sample] = {} + for allocation in task.inputs: + identity = allocation.implementation.identity + if identity in wells: + continue + if allocation.producer is None: + if allocation.form.counted: + well = plates["external_inputs"][f"A{count_index + 1}"] + count_index += 1 + else: + names = plates["reagents"].wells + if stock_index >= len(names): + raise ValueError("Inputs exceed the reagent container's number of wells") + well = plates["reagents"][names[stock_index]] + stock_index += 1 + else: + producer, source, previous_container = produced_at[identity] + name = f"upstream_{len(handoffs) + 1}" + container = Container( + id=name, + labware=previous_container.labware, + site=DeckSite.MORE_PLATES + if sum( + isinstance(item, Container) and item.site is DeckSite.PLATES + for item in containers + ) + >= 2 + else DeckSite.PLATES, + ) + containers.append(container) + plate = protocol.plate( + name, + shape=(container.labware.rows, container.labware.columns), + capacity=container.labware.capacity_ul * uL, + ) + well = plate[source.well] + handoffs.append( + MaterialHandoff( + producer=producer, + implementation=allocation.implementation, + source=source, + destination=Location(well.resource, well.name), + volume_ul=Decimal(0) if allocation.form.counted else remaining[identity], + count=int(remaining[identity]) if allocation.form.counted else None, + ) + ) + wells[identity] = well + if not allocation.form.counted: + protocol.load(well, identity, volume=remaining[identity] * uL) + sample = Sample( + id=protocol_id + f"/input_{len(samples) + 1}", + material_identity=allocation.design.identity, + label=allocation.design.identity, + design=allocation.design, + implementation=allocation.implementation, + form=allocation.form, + count=int(remaining[identity]) if allocation.form.counted else None, + ) + samples[identity] = sample + protocol.add_sample(sample, at=well, is_input=True) + output = plates[product_container.id]["A1"] + + if isinstance(method, AssemblyMethod): + _transfers(protocol, wells, task.inputs, output) + protocol.mix(output, volume=method.mix_volume_ul * uL, cycles=method.mix_cycles) + protocol.thermocycle( + plates[product_container.id], + tuple((celsius(h.celsius), h.seconds * seconds) for h in method.profile), + cycles=method.cycles, + lid_temperature=None if method.lid_celsius is None else celsius(method.lid_celsius), + ) + physical_output = method.reaction_volume_ul + elif isinstance(method, TransformationMethod): + assert isinstance(task.recipe, TransformationRecipe) + if method.initial_celsius is not None: + protocol.set_temperature( + plates[product_container.id], celsius(method.initial_celsius) + ) + _transfers(protocol, wells, _role(task.inputs, "cells"), output) + protocol.mix( + output, volume=method.cell_mix_volume_ul * uL, cycles=method.cell_mix_cycles + ) + for allocation in task.inputs: + if allocation.role.startswith("dna_"): + protocol.mix( + wells[allocation.implementation.identity], + volume=allocation.volume_ul * uL, + cycles=method.dna_mix_cycles, + ) + _transfers(protocol, wells, (allocation,), output) + physical_output = method.reaction_volume(len(task.recipe.plasmids)) + protocol.thermocycle( + plates[product_container.id], + tuple((celsius(h.celsius), h.seconds * seconds) for h in method.profile), + block_volume=(physical_output - method.recovery.volume_ul) * uL, + ) + _transfers(protocol, wells, _role(task.inputs, "recovery"), output) + protocol.thermocycle( + plates[product_container.id], + tuple((celsius(h.celsius), h.seconds * seconds) for h in method.recovery_profile), + block_volume=physical_output * uL, + ) + elif isinstance(method, PlatingMethod): + if len(method.dilution_factors) > len(plates["dilutions"].wells): + raise ValueError("Dilution series exceeds the plate's number of wells") + previous = None + previous_sample = None + for index, _factor in enumerate(method.dilution_factors): + destination = plates["dilutions"][plates["dilutions"].wells[index]] + additions = _role(task.inputs, f"diluent_{index + 1}") + _transfers(protocol, wells, additions, destination) + if previous is None: + _transfers(protocol, wells, _role(task.inputs, "culture"), destination) + parents = tuple( + samples[a.implementation.identity].id + for a in (*additions, *_role(task.inputs, "culture")) + ) + else: + protocol.transfer(previous, destination, volume=method.transfer_volume_ul * uL) + parents = ( + *(samples[a.implementation.identity].id for a in additions), + str(previous_sample), + ) + protocol.mix( + destination, volume=method.mix_volume_ul * uL, cycles=method.mix_cycles + ) + material = Implementation( + identity=task.identity + f"/dilution_{index + 1}", + derived_from=(task.design,), + generated_by=(document.get(task.identity, Activity).ref,), + evidence_state=EvidenceState.PLANNED, + ) + document.add(material) + sample = Sample( + id=material.identity, + material_identity=task.design.identity, + label=task.design.identity, + design=task.design, + implementation=material.ref, + role="dilution", + dilution=index + 1, + parent_ids=tuple(dict.fromkeys(parents)), + ) + protocol.add_sample(sample, at=destination) + previous, previous_sample = destination, sample.id + assert previous is not None + protocol.transfer( + previous, + output, + volume=method.spot_volume_ul * uL, + destination_height=method.spot_height_mm * units.millimeter, + ) + physical_output = Decimal(1) + else: + consumed: dict[str, Decimal] = defaultdict(Decimal) + for allocation in task.inputs: + consumed[allocation.implementation.identity] += allocation.amount + ports = tuple( + MaterialPort( + location=Location(wells[identity].resource, wells[identity].name), + volume_ul=Decimal(0) if samples[identity].count is not None else amount, + count=int(amount) if samples[identity].count is not None else 0, + ) + for identity, amount in consumed.items() + ) + protocol.external_preparation( + procedure=method.procedure, + instructions=method.instructions, + inputs=ports, + outputs=( + MaterialPort( + location=Location(output.resource, output.name), + volume_ul=method.output_volume_ul, + count=method.output_count, + ), + ), + ) + physical_output = ( + Decimal(method.output_count) + if task.output_form.counted + else method.output_volume_ul + ) + for allocation in task.inputs: + remaining[allocation.implementation.identity] -= allocation.amount + protocol.add_sample( + Sample( + id=task.output.identity, + material_identity=task.design.identity, + label=task.design.identity, + design=task.design, + implementation=task.output, + form=task.output_form, + count=task.output_count if task.output_form.counted else None, + parent_ids=(str(previous_sample),) + if plating + else tuple(sample.id for sample in samples.values()), + role=task.kind.value, + ), + at=output, + is_output=True, + ) + remaining[task.output.identity] = physical_output + produced_at[task.output.identity] = ( + task.identity, + Location(output.resource, output.name), + product_container, + ) + recorded = protocol.snapshot() + specification = Plan( + identity=task.identity + "/specification", + protocol=recorded.identity, + derived_from=(document.get(method.identity, Plan).ref,), + ) + document.add(specification) + activity = document.get(task.identity, Activity) + document.replace( + replace( + activity, + association=tuple( + replace(association, plan=specification.ref) + for association in activity.association + ), + ) + ) + stages.append( + ProtocolStage( + identity=task.identity, + protocol=recorded, + deck=Deck(containers=tuple(containers)), + depends_on=task.depends_on, + handoffs=tuple(handoffs), + external=external, + ) + ) + return ExperimentPlan( + identity=plan.request.identity, + provenance=document.freeze(), + stages=tuple(stages), + build_json=plan.plan_json, + inputs=( + SourceArtifact(name="build-provenance.ttl", text=plan.document.to_turtle()), + SourceArtifact( + name="inventory.json", + text=canonical_json( + {"format": "lab.inventory.v1", "inventory": plan.inventory}, + ), + ), + SourceArtifact( + name="catalog.json", + text=canonical_json( + {"format": "lab.catalog.v1", "catalog": plan.catalog}, + ), + ), + SourceArtifact( + name="system.json", + text=canonical_json( + {"format": "lab.cloning-system.v1", "system": plan.system}, + ), + ), + SourceArtifact( + name="methods.json", + text=canonical_json( + {"format": "lab.cloning-methods.v1", "methods": plan.methods}, + ), + ), + ), + ) diff --git a/tests/cloning_integration_fixture.py b/tests/cloning_integration_fixture.py new file mode 100644 index 0000000..efa3ac8 --- /dev/null +++ b/tests/cloning_integration_fixture.py @@ -0,0 +1,139 @@ +"""Synthetic compiler fixtures; the arbitrary parameters are not wet-lab methods.""" + +from dataclasses import replace +from decimal import Decimal + +from lab import uL +from lab.experiments.cloning.methods import ( + ExternalPreparationMethod, + PlatingMethod, + Reagent, + TransformationMethod, +) +from lab.experiments.cloning.planning import BuildRequest, CountTarget +from lab.experiments.cloning.systems import ( + CloningSystem, + ExternalPreparationRecipe, + PlatingRecipe, + TransformationRecipe, +) +from lab.inventory import CountedStock, Inventory, MaterialForm, Stock +from lab.operations import Hold +from lab.provenance import Component, Document, EvidenceState, Implementation +from lab.provenance.vocabulary import SMALL_MOLECULE +from tests.planning_fixture import NS, planning_case + + +def integrated_case(count=2): + _, inputs = planning_case() + doc = Document.from_snapshot(inputs["document"]) + vector = doc.get(NS + "/vector", Component) + target = doc.get(NS + "/target", Component) + cells = Component(identity=NS + "/cells", types=("https://example.org/cell",)) + strain = Component(identity=NS + "/strain", types=("https://example.org/cell",)) + broth = Component(identity=NS + "/broth", types=(SMALL_MOLECULE,)) + substrate = Component(identity=NS + "/substrate", types=(SMALL_MOLECULE,)) + doc.add(cells, strain, broth, substrate) + stocks = [s for s in inputs["inventory"].stocks if s.design != vector.ref] + for design, form in ((cells, MaterialForm.COMPETENT_CELLS), (broth, MaterialForm.REAGENT)): + material = Implementation( + identity=design.identity + "/material", + derived_from=(design.ref,), + evidence_state=EvidenceState.RECORDED, + ) + doc.add(material) + stocks.append( + Stock( + identity=design.identity + "/stock", + design=design.ref, + implementation=material.ref, + form=form, + quantity=20 * uL, + ) + ) + material = Implementation( + identity=NS + "/received_stab", + derived_from=(vector.ref,), + evidence_state=EvidenceState.RECORDED, + ) + doc.add(material) + stocks.append( + CountedStock( + identity=NS + "/stab_stock", + design=vector.ref, + implementation=material.ref, + form=MaterialForm.BACTERIAL_STAB, + count=1, + ) + ) + preparation = ExternalPreparationMethod( + identity=NS + "/preparation_method", + procedure=NS + "/sop", + instructions=( + "Apply the supplied test procedure and record the resulting material and quantity." + ), + source_count=1, + output_volume_ul=Decimal(10), + ) + transformation = TransformationMethod( + identity=NS + "/transformation_method", + cell_volume_ul=Decimal(2), + dna_volume_ul=Decimal(1), + recovery=Reagent(component=broth.ref, volume_ul=Decimal(2)), + profile=(Hold(Decimal(25), Decimal(1)),), + recovery_profile=(Hold(Decimal(25), Decimal(1)),), + output_volume_ul=Decimal(5), + cell_mix_volume_ul=Decimal(1), + cell_mix_cycles=1, + dna_mix_cycles=1, + initial_celsius=Decimal(25), + ) + plating = PlatingMethod( + identity=NS + "/plating_method", + substrate=substrate.ref, + diluent=broth.ref, + transfer_volume_ul=Decimal(1), + dilution_factors=(Decimal(2), Decimal(3)), + spot_volume_ul=Decimal(1), + spot_height_mm=Decimal(2), + mix_volume_ul=Decimal(1), + mix_cycles=1, + ) + system = CloningSystem( + identity=NS + "/complete_system", + recipes=( + *inputs["system"].recipes, + ExternalPreparationRecipe( + identity=NS + "/preparation_recipe", + product=vector.ref, + source=vector.ref, + source_form=MaterialForm.BACTERIAL_STAB, + output_form=MaterialForm.DNA, + method=preparation.identity, + ), + TransformationRecipe( + identity=NS + "/transformation_recipe", + product=strain.ref, + chassis=cells.ref, + plasmids=(target.ref,), + method=transformation.identity, + ), + PlatingRecipe( + identity=NS + "/plating_recipe", product=strain.ref, method=plating.identity + ), + ), + ) + return BuildRequest( + identity=NS + "/integrated_request", targets=(CountTarget(design=strain.ref, count=count),) + ), { + **inputs, + "document": doc.freeze(), + "inventory": Inventory(identity=NS + "/integrated_inventory", stocks=tuple(stocks)), + "system": system, + "methods": replace( + inputs["methods"], + transformations=(transformation,), + platings=(plating,), + preparations=(preparation,), + ), + } diff --git a/tests/test_cloning_integration.py b/tests/test_cloning_integration.py new file mode 100644 index 0000000..ffa2f6e --- /dev/null +++ b/tests/test_cloning_integration.py @@ -0,0 +1,298 @@ +from dataclasses import replace +from decimal import Decimal +from io import StringIO + +import pytest + +import lab +from lab.experiments import cloning +from lab.inventory import CountedStock, MaterialForm +from lab.operations import ExternalPreparation, Thermocycle, Transfer +from lab.provenance import Component, Document, EvidenceState, Implementation +from lab.targets import LiquidHandler, Manual +from tests.cloning_integration_fixture import integrated_case +from tests.planning_fixture import NS + + +def test_all_cloning_routes_share_material_and_evidence_accounting(): + request, inputs = integrated_case() + planned = cloning.plan(request, **inputs) + planned.require_ready() + assert [task.kind.value for task in planned.tasks] == [ + "preparation", + "assembly", + "transformation", + "plating", + "plating", + ] + assert sum(product.count for product in planned.products) == 2 + assert all( + product.volume_ul == 0 and product.form is MaterialForm.PLATED_SAMPLE + for product in planned.products + ) + experiment = cloning.build(planned) + assert len(experiment.stages) == 5 + assert experiment.stages[0].external + assert isinstance(experiment.stages[0].protocol.steps[0], ExternalPreparation) + transformation = experiment.stages[2] + assert ( + len([step for step in transformation.protocol.steps if isinstance(step, Thermocycle)]) == 2 + ) + first, second = experiment.stages[-2:] + assert first.handoffs[0].implementation == second.handoffs[0].implementation + assert (first.handoffs[0].volume_ul, second.handoffs[0].volume_ul) == (Decimal(5), Decimal(4)) + for task in planned.tasks: + material = experiment.provenance.resolve(task.output) + assert material.built is None and material.evidence_state is EvidenceState.PLANNED + compilation = lab.compile(experiment, Manual()) + assert "Expected output" in compilation.files["methods.md"] + + +def test_all_method_recipe_and_stock_inputs_round_trip(tmp_path): + request, inputs = integrated_case() + assert ( + cloning.CloningMethods.read(inputs["methods"].write(tmp_path / "methods.json")) + == inputs["methods"] + ) + assert ( + cloning.CloningSystem.read(inputs["system"].write(tmp_path / "system.json")) + == inputs["system"] + ) + inventory = inputs["inventory"] + assert ( + type(inventory).read( + inventory.write(tmp_path / "inventory.json"), document=inputs["document"] + ) + == inventory + ) + + +def test_missing_preparation_never_becomes_an_implicit_yield(): + request, inputs = integrated_case() + system = replace( + inputs["system"], + recipes=tuple( + recipe for recipe in inputs["system"].recipes if recipe.kind != "preparation" + ), + ) + planned = cloning.plan(request, **{**inputs, "system": system}) + assert not planned.ready + assert any(requirement.kind.value == "preparation" for requirement in planned.requirements) + with pytest.raises(ValueError, match="not ready"): + cloning.build(planned) + + +def test_counted_stock_is_not_reused_when_repeating_external_preparation(): + _, inputs = integrated_case() + vector = inputs["document"].get(NS + "/vector", Component) + request = cloning.BuildRequest( + identity=NS + "/repeat_preparation", + targets=(cloning.BuildTarget(design=vector.ref, volume_ul=Decimal(11)),), + ) + planned = cloning.plan(request, **inputs) + assert not planned.ready + assert ( + sum( + allocation.count + for allocation in planned.allocations + if allocation.stock == NS + "/stab_stock" + ) + == 1 + ) + assert sum(requirement.count for requirement in planned.requirements) == 1 + assert all(requirement.volume_ul == 0 for requirement in planned.requirements) + assert planned.acquisitions[0].count == 1 + + +def test_counted_output_handoff_can_feed_an_explicit_external_procedure(): + _, inputs = integrated_case() + doc = Document.from_snapshot(inputs["document"]) + strain = doc.get(NS + "/strain", Component) + product = Component(identity=NS + "/extracted", types=("https://example.org/dna",)) + doc.add(product) + method = cloning.ExternalPreparationMethod( + identity=NS + "/extract_method", + procedure=NS + "/extract_sop", + instructions="Apply the supplied external test procedure; record the output.", + source_count=1, + output_volume_ul=Decimal(5), + ) + recipe = cloning.ExternalPreparationRecipe( + identity=NS + "/extract_recipe", + product=product.ref, + source=strain.ref, + source_form=MaterialForm.PLATED_SAMPLE, + output_form=MaterialForm.DNA, + method=method.identity, + ) + request = cloning.BuildRequest( + identity=NS + "/extract_request", + targets=(cloning.BuildTarget(design=product.ref, volume_ul=Decimal(2)),), + ) + planned = cloning.plan( + request, + **{ + **inputs, + "document": doc.freeze(), + "system": replace(inputs["system"], recipes=(*inputs["system"].recipes, recipe)), + "methods": replace( + inputs["methods"], preparations=(*inputs["methods"].preparations, method) + ), + }, + ) + planned.require_ready() + experiment = cloning.build(planned) + stage = experiment.stages[-1] + assert stage.external and stage.handoffs[0].count == 1 + assert stage.handoffs[0].volume_ul == 0 + compilation = lab.compile(experiment, Manual()) + assert "1 unit(s)" in compilation.files["methods.md"] + with pytest.raises(ValueError, match="volume|count"): + replace( + experiment, + stages=( + *experiment.stages[:-1], + replace(stage, handoffs=(replace(stage.handoffs[0], count=2),)), + ), + ) + + +@pytest.mark.integration +@pytest.mark.parametrize("handler", [LiquidHandler.OT2, LiquidHandler.FLEX]) +def test_integrated_workflow_runs_each_robot_stage_in_official_simulator(handler): + simulator = pytest.importorskip("opentrons.simulate") + request, inputs = integrated_case(count=1) + experiment = cloning.build(cloning.plan(request, **inputs)) + compilation = lab.compile(experiment, handler) + assert compilation.stages[0].target.name == "Manual" + for stage in compilation.stages[1:]: + simulator.simulate(StringIO(stage.files["protocol.py"])) + assert len(stage.source_map) == len(stage.protocol.steps) + assert ".bottom(z=2)" in compilation.stages[-1].files["protocol.py"] + assert ( + compilation.files["protocol.labop.ttl"] + == lab.compile(experiment, Manual()).files["protocol.labop.ttl"] + ) + + +@pytest.mark.integration +async def test_integrated_star_preview_carries_temperature_and_spot_height(): + backends = pytest.importorskip("pylabrobot.liquid_handling.backends") + resources = pytest.importorskip("pylabrobot.resources") + events = [] + + class Recorder(backends.SerializingBackend): + async def send_command(self, command, data=None): + events.append((command, data)) + + async def thermal(plate, *, profile, cycles, lid_temperature): + events.append(("thermal", profile)) + + async def hold(plate, *, temperature): + events.append(("hold", temperature)) + + request, inputs = integrated_case(count=1) + compilation = lab.compile(cloning.build(cloning.plan(request, **inputs)), LiquidHandler.STAR) + resources.set_volume_tracking(True) + try: + for stage in compilation.stages[1:]: + namespace = {"__name__": "_generated"} + exec(compile(stage.files["protocol.py"], "preview.py", "exec"), namespace) + await namespace["run"]( + Recorder(num_channels=8), + **( + {"thermocycle": thermal, "set_temperature": hold} + if any(isinstance(step, Thermocycle) for step in stage.protocol.steps) + else {} + ), + ) + finally: + resources.set_volume_tracking(False) + assert ("hold", 25) in events + spots = [ + data + for command, data in events + if command == "dispense" + and any(channel.get("liquid_height") == 2 for channel in data["channels"]) + ] + assert spots + + +def test_external_material_accounting_rejects_overdrawn_counts_and_yields(): + request, inputs = integrated_case(count=1) + experiment = cloning.build(cloning.plan(request, **inputs)) + stage = experiment.stages[0] + step = stage.protocol.steps[0] + overdraw = replace(step, inputs=(replace(step.inputs[0], count=2),)) + with pytest.raises(ValueError, match="available material count"): + replace( + experiment, + stages=( + replace(stage, protocol=replace(stage.protocol, steps=(overdraw,))), + *experiment.stages[1:], + ), + ) + overflow = replace(step, outputs=(replace(step.outputs[0], volume_ul=Decimal(101)),)) + with pytest.raises(ValueError, match="capacity"): + replace( + experiment, + stages=( + replace(stage, protocol=replace(stage.protocol, steps=(overflow,))), + *experiment.stages[1:], + ), + ) + with pytest.raises(ValueError, match="positive volume or count"): + replace(step.inputs[0], volume_ul=Decimal(1)) + pipetting = Transfer(step.inputs[0].location, step.outputs[0].location, Decimal(1), step.origin) + with pytest.raises(ValueError, match="Counted material"): + replace( + experiment, + stages=( + replace(stage, protocol=replace(stage.protocol, steps=(pipetting,))), + *experiment.stages[1:], + ), + ) + + +def test_count_targets_reuse_recorded_inventory_without_claiming_colonies(): + request, inputs = integrated_case(count=2) + doc = Document.from_snapshot(inputs["document"]) + design = request.targets[0].design + material = Implementation( + identity=NS + "/existing_spots", + derived_from=(design,), + evidence_state=EvidenceState.RECORDED, + ) + doc.add(material) + stock = CountedStock( + identity=NS + "/spot_stock", + design=design, + implementation=material.ref, + form=MaterialForm.PLATED_SAMPLE, + count=2, + ) + planned = cloning.plan( + request, + **{ + **inputs, + "document": doc.freeze(), + "inventory": replace(inputs["inventory"], stocks=(*inputs["inventory"].stocks, stock)), + }, + ) + assert planned.ready and not planned.tasks + assert planned.products[0].implementation == material.ref and planned.products[0].count == 2 + assert not doc.freeze().resolve(material.ref).built + + +def test_external_preparation_needs_matching_quantity_kinds_and_explicit_target(): + request, inputs = integrated_case(count=1) + method = inputs["methods"].preparations[0] + wrong = replace(method, source_count=0, source_volume_ul=Decimal(1)) + with pytest.raises(ValueError, match="quantity kinds"): + cloning.plan( + request, **{**inputs, "methods": replace(inputs["methods"], preparations=(wrong,))} + ) + experiment = cloning.build(cloning.plan(request, **inputs)) + targets = {stage.identity: stage.deck for stage in experiment.stages} + with pytest.raises(ValueError, match="Manual targets"): + lab.compile(experiment, targets, liquid_handler=LiquidHandler.OT2) diff --git a/tests/test_experiments.py b/tests/test_experiments.py new file mode 100644 index 0000000..ae074af --- /dev/null +++ b/tests/test_experiments.py @@ -0,0 +1,121 @@ +import builtins +import hashlib +import json +from dataclasses import replace +from decimal import Decimal +from io import StringIO + +import pytest + +import lab +from lab.experiments import cloning +from lab.labop import export +from lab.provenance import Document, EvidenceState, Implementation +from lab.targets import LiquidHandler, Manual +from tests.planning_fixture import multilevel_case, planning_case + + +def test_build_freezes_protocol_and_material_identity_before_hardware_selection(tmp_path): + request, inputs = planning_case() + experiment = cloning.build(cloning.plan(request, **inputs)) + artifact = lab.compile(experiment, Manual()) + assert all( + sample.design and sample.implementation for sample in artifact.stages[0].protocol.samples + ) + assert all(step.identity for step in artifact.stages[0].protocol.steps) + files = artifact.files + build_data = json.loads(files["build.json"]) + assert ( + hashlib.sha256(files["inputs/build-provenance.ttl"].encode()).hexdigest() + == (build_data["provenance_sha256"]) + ) + manifest = json.loads(files["bundle.json"]) + for name, checksum in manifest["sha256"].items(): + assert hashlib.sha256(files[name].encode()).hexdigest() == checksum + path = artifact.write(tmp_path) + assert Document.read(path / "provenance.ttl").freeze().digest == experiment.provenance.digest + assert artifact.write(tmp_path) == path + assert "Planned methods" in files["methods.md"] + outputs = tuple( + obj + for obj in experiment.provenance.objects + if isinstance(obj, Implementation) and obj.evidence_state is EvidenceState.PLANNED + ) + assert outputs and all(obj.built is None for obj in outputs) + + +def test_multilevel_handoffs_use_remaining_physical_volume(): + request, inputs = multilevel_case() + experiment = cloning.build(cloning.plan(request, **inputs)) + assert len(experiment.stages) == 3 + first, second, third = experiment.stages + assert second.depends_on == third.depends_on == (first.identity,) + assert second.handoffs[0].implementation == third.handoffs[0].implementation + assert second.handoffs[0].volume_ul == Decimal(5) + assert third.handoffs[0].volume_ul == Decimal(4) + with pytest.raises(ValueError, match="requires an upstream"): + replace(experiment, stages=(first, replace(second, handoffs=()), third)) + with pytest.raises(ValueError, match="volume"): + replace( + experiment, + stages=( + first, + second, + replace(third, handoffs=(replace(third.handoffs[0], volume_ul=Decimal(5)),)), + ), + ) + assert export(experiment).text == export(experiment).text + + +def test_missing_inputs_block_executable_generation(): + request, inputs = planning_case(stock_volume="0") + with pytest.raises(ValueError, match="not ready"): + cloning.build(cloning.plan(request, **inputs)) + + +@pytest.mark.integration +@pytest.mark.parametrize("handler", [LiquidHandler.OT2, LiquidHandler.FLEX]) +def test_assembly_robot_code_runs_in_official_simulator(handler): + simulator = pytest.importorskip("opentrons.simulate") + request, inputs = planning_case() + experiment = cloning.build(cloning.plan(request, **inputs)) + artifact = lab.compile(experiment, handler) + compilation = artifact.stages[0] + log, _ = simulator.simulate(StringIO(compilation.files["protocol.py"])) + assert any("Aspirating 1.0" in event["payload"]["text"] for event in log) + assert [item["step"] for item in compilation.source_map] == [ + step.identity for step in compilation.protocol.steps + ] + manual = lab.compile(experiment, Manual()) + for name in ("experiment.json", "protocol.labop.ttl", "provenance.ttl", "methods.md"): + assert artifact.files[name] == manual.files[name] + + +@pytest.mark.integration +async def test_assembly_star_preview_matches_planned_final_volumes(): + backend_module = pytest.importorskip("pylabrobot.liquid_handling.backends") + resources = pytest.importorskip("pylabrobot.resources") + events = [] + + class Recorder(backend_module.SerializingBackend): + async def send_command(self, command, data=None): + events.append(command) + + async def thermal(plate, *, profile, cycles, lid_temperature): + assert profile == [(25, 1)] + assert plate.get_item("A1").tracker.get_used_volume() == 5 + events.append("thermal") + + request, inputs = planning_case() + experiment = cloning.build(cloning.plan(request, **inputs)) + artifact = lab.compile(experiment, LiquidHandler.STAR) + compilation = artifact.stages[0] + namespace = {"__name__": "_generated"} + exec(builtins.compile(compilation.files["protocol.py"], "preview.py", "exec"), namespace) + resources.set_volume_tracking(True) + try: + await namespace["run"](Recorder(num_channels=8), thermocycle=thermal) + finally: + resources.set_volume_tracking(False) + assert events[-2:] == ["thermal", "stop"] + assert len(compilation.source_map) == len(compilation.protocol.steps) diff --git a/tests/test_labop.py b/tests/test_labop.py index 99b79ec..173d649 100644 --- a/tests/test_labop.py +++ b/tests/test_labop.py @@ -1,3 +1,4 @@ +import hashlib import json from importlib.resources import files from urllib.parse import unquote @@ -7,42 +8,53 @@ from lab import ExperimentPlan, Protocol, ProtocolStage, celsius, seconds, uL from lab.deck import Container, Deck, DeckSite +from lab.experiments import cloning 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 +from tests.cloning_integration_fixture import integrated_case +from tests.planning_fixture import multilevel_case, planning_case -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), - ) +def artifact(*, operations=False): + if operations == "cloning": + request, inputs = integrated_case(count=1) + experiment = cloning.build(cloning.plan(request, **inputs)) + return experiment, export(experiment) + if operations: + 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) + request, inputs = planning_case() + experiment = cloning.build(cloning.plan(request, **inputs)) return experiment, export(experiment) -@pytest.mark.parametrize("operations", [True]) +@pytest.mark.parametrize("operations", [False, True, "cloning"]) def test_labop_has_protocols_exact_semantic_actions_and_no_execution_claims(operations): experiment, result = artifact(operations=operations) graph = result.graph() @@ -73,7 +85,7 @@ def test_labop_has_protocols_exact_semantic_actions_and_no_execution_claims(oper assert amounts and all(graph.value(node, OM.hasUnit) for node in amounts) -@pytest.mark.parametrize("operations", [True]) +@pytest.mark.parametrize("operations", [False, True, "cloning"]) def test_calls_follow_upstream_required_parameter_names_and_directions(operations): _, result = artifact(operations=operations) graph = result.graph() @@ -102,7 +114,7 @@ def test_calls_follow_upstream_required_parameter_names_and_directions(operation assert name in supplied, (behavior, name) -@pytest.mark.parametrize("operations", [True]) +@pytest.mark.parametrize("operations", [False, True, "cloning"]) def test_generated_nodes_satisfy_pinned_ontology_cardinalities(operations): _, result = artifact(operations=operations) graph = result.graph() @@ -139,3 +151,18 @@ def test_generated_nodes_satisfy_pinned_ontology_cardinalities(operations): for predicate in set(graph.predicates()): if str(predicate).startswith((str(LABOP), str(UML))): assert predicate in declared, predicate + + +def test_multilevel_calls_have_explicit_material_flows_and_pinned_resources(): + request, inputs = multilevel_case() + experiment = cloning.build(cloning.plan(request, **inputs)) + graph = export(experiment).graph() + handoff_pins = tuple(graph.subjects(LAB.handoffs)) + assert len(handoff_pins) == 2 + for pin in handoff_pins: + edge = graph.value(predicate=UML.target, object=pin) + assert (edge, RDF.type, UML.ObjectFlow) in graph + manifest = json.loads(files("lab.labop").joinpath("resources/upstream.json").read_text()) + for name, item in manifest["files"].items(): + data = files("lab.labop").joinpath("resources", name).read_bytes() + assert hashlib.sha256(data).hexdigest() == item["sha256"]