diff --git a/src/buildcompiler/sbol/__init__.py b/src/buildcompiler/sbol/__init__.py index 45d9597..d9e6f2c 100644 --- a/src/buildcompiler/sbol/__init__.py +++ b/src/buildcompiler/sbol/__init__.py @@ -1,5 +1,12 @@ """SBOL package exports for clean architecture contracts.""" +from .assembly import AssemblyJob, AssemblySbolResult, AssemblyService from .resolver import PullPolicy, SbolResolver -__all__ = ["PullPolicy", "SbolResolver"] +__all__ = [ + "AssemblyJob", + "AssemblySbolResult", + "AssemblyService", + "PullPolicy", + "SbolResolver", +] diff --git a/src/buildcompiler/sbol/assembly.py b/src/buildcompiler/sbol/assembly.py new file mode 100644 index 0000000..fa877d9 --- /dev/null +++ b/src/buildcompiler/sbol/assembly.py @@ -0,0 +1,185 @@ +"""SBOL assembly service wrapping legacy Golden Gate behavior.""" + +from dataclasses import dataclass, field + +import sbol2 + +from buildcompiler.domain import ( + BuildStage, + IndexedBackbone, + IndexedPlasmid, + IndexedReagent, + MaterialState, +) +from buildcompiler.sbol2build import Assembly + + +@dataclass +class _LegacyPlasmidAdapter: + plasmid_definition: sbol2.ComponentDefinition + plasmid_implementations: list[sbol2.Implementation] + + +@dataclass +class AssemblyJob: + """Normalized assembly inputs plus source/target SBOL documents.""" + + stage: BuildStage + product_identity: str + product_display_id: str + part_plasmids: list[IndexedPlasmid] + backbone: IndexedBackbone + restriction_enzyme: IndexedReagent + ligase: IndexedReagent + source_document: sbol2.Document + target_document: sbol2.Document + include_extracted_parts: bool = False + + +@dataclass +class AssemblySbolResult: + """Assembly output contract: normalized products plus stage document.""" + + products: list[IndexedPlasmid] + stage_document: sbol2.Document + activity_identity: str + logs: list[str] = field(default_factory=list) + + +class AssemblyService: + """Service wrapper that preserves legacy assembly internals behind normalized contracts.""" + + def run(self, job: AssemblyJob) -> AssemblySbolResult: + if not job.part_plasmids: + raise ValueError("AssemblyJob.part_plasmids must contain at least one plasmid") + + legacy_parts = [ + self._record_to_legacy_plasmid(record, job.source_document, "part_plasmids") + for record in job.part_plasmids + ] + legacy_backbone = self._record_to_legacy_plasmid( + IndexedPlasmid( + identity=job.backbone.identity, + display_id=job.backbone.display_id, + name=job.backbone.name, + metadata=job.backbone.metadata, + sbol_component=job.backbone.sbol_component, + ), + job.source_document, + "backbone", + ) + + restriction_impl = self._implementation_from_record( + job.restriction_enzyme, job.source_document + ) + ligase_impl = self._implementation_from_record(job.ligase, job.source_document) + + composite_prefix = job.product_display_id or job.product_identity.split("/")[-1] + legacy_assembly = Assembly( + part_plasmids=legacy_parts, + backbone_plasmid=legacy_backbone, + restriction_enzyme=restriction_impl, + ligase=ligase_impl, + source_document=job.source_document, + final_document=job.target_document, + composite_prefix=composite_prefix, + ) + legacy_products, final_doc = legacy_assembly.run( + include_extracted_parts=job.include_extracted_parts + ) + + products = [ + self._indexed_product_from_legacy_product(plasmid, job) + for plasmid in legacy_products + ] + logs = [ + f"Assembled {len(products)} product(s) at stage {job.stage.value}.", + f"Assembly activity: {legacy_assembly.assembly_activity.identity}", + ] + + return AssemblySbolResult( + products=products, + stage_document=final_doc, + activity_identity=legacy_assembly.assembly_activity.identity, + logs=logs, + ) + + def _record_to_legacy_plasmid( + self, + record: IndexedPlasmid, + source_document: sbol2.Document, + field_name: str, + ) -> _LegacyPlasmidAdapter: + component = self._component_from_record(record, source_document, field_name) + implementation = self._implementation_from_plasmid_record(record, source_document) + return _LegacyPlasmidAdapter(component, [implementation]) + + def _component_from_record( + self, + record: IndexedPlasmid, + source_document: sbol2.Document, + field_name: str, + ) -> sbol2.ComponentDefinition: + component = record.sbol_component or source_document.find(record.identity) + if component is None: + raise ValueError( + f"Missing SBOL ComponentDefinition for {field_name} record {record.identity}" + ) + if not isinstance(component, sbol2.ComponentDefinition): + raise ValueError( + f"{field_name} record {record.identity} must resolve to sbol2.ComponentDefinition" + ) + return component + + def _implementation_from_plasmid_record( + self, record: IndexedPlasmid, source_document: sbol2.Document + ) -> sbol2.Implementation: + impl_identity = record.metadata.get("implementation_identity") + implementation = source_document.find(impl_identity) if impl_identity else None + + if implementation is None: + component = self._component_from_record(record, source_document, "plasmid") + matches = [ + impl + for impl in source_document.implementations + if isinstance(impl, sbol2.Implementation) and impl.built == component.identity + ] + implementation = matches[0] if matches else None + + if implementation is None: + raise ValueError( + f"Missing SBOL Implementation for plasmid {record.identity}; " + "set metadata['implementation_identity'] or include implementation in source_document" + ) + return implementation + + def _implementation_from_record( + self, record: IndexedReagent, source_document: sbol2.Document + ) -> sbol2.Implementation: + impl_identity = record.metadata.get("implementation_identity") or record.identity + implementation = source_document.find(impl_identity) + if not isinstance(implementation, sbol2.Implementation): + raise ValueError( + "Missing SBOL Implementation for reagent " + f"{record.identity}; expected metadata['implementation_identity'] or identity to resolve" + ) + return implementation + + def _indexed_product_from_legacy_product( + self, product, job: AssemblyJob + ) -> IndexedPlasmid: + component = product.plasmid_definition + return IndexedPlasmid( + identity=component.identity, + display_id=component.displayId, + name=component.name, + state=MaterialState.GENERATED, + roles=list(component.roles), + metadata={ + "source_stage": job.stage.value, + "source_product_identity": job.product_identity, + "source_product_display_id": job.product_display_id, + "assembly_activity_identity": product.plasmid_implementations[0].wasGeneratedBy, + }, + sbol_component=component, + ) diff --git a/tests/unit/sbol/test_assembly_service.py b/tests/unit/sbol/test_assembly_service.py new file mode 100644 index 0000000..2ef8475 --- /dev/null +++ b/tests/unit/sbol/test_assembly_service.py @@ -0,0 +1,81 @@ +import sbol2 +import pytest + +from buildcompiler.domain import ( + BuildStage, + IndexedBackbone, + IndexedPlasmid, + IndexedReagent, + MaterialState, +) +from buildcompiler.sbol import AssemblyJob, AssemblySbolResult, AssemblyService + + +def test_assembly_service_runs_and_returns_normalized_products(monkeypatch): + source = sbol2.Document() + product_component = sbol2.ComponentDefinition("assembled_product") + source.add(product_component) + product_impl = sbol2.Implementation("assembled_product_impl") + product_impl.built = product_component.identity + source.add(product_impl) + product_impl.wasGeneratedBy = "https://example.org/activity/assembly" + + class FakeLegacyAssembly: + def __init__(self, **kwargs): + self.assembly_activity = sbol2.Activity("fake_assembly") + + def run(self, include_extracted_parts=False): + return [type("LegacyProduct", (), {"plasmid_definition": product_component, "plasmid_implementations": [product_impl]})()], source + + monkeypatch.setattr("buildcompiler.sbol.assembly.Assembly", FakeLegacyAssembly) + + service = AssemblyService() + result = service.run( + AssemblyJob( + stage=BuildStage.ASSEMBLY_LVL1, + product_identity="https://example.org/products/p001", + product_display_id="p001", + part_plasmids=[ + IndexedPlasmid( + identity=product_component.identity, + sbol_component=product_component, + metadata={"implementation_identity": product_impl.identity}, + ) + ], + backbone=IndexedBackbone( + identity=product_component.identity, + sbol_component=product_component, + metadata={"implementation_identity": product_impl.identity}, + ), + restriction_enzyme=IndexedReagent(identity=product_impl.identity), + ligase=IndexedReagent(identity=product_impl.identity), + source_document=source, + target_document=sbol2.Document(), + ) + ) + + assert isinstance(result, AssemblySbolResult) + assert result.products + assert isinstance(result.products[0], IndexedPlasmid) + assert result.products[0].state == MaterialState.GENERATED + assert result.activity_identity + + +def test_assembly_service_raises_clear_error_for_missing_component(): + doc = sbol2.Document() + service = AssemblyService() + + with pytest.raises(ValueError, match="Missing SBOL ComponentDefinition"): + service.run( + AssemblyJob( + stage=BuildStage.ASSEMBLY_LVL1, + product_identity="https://example.org/products/p001", + product_display_id="p001", + part_plasmids=[IndexedPlasmid(identity="https://example.org/missing")], + backbone=IndexedBackbone(identity="https://example.org/backbone"), + restriction_enzyme=IndexedReagent(identity="https://example.org/reagent/re"), + ligase=IndexedReagent(identity="https://example.org/reagent/ligase"), + source_document=doc, + target_document=sbol2.Document(), + ) + )