diff --git a/pyproject.toml b/pyproject.toml index 9403ad6..db827fe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,6 +24,7 @@ dependencies = [ "pint>=0.24,<0.27", "sbol3==1.2.0.post0", "rdflib>=6.1.1,<7", + "biopython==1.84", ] [project.urls] @@ -91,5 +92,5 @@ check_untyped_defs = true disallow_untyped_defs = true [[tool.mypy.overrides]] -module = ["opentrons.*", "pylabrobot.*", "sbol3.*"] +module = ["opentrons.*", "pylabrobot.*", "sbol3.*", "Bio.*"] ignore_missing_imports = true diff --git a/scripts/check_install.py b/scripts/check_install.py index 1f531b7..6c917e8 100644 --- a/scripts/check_install.py +++ b/scripts/check_install.py @@ -20,6 +20,7 @@ def main() -> None: # Verify public modules and subpackages are present in the installed distribution. import_module("lab.experiments.cloning") + import_module("lab.experiments.cloning.routes") import_module("lab.part") import_module("lab.samples") import_module("lab.provenance") diff --git a/src/lab/experiments/cloning/__init__.py b/src/lab/experiments/cloning/__init__.py index 8a350c5..fb160a6 100644 --- a/src/lab/experiments/cloning/__init__.py +++ b/src/lab/experiments/cloning/__init__.py @@ -5,6 +5,14 @@ """ from lab.experiments.cloning.decks import assembly_deck, plating_deck, transformation_deck +from lab.experiments.cloning.methods import ( + AssemblyMethod, + CloningMethods, + ExternalPreparationMethod, + PlatingMethod, + Reagent, + TransformationMethod, +) from lab.experiments.cloning.stages import ( AssemblyLayout, AssemblyReaction, @@ -21,6 +29,14 @@ record_plating, record_transformation, ) +from lab.experiments.cloning.systems import ( + AssemblyRecipe, + CloningSystem, + ExternalPreparationRecipe, + FragmentSelection, + PlatingRecipe, + TransformationRecipe, +) from lab.experiments.cloning.types import ( BSAI, Assembly, @@ -32,6 +48,18 @@ from lab.experiments.cloning.workflow import golden_gate __all__ = [ + "AssemblyRecipe", + "CloningSystem", + "ExternalPreparationRecipe", + "FragmentSelection", + "PlatingRecipe", + "TransformationRecipe", + "AssemblyMethod", + "CloningMethods", + "ExternalPreparationMethod", + "PlatingMethod", + "Reagent", + "TransformationMethod", "Assembly", "AssemblyLayout", "AssemblyReaction", diff --git a/src/lab/experiments/cloning/methods.py b/src/lab/experiments/cloning/methods.py new file mode 100644 index 0000000..407b3be --- /dev/null +++ b/src/lab/experiments/cloning/methods.py @@ -0,0 +1,405 @@ +"""Resolved method parameters, supplied independently of provenance graphs.""" + +import json +from dataclasses import dataclass +from decimal import Decimal +from pathlib import Path +from typing import TypeVar + +from lab.artifacts import canonical_json, write_bundle +from lab.operations import Hold +from lab.provenance import Component, Ref +from lab.provenance.types import require_iri + + +def positive(value: Decimal, label: str) -> None: + if not isinstance(value, Decimal) or not value.is_finite() or value <= 0: + raise ValueError(f"{label} must be a positive finite Decimal") + + +@dataclass(frozen=True, kw_only=True) +class Reagent: + component: Ref[Component] + volume_ul: Decimal + + def __post_init__(self) -> None: + if not isinstance(self.component, Ref): + raise TypeError("Reagents need a component reference") + positive(self.volume_ul, "Reagent volume") + + +@dataclass(frozen=True, kw_only=True) +class AssemblyMethod: + """All volumes are microlitres; all thermal holds use seconds and Celsius. + + ``output_volume_ul`` is an explicit planned usable yield, not a measurement. + The enzyme is an ordinary explicit reagent in this method. + """ + + identity: str + enzyme: str + dna_volume_ul: Decimal + reaction_volume_ul: Decimal + output_volume_ul: Decimal + reagents: tuple[Reagent, ...] + diluent: Ref[Component] + profile: tuple[Hold, ...] + cycles: int + mix_volume_ul: Decimal + mix_cycles: int + lid_celsius: Decimal | None = None + + def __post_init__(self) -> None: + require_iri(self.identity) + for name in ("dna_volume_ul", "reaction_volume_ul", "output_volume_ul", "mix_volume_ul"): + positive(getattr(self, name), name) + if ( + self.output_volume_ul > self.reaction_volume_ul + or self.mix_volume_ul > self.reaction_volume_ul + ): + raise ValueError("Usable output and mixing volumes cannot exceed reaction volume") + if not isinstance(self.reagents, tuple) or not all( + isinstance(item, Reagent) for item in self.reagents + ): + raise TypeError("Reagents must be a tuple of Reagent objects") + if not isinstance(self.diluent, Ref): + raise TypeError("Diluent must be a component reference") + if ( + not isinstance(self.profile, tuple) + or not self.profile + or not all(isinstance(hold, Hold) for hold in self.profile) + ): + raise ValueError("Supply an explicit thermal profile") + for hold in self.profile: + if ( + not isinstance(hold.celsius, Decimal) + or not hold.celsius.is_finite() + or hold.celsius < Decimal("-273.15") + ): + raise ValueError("Thermal temperatures must be finite Decimal Celsius values") + positive(hold.seconds, "Hold duration") + for value in (self.cycles, self.mix_cycles): + if type(value) is not int or value < 1: + raise ValueError("Cycle counts must be positive integers") + if self.lid_celsius is not None and ( + not isinstance(self.lid_celsius, Decimal) + or not self.lid_celsius.is_finite() + or self.lid_celsius < 0 + ): + raise ValueError("Lid temperature must be nonnegative finite Decimal Celsius") + + def additions(self, components: tuple[Ref[Component], ...]) -> tuple[Reagent, ...]: + dna = tuple( + Reagent(component=component, volume_ul=self.dna_volume_ul) for component in components + ) + water = self.reaction_volume_ul - sum( + (item.volume_ul for item in (*dna, *self.reagents)), Decimal(0) + ) + if water < 0: + raise ValueError("Method additions exceed the reaction volume") + return ( + *self.reagents, + *dna, + *((Reagent(component=self.diluent, volume_ul=water),) if water else ()), + ) + + +def thermal_profile(profile: tuple[Hold, ...]) -> None: + if not isinstance(profile, tuple) or not profile: + raise ValueError("Supply an explicit nonempty thermal profile") + for hold in profile: + if not isinstance(hold, Hold): + raise TypeError("Thermal profiles contain Hold values") + if ( + not isinstance(hold.celsius, Decimal) + or not hold.celsius.is_finite() + or hold.celsius < Decimal("-273.15") + ): + raise ValueError("Invalid Celsius temperature") + positive(hold.seconds, "Hold duration") + + +@dataclass(frozen=True, kw_only=True) +class TransformationMethod: + """Explicit cell/DNA additions, treatment, medium addition, and recovery. + + The output is an unverified recovery mixture, never a confirmed clone. + No temperatures, durations, or biological yields are inferred. + """ + + identity: str + cell_volume_ul: Decimal + dna_volume_ul: Decimal + recovery: Reagent + profile: tuple[Hold, ...] + recovery_profile: tuple[Hold, ...] + output_volume_ul: Decimal + cell_mix_volume_ul: Decimal + cell_mix_cycles: int + dna_mix_cycles: int + initial_celsius: Decimal | None = None + + def __post_init__(self) -> None: + require_iri(self.identity) + for name in ("cell_volume_ul", "dna_volume_ul", "output_volume_ul", "cell_mix_volume_ul"): + positive(getattr(self, name), name) + if self.cell_mix_volume_ul > self.cell_volume_ul: + raise ValueError("Cell mixing volume cannot exceed the allocated cell volume") + if not isinstance(self.recovery, Reagent): + raise TypeError("Recovery medium must be a Reagent") + thermal_profile(self.profile) + thermal_profile(self.recovery_profile) + for value in (self.cell_mix_cycles, self.dna_mix_cycles): + if type(value) is not int or value < 1: + raise ValueError("Mixing cycles must be positive integers") + if self.initial_celsius is not None: + thermal_profile((Hold(self.initial_celsius, Decimal(1)),)) + + def reaction_volume(self, plasmids: int) -> Decimal: + volume = self.cell_volume_ul + self.dna_volume_ul * plasmids + self.recovery.volume_ul + if self.output_volume_ul > volume: + raise ValueError("Usable transformation output cannot exceed the added liquid") + return volume + + +@dataclass(frozen=True, kw_only=True) +class PlatingMethod: + """One deposited sample after an explicit dilution series. + + Each planned reaction produces one counted spot at the final dilution. + Repeated targets produce independent series; no colony yield is predicted. + The substrate is a supplied plate precondition, like other deck consumables. + """ + + identity: str + substrate: Ref[Component] + diluent: Ref[Component] + transfer_volume_ul: Decimal + dilution_factors: tuple[Decimal, ...] + spot_volume_ul: Decimal + mix_volume_ul: Decimal + mix_cycles: int + spot_height_mm: Decimal + + def __post_init__(self) -> None: + require_iri(self.identity) + if not isinstance(self.substrate, Ref) or not isinstance(self.diluent, Ref): + raise TypeError("Plating substrate and diluent must be component references") + for name in ("transfer_volume_ul", "spot_volume_ul", "mix_volume_ul", "spot_height_mm"): + positive(getattr(self, name), name) + if not isinstance(self.dilution_factors, tuple) or not self.dilution_factors: + raise ValueError("Supply an explicit dilution series") + for factor in self.dilution_factors: + positive(factor, "Dilution factor") + if factor <= 1 or self.mix_volume_ul > self.transfer_volume_ul * factor: + raise ValueError("Dilutions must increase volume and cover the mixing volume") + if self.spot_volume_ul > self.transfer_volume_ul * self.dilution_factors[-1]: + raise ValueError("The last dilution cannot supply the specified spot volume") + if type(self.mix_cycles) is not int or self.mix_cycles < 1: + raise ValueError("Mixing cycles must be positive integers") + if ( + not isinstance(self.spot_height_mm, Decimal) + or not self.spot_height_mm.is_finite() + or self.spot_height_mm < 0 + ): + raise ValueError("Specify a nonnegative spotting height above the well bottom") + + +@dataclass(frozen=True, kw_only=True) +class ExternalPreparationMethod: + """A caller-specified external procedure and prospective material balance. + + Exactly one input and output quantity kind is required. Other consumed + liquids can be named as reagents. The procedure must specify handling, + waste, and losses; a declared yield is not a measurement or a robot action. + """ + + identity: str + procedure: str + instructions: str + source_volume_ul: Decimal = Decimal(0) + source_count: int = 0 + output_volume_ul: Decimal = Decimal(0) + output_count: int = 0 + reagents: tuple[Reagent, ...] = () + + def __post_init__(self) -> None: + require_iri(self.identity) + require_iri(self.procedure) + if not isinstance(self.instructions, str) or not self.instructions.strip(): + raise ValueError("External preparation needs explicit instructions") + for volume, count in ( + (self.source_volume_ul, self.source_count), + (self.output_volume_ul, self.output_count), + ): + if ( + not isinstance(volume, Decimal) + or not volume.is_finite() + or volume < 0 + or type(count) is not int + or count < 0 + or (volume > 0) == (count > 0) + ): + raise ValueError("Specify either a positive volume or a positive count") + if not isinstance(self.reagents, tuple) or not all( + isinstance(r, Reagent) for r in self.reagents + ): + raise TypeError("Preparation reagents must be an immutable tuple") + + +Method = AssemblyMethod | TransformationMethod | PlatingMethod | ExternalPreparationMethod +M = TypeVar("M", bound=Method) + + +@dataclass(frozen=True, kw_only=True) +class CloningMethods: + assemblies: tuple[AssemblyMethod, ...] = () + transformations: tuple[TransformationMethod, ...] = () + platings: tuple[PlatingMethod, ...] = () + preparations: tuple[ExternalPreparationMethod, ...] = () + + def __post_init__(self) -> None: + for name, cls in ( + ("assemblies", AssemblyMethod), + ("transformations", TransformationMethod), + ("platings", PlatingMethod), + ("preparations", ExternalPreparationMethod), + ): + items = getattr(self, name) + if not isinstance(items, tuple) or not all(isinstance(item, cls) for item in items): + raise TypeError(f"{name} must contain an immutable tuple of {cls.__name__}") + object.__setattr__(self, name, tuple(sorted(items, key=lambda item: item.identity))) + if len({method.identity for method in self.all}) != len(self.all): + raise ValueError("Method identities must be unique") + + @property + def all(self) -> tuple[Method, ...]: + return (*self.assemblies, *self.transformations, *self.platings, *self.preparations) + + def get(self, identity: str, kind: type[M]) -> M: + for method in self.all: + if method.identity == identity: + if not isinstance(method, kind): + raise TypeError(f"Method {identity} must be a {kind.__name__}") + return method + raise KeyError(identity) + + def write(self, path: str | Path) -> Path: + path = Path(path) + write_bundle( + path.parent, + { + path.name: canonical_json( + { + "format": "lab.cloning-methods.v1", + "methods": self, + } + ) + }, + ) + return path + + @classmethod + def read(cls, path: str | Path) -> "CloningMethods": + data = json.loads(Path(path).read_text(encoding="utf-8"), parse_float=Decimal) + if data.get("format") != "lab.cloning-methods.v1": + raise ValueError("Expected lab.cloning-methods.v1") + methods = [] + for row in data["methods"]["assemblies"]: + decimal_fields = ( + "dna_volume_ul", + "reaction_volume_ul", + "output_volume_ul", + "mix_volume_ul", + ) + methods.append( + AssemblyMethod( + **{ + **row, + **{name: Decimal(row[name]) for name in decimal_fields}, + "lid_celsius": None + if row["lid_celsius"] is None + else Decimal(row["lid_celsius"]), + "reagents": tuple( + Reagent( + component=Ref(reagent["component"]["identity"]), + volume_ul=Decimal(reagent["volume_ul"]), + ) + for reagent in row["reagents"] + ), + "diluent": Ref(row["diluent"]["identity"]), + "profile": tuple( + Hold(Decimal(hold["celsius"]), Decimal(hold["seconds"])) + for hold in row["profile"] + ), + } + ) + ) + + def reagent(row: dict) -> Reagent: + return Reagent( + component=Ref(row["component"]["identity"]), volume_ul=Decimal(row["volume_ul"]) + ) + + def profile(rows: list) -> tuple[Hold, ...]: + return tuple(Hold(Decimal(row["celsius"]), Decimal(row["seconds"])) for row in rows) + + return cls( + assemblies=tuple(methods), + transformations=tuple( + TransformationMethod( + **{ + **row, + **{ + name: Decimal(row[name]) + for name in ( + "cell_volume_ul", + "dna_volume_ul", + "output_volume_ul", + "cell_mix_volume_ul", + ) + }, + "initial_celsius": None + if row["initial_celsius"] is None + else Decimal(row["initial_celsius"]), + "recovery": reagent(row["recovery"]), + "profile": profile(row["profile"]), + "recovery_profile": profile(row["recovery_profile"]), + } + ) + for row in data["methods"]["transformations"] + ), + platings=tuple( + PlatingMethod( + **{ + **row, + "substrate": Ref(row["substrate"]["identity"]), + "diluent": Ref(row["diluent"]["identity"]), + **{ + name: Decimal(row[name]) + for name in ( + "transfer_volume_ul", + "spot_volume_ul", + "mix_volume_ul", + "spot_height_mm", + ) + }, + "dilution_factors": tuple( + Decimal(factor) for factor in row["dilution_factors"] + ), + } + ) + for row in data["methods"]["platings"] + ), + preparations=tuple( + ExternalPreparationMethod( + **{ + **row, + "source_volume_ul": Decimal(row["source_volume_ul"]), + "output_volume_ul": Decimal(row["output_volume_ul"]), + "reagents": tuple(reagent(r) for r in row["reagents"]), + } + ) + for row in data["methods"]["preparations"] + ), + ) diff --git a/src/lab/experiments/cloning/routes.py b/src/lab/experiments/cloning/routes.py new file mode 100644 index 0000000..faef6b2 --- /dev/null +++ b/src/lab/experiments/cloning/routes.py @@ -0,0 +1,135 @@ +"""Resolve typed recipes into explicit material demands, without choosing a route.""" + +from dataclasses import dataclass +from decimal import Decimal + +from lab.experiments.cloning.methods import ( + AssemblyMethod, + CloningMethods, + ExternalPreparationMethod, + Method, + PlatingMethod, + TransformationMethod, +) +from lab.experiments.cloning.systems import ( + AssemblyRecipe, + ExternalPreparationRecipe, + PlatingRecipe, + Recipe, + TransformationRecipe, +) +from lab.inventory import MaterialForm +from lab.provenance import Component, Ref + + +@dataclass(frozen=True) +class Demand: + design: Ref[Component] + form: MaterialForm + amount: Decimal + role: str + + +@dataclass(frozen=True) +class ResolvedRecipe: + method: Method + inputs: tuple[Demand, ...] + output_amount: Decimal + + +def resolve_recipe(recipe: Recipe, methods: CloningMethods) -> ResolvedRecipe: + if isinstance(recipe, AssemblyRecipe): + assembly = methods.get(recipe.method, AssemblyMethod) + if recipe.enzyme != assembly.enzyme: + raise ValueError(f"Recipe {recipe.identity} and its method use different enzymes") + inputs = tuple( + Demand( + addition.component, + MaterialForm.DNA + if len(assembly.reagents) <= index < len(assembly.reagents) + len(recipe.fragments) + else MaterialForm.REAGENT, + addition.volume_ul, + f"addition_{index + 1}", + ) + for index, addition in enumerate( + assembly.additions(tuple(f.component for f in recipe.fragments)) + ) + ) + return ResolvedRecipe(assembly, inputs, assembly.output_volume_ul) + if isinstance(recipe, TransformationRecipe): + transformation = methods.get(recipe.method, TransformationMethod) + transformation.reaction_volume(len(recipe.plasmids)) + return ResolvedRecipe( + transformation, + ( + Demand( + recipe.chassis, + MaterialForm.COMPETENT_CELLS, + transformation.cell_volume_ul, + "cells", + ), + *( + Demand( + plasmid, MaterialForm.DNA, transformation.dna_volume_ul, f"dna_{index + 1}" + ) + for index, plasmid in enumerate(recipe.plasmids) + ), + Demand( + transformation.recovery.component, + MaterialForm.REAGENT, + transformation.recovery.volume_ul, + "recovery", + ), + ), + transformation.output_volume_ul, + ) + if isinstance(recipe, PlatingRecipe): + plating = methods.get(recipe.method, PlatingMethod) + return ResolvedRecipe( + plating, + ( + Demand(recipe.product, MaterialForm.CULTURE, plating.transfer_volume_ul, "culture"), + *( + Demand( + plating.diluent, + MaterialForm.REAGENT, + plating.transfer_volume_ul * (factor - 1), + f"diluent_{index + 1}", + ) + for index, factor in enumerate(plating.dilution_factors) + ), + ), + Decimal(1), + ) + if isinstance(recipe, ExternalPreparationRecipe): + preparation = methods.get(recipe.method, ExternalPreparationMethod) + if recipe.source_form.counted != bool( + preparation.source_count + ) or recipe.output_form.counted != bool(preparation.output_count): + raise ValueError("Preparation quantity kinds must match the declared material forms") + return ResolvedRecipe( + preparation, + ( + Demand( + recipe.source, + recipe.source_form, + Decimal(preparation.source_count) + if recipe.source_form.counted + else preparation.source_volume_ul, + "source", + ), + *( + Demand( + reagent.component, + MaterialForm.REAGENT, + reagent.volume_ul, + f"reagent_{index + 1}", + ) + for index, reagent in enumerate(preparation.reagents) + ), + ), + Decimal(preparation.output_count) + if recipe.output_form.counted + else preparation.output_volume_ul, + ) + raise TypeError(f"Unsupported recipe {type(recipe).__name__}") diff --git a/src/lab/experiments/cloning/systems.py b/src/lab/experiments/cloning/systems.py new file mode 100644 index 0000000..3125092 --- /dev/null +++ b/src/lab/experiments/cloning/systems.py @@ -0,0 +1,214 @@ +"""Explicit fragment selections and allowed assembly routes.""" + +import json +from dataclasses import dataclass, field +from pathlib import Path + +from Bio.Restriction.Restriction import RestrictionBatch + +from lab.artifacts import canonical_json, write_bundle +from lab.inventory import MaterialForm +from lab.provenance import Component, Ref +from lab.provenance.types import require_iri + + +@dataclass(frozen=True, kw_only=True) +class FragmentSelection: + """Select a digest fragment by zero-based Watson cut positions. + + ``None`` denotes an end of a linear molecule. For a circular molecule, equal + cut positions select a single-cut linearization. Reversal is explicit. + """ + + component: Ref[Component] + left_cut: int | None + right_cut: int | None + reverse_complement: bool = False + + def __post_init__(self) -> None: + if not isinstance(self.component, Ref): + raise TypeError("Fragment component must be a Ref[Component]") + for position in (self.left_cut, self.right_cut): + if position is not None and (type(position) is not int or position < 0): + raise ValueError("Cut positions must be nonnegative integers or None") + if type(self.reverse_complement) is not bool: + raise TypeError("reverse_complement must be a bool") + + +@dataclass(frozen=True, kw_only=True) +class AssemblyRecipe: + kind: str = field(default="assembly", init=False) + identity: str + product: Ref[Component] + enzyme: str + fragments: tuple[FragmentSelection, ...] + method: str + circular: bool = True + + @property + def output_form(self) -> MaterialForm: + return MaterialForm.DNA + + def __post_init__(self) -> None: + require_iri(self.identity) + require_iri(self.method) + if not isinstance(self.product, Ref): + raise TypeError("Recipe product must be a Ref[Component]") + if ( + not isinstance(self.fragments, tuple) + or not self.fragments + or not all(isinstance(item, FragmentSelection) for item in self.fragments) + ): + raise ValueError("Recipes need an ordered tuple of selected fragments") + if len(RestrictionBatch([self.enzyme])) != 1: + raise ValueError("Select one restriction enzyme") + if type(self.circular) is not bool: + raise TypeError("circular must be a bool") + + +@dataclass(frozen=True, kw_only=True) +class TransformationRecipe: + kind: str = field(default="transformation", init=False) + identity: str + product: Ref[Component] + chassis: Ref[Component] + plasmids: tuple[Ref[Component], ...] + method: str + + @property + def output_form(self) -> MaterialForm: + return MaterialForm.CULTURE + + def __post_init__(self) -> None: + require_iri(self.identity) + require_iri(self.method) + if not isinstance(self.product, Ref) or not isinstance(self.chassis, Ref): + raise TypeError("Transformation product and chassis must be component references") + if ( + not isinstance(self.plasmids, tuple) + or not self.plasmids + or not all(isinstance(p, Ref) for p in self.plasmids) + ): + raise ValueError("Supply an ordered, nonempty tuple of plasmid references") + if len(set(self.plasmids)) != len(self.plasmids): + raise ValueError("Transformation plasmids must be distinct") + + +@dataclass(frozen=True, kw_only=True) +class PlatingRecipe: + kind: str = field(default="plating", init=False) + identity: str + product: Ref[Component] + method: str + + @property + def output_form(self) -> MaterialForm: + return MaterialForm.PLATED_SAMPLE + + def __post_init__(self) -> None: + require_iri(self.identity) + require_iri(self.method) + if not isinstance(self.product, Ref): + raise TypeError("Plating design must be a component reference") + + +@dataclass(frozen=True, kw_only=True) +class ExternalPreparationRecipe: + kind: str = field(default="preparation", init=False) + identity: str + product: Ref[Component] + source: Ref[Component] + source_form: MaterialForm + output_form: MaterialForm + method: str + + def __post_init__(self) -> None: + require_iri(self.identity) + require_iri(self.method) + if not isinstance(self.product, Ref) or not isinstance(self.source, Ref): + raise TypeError("Preparation source and product must be component references") + if not isinstance(self.source_form, MaterialForm) or not isinstance( + self.output_form, MaterialForm + ): + raise TypeError("Preparation needs explicit source and output material forms") + + +Recipe = AssemblyRecipe | TransformationRecipe | PlatingRecipe | ExternalPreparationRecipe + + +@dataclass(frozen=True, kw_only=True) +class CloningSystem: + identity: str + recipes: tuple[Recipe, ...] + + def __post_init__(self) -> None: + require_iri(self.identity) + if not isinstance(self.recipes, tuple) or not all( + isinstance( + recipe, + (AssemblyRecipe, TransformationRecipe, PlatingRecipe, ExternalPreparationRecipe), + ) + for recipe in self.recipes + ): + raise TypeError("Recipes must be an immutable tuple of typed cloning recipes") + if len({recipe.identity for recipe in self.recipes}) != len(self.recipes): + raise ValueError("Recipe identities must be unique") + object.__setattr__( + self, "recipes", tuple(sorted(self.recipes, key=lambda item: item.identity)) + ) + + def routes( + self, product: Ref[Component], form: MaterialForm = MaterialForm.DNA + ) -> tuple[Recipe, ...]: + return tuple( + recipe + for recipe in self.recipes + if recipe.product == product and recipe.output_form is form + ) + + def write(self, path: str | Path) -> Path: + path = Path(path) + write_bundle( + path.parent, + { + path.name: canonical_json( + { + "format": "lab.cloning-system.v1", + "system": self, + } + ) + }, + ) + return path + + @classmethod + def read(cls, path: str | Path) -> "CloningSystem": + data = json.loads(Path(path).read_text(encoding="utf-8")) + if data.get("format") != "lab.cloning-system.v1": + raise ValueError("Expected lab.cloning-system.v1") + recipes: list[Recipe] = [] + for row in data["system"]["recipes"]: + kind = row.pop("kind") + row["product"] = Ref(row["product"]["identity"]) + if kind == "assembly": + row["fragments"] = tuple( + FragmentSelection( + **{**fragment, "component": Ref(fragment["component"]["identity"])} + ) + for fragment in row["fragments"] + ) + recipes.append(AssemblyRecipe(**row)) + elif kind == "transformation": + row["chassis"] = Ref(row["chassis"]["identity"]) + row["plasmids"] = tuple(Ref(p["identity"]) for p in row["plasmids"]) + recipes.append(TransformationRecipe(**row)) + elif kind == "plating": + recipes.append(PlatingRecipe(**row)) + elif kind == "preparation": + row["source"] = Ref(row["source"]["identity"]) + row["source_form"] = MaterialForm(row["source_form"]) + row["output_form"] = MaterialForm(row["output_form"]) + recipes.append(ExternalPreparationRecipe(**row)) + else: + raise ValueError(f"Unknown recipe kind: {kind}") + return cls(identity=data["system"]["identity"], recipes=tuple(recipes)) diff --git a/tests/test_cloning_recipes.py b/tests/test_cloning_recipes.py new file mode 100644 index 0000000..26385c7 --- /dev/null +++ b/tests/test_cloning_recipes.py @@ -0,0 +1,41 @@ +from decimal import Decimal + +from lab.experiments import cloning +from lab.experiments.cloning.routes import resolve_recipe +from lab.inventory import MaterialForm +from lab.operations import Hold +from lab.provenance import Ref + + +def test_recipe_and_method_round_trip_preserves_material_demands(tmp_path): + root = "https://example.org/recipes/" + method = cloning.TransformationMethod( + identity=root + "method", + cell_volume_ul=Decimal(2), + dna_volume_ul=Decimal(1), + recovery=cloning.Reagent(component=Ref(root + "medium"), volume_ul=Decimal(5)), + output_volume_ul=Decimal(6), + profile=(Hold(Decimal(25), Decimal(1)),), + recovery_profile=(Hold(Decimal(25), Decimal(2)),), + cell_mix_volume_ul=Decimal(2), + cell_mix_cycles=1, + dna_mix_cycles=1, + ) + recipe = cloning.TransformationRecipe( + identity=root + "recipe", + product=Ref(root + "product"), + chassis=Ref(root + "cells"), + plasmids=(Ref(root + "plasmid"),), + method=method.identity, + ) + methods = cloning.CloningMethods(transformations=(method,)) + system = cloning.CloningSystem(identity=root + "system", recipes=(recipe,)) + assert cloning.CloningMethods.read(methods.write(tmp_path / "methods.json")) == methods + assert cloning.CloningSystem.read(system.write(tmp_path / "system.json")) == system + demands = resolve_recipe(recipe, methods) + assert demands.output_amount == 6 + assert [(demand.form, demand.amount) for demand in demands.inputs] == [ + (MaterialForm.COMPETENT_CELLS, Decimal(2)), + (MaterialForm.DNA, Decimal(1)), + (MaterialForm.REAGENT, Decimal(5)), + ] diff --git a/uv.lock b/uv.lock index 1cdfdd6..b7d7c41 100644 --- a/uv.lock +++ b/uv.lock @@ -55,6 +55,29 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b9/fa/123043af240e49752f1c4bd24da5053b6bd00cad78c2be53c0d1e8b975bc/backports.tarfile-1.2.0-py3-none-any.whl", hash = "sha256:77e284d754527b01fb1e6fa8a1afe577858ebe4e9dad8919e34c862cb399bc34", size = 30181 }, ] +[[package]] +name = "biopython" +version = "1.84" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "numpy" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/9e/7f/eaca4de03f0ee06c9d578d2730fd55858a57cee3620c62d3bc17b5da5447/biopython-1.84.tar.gz", hash = "sha256:60fbe6f996e8a6866a42698c17e552127d99a9aab3259d6249fbaabd0e0cc7b4", size = 25793001 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c5/7b/c1e9f66e23b01958ae0284a437a0e586ce20387fc6ea0382c21230ac59bc/biopython-1.84-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:2d4ed30aebd96b4aadeb1f04adce92795c696f5bd56d1fd45517b89059918dd4", size = 2754460 }, + { url = "https://files.pythonhosted.org/packages/5f/49/c9ffacca2e26259e28215e0ac599db21adf4070359d2aa9d006f3ecf1051/biopython-1.84-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:c792508988fc3ccf18eaae2a826c9cd97f1c27fb55bb87bdce6a101fee9f5a0c", size = 2737878 }, + { url = "https://files.pythonhosted.org/packages/b4/d6/5aae16c1dd91284a40b769926cd69214ddbb986e710f6d44dbe1f6f20c34/biopython-1.84-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:507ac3956f3107e77fee362ecb048dafb5f97cbcf110012d091418430c3227c7", size = 3174779 }, + { url = "https://files.pythonhosted.org/packages/4c/3c/cecf231afa65e7194ac06ba981631a9870515bb7a37a15cad1ab414325c4/biopython-1.84-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:894ee7533cca7f5f9769e2595fbe7b0dba5018f39a2170753d101a13e7585ff4", size = 3192041 }, + { url = "https://files.pythonhosted.org/packages/0f/6c/3e8f01ddea31eeba4cddaa6dbb37b7978ea0164bd35b783fda9f5be59cc4/biopython-1.84-cp311-cp311-win32.whl", hash = "sha256:7f4c746825721ec367c2f2d6a8cda3bc6495a1e084e5b2fbab26e9467706603f", size = 2755506 }, + { url = "https://files.pythonhosted.org/packages/89/38/e45df36e10de29141ba0e5d94bfb942925e62cb6ad182e58b74fa5edcfcc/biopython-1.84-cp311-cp311-win_amd64.whl", hash = "sha256:2cb8e839ab472244b6082635ad1df67c94c05df0bd02a023103ed00ea66c4d20", size = 2792261 }, + { url = "https://files.pythonhosted.org/packages/f7/f6/a61af0d2c8c04e446bce4727e8124797132858f518b6d6543d0e7213abed/biopython-1.84-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:ba58a6d76288333c5f178a426116953fa68204bd0cfc401694087dd4f96d0059", size = 2755863 }, + { url = "https://files.pythonhosted.org/packages/e9/1a/25c7df41987383070987f7b9842f48d3a33b0a78a85c2ca9d93ed810fa2a/biopython-1.84-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:ee3566f6dc3acf20e238540daf896f0af20cff531521bf41fdf5143f73e209ae", size = 2738072 }, + { url = "https://files.pythonhosted.org/packages/a2/b2/c7f2a0a151208c634ac1eaa5d6345899659b1d5a700a84ef2e4f2b0e80a9/biopython-1.84-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:89ef3967f5a88b5bb6344bef75ae83386de53fed3966d5c8c334ad885f8db08a", size = 3186633 }, + { url = "https://files.pythonhosted.org/packages/46/37/7db2bcbb396edba3f767dd89ac23ef5adc35c7a92ef3912c06d1e71469e1/biopython-1.84-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:61765b71f84814a1eeb55ab222f43330aa7ad3e55ab91e8b444706149c67a281", size = 3206061 }, + { url = "https://files.pythonhosted.org/packages/b2/12/6c9d73cbb8c9d19ab4187aaf187f967de6e83738947b7180fdd8bc9211a2/biopython-1.84-cp312-cp312-win32.whl", hash = "sha256:52b6098f47d6b90fc8a5e8579b81ee50047e9108f0976e69c891ae0c4817e42d", size = 2756622 }, + { url = "https://files.pythonhosted.org/packages/d1/53/91d12cc254a804c797afaefec91ede04bc1f7cbd788a04ebbea9e31ee0cf/biopython-1.84-cp312-cp312-win_amd64.whl", hash = "sha256:ecff2fcf5da29b600474c0bfcdbbac0f98b25e22fe60a853d0ee798c00f7396c", size = 2792652 }, +] + [[package]] name = "certifi" version = "2026.7.22" @@ -382,6 +405,7 @@ wheels = [ name = "lab-compiler" source = { editable = "." } dependencies = [ + { name = "biopython" }, { name = "pint", version = "0.25.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" }, { name = "pint", version = "0.26.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" }, { name = "rdflib" }, @@ -408,6 +432,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "biopython", specifier = "==1.84" }, { name = "opentrons", marker = "extra == 'opentrons'", specifier = "==8.8.2" }, { name = "pint", specifier = ">=0.24,<0.27" }, { name = "pylabrobot", marker = "extra == 'star'", specifier = "==0.2.1" },