From 1bd79749e987dec27809bf2c42a34cbcde670409 Mon Sep 17 00:00:00 2001 From: Mike Arpaia Date: Wed, 30 Sep 2026 22:22:35 -0600 Subject: [PATCH] Plan cloning routes against quantities, recipes, and acquisition requirements --- src/lab/experiments/cloning/__init__.py | 30 +- src/lab/experiments/cloning/planning.py | 701 ++++++++++++++++++++++++ tests/planning_fixture.py | 124 +++++ tests/test_planning.py | 310 +++++++++++ tests/test_suppliers.py | 84 ++- 5 files changed, 1234 insertions(+), 15 deletions(-) create mode 100644 src/lab/experiments/cloning/planning.py create mode 100644 tests/planning_fixture.py create mode 100644 tests/test_planning.py diff --git a/src/lab/experiments/cloning/__init__.py b/src/lab/experiments/cloning/__init__.py index fb160a6..fb3c438 100644 --- a/src/lab/experiments/cloning/__init__.py +++ b/src/lab/experiments/cloning/__init__.py @@ -13,6 +13,15 @@ Reagent, TransformationMethod, ) +from lab.experiments.cloning.planning import ( + BuildPlan, + BuildRequest, + BuildTarget, + CountTarget, + EditablePositions, + PlanningPolicy, + plan, +) from lab.experiments.cloning.stages import ( AssemblyLayout, AssemblyReaction, @@ -48,19 +57,24 @@ from lab.experiments.cloning.workflow import golden_gate __all__ = [ + "Assembly", + "AssemblyMethod", "AssemblyRecipe", "CloningSystem", - "ExternalPreparationRecipe", - "FragmentSelection", - "PlatingRecipe", - "TransformationRecipe", - "AssemblyMethod", + "BuildPlan", + "BuildRequest", + "BuildTarget", "CloningMethods", + "CountTarget", "ExternalPreparationMethod", + "ExternalPreparationRecipe", + "EditablePositions", + "FragmentSelection", + "PlanningPolicy", "PlatingMethod", + "PlatingRecipe", "Reagent", - "TransformationMethod", - "Assembly", + "plan", "AssemblyLayout", "AssemblyReaction", "AssemblyRequest", @@ -69,6 +83,8 @@ "PlatingRequest", "PlatingVolumes", "Transformation", + "TransformationMethod", + "TransformationRecipe", "TransformationLayout", "TransformationRequest", "assembly_deck", diff --git a/src/lab/experiments/cloning/planning.py b/src/lab/experiments/cloning/planning.py new file mode 100644 index 0000000..57be116 --- /dev/null +++ b/src/lab/experiments/cloning/planning.py @@ -0,0 +1,701 @@ +"""Pure inventory allocation and dependency-aware assembly route planning.""" + +import hashlib +from dataclasses import dataclass, field, replace +from decimal import Decimal +from enum import StrEnum +from pathlib import Path + +from lab._version import __version__ +from lab.artifacts import canonical_json, digest, write_bundle +from lab.experiments.cloning.domestication import EditProposal, propose_edits +from lab.experiments.cloning.methods import ( + CloningMethods, + ExternalPreparationMethod, + Method, + PlatingMethod, + positive, +) +from lab.experiments.cloning.routes import resolve_recipe +from lab.experiments.cloning.sequences import calculate_assembly +from lab.experiments.cloning.systems import ( + AssemblyRecipe, + CloningSystem, + Recipe, +) +from lab.inventory import Inventory, MaterialForm +from lab.provenance import ( + Activity, + Agent, + AgentKind, + Association, + Component, + Document, + DocumentSnapshot, + EvidenceState, + Implementation, + Plan, + Ref, + Usage, +) +from lab.provenance.types import require_iri +from lab.provenance.vocabulary import LAB +from lab.suppliers.types import AcquisitionRequest, Catalog, Receipt + + +class TaskKind(StrEnum): + ASSEMBLY = "assembly" + TRANSFORMATION = "transformation" + PLATING = "plating" + ACQUISITION = "acquisition" + PREPARATION = "preparation" + + +class RequirementKind(StrEnum): + ACQUISITION = "acquisition" + PREPARATION = "preparation" + DESIGN = "design" + + +@dataclass(frozen=True, kw_only=True) +class BuildTarget: + design: Ref[Component] + volume_ul: Decimal + form: MaterialForm = MaterialForm.DNA + + def __post_init__(self) -> None: + if not isinstance(self.design, Ref) or not isinstance(self.form, MaterialForm): + raise TypeError("Targets need a design reference and a MaterialForm") + positive(self.volume_ul, "Target volume") + if self.form.counted: + raise ValueError("Use CountTarget for counted material forms") + + @property + def amount(self) -> Decimal: + return self.volume_ul + + +@dataclass(frozen=True, kw_only=True) +class CountTarget: + design: Ref[Component] + count: int + form: MaterialForm = MaterialForm.PLATED_SAMPLE + + def __post_init__(self) -> None: + if ( + not isinstance(self.design, Ref) + or not isinstance(self.form, MaterialForm) + or not self.form.counted + ): + raise TypeError("Count targets need a design reference and a counted material form") + if type(self.count) is not int or self.count < 1: + raise ValueError("Target count must be a positive integer") + + @property + def amount(self) -> Decimal: + return Decimal(self.count) + + +@dataclass(frozen=True, kw_only=True) +class BuildRequest: + identity: str + targets: tuple[BuildTarget | CountTarget, ...] + + def __post_init__(self) -> None: + require_iri(self.identity) + if ( + not isinstance(self.targets, tuple) + or not self.targets + or not all(isinstance(target, (BuildTarget, CountTarget)) for target in self.targets) + ): + raise ValueError("A build request needs a nonempty tuple of targets") + if len({(target.design, target.form) for target in self.targets}) != len(self.targets): + raise ValueError("Combine quantities for duplicate targets") + + +@dataclass(frozen=True, kw_only=True) +class EditablePositions: + component: Ref[Component] + positions: tuple[int, ...] + + def __post_init__(self) -> None: + if not isinstance(self.component, Ref): + raise TypeError("Editable component must be a reference") + if ( + not isinstance(self.positions, tuple) + or any(type(position) is not int or position < 0 for position in self.positions) + or len(set(self.positions)) != len(self.positions) + ): + raise ValueError("Editable positions must be a tuple of unique nonnegative integers") + + +@dataclass(frozen=True, kw_only=True) +class PlanningPolicy: + """Reuse available stock, then minimize missing work, depth, and reactions. + + Route combinations are explored across the entire request. Exceeding the + explicit search limit is an error, never an undisclosed greedy fallback. + """ + + max_states: int = 4096 + editable_positions: tuple[EditablePositions, ...] = () + + def __post_init__(self) -> None: + if type(self.max_states) is not int or self.max_states < 1: + raise ValueError("max_states must be positive") + if not isinstance(self.editable_positions, tuple) or not all( + isinstance(item, EditablePositions) for item in self.editable_positions + ): + raise TypeError("Editable positions must be an immutable tuple") + if len({item.component for item in self.editable_positions}) != len( + self.editable_positions + ): + raise ValueError("Specify editable positions once per component") + + +@dataclass(frozen=True, kw_only=True) +class Allocation: + consumer: str + design: Ref[Component] + implementation: Ref[Implementation] + volume_ul: Decimal + stock: str | None + producer: str | None + form: MaterialForm = MaterialForm.DNA + count: int = 0 + role: str = "material" + + @property + def amount(self) -> Decimal: + return Decimal(self.count) if self.form.counted else self.volume_ul + + +@dataclass(frozen=True, kw_only=True) +class Requirement: + kind: RequirementKind + design: Ref[Component] + form: MaterialForm + volume_ul: Decimal + message: str + task: str | None = None + count: int = 0 + + +@dataclass(frozen=True, kw_only=True) +class BuildTask: + identity: str + kind: TaskKind + requested_design: Ref[Component] + design: Ref[Component] + output: Ref[Implementation] + output_volume_ul: Decimal + depends_on: tuple[str, ...] = () + inputs: tuple[Allocation, ...] = () + recipe: Recipe | None = None + method: Method | None = None + output_form: MaterialForm = MaterialForm.DNA + output_count: int = 0 + + +@dataclass(frozen=True, kw_only=True) +class BuildPlan: + request: BuildRequest + document: DocumentSnapshot + inventory: Inventory + system: CloningSystem + methods: CloningMethods + policy: PlanningPolicy + tasks: tuple[BuildTask, ...] + allocations: tuple[Allocation, ...] + products: tuple[Allocation, ...] + requirements: tuple[Requirement, ...] + catalog: Catalog = Catalog() + receipts: tuple[Receipt, ...] = () + acquisitions: tuple[AcquisitionRequest, ...] = () + edit_proposals: tuple[EditProposal, ...] = () + + @property + def ready(self) -> bool: + return not self.requirements + + def require_ready(self) -> None: + if not self.ready: + raise ValueError( + "Build is not ready:\n" + "\n".join(item.message for item in self.requirements) + ) + + @property + def plan_json(self) -> str: + return canonical_json( + { + "format": "lab.build.v1", + "request": self.request, + "provenance_sha256": self.document.digest, + "inventory_sha256": self.inventory.digest, + "system": self.system, + "methods": self.methods, + "policy": self.policy, + "tasks": self.tasks, + "allocations": self.allocations, + "products": self.products, + "requirements": self.requirements, + "catalog_sha256": self.catalog.digest, + "receipts": self.receipts, + "acquisitions": self.acquisitions, + "edit_proposals": self.edit_proposals, + } + ) + + @property + def digest(self) -> str: + return hashlib.sha256(self.plan_json.encode()).hexdigest() + + def write(self, directory: str | Path) -> Path: + return write_bundle( + directory, + { + "build.json": self.plan_json, + "provenance.ttl": self.document.to_turtle(), + "inventory.json": canonical_json( + {"format": "lab.inventory.v1", "inventory": self.inventory} + ), + "catalog.json": canonical_json( + {"format": "lab.catalog.v1", "catalog": self.catalog} + ), + }, + ) + + +@dataclass(frozen=True) +class _Supply: + requested: Ref[Component] + design: Ref[Component] + implementation: Ref[Implementation] + form: MaterialForm + remaining: Decimal + stock: str | None = None + producer: str | None = None + + +@dataclass +class _State: + document: DocumentSnapshot + supplies: dict[str, _Supply] + tasks: tuple[BuildTask, ...] = () + allocations: tuple[Allocation, ...] = () + requirements: tuple[Requirement, ...] = () + proposals: tuple[EditProposal, ...] = () + resolved: dict[str, Ref[Component]] = field(default_factory=dict) + + def copy(self) -> "_State": + return replace(self, supplies=dict(self.supplies), resolved=dict(self.resolved)) + + +def _take( + state: _State, + consumer: str, + design: Ref[Component], + form: MaterialForm, + amount: Decimal, + role: str = "material", +) -> Decimal: + for key, supply in sorted( + state.supplies.items(), key=lambda item: (item[1].stock is None, item[0]) + ): + if supply.requested != design or supply.form != form or not supply.remaining: + continue + taken = min(amount, supply.remaining) + allocation = Allocation( + consumer=consumer, + design=supply.design, + implementation=supply.implementation, + volume_ul=Decimal(0) if form.counted else taken, + count=int(taken) if form.counted else 0, + form=form, + role=role, + stock=supply.stock, + producer=supply.producer, + ) + state.allocations += (allocation,) + state.supplies[key] = replace(supply, remaining=supply.remaining - taken) + amount -= taken + if not amount: + break + return amount + + +def _score(state: _State) -> tuple[object, ...]: + depths: dict[str, int] = {} + for task in state.tasks: + depths[task.identity] = max((depths[parent] for parent in task.depends_on), default=0) + ( + task.recipe is not None + ) + return ( + sum(item.kind is RequirementKind.DESIGN for item in state.requirements), + sum(item.kind is RequirementKind.ACQUISITION for item in state.requirements), + sum(item.kind is RequirementKind.PREPARATION for item in state.requirements), + max(depths.values(), default=0), + sum(task.recipe is not None for task in state.tasks), + tuple( + task.recipe.identity if task.recipe else task.design.identity for task in state.tasks + ), + ) + + +def plan( + request: BuildRequest, + *, + document: DocumentSnapshot | Document, + inventory: Inventory, + system: CloningSystem, + methods: CloningMethods, + policy: PlanningPolicy | None = None, + catalog: Catalog | None = None, + receipts: tuple[Receipt, ...] = (), +) -> BuildPlan: + """Allocate a whole request and choose a deterministic feasible route. + + Planning never edits input sequences, depletes inventory, orders materials, + or operates a device. A plan with outstanding requirements is inspectable + but cannot be built into executable protocols. + """ + policy = policy or PlanningPolicy() + catalog = catalog or Catalog() + snapshot = document.freeze() if isinstance(document, Document) else document + snapshot.validate().raise_for_errors() + inventory.validate(snapshot) + for entry in catalog.entries: + snapshot.resolve(entry.design) + if not isinstance(receipts, tuple) or not all(isinstance(item, Receipt) for item in receipts): + raise TypeError("Receipts must be an immutable tuple") + for receipt in receipts: + material = snapshot.resolve(receipt.implementation) + if ( + material.evidence_state is not EvidenceState.RECORDED + or receipt.design not in material.derived_from + ): + raise ValueError("Record received material in provenance before planning with it") + for target in request.targets: + snapshot.get(target.design.identity, Component) + for recipe in system.recipes: + snapshot.get(recipe.product.identity, Component) + resolved_recipe = resolve_recipe(recipe, methods) + for demand in resolved_recipe.inputs: + snapshot.get(demand.design.identity, Component) + if isinstance(resolved_recipe.method, PlatingMethod): + snapshot.resolve(resolved_recipe.method.substrate) + initial = _State( + snapshot, + { + stock.identity: _Supply( + requested=stock.design, + design=stock.design, + implementation=stock.implementation, + form=stock.form, + remaining=stock.amount, + stock=stock.identity, + ) + for stock in inventory.stocks + }, + ) + explored = 0 + + def satisfy( + state: _State, + consumer: str, + design: Ref[Component], + form: MaterialForm, + amount: Decimal, + stack: tuple[tuple[str, MaterialForm], ...], + role: str = "material", + ) -> list[_State]: + nonlocal explored + explored += 1 + if explored > policy.max_states: + raise ValueError( + f"Planning exceeded max_states={policy.max_states}; " + "narrow the route set or raise the limit" + ) + state = state.copy() + missing = _take(state, consumer, design, form, amount, role) + if not missing: + return [state] + if (design.identity, form) in stack: + state.requirements += ( + Requirement( + kind=RequirementKind.DESIGN, + design=design, + form=form, + volume_ul=Decimal(0) if form.counted else missing, + count=int(missing) if form.counted else 0, + message=f"Cyclic build dependency for {design.identity}", + ), + ) + return [state] + routes = system.routes(design, form) + if not routes: + candidates = [ + stock + for stock in inventory.stocks + if stock.design == design and stock.form != form and stock.amount > 0 + ] + received = tuple(receipt for receipt in receipts if receipt.design == design) + kind = TaskKind.PREPARATION if candidates or received else TaskKind.ACQUISITION + task_id = request.identity + f"/task_{len(state.tasks) + 1}" + implementation = Implementation( + identity=task_id + "/output", + derived_from=(design,), + evidence_state=EvidenceState.PLANNED, + ) + doc = Document.from_snapshot(state.document) + doc.add(implementation) + state.document = doc.freeze() + state.tasks += ( + BuildTask( + identity=task_id, + kind=kind, + requested_design=design, + design=design, + output=implementation.ref, + output_volume_ul=Decimal(0) if form.counted else missing, + output_count=int(missing) if form.counted else 0, + output_form=form, + ), + ) + reason = ( + ( + f"Prepare {design.identity} as {form.value}; available forms: " + + ", ".join( + sorted( + {stock.form.value for stock in candidates} + | {receipt.form.value for receipt in received} + ) + ) + ) + if candidates or received + else ( + f"Acquire {missing} {'unit(s)' if form.counted else 'µL'} " + f"of {design.identity} as {form.value}" + ) + ) + state.requirements += ( + Requirement( + kind=RequirementKind(kind.value), + design=design, + form=form, + volume_ul=Decimal(0) if form.counted else missing, + count=int(missing) if form.counted else 0, + message=reason, + task=task_id, + ), + ) + state.supplies[implementation.identity] = _Supply( + design, design, implementation.ref, form, missing, producer=task_id + ) + _take(state, consumer, design, form, missing, role) + return [state] + outcomes: list[_State] = [] + for recipe in routes: + branch = state.copy() + resolved_recipe = resolve_recipe(recipe, methods) + method = resolved_recipe.method + task_id = ( + request.identity + + "/" + + recipe.kind + + "_" + + str(len(branch.tasks) + 1) + + "_" + + digest((recipe.identity, consumer))[:12] + ) + states = [branch] + for demand in resolved_recipe.inputs: + states = [ + next_state + for current in states + for next_state in satisfy( + current, + task_id, + demand.design, + demand.form, + demand.amount, + (*stack, (design.identity, form)), + demand.role, + ) + ] + for current in states: + try: + calculated = ( + calculate_assembly( + recipe, document=current.document, resolved=current.resolved + ) + if isinstance(recipe, AssemblyRecipe) + else None + ) + except (ValueError, KeyError) as error: + current.requirements += ( + Requirement( + kind=RequirementKind.DESIGN, + design=design, + form=form, + volume_ul=Decimal(0) if form.counted else missing, + count=int(missing) if form.counted else 0, + message=str(error), + ), + ) + for editable in policy.editable_positions: + if isinstance(recipe, AssemblyRecipe) and editable.component in tuple( + fragment.component for fragment in recipe.fragments + ): + current.proposals += propose_edits( + editable.component, + document=current.document, + enzyme=recipe.enzyme, + editable_positions=editable.positions, + ) + outcomes.append(current) + continue + inputs = tuple(item for item in current.allocations if item.consumer == task_id) + parents = tuple( + sorted({item.producer for item in inputs if item.producer is not None}) + ) + product = calculated.product.ref if calculated else design + planner = Agent( + identity=request.identity + "/planner", + kind=AgentKind.SOFTWARE, + name="Lab planner", + software_version=__version__, + ) + method_plan = Plan( + identity=method.identity, + protocol=method.procedure + if isinstance(method, ExternalPreparationMethod) + else None, + description=method.instructions + if isinstance(method, ExternalPreparationMethod) + else None, + ) + activity = Activity( + identity=task_id, + types=(LAB + recipe.kind,), + evidence_state=EvidenceState.PLANNED, + usage=tuple( + Usage(entity=ref) + for ref in dict.fromkeys(item.implementation for item in inputs) + ) + + ( + (Usage(entity=method.substrate, roles=(LAB + "substrate",)),) + if isinstance(method, PlatingMethod) + else () + ), + informed_by=tuple( + Ref(parent) + for parent in parents + if any( + task.identity == parent and task.recipe is not None + for task in current.tasks + ) + ), + association=( + Association( + agent=planner.ref, + plan=method_plan.ref, + roles=(LAB + "planner",), + ), + ), + ) + implementation = Implementation( + identity=task_id + "/output", + derived_from=(product,), + generated_by=(activity.ref,), + evidence_state=EvidenceState.PLANNED, + ) + doc = Document.from_snapshot(current.document) + doc.add( + *(calculated.objects if calculated else ()), + planner, + method_plan, + activity, + implementation, + ) + current.document = doc.freeze() + current.tasks += ( + BuildTask( + identity=task_id, + kind=TaskKind(recipe.kind), + requested_design=design, + design=product, + output=implementation.ref, + output_volume_ul=Decimal(0) + if form.counted + else resolved_recipe.output_amount, + output_count=int(resolved_recipe.output_amount) if form.counted else 0, + output_form=form, + depends_on=parents, + inputs=inputs, + recipe=recipe, + method=method, + ), + ) + if calculated is not None: + current.resolved[design.identity] = product + current.supplies[implementation.identity] = _Supply( + design, + product, + implementation.ref, + form, + resolved_recipe.output_amount, + producer=task_id, + ) + remainder = _take(current, consumer, design, form, missing, role) + if remainder: + outcomes.extend( + satisfy(current, consumer, design, form, remainder, stack, role) + ) + else: + outcomes.append(current) + return outcomes + + states = [initial] + for index, target in enumerate(request.targets): + consumer = request.identity + f"/target_{index + 1}" + states = [ + next_state + for state in states + for next_state in satisfy( + state, consumer, target.design, target.form, target.amount, () + ) + ] + winner = min(states, key=_score) + products = tuple( + item + for item in winner.allocations + if item.consumer.startswith(request.identity + "/target_") + ) + return BuildPlan( + request=request, + document=winner.document, + inventory=inventory, + system=system, + methods=methods, + policy=policy, + tasks=winner.tasks, + allocations=winner.allocations, + products=products, + requirements=winner.requirements, + catalog=catalog, + receipts=receipts, + acquisitions=tuple( + AcquisitionRequest( + identity=request.identity + f"/acquisition_{index + 1}", + design=item.design, + required_form=item.form, + volume_ul=item.volume_ul, + count=item.count, + candidates=catalog.candidates(item.design), + ) + for index, item in enumerate(winner.requirements) + if item.kind is RequirementKind.ACQUISITION + ), + edit_proposals=tuple(dict.fromkeys(winner.proposals)), + ) diff --git a/tests/planning_fixture.py b/tests/planning_fixture.py new file mode 100644 index 0000000..b129261 --- /dev/null +++ b/tests/planning_fixture.py @@ -0,0 +1,124 @@ +from dataclasses import replace +from decimal import Decimal + +from lab import uL +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 + +NS = "https://example.org/planning" + + +def planning_case(*, volume="3", stock_volume="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 multilevel_case(): + request, inputs = planning_case() + doc = Document.from_snapshot(inputs["document"]) + intermediate = doc.get(NS + "/target", Component) + finals = tuple(replace(intermediate, identity=NS + f"/final_{i}") for i in range(2)) + doc.add(*finals) + first = inputs["system"].recipes[0] + recipes = tuple( + replace( + first, + identity=NS + f"/second_recipe_{index}", + product=final.ref, + fragments=( + first.fragments[0], + FragmentSelection(component=intermediate.ref, left_cut=14, right_cut=0), + ), + ) + for index, final in enumerate(finals) + ) + return ( + replace( + request, + targets=tuple(BuildTarget(design=final.ref, volume_ul=Decimal(3)) for final in finals), + ), + { + **inputs, + "document": doc.freeze(), + "system": replace(inputs["system"], recipes=(first, *recipes)), + }, + ) diff --git a/tests/test_planning.py b/tests/test_planning.py new file mode 100644 index 0000000..4033aee --- /dev/null +++ b/tests/test_planning.py @@ -0,0 +1,310 @@ +import json +from dataclasses import replace +from decimal import Decimal + +import pytest + +from lab import uL +from lab.experiments.cloning.domestication import propose_edits +from lab.experiments.cloning.methods import CloningMethods +from lab.experiments.cloning.planning import ( + BuildTarget, + PlanningPolicy, + RequirementKind, + TaskKind, + plan, +) +from lab.experiments.cloning.sequences import calculate_assembly, digest_fragments +from lab.experiments.cloning.systems import CloningSystem, FragmentSelection +from lab.inventory import Inventory, MaterialForm, Stock +from lab.provenance import Component, Document, EvidenceState, Implementation, Sequence +from tests.planning_fixture import NS, planning_case + + +def test_explicit_fragments_calculate_product_and_computational_provenance(): + _, inputs = planning_case() + recipe = inputs["system"].recipes[0] + fragments = digest_fragments( + recipe.fragments[1].component, document=inputs["document"], enzyme="EcoRI" + ) + assert [ + (f.selection.left_cut, f.selection.right_cut, f.watson, f.crick, f.overhang) + for f in fragments + ] == [ + (None, 3, "GGG", "AATTCCC", 0), + (3, 12, "AATTCCCCG", "AATTCGGGG", -4), + (12, None, "AATTCGG", "CCG", -4), + ] + calculated = calculate_assembly(recipe, document=inputs["document"]) + assert calculated.sequence.elements == "AATTCTTTTAAAAGAATTCCCCG" + assert calculated.activity.evidence_state is EvidenceState.RECORDED + doc = Document.from_snapshot(inputs["document"]) + doc.add(*calculated.objects) + assert not doc.to_sbol3().validate().errors + + +def test_planner_allocates_specific_implementations_and_emits_planned_outputs(tmp_path): + request, inputs = planning_case() + build = plan(request, **inputs) + build.require_ready() + assert len(build.tasks) == 1 + task = build.tasks[0] + assert task.kind is TaskKind.ASSEMBLY + assert len(task.inputs) == 4 + assert sum(item.volume_ul for item in task.inputs) == 5 + assert all(item.stock is not None for item in task.inputs) + output = build.document.resolve(task.output) + assert output.built is None and output.evidence_state is EvidenceState.PLANNED + assert sum(item.volume_ul for item in build.products) == 3 + assert inputs["inventory"].stocks[0].quantity_ul == 10 + assert plan(request, **inputs).digest == build.digest + path = build.write(tmp_path) + assert ( + json.loads((path / "build.json").read_text())["provenance_sha256"] == build.document.digest + ) + assert Document.read(path / "provenance.ttl").freeze().digest == build.document.digest + + +def test_inventory_is_reused_before_assembling(): + request, inputs = planning_case() + doc = Document.from_snapshot(inputs["document"]) + implementation = Implementation( + identity=NS + "/existing", + derived_from=(request.targets[0].design,), + evidence_state=EvidenceState.RECORDED, + ) + doc.add(implementation) + inventory = Inventory( + identity=NS + "/available", + stocks=( + *inputs["inventory"].stocks, + Stock( + identity=NS + "/available_stock", + implementation=implementation.ref, + design=request.targets[0].design, + form=MaterialForm.DNA, + quantity=20 * uL, + ), + ), + ) + build = plan(request, **{**inputs, "document": doc.freeze(), "inventory": inventory}) + assert build.ready and build.tasks == () + assert build.products[0].implementation == implementation.ref + + +def test_multiple_batches_do_not_double_count_inventory_or_reuse_output_identity(): + request, inputs = planning_case(volume="11", stock_volume="3") + build = plan(request, **inputs) + assert sum(task.kind is TaskKind.ASSEMBLY for task in build.tasks) == 3 + assert len({task.output for task in build.tasks}) == len(build.tasks) + # Three batches require six microlitres of water; inventory contains three. + water = [item for item in build.requirements if item.design.identity == NS + "/water"] + assert sum(item.volume_ul for item in water) == 3 + for stock in inputs["inventory"].stocks: + assert ( + sum(item.volume_ul for item in build.allocations if item.stock == stock.identity) + <= stock.quantity_ul + ) + with pytest.raises(ValueError, match="not ready"): + build.require_ready() + + +def test_wrong_material_form_requires_preparation(): + request, inputs = planning_case() + original = inputs["inventory"].stocks + stocks = tuple( + Stock( + identity=stock.identity, + implementation=stock.implementation, + design=stock.design, + form=MaterialForm.CULTURE if stock.design.identity == NS + "/vector" else stock.form, + quantity=stock.quantity_ul * uL, + ) + for stock in original + ) + build = plan( + request, **{**inputs, "inventory": Inventory(identity=NS + "/other", stocks=stocks)} + ) + assert any(item.kind is RequirementKind.PREPARATION for item in build.requirements) + + +def test_invalid_fragment_selection_does_not_silently_choose_another_fragment(): + request, inputs = planning_case() + recipe = inputs["system"].recipes[0] + invalid = replace( + recipe, fragments=(replace(recipe.fragments[0], left_cut=4), *recipe.fragments[1:]) + ) + build = plan(request, **{**inputs, "system": replace(inputs["system"], recipes=(invalid,))}) + assert not build.ready + assert any("does not select exactly one" in item.message for item in build.requirements) + + +def test_route_tie_breaking_is_stable_and_search_limits_are_explicit(): + request, inputs = planning_case() + recipe = inputs["system"].recipes[0] + alternate = replace(recipe, identity=NS + "/z_recipe") + system = replace(inputs["system"], recipes=(alternate, recipe)) + build = plan(request, **{**inputs, "system": system}) + assert build.tasks[0].recipe.identity == recipe.identity + with pytest.raises(ValueError, match="max_states"): + plan(request, **inputs, policy=PlanningPolicy(max_states=1)) + + +def test_edits_are_proposals_until_explicitly_applied_to_a_new_design(): + _, inputs = planning_case() + document = inputs["document"] + component = document.get(NS + "/insert", Component) + proposals = propose_edits( + component.ref, document=document, enzyme="EcoRI", editable_positions=(3,) + ) + assert proposals + original = document.resolve(component.sequences[0]) + edited = proposals[0].apply(document, identity=NS + "/selected_edit") + assert document.resolve(component.sequences[0]) == original + assert edited.get(NS + "/selected_edit/sequence", Sequence).elements != original.elements + assert edited.get(NS + "/selected_edit", Component).derived_from == (component.ref,) + assert len(edited.objects) == len(document.objects) + 3 + + +def test_inventory_round_trip_and_recorded_evidence_requirement(tmp_path): + _, inputs = planning_case() + inventory = inputs["inventory"] + path = inventory.write(tmp_path / "inventory.json") + assert Inventory.read(path, document=inputs["document"]) == inventory + implementation = Implementation(identity=NS + "/planned", evidence_state=EvidenceState.PLANNED) + doc = Document.from_snapshot(inputs["document"]) + doc.add(implementation) + stock = Stock( + identity=NS + "/stock", + implementation=implementation.ref, + design=inventory.stocks[0].design, + form=MaterialForm.DNA, + quantity=1 * uL, + ) + with pytest.raises(ValueError, match="recorded"): + Inventory(identity=NS + "/inventory", stocks=(stock,)).validate(doc.freeze()) + + +def test_stock_quantities_are_normalized_and_detached_from_mutable_pint_values(): + _, inputs = planning_case() + source = inputs["inventory"].stocks[0] + quantity = 10 * uL + stock = Stock( + identity=source.identity, + implementation=source.implementation, + design=source.design, + form=source.form, + quantity=quantity, + ) + quantity *= 2 + assert stock.quantity_ul == Decimal(10) + + +def test_multilevel_assembly_consumes_a_planned_intermediate(): + request, inputs = planning_case() + doc = Document.from_snapshot(inputs["document"]) + intermediate = doc.get(NS + "/target", Component) + final = replace(intermediate, identity=NS + "/final") + doc.add(final) + first = inputs["system"].recipes[0] + second = replace( + first, + identity=NS + "/second_recipe", + product=final.ref, + fragments=( + first.fragments[0], + FragmentSelection(component=intermediate.ref, left_cut=14, right_cut=0), + ), + ) + request = replace(request, targets=(BuildTarget(design=final.ref, volume_ul=Decimal(3)),)) + build = plan( + request, + **{ + **inputs, + "document": doc.freeze(), + "system": replace(inputs["system"], recipes=(first, second)), + }, + ) + build.require_ready() + assert len(build.tasks) == 2 + assert build.tasks[1].depends_on == (build.tasks[0].identity,) + assert any(item.implementation == build.tasks[0].output for item in build.tasks[1].inputs) + assert sum(item.volume_ul for item in build.products) == 3 + + +def test_route_search_preserves_scarce_inventory_for_another_target(): + request, inputs = planning_case() + doc = Document.from_snapshot(inputs["document"]) + vector = doc.get(NS + "/vector", Component) + other_vector = replace(vector, identity=NS + "/other_vector") + other_target = replace(doc.get(NS + "/target", Component), identity=NS + "/other_target") + stock_impl = Implementation( + identity=NS + "/other_vector_impl", + derived_from=(other_vector.ref,), + evidence_state=EvidenceState.RECORDED, + ) + doc.add(other_vector, other_target, stock_impl) + stocks = tuple( + Stock( + identity=stock.identity, + implementation=stock.implementation, + design=stock.design, + form=stock.form, + quantity=(1 if stock.design == vector.ref else 10) * uL, + ) + for stock in inputs["inventory"].stocks + ) + alternate_stock = Stock( + identity=NS + "/other_vector_stock", + implementation=stock_impl.ref, + design=other_vector.ref, + form=MaterialForm.DNA, + quantity=1 * uL, + ) + first = inputs["system"].recipes[0] + alternative = replace( + first, + identity=NS + "/z_alternative", + fragments=(replace(first.fragments[0], component=other_vector.ref), first.fragments[1]), + ) + second = replace(first, identity=NS + "/second", product=other_target.ref) + build = plan( + replace( + request, + targets=(*request.targets, BuildTarget(design=other_target.ref, volume_ul=Decimal(3))), + ), + **{ + **inputs, + "document": doc.freeze(), + "inventory": Inventory(identity=NS + "/scarce", stocks=(*stocks, alternate_stock)), + "system": replace(inputs["system"], recipes=(first, alternative, second)), + }, + ) + build.require_ready() + assert build.tasks[0].recipe.identity == alternative.identity + assert build.tasks[1].recipe.identity == second.identity + + +def test_typed_recipe_and_method_files_round_trip(tmp_path): + _, inputs = planning_case() + system = inputs["system"] + methods = inputs["methods"] + assert CloningSystem.read(system.write(tmp_path / "system.json")) == system + assert CloningMethods.read(methods.write(tmp_path / "methods.json")) == methods + + +def test_authoring_can_add_frozen_objects_idempotently_and_explicitly_revise(): + _, inputs = planning_case() + doc = Document.from_snapshot(inputs["document"]) + authored = Component(identity=NS + "/new", types=("https://example.org/type",)) + doc.add(authored) + snapshot = doc.freeze() + doc = Document.from_snapshot(snapshot) + doc.add(authored) + assert doc.freeze().digest == snapshot.digest + revised = replace(authored, name="A deliberate revision") + with pytest.raises(ValueError, match="Conflicting"): + doc.add(revised) + doc.replace(revised) + assert doc.freeze().get(authored.identity, Component).name == revised.name + assert snapshot.get(authored.identity, Component).name is None diff --git a/tests/test_suppliers.py b/tests/test_suppliers.py index 31d21b0..7f61706 100644 --- a/tests/test_suppliers.py +++ b/tests/test_suppliers.py @@ -6,15 +6,20 @@ import pytest -from lab.inventory import MaterialForm -from lab.provenance import Ref +from lab import uL +from lab.experiments.cloning.planning import RequirementKind, plan +from lab.inventory import Inventory, MaterialForm +from lab.provenance import Agent, Document, EvidenceState from lab.suppliers import ( AddgeneClient, Catalog, CatalogEntry, + OrderReference, + Receipt, SequenceSource, parse_plasmid, ) +from tests.planning_fixture import NS, planning_case TIME = datetime(2026, 9, 27, tzinfo=UTC) @@ -59,7 +64,8 @@ def catalog(design): def test_catalog_preserves_source_completeness_and_original_metadata(tmp_path): - snapshot = catalog(Ref("https://example.org/catalog/design")) + request, _ = planning_case() + snapshot = catalog(request.targets[0].design) item = snapshot.entries[0].item assert item.sequences[0].source is SequenceSource.DEPOSITOR assert item.sequences[0].complete @@ -91,13 +97,75 @@ def test_catalog_http_request_matches_documented_auth_and_endpoint(): client.plasmid(43) +def test_catalog_candidate_does_not_count_as_inventory(): + request, inputs = planning_case() + missing = next( + stock for stock in inputs["inventory"].stocks if stock.design.identity == NS + "/vector" + ) + inventory = replace( + inputs["inventory"], + stocks=tuple(stock for stock in inputs["inventory"].stocks if stock != missing), + ) + build = plan(request, **{**inputs, "inventory": inventory}, catalog=catalog(missing.design)) + assert not build.ready + assert build.acquisitions[0].candidates[0].form is MaterialForm.BACTERIAL_STAB + assert build.acquisitions[0].required_form is MaterialForm.DNA + assert build.acquisitions[0].volume_ul == 1 + + +def test_receipt_records_physical_arrival_without_claiming_sequence_verification(): + request, inputs = planning_case() + design = inputs["system"].recipes[0].fragments[0].component + person = Agent(identity=NS + "/person") + doc = Document.from_snapshot(inputs["document"]) + doc.add(person) + item = catalog(design).entries[0].item + order = OrderReference( + identity=NS + "/order", + acquisition=NS + "/acquisition", + item=item.identity, + reference="external-order-42", + placed_at=TIME, + ) + receipt = Receipt( + identity=NS + "/receipt", + order=order, + design=design, + form=MaterialForm.BACTERIAL_STAB, + received_at=TIME, + received_by=person.ref, + packages=1, + ) + recorded = receipt.record(doc.freeze()) + assert recorded.resolve(receipt.implementation).built is None + assert recorded.resolve(receipt.implementation).evidence_state is EvidenceState.RECORDED + counted = receipt.counted_stock(identity=NS + "/counted_stock", count=1) + Inventory(identity=NS + "/counted_inventory", stocks=(counted,)).validate(recorded) + assert counted.count == 1 and counted.implementation == receipt.implementation + with pytest.raises(ValueError, match="preparation"): + receipt.stock(identity=NS + "/stock", quantity=1 * uL) + inventory = replace( + inputs["inventory"], + stocks=tuple(stock for stock in inputs["inventory"].stocks if stock.design != design), + ) + build = plan( + request, **{**inputs, "document": recorded, "inventory": inventory}, receipts=(receipt,) + ) + assert any(item.kind is RequirementKind.PREPARATION for item in build.requirements) + assert not build.acquisitions + + liquid = replace(receipt, identity=NS + "/liquid_receipt", form=MaterialForm.DNA) + stock = liquid.stock(identity=NS + "/received_stock", quantity=10 * uL) + Inventory(identity=NS + "/received_inventory", stocks=(stock,)).validate( + liquid.record(doc.freeze()) + ) + + def test_addgene_is_preferred_without_automatic_sequence_selection(): - entry = catalog(Ref("https://example.org/catalog/design")).entries[0] + request, _ = planning_case() + entry = catalog(request.targets[0].design).entries[0] alternative = replace( - entry, - item=replace( - entry.item, supplier="Example", identity="https://example.org/catalog/supplier" - ), + entry, item=replace(entry.item, supplier="Example", identity=NS + "/supplier") ) snapshot = Catalog(entries=(alternative, entry)) assert snapshot.candidates(entry.design) == (entry, alternative)