Implement BuildCompiler.plating with SBOL provenance, protocol artifacts, and robot helpers

pyproject.toml

@@ -34,6 +34,11 @@ test = [
   "pytest < 5.0.0",
   "pytest-cov[all]"
 ]
+automation = [
+  "pudupy",
+  "opentrons",
+  "SBOLInventory @ git+https://github.com/DRAGGON-Lab/SBOLInventory.git ; python_version >= '3.10'",
+]
 
 [project.urls]
 "Homepage" = "https://github.com/MyersResearchGroup/BuildCompiler"

src/buildcompiler/buildcompiler.py

@@ -1,6 +1,8 @@
 import sbol2
 import random
 import warnings
+import urllib.parse
+from pathlib import Path
 from typing import Any, Dict, List
 
 from buildcompiler.plasmid import Plasmid
@@ -9,7 +11,16 @@
     get_or_pull,
     get_compatible_plasmids,
 )
-from .robotutils import assembly_plan_RDF_to_JSON
+from .robotutils import (
+    assembly_plan_RDF_to_JSON,
+    generate_96_well_positions,
+    normalize_plating_input,
+    run_opentrons_script_to_zip,
+    write_manual_plating_protocol,
+    write_plate_map_csv,
+    write_plate_map_json,
+    write_plating_protocol_script,
+)
 from .constants import (
     AMP,
     KAN,
@@ -21,6 +32,7 @@
     ENGINEERED_PLASMID,
     PLASMID_CLONING_VECTOR,
     ORGANISM_STRAIN,
+    PLATING_ACTIVITY_ROLE,
 )
 
 
@@ -502,6 +514,228 @@ def transformation(
             },
         }
 
