diff --git a/src/buildcompiler/adapters/pudu/__init__.py b/src/buildcompiler/adapters/pudu/__init__.py index 5e19692..a31fbde 100644 --- a/src/buildcompiler/adapters/pudu/__init__.py +++ b/src/buildcompiler/adapters/pudu/__init__.py @@ -1 +1,5 @@ -"""Package scaffolding for clean architecture.""" +"""PUDU adapter exports.""" + +from .assembly_json import assembly_route_to_pudu_json, assembly_routes_to_pudu_json + +__all__ = ["assembly_route_to_pudu_json", "assembly_routes_to_pudu_json"] diff --git a/src/buildcompiler/adapters/pudu/assembly_json.py b/src/buildcompiler/adapters/pudu/assembly_json.py new file mode 100644 index 0000000..9d5fbe8 --- /dev/null +++ b/src/buildcompiler/adapters/pudu/assembly_json.py @@ -0,0 +1,64 @@ +"""In-memory adapter for compiler-level PUDU assembly JSON payloads.""" + +from collections.abc import Sequence + +from buildcompiler.domain import IndexedBackbone, IndexedPlasmid, IndexedReagent + + +def _stable_identifier(identity: str, display_id: str | None) -> str: + return identity or display_id or "" + + +def assembly_route_to_pudu_json( + *, + product_identity: str, + part_plasmids: Sequence[IndexedPlasmid], + backbone: IndexedBackbone, + restriction_enzyme: IndexedReagent, +) -> dict[str, object]: + """Adapt a selected lvl1 route into legacy-compatible assembly JSON keys.""" + + parts_list = [ + _stable_identifier(identity=part.identity, display_id=part.display_id) + for part in part_plasmids + ] + return { + "Product": product_identity, + "Backbone": _stable_identifier( + identity=backbone.identity, display_id=backbone.display_id + ), + "PartsList": parts_list, + "Restriction Enzyme": ( + restriction_enzyme.name + or _stable_identifier( + identity=restriction_enzyme.identity, + display_id=restriction_enzyme.display_id, + ) + ), + } + + +def assembly_routes_to_pudu_json( + *, + product_identities: Sequence[str], + part_plasmid_routes: Sequence[Sequence[IndexedPlasmid]], + backbones: Sequence[IndexedBackbone], + restriction_enzymes: Sequence[IndexedReagent], +) -> list[dict[str, object]]: + """Batch helper for deterministic in-memory assembly JSON payloads.""" + + return [ + assembly_route_to_pudu_json( + product_identity=product_identity, + part_plasmids=part_plasmids, + backbone=backbone, + restriction_enzyme=restriction_enzyme, + ) + for product_identity, part_plasmids, backbone, restriction_enzyme in zip( + product_identities, + part_plasmid_routes, + backbones, + restriction_enzymes, + strict=True, + ) + ] diff --git a/src/buildcompiler/stages/__init__.py b/src/buildcompiler/stages/__init__.py index 5e19692..b024f42 100644 --- a/src/buildcompiler/stages/__init__.py +++ b/src/buildcompiler/stages/__init__.py @@ -1 +1,5 @@ -"""Package scaffolding for clean architecture.""" +"""Stage exports.""" + +from .assembly_lvl1 import AssemblyLvl1Stage + +__all__ = ["AssemblyLvl1Stage"] diff --git a/src/buildcompiler/stages/assembly_lvl1.py b/src/buildcompiler/stages/assembly_lvl1.py new file mode 100644 index 0000000..3332eb7 --- /dev/null +++ b/src/buildcompiler/stages/assembly_lvl1.py @@ -0,0 +1,239 @@ +"""Thin lvl1 assembly stage orchestration.""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +import sbol2 + +from buildcompiler.adapters.pudu import assembly_route_to_pudu_json +from buildcompiler.api import BuildOptions +from buildcompiler.domain import ( + BuildRequest, + BuildStage, + BuildWarning, + MissingBuildInput, + StageResult, + StageStatus, +) +from buildcompiler.inventory import CompatibilitySelector, Inventory +from buildcompiler.sbol import AssemblyJob, AssemblyService + + +class AssemblyLvl1Stage: + def __init__( + self, + *, + inventory: Inventory, + selector: CompatibilitySelector | None = None, + assembly_service: AssemblyService | None = None, + options: BuildOptions | None = None, + ) -> None: + self.inventory = inventory + self.options = options or BuildOptions() + self.selector = selector or CompatibilitySelector( + inventory, options=self.options + ) + self.assembly_service = assembly_service or AssemblyService() + + def run( + self, + request: BuildRequest, + *, + source_document: sbol2.Document, + target_document: sbol2.Document, + ) -> StageResult: + constraints = request.constraints or {} + warnings = self._extract_warnings(request=request, constraints=constraints) + part_identities = self._extract_part_identities(constraints) + if not part_identities: + return StageResult( + id=f"{request.id}:{BuildStage.ASSEMBLY_LVL1.value}", + stage=BuildStage.ASSEMBLY_LVL1, + status=StageStatus.FAILED, + request_ids=[request.id], + warnings=warnings, + logs=[ + "Missing ordered_part_identities/part_identities constraint for lvl1 assembly." + ], + ) + + route_selection = self.selector.select_lvl1_route( + request_id=request.id, + part_identities=part_identities, + constraints=constraints, + ) + route = route_selection.selected + if route is None: + return StageResult( + id=f"{request.id}:{BuildStage.ASSEMBLY_LVL1.value}", + stage=BuildStage.ASSEMBLY_LVL1, + status=StageStatus.BLOCKED, + request_ids=[request.id], + warnings=warnings, + logs=["No lvl1 route selected by CompatibilitySelector."], + ) + + missing_inputs: list[MissingBuildInput] = [] + for missing_identity in route.missing_part_identities: + missing_inputs.append( + MissingBuildInput( + source_stage=BuildStage.ASSEMBLY_LVL1, + source_design_identity=request.source_identity, + missing_identity=missing_identity, + missing_display_id=missing_identity.rsplit("/", 1)[-1], + missing_kind=self._infer_missing_kind(missing_identity), + required_stage=BuildStage.DOMESTICATION, + reason="No compatible lvl1 part plasmid found in inventory.", + ) + ) + + if route.backbone is None: + missing_inputs.append( + MissingBuildInput( + source_stage=BuildStage.ASSEMBLY_LVL1, + source_design_identity=request.source_identity, + missing_identity="backbone", + missing_display_id=None, + missing_kind="backbone", + required_stage="fatal", + reason="No compatible lvl1 backbone found in inventory.", + ) + ) + + restriction_enzyme_name = self.options.reagents.default_restriction_enzyme + restriction_enzyme = self.inventory.find_restriction_enzyme( + restriction_enzyme_name + ) + if restriction_enzyme is None: + missing_inputs.append( + MissingBuildInput( + source_stage=BuildStage.ASSEMBLY_LVL1, + source_design_identity=request.source_identity, + missing_identity=restriction_enzyme_name, + missing_display_id=restriction_enzyme_name, + missing_kind="restriction_enzyme", + required_stage="fatal", + reason="Required restriction enzyme is missing from inventory.", + ) + ) + + ligase_name = self.options.reagents.default_ligase + ligase = self.inventory.find_ligase(ligase_name) + if ligase is None: + missing_inputs.append( + MissingBuildInput( + source_stage=BuildStage.ASSEMBLY_LVL1, + source_design_identity=request.source_identity, + missing_identity=ligase_name, + missing_display_id=ligase_name, + missing_kind="ligase", + required_stage="fatal", + reason="Required ligase is missing from inventory.", + ) + ) + + if missing_inputs: + return StageResult( + id=f"{request.id}:{BuildStage.ASSEMBLY_LVL1.value}", + stage=BuildStage.ASSEMBLY_LVL1, + status=StageStatus.BLOCKED, + request_ids=[request.id], + missing_inputs=missing_inputs, + warnings=warnings, + logs=[ + f"Blocked lvl1 assembly for {request.id}; missing {len(missing_inputs)} required input(s)." + ], + ) + + product_identity = ( + constraints.get("product_identity") or request.source_identity + ) + product_display_id = ( + constraints.get("product_display_id") + or request.source_display_id + or product_identity.rsplit("/", 1)[-1] + ) + + assembly_result = self.assembly_service.run( + AssemblyJob( + stage=BuildStage.ASSEMBLY_LVL1, + product_identity=product_identity, + product_display_id=product_display_id, + part_plasmids=list(route.selected_part_plasmids), + backbone=route.backbone, + restriction_enzyme=restriction_enzyme, + ligase=ligase, + source_document=source_document, + target_document=target_document, + ) + ) + + for product in assembly_result.products: + insert_identities = list(product.metadata.get("insert_identities", [])) + if request.source_identity not in insert_identities: + insert_identities.append(request.source_identity) + product.metadata["insert_identities"] = insert_identities + self.inventory.add_generated_product(product) + + json_intermediate = assembly_route_to_pudu_json( + product_identity=product_identity, + part_plasmids=route.selected_part_plasmids, + backbone=route.backbone, + restriction_enzyme=restriction_enzyme, + ) + + logs = [ + f"Selected lvl1 route with {len(route.selected_part_plasmids)} part plasmid(s).", + *assembly_result.logs, + ] + return StageResult( + id=f"{request.id}:{BuildStage.ASSEMBLY_LVL1.value}", + stage=BuildStage.ASSEMBLY_LVL1, + status=StageStatus.SUCCESS, + request_ids=[request.id], + products=assembly_result.products, + warnings=warnings, + sbol_document=assembly_result.stage_document, + json_intermediate=json_intermediate, + logs=logs, + ) + + def _extract_part_identities(self, constraints: Mapping[str, Any]) -> list[str]: + ordered = constraints.get("ordered_part_identities") + if ordered: + return list(ordered) + planner_order = constraints.get("part_order") + if planner_order: + return list(planner_order) + unordered = constraints.get("part_identities") + if unordered: + return list(unordered) + return [] + + def _extract_warnings( + self, *, request: BuildRequest, constraints: Mapping[str, Any] + ) -> list[BuildWarning]: + warnings: list[BuildWarning] = [] + for item in constraints.get("ordering_warnings", []): + if isinstance(item, BuildWarning): + warnings.append(item) + elif isinstance(item, Mapping): + warnings.append( + BuildWarning( + code=str(item.get("code", "ordering_warning")), + message=str(item.get("message", "Ordering warning.")), + stage=BuildStage.ASSEMBLY_LVL1, + source_identity=request.source_identity, + metadata=dict(item.get("metadata", {})), + ) + ) + return warnings + + def _infer_missing_kind(self, part_identity: str) -> str: + text = part_identity.lower() + for role in ("promoter", "rbs", "cds", "terminator"): + if role in text: + return role + return "reagent" diff --git a/tests/unit/adapters/pudu/test_assembly_json.py b/tests/unit/adapters/pudu/test_assembly_json.py new file mode 100644 index 0000000..5b0859c --- /dev/null +++ b/tests/unit/adapters/pudu/test_assembly_json.py @@ -0,0 +1,26 @@ +from buildcompiler.adapters.pudu import assembly_route_to_pudu_json +from buildcompiler.domain import IndexedBackbone, IndexedPlasmid, IndexedReagent + + +def test_assembly_route_to_pudu_json_shape_and_values(): + payload = assembly_route_to_pudu_json( + product_identity="https://example.org/products/p1", + part_plasmids=[ + IndexedPlasmid(identity="https://example.org/plasmids/part1"), + IndexedPlasmid(identity="https://example.org/plasmids/part2"), + ], + backbone=IndexedBackbone(identity="https://example.org/backbones/b1"), + restriction_enzyme=IndexedReagent( + identity="https://example.org/reagents/re1", name="BsaI" + ), + ) + + assert payload == { + "Product": "https://example.org/products/p1", + "Backbone": "https://example.org/backbones/b1", + "PartsList": [ + "https://example.org/plasmids/part1", + "https://example.org/plasmids/part2", + ], + "Restriction Enzyme": "BsaI", + } diff --git a/tests/unit/stages/test_assembly_lvl1.py b/tests/unit/stages/test_assembly_lvl1.py new file mode 100644 index 0000000..1563525 --- /dev/null +++ b/tests/unit/stages/test_assembly_lvl1.py @@ -0,0 +1,186 @@ +import sbol2 + +from buildcompiler.domain import ( + BuildRequest, + BuildStage, + BuildWarning, + DesignKind, + IndexedBackbone, + IndexedPlasmid, + IndexedReagent, + MaterialState, + StageStatus, +) +from buildcompiler.inventory import Inventory +from buildcompiler.sbol import AssemblySbolResult +from buildcompiler.stages import AssemblyLvl1Stage + + +class _FakeAssemblyService: + def __init__(self, products): + self.products = products + + def run(self, job): + return AssemblySbolResult( + products=self.products, + stage_document=job.target_document, + activity_identity="https://example.org/activity/a1", + logs=["fake-assembly-service-ran"], + ) + + +def _inventory(*, include_parts=True, include_backbone=True, include_reagents=True): + plasmids = [] + if include_parts: + plasmids = [ + IndexedPlasmid( + identity="https://example.org/plasmids/p-prom", + metadata={ + "insert_identities": ["https://example.org/parts/promoter"], + "antibiotic": "Ampicillin", + }, + ), + IndexedPlasmid( + identity="https://example.org/plasmids/p-rbs", + metadata={ + "insert_identities": ["https://example.org/parts/rbs"], + "antibiotic": "Ampicillin", + }, + ), + IndexedPlasmid( + identity="https://example.org/plasmids/p-cds", + metadata={ + "insert_identities": ["https://example.org/parts/cds"], + "antibiotic": "Ampicillin", + }, + ), + IndexedPlasmid( + identity="https://example.org/plasmids/p-term", + metadata={ + "insert_identities": ["https://example.org/parts/terminator"], + "antibiotic": "Ampicillin", + }, + ), + ] + backbones = [] + if include_backbone: + backbones = [ + IndexedBackbone( + identity="https://example.org/backbones/lvl1", + metadata={ + "fusion_sites": ("A", "B"), + "antibiotic": "Ampicillin", + "stage": BuildStage.ASSEMBLY_LVL1.value, + }, + ) + ] + reagents = [] + if include_reagents: + reagents = [ + IndexedReagent( + identity="https://example.org/reagents/bsaI", + name="BsaI", + reagent_type="restriction_enzyme", + ), + IndexedReagent( + identity="https://example.org/reagents/ligase", + name="T4_DNA_ligase", + reagent_type="ligase", + ), + ] + return Inventory(plasmids=plasmids, backbones=backbones, reagents=reagents) + + +def _request(): + return BuildRequest( + id="req-1", + stage=BuildStage.ASSEMBLY_LVL1, + source_identity="https://example.org/designs/d1", + source_display_id="d1", + source_kind=DesignKind.COMPONENT_DEFINITION, + constraints={ + "ordered_part_identities": [ + "https://example.org/parts/promoter", + "https://example.org/parts/rbs", + "https://example.org/parts/cds", + "https://example.org/parts/terminator", + ], + "fusion_sites": ["A", "B"], + "antibiotic": "Ampicillin", + "ordering_warnings": [ + { + "code": "planner.ordering", + "message": "planner warning", + "metadata": {"source": "planner"}, + } + ], + }, + ) + + +def test_assembly_lvl1_success_returns_products_json_and_indexes_generated(): + inv = _inventory() + generated = IndexedPlasmid( + identity="https://example.org/plasmids/generated-1", + state=MaterialState.GENERATED, + metadata={"insert_identities": ["https://example.org/parts/promoter"]}, + ) + stage = AssemblyLvl1Stage( + inventory=inv, assembly_service=_FakeAssemblyService([generated]) + ) + + result = stage.run( + _request(), source_document=sbol2.Document(), target_document=sbol2.Document() + ) + + assert result.status == StageStatus.SUCCESS + assert result.products + assert result.products[0].identity == generated.identity + assert ( + result.json_intermediate + and result.json_intermediate["Product"] == "https://example.org/designs/d1" + ) + assert result.sbol_document is not None + assert inv.find_lvl1_region_plasmids("https://example.org/designs/d1") + assert result.logs + assert result.warnings and isinstance(result.warnings[0], BuildWarning) + + +def test_assembly_lvl1_missing_inputs_return_blocked_with_structured_kinds(): + inv = _inventory( + include_parts=False, include_backbone=False, include_reagents=False + ) + stage = AssemblyLvl1Stage(inventory=inv, assembly_service=_FakeAssemblyService([])) + + result = stage.run( + _request(), source_document=sbol2.Document(), target_document=sbol2.Document() + ) + + assert result.status == StageStatus.BLOCKED + kinds = {missing.missing_kind for missing in result.missing_inputs} + assert { + "promoter", + "rbs", + "cds", + "terminator", + "backbone", + "restriction_enzyme", + "ligase", + }.issubset(kinds) + + +def test_assembly_lvl1_accepts_planner_part_order_constraint(): + inv = _inventory() + stage = AssemblyLvl1Stage(inventory=inv, assembly_service=_FakeAssemblyService([])) + request = _request() + request.constraints = { + "part_order": request.constraints["ordered_part_identities"], + "fusion_sites": request.constraints["fusion_sites"], + "antibiotic": request.constraints["antibiotic"], + } + + result = stage.run( + request, source_document=sbol2.Document(), target_document=sbol2.Document() + ) + + assert result.status == StageStatus.SUCCESS