From 6bce1f03ef967018877723dd8f3627ce82559fe3 Mon Sep 17 00:00:00 2001 From: Gonzalo Vidal <35148159+Gonza10V@users.noreply.github.com> Date: Mon, 27 Apr 2026 23:55:13 -0600 Subject: [PATCH] Add plating workflow with SBOL/protocol artifact generation --- pyproject.toml | 5 + src/buildcompiler/buildcompiler.py | 236 +++++++++++++++++++++++++++- src/buildcompiler/constants.py | 1 + src/buildcompiler/robotutils.py | 241 +++++++++++++++++++++++++++++ tests/test_plating.py | 130 ++++++++++++++++ 5 files changed, 612 insertions(+), 1 deletion(-) create mode 100644 tests/test_plating.py diff --git a/pyproject.toml b/pyproject.toml index 7130b71..a0c13bb 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/buildcompiler/buildcompiler.py b/src/buildcompiler/buildcompiler.py index aa0511b..7e4005f 100644 --- a/src/buildcompiler/buildcompiler.py +++ b/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, diff --git a/src/buildcompiler/constants.py b/src/buildcompiler/constants.py index 19c25a2..1fed16a 100644 --- a/src/buildcompiler/constants.py +++ b/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" diff --git a/src/buildcompiler/robotutils.py b/src/buildcompiler/robotutils.py index 5b91cf9..6db50d6 100644 --- a/src/buildcompiler/robotutils.py +++ b/src/buildcompiler/robotutils.py @@ -1,10 +1,251 @@ import sbol2 import json +import csv import shutil import subprocess import tempfile import zipfile from pathlib import Path +from typing import Any, Dict, List + + +def load_json_or_dict(value): + """Load a JSON file path/string into a dict, or return dict-like input as-is.""" + if isinstance(value, dict): + return value + if isinstance(value, Path): + return json.loads(value.read_text(encoding="utf-8")) + if isinstance(value, str): + path = Path(value) + if path.exists(): + return json.loads(path.read_text(encoding="utf-8")) + return json.loads(value) + raise ValueError("Expected dict, JSON string, or JSON file path.") + + +def normalize_plating_input(transformation_results, doc=None): + """Normalize transformation/plating inputs to a deterministic list of entries.""" + payload = load_json_or_dict(transformation_results) + normalized = [] + + if isinstance(payload, dict) and isinstance(payload.get("sbol_artifacts"), list): + for idx, artifact in enumerate(payload["sbol_artifacts"], start=1): + if not isinstance(artifact, dict): + continue + impl_uri = artifact.get("transformed_strain_implementation") + module_uri = artifact.get("transformed_strain_module") + if not impl_uri and not module_uri: + continue + normalized.append( + { + "order": idx, + "well_hint": None, + "source_impl_uri": impl_uri, + "strain_module_uri": module_uri, + } + ) + elif isinstance(payload, dict) and isinstance(payload.get("strain_locations"), dict): + for idx, well in enumerate(sorted(payload["strain_locations"]), start=1): + impl_uri = payload["strain_locations"][well] + normalized.append( + { + "order": idx, + "well_hint": well, + "source_impl_uri": impl_uri, + "strain_module_uri": None, + } + ) + elif isinstance(payload, dict) and isinstance( + payload.get("bacterium_locations"), dict + ): + for idx, well in enumerate(sorted(payload["bacterium_locations"]), start=1): + impl_uri = payload["bacterium_locations"][well] + normalized.append( + { + "order": idx, + "well_hint": well, + "source_impl_uri": impl_uri, + "strain_module_uri": None, + } + ) + else: + raise ValueError( + "Unsupported plating input shape. Expected transformation output, " + "strain_locations, or bacterium_locations." + ) + + if doc is not None: + for item in normalized: + if item["strain_module_uri"] is None and item["source_impl_uri"]: + impl = doc.find(item["source_impl_uri"]) + if impl is not None and getattr(impl, "built", None): + item["strain_module_uri"] = impl.built + + if not normalized: + raise ValueError("No transformed strains found in plating input.") + + return sorted(normalized, key=lambda x: x["order"]) + + +def generate_96_well_positions(limit=96): + wells = [f"{row}{column}" for row in "ABCDEFGH" for column in range(1, 13)] + if limit < 0: + raise ValueError("limit must be non-negative") + return wells[:limit] + + +def write_plate_map_json(path, data): + path = Path(path) + path.write_text(json.dumps(data, indent=2), encoding="utf-8") + return path + + +def write_plate_map_csv(path, rows: List[Dict[str, Any]]): + path = Path(path) + fields = [ + "well", + "source_transformed_strain_implementation", + "strain_module", + "plated_strain_implementation", + "strain_display_name", + ] + with path.open("w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=fields) + writer.writeheader() + for row in rows: + writer.writerow( + { + "well": row.get("well"), + "source_transformed_strain_implementation": row.get( + "source_transformed_strain_implementation" + ), + "strain_module": row.get("strain_module"), + "plated_strain_implementation": row.get( + "plated_strain_implementation" + ), + "strain_display_name": row.get("strain_display_name"), + } + ) + return path + + +def write_manual_plating_protocol(path, plate_id, plate_rows, advanced_params): + path = Path(path) + params = advanced_params or {} + table_lines = [ + "| Well | Source transformed strain implementation | Plated strain implementation | Strain module |", + "|---|---|---|---|", + ] + for row in plate_rows: + table_lines.append( + f"| {row['well']} | {row.get('source_transformed_strain_implementation','')} " + f"| {row.get('plated_strain_implementation','')} | {row.get('strain_module','')} |" + ) + + param_lines = "\n".join([f"- **{k}**: {v}" for k, v in sorted(params.items())]) or "- (none)" + strain_lines = "\n".join( + [ + f"- {row.get('strain_display_name', row.get('source_transformed_strain_implementation'))}" + for row in plate_rows + ] + ) + content = ( + "# BuildCompiler Plating Protocol\n\n" + f"## Plate\n- Plate ID: `{plate_id}`\n- Protocol type: `manual`\n\n" + "## Input transformed strains\n" + f"{strain_lines}\n\n" + "## Parameters\n" + f"{param_lines}\n\n" + "## 96-well plate map\n" + f"{chr(10).join(table_lines)}\n\n" + "## Steps\n" + "1. Prepare one sterile solid-media 96-well plate.\n" + "2. Label the plate with the plate ID and date.\n" + "3. Transfer each transformed strain to the destination well shown in the map.\n" + "4. Incubate according to lab defaults or parameters above.\n" + ) + path.write_text(content, encoding="utf-8") + return path + + +def write_plating_protocol_script(path, plating_data, advanced_params): + path = Path(path) + script = ( + "from pudu.plating import Plating\n" + "from opentrons import protocol_api\n\n" + "metadata = {\n" + ' "protocolName": "BuildCompiler Plating",\n' + ' "author": "BuildCompiler",\n' + ' "description": "Automated plating protocol generated from BuildCompiler transformation results",\n' + ' "apiLevel": "2.21",\n' + "}\n\n" + f"PLATING_DATA = {json.dumps(plating_data, indent=4)}\n" + f"ADVANCED_PARAMS = {json.dumps(advanced_params or {}, indent=4)}\n\n" + "def run(protocol: protocol_api.ProtocolContext):\n" + " plating = Plating(\n" + " plating_data=PLATING_DATA,\n" + " json_params=ADVANCED_PARAMS,\n" + " )\n" + " plating.run(protocol)\n" + ) + path.write_text(script, encoding="utf-8") + return path + + +def run_opentrons_script_to_zip( + opentrons_script_path, + plating_json_path, + zip_name=None, + overwrite=False, +): + script_path = Path(opentrons_script_path).resolve() + json_path = Path(plating_json_path).resolve() + if not script_path.exists(): + raise FileNotFoundError(f"Opentrons script not found: {script_path}") + if not json_path.exists(): + raise FileNotFoundError(f"JSON file not found: {json_path}") + + out_zip = script_path.parent / (zip_name or f"{script_path.stem}_simulation.zip") + if out_zip.exists() and not overwrite: + stem = out_zip.stem + suffix = out_zip.suffix + i = 1 + while True: + candidate = out_zip.parent / f"{stem}_{i}{suffix}" + if not candidate.exists(): + out_zip = candidate + break + i += 1 + + with tempfile.TemporaryDirectory() as tmpdirname: + tmpdir = Path(tmpdirname) + tmp_script = tmpdir / script_path.name + tmp_json = tmpdir / json_path.name + shutil.copy2(script_path, tmp_script) + shutil.copy2(json_path, tmp_json) + + proc = subprocess.run( + ["opentrons_simulate", str(tmp_script)], + capture_output=True, + text=True, + cwd=tmpdir, + ) + (tmpdir / "simulate_stdout.txt").write_text( + proc.stdout or "", encoding="utf-8", errors="replace" + ) + (tmpdir / "simulate_stderr.txt").write_text( + proc.stderr or "", encoding="utf-8", errors="replace" + ) + (tmpdir / "simulate_returncode.txt").write_text( + str(proc.returncode), encoding="utf-8" + ) + + with zipfile.ZipFile(out_zip, "w", compression=zipfile.ZIP_DEFLATED) as zf: + for file_path in tmpdir.rglob("*"): + if file_path.is_file(): + zf.write(file_path, arcname=file_path.relative_to(tmpdir)) + + return out_zip def assembly_plan_RDF_to_JSON(file, output_path: str | Path | None = None): if isinstance(file, sbol2.Document): diff --git a/tests/test_plating.py b/tests/test_plating.py new file mode 100644 index 0000000..e0c1684 --- /dev/null +++ b/tests/test_plating.py @@ -0,0 +1,130 @@ +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import sbol2 + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../src"))) + +from buildcompiler.buildcompiler import BuildCompiler +from buildcompiler.constants import ENGINEERED_PLASMID, ORGANISM_STRAIN +from buildcompiler.robotutils import generate_96_well_positions, normalize_plating_input + + +class _ProcResult: + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +class TestPlating(unittest.TestCase): + def setUp(self): + self.doc = sbol2.Document() + self.compiler = BuildCompiler( + collections=[], + sbh_registry="https://synbiohub.org", + auth_token="", + sbol_doc=self.doc, + ) + self._id_seed = 0 + + def _make_transform_result(self, count=1): + artifacts = [] + self._id_seed += 1 + prefix = f"t{self._id_seed}_" + for i in range(1, count + 1): + plasmid = sbol2.ComponentDefinition(f"{prefix}p{i}") + plasmid.roles = [ENGINEERED_PLASMID] + self.doc.add(plasmid) + transform_module = sbol2.ModuleDefinition(f"{prefix}strain_{i}") + transform_module.roles = [ORGANISM_STRAIN] + self.doc.add(transform_module) + impl = sbol2.Implementation(f"{prefix}strain_{i}_impl") + impl.built = transform_module.identity + self.doc.add(impl) + artifacts.append( + { + "transformed_strain_module": transform_module.identity, + "transformed_strain_implementation": impl.identity, + } + ) + return {"stage": "transformation", "sbol_artifacts": artifacts} + + def test_normalization_shapes(self): + result = self._make_transform_result(2) + normalized = normalize_plating_input(result, doc=self.doc) + self.assertEqual(len(normalized), 2) + + sloc = normalize_plating_input({"strain_locations": {"A1": "x", "A2": "y"}}) + self.assertEqual(len(sloc), 2) + + bloc = normalize_plating_input({"bacterium_locations": {"A1": "x"}}) + self.assertEqual(len(bloc), 1) + + with self.assertRaises(ValueError): + normalize_plating_input({"invalid": True}) + + def test_plate_well_mapping_limits(self): + self.assertEqual(generate_96_well_positions(1), ["A1"]) + self.assertEqual(generate_96_well_positions(13)[-1], "B1") + self.assertEqual(len(generate_96_well_positions(96)), 96) + + with tempfile.TemporaryDirectory() as tmpdir: + ok = self.compiler.plating( + self._make_transform_result(96), Path(tmpdir), protocol_type="manual" + ) + self.assertEqual(len(ok["plate"]["plate_map"]), 96) + + with self.assertRaises(ValueError): + self.compiler.plating( + self._make_transform_result(97), Path(tmpdir), protocol_type="manual" + ) + + def test_plating_manual_outputs_and_provenance(self): + with tempfile.TemporaryDirectory() as tmpdir: + result = self.compiler.plating( + transformation_results=self._make_transform_result(2), + results_dir=tmpdir, + protocol_type="manual", + advanced_params={"incubation_temperature_c": 37}, + ) + self.assertEqual(result["stage"], "plating") + self.assertEqual(result["protocol_type"], "manual") + self.assertEqual(len(result["plate"]["plate_map"]), 2) + self.assertTrue( + Path(result["protocol_artifacts"]["manual_protocol_markdown"]).exists() + ) + self.assertTrue(Path(result["protocol_artifacts"]["plate_map_json"]).exists()) + self.assertTrue(Path(result["protocol_artifacts"]["plate_map_csv"]).exists()) + + activity = self.doc.find(result["sbol_artifacts"]["plating_activity"]) + self.assertIsNotNone(activity) + self.assertEqual(len(activity.usages), 2) + self.assertTrue(len(activity.associations) >= 1) + + @patch("buildcompiler.robotutils.subprocess.run") + def test_plating_automated_script_and_sim_zip(self, mock_run): + mock_run.return_value = _ProcResult(returncode=0, stdout="ok", stderr="") + with tempfile.TemporaryDirectory() as tmpdir: + result = self.compiler.plating( + transformation_results=self._make_transform_result(1), + results_dir=tmpdir, + protocol_type="automated", + advanced_params={"replicates": 1}, + overwrite=True, + ) + script_path = Path(result["protocol_artifacts"]["ot2_script"]) + script = script_path.read_text(encoding="utf-8") + self.assertIn("def run(protocol: protocol_api.ProtocolContext):", script) + self.assertIn("from opentrons import protocol_api", script) + self.assertIn("json_params=ADVANCED_PARAMS", script) + self.assertNotIn("advanced_params=ADVANCED_PARAMS", script) + self.assertTrue(Path(result["protocol_artifacts"]["simulation_zip"]).exists()) + + +if __name__ == "__main__": + unittest.main()