+    def plating(
+        self,
+        transformation_results: dict,
+        results_dir: str | Path,
+        protocol_type: str = "manual",
+        advanced_params: dict | None = None,
+        plate_name: str | None = None,
+        plating_doc: sbol2.Document | None = None,
+        overwrite: bool = False,
+    ) -> Dict[str, Any]:
+        """Generate a plated 96-well output and protocol artifacts."""
+        if protocol_type not in {"manual", "automated"}:
+            raise ValueError("protocol_type must be one of: 'manual', 'automated'.")
+        if plating_doc is None:
+            plating_doc = self.sbol_doc
+        advanced_params = advanced_params or {}
+
+        normalized = normalize_plating_input(transformation_results, doc=plating_doc)
+        if len(normalized) > 96:
+            raise ValueError("plating supports up to 96 transformed strains.")
+
+        wells = generate_96_well_positions(limit=len(normalized))
+        results_path = Path(results_dir)
+        results_path.mkdir(parents=True, exist_ok=True)
+
+        plate_id = plate_name or "solid_96_well_plate"
+        plate_impl = sbol2.Implementation(plate_id)
+        plate_md = plating_doc.find("solid_96_well_plate_md") or sbol2.ModuleDefinition(
+            "solid_96_well_plate_md"
+        )
+        plate_md.name = "Solid 96-well plate"
+        self._add_if_absent(plating_doc, plate_md)
+        plate_impl.built = plate_md.identity
+        self._add_if_absent(plating_doc, plate_impl)
+
+        # Optional SBOLInventory integration with fallback behavior.
+        try:
+            from sbol_inventory import (  # type: ignore
+                make_solid_96_well_plate,
+                make_plated_strain,
+                place_in_plate,
+            )
+
+            inventory_enabled = True
+            inventory_plate = make_solid_96_well_plate(
+                uri=plate_impl.identity, plate_md_uri=plate_md.identity
+            )
+        except Exception:
+            inventory_enabled = False
+            inventory_plate = None
+
+        activity_id = f"plating_{protocol_type}_{plate_id}"
+        plating_activity = sbol2.Activity(activity_id)
+        plating_activity.name = f"Plating activity for {plate_id}"
+        plating_activity.types = "http://sbols.org/v2#build"
+        self._add_if_absent(plating_doc, plating_activity)
+
+        agent_id = (
+            "manual_plating_agent"
+            if protocol_type == "manual"
+            else "opentrons_plating_agent"
+        )
+        agent = plating_doc.find(agent_id) or sbol2.Agent(agent_id)
+        agent.name = "Manual plating agent" if protocol_type == "manual" else "Opentrons plating agent"
+        self._add_if_absent(plating_doc, agent)
+
+        plan_id = f"{plate_id}_{protocol_type}_plating_plan"
+        plan = plating_doc.find(plan_id) or sbol2.Plan(plan_id)
+        plan.name = f"{protocol_type.title()} plating plan for {plate_id}"
+        self._add_if_absent(plating_doc, plan)
+
+        association = sbol2.Association(
+            uri=f"{activity_id}_association",
+            agent=agent.identity,
+            role="http://sbols.org/v2#build",
+        )
+        association.plan = plan.identity
+        plating_activity.associations = [association]
+        self._add_if_absent(plating_doc, association)
+
+        plate_rows = []
+        plate_map = {}
+        bacterium_locations = {}
+        plated_impls = []
+
+        for idx, entry in enumerate(normalized):
+            well = wells[idx]
+            source_impl_uri = entry.get("source_impl_uri")
+            source_impl = plating_doc.find(source_impl_uri) if source_impl_uri else None
+            strain_module_uri = entry.get("strain_module_uri")
+            if strain_module_uri is None and source_impl is not None:
+                strain_module_uri = getattr(source_impl, "built", None)
+
+            display_source = source_impl_uri or strain_module_uri or f"strain_{idx+1}"
+            parsed = urllib.parse.urlparse(display_source)
+            slug = parsed.path.split("/")[-1] if parsed.path else display_source
+            slug = slug.replace("#", "_").replace(":", "_")
+
+            plated_module_id = f"{slug}_plated_{well}_md"
+            plated_module = plating_doc.find(plated_module_id) or sbol2.ModuleDefinition(
+                plated_module_id
+            )
+            plated_module.roles = [ORGANISM_STRAIN]
+            plated_module.name = f"Plated strain {slug} at {well}"
+            if strain_module_uri:
+                plated_module.wasDerivedFrom = strain_module_uri
+            self._add_if_absent(plating_doc, plated_module)
+
+            plated_impl_id = f"{slug}_plated_{well}_impl"
+            plated_impl = plating_doc.find(plated_impl_id) or sbol2.Implementation(
+                plated_impl_id
+            )
+            plated_impl.built = plated_module.identity
+            plated_impl.wasGeneratedBy = plating_activity.identity
+            if source_impl_uri:
+                plated_impl.wasDerivedFrom = source_impl_uri
+            self._add_if_absent(plating_doc, plated_impl)
+            plated_impls.append(plated_impl.identity)
+
+            usage = sbol2.Usage(
+                uri=f"{activity_id}_usage_{idx+1}",
+                entity=source_impl_uri or plated_module.identity,
+                role=PLATING_ACTIVITY_ROLE,
+            )
+            self._add_if_absent(plating_doc, usage)
+            current_usages = list(plating_activity.usages)
+            current_usages.append(usage)
+            plating_activity.usages = current_usages
+
+            if inventory_enabled:
+                try:
+                    inventory_plated = make_plated_strain(
+                        uri=plated_impl.identity,
+                        strain_md_uri=strain_module_uri or plated_module.identity,
+                        design_uri=source_impl_uri,
+                    )
+                    place_in_plate(inventory_plate, inventory_plated, well)
+                except Exception:
+                    inventory_enabled = False
+
+            plate_map[well] = plated_impl.identity
+            display_name = plated_module.displayId
+            bacterium_locations[well] = display_name
+            plate_rows.append(
+                {
+                    "well": well,
+                    "source_transformed_strain_implementation": source_impl_uri,
+                    "strain_module": strain_module_uri,
+                    "plated_strain_implementation": plated_impl.identity,
+                    "strain_display_name": display_name,
+                }
+            )
+
+        plate_map_json_path = write_plate_map_json(
+            results_path / "plate_map.json",
+            {
+                "plate_implementation": plate_impl.identity,
+                "protocol_type": protocol_type,
+                "well_map": plate_rows,
+            },
+        )
+        plate_map_csv_path = write_plate_map_csv(results_path / "plate_map.csv", plate_rows)
+        plating_input_json_path = write_plate_map_json(
+            results_path / "plating_input.json",
+            {"bacterium_locations": bacterium_locations},
+        )
+
+        logs = []
+        protocol_artifacts: Dict[str, Any] = {
+            "plate_map_json": str(plate_map_json_path),
+            "plate_map_csv": str(plate_map_csv_path),
+            "logs": logs,
+        }
+
+        if protocol_type == "manual":
+            md_path = write_manual_plating_protocol(
+                results_path / "manual_plating_protocol.md",
+                plate_id=plate_impl.displayId,
+                plate_rows=plate_rows,
+                advanced_params=advanced_params,
+            )
+            protocol_artifacts["manual_protocol_markdown"] = str(md_path)
+            plan.description = f"Manual protocol file: {md_path}"
+        else:
+            script_path = write_plating_protocol_script(
+                results_path / "plating_ot2.py",
+                plating_data={"bacterium_locations": bacterium_locations},
+                advanced_params=advanced_params,
+            )
+            protocol_artifacts["ot2_script"] = str(script_path)
+            plan.description = f"Automated protocol script: {script_path}"
+            try:
+                sim_zip = run_opentrons_script_to_zip(
+                    script_path,
+                    plating_input_json_path,
+                    overwrite=overwrite,
+                )
+                protocol_artifacts["simulation_zip"] = str(sim_zip)
+            except Exception as exc:
+                logs.append(f"Opentrons simulation skipped: {exc}")
+
+        return {
+            "stage": "plating",
+            "protocol_type": protocol_type,
+            "plate": {
+                "plate_implementation": plate_impl.identity,
+                "plate_map": plate_map,
+            },
+            "sbol_artifacts": {
+                "plating_activity": plating_activity.identity,
+                "agent": agent.identity,
+                "plan": plan.identity,
+                "plate_implementation": plate_impl.identity,
+                "plated_strain_implementations": plated_impls,
+            },
+            "json_intermediate": {
+                "plating_data": {"bacterium_locations": bacterium_locations},
+                "advanced_params": advanced_params,
+            },
+            "protocol_artifacts": protocol_artifacts,
+        }
+
     def _extract_plasmids_from_strain(
         self,
         strain: sbol2.ModuleDefinition,

src/buildcompiler/constants.py

@@ -35,6 +35,7 @@
 RESTRICTION_ENZYME = "http://identifiers.org/obi/OBI_0000732"
 RESTRICTION_ENZYME_ASSEMBLY_SCAR = "http://identifiers.org/so/SO:0001953"
 ORGANISM_STRAIN = "https://identifiers.org/ncit/NCIT:C14419"
+PLATING_ACTIVITY_ROLE = "https://w3id.org/buildcompiler/roles/plating_input"
 
 FIVE_PRIME_OVERHANG = "http://identifiers.org/so/SO:0001932"
 THREE_PRIME_OVERHANG = "http://identifiers.org/so/SO:0001933"