Research: benchmark analysis and expansion candidates

docker/sim/Dockerfile

@@ -0,0 +1,15 @@
+# Text2WetLab simulator image: runs third-party lab-automation protocols with no hardware and no network.
+#   Opentrons OT-2 / Flex : opentrons_simulate (Opentrons 7.5.0, the benchmark's grader version)
+#   Hamilton STAR / STARlet, Tecan EVO and others : PyLabRobot simulated backends
+# Run untrusted protocol code with:  docker run --rm --network none -v <repo>:/work:ro -v <out>:/out t2w-sim ...
+FROM python:3.10-slim
+ENV PIP_NO_CACHE_DIR=1 PYTHONDONTWRITEBYTECODE=1 OT_SMOOTHIE_ID=AMA
+RUN pip install "opentrons==7.5.0" "pylabrobot" numpy pandas
+# Opentrons fetches nothing at simulate time once its shared data is installed; warm it so --network none works.
+RUN python -c "from opentrons import simulate; print('opentrons ok')" && python -c "import pylabrobot; print('pylabrobot ok')"
+COPY runlog.py mylog.py /t2w/
+COPY simulate_all.py hamilton_ape.py /t2w/
+RUN useradd -m sim
+USER sim
+WORKDIR /tmp
+ENTRYPOINT ["python", "/t2w/simulate_all.py"]

docker/sim/hamilton_ape.py

@@ -0,0 +1,126 @@
+"""Run the har1eyk STARlet "Automated Protein Extraction" protocol on PyLabRobot's simulated backend and log it.
+
+usage (in the container): python /t2w/hamilton_ape.py <protocol_dir> <out.json>
+The deck is rebuilt from the authors' APE_96w notebook (rails 25 tips, 13 and 19 MFX carriers). Every
+liquid-handler call is recorded with resource names so the host can replay it on a 3D STARlet; waits,
+manual pauses and prints become comment events instead of real delays.
+"""
+import asyncio, builtins, json, sys, time
+
+from pylabrobot.liquid_handling import LiquidHandler
+from pylabrobot.liquid_handling.backends import LiquidHandlerChatterboxBackend
+from pylabrobot.resources.hamilton import STARLetDeck, MFX_CAR_L5_base, TIP_CAR_480_A00
+from pylabrobot.resources.hamilton.mfx_modules import Hamilton_MFX_plateholder_DWP_metal_tapped
+from pylabrobot.resources.diy.grindbio.modules import Hamilton_MFX_plateholder_DWP_metal_tapped_10mm_3dprint
+from pylabrobot.resources.alpaqua import Alpaqua_96_magnum_flx
+from pylabrobot.resources.bioer.plates import BioER_96_wellplate_Vb_2200uL
+from pylabrobot.resources.agenbio.plates import AGenBio_1_troughplate_100000uL_Fl
+from pylabrobot.resources import hamilton_96_tiprack_1000uL, set_tip_tracking, set_volume_tracking
+
+EVENTS = []
+
+
+class SimulatedSTAR(LiquidHandlerChatterboxBackend):
+  """The generic simulated backend, answering any STAR-only firmware command (CoRe gripper tools, sensors) as a no-op."""
+
+  def __getattr__(self, name):
+    if name.startswith("_"):
+      raise AttributeError(name)
+
+    async def no_op(*a, **k):
+      return True
+    return no_op
+
+
+def names(xs):
+  return [getattr(x, "name", str(x)) for x in xs]
+
+
+def build(lh):
+  tip_car = TIP_CAR_480_A00("tip_car")
+  lh.deck.assign_child_resource(tip_car, rails=25)
+  racks = [hamilton_96_tiprack_1000uL(f"tips_0{i}") for i in range(3)]
+  for i, r in enumerate(racks):
+    tip_car[i] = r
+  m13 = {i: Hamilton_MFX_plateholder_DWP_metal_tapped_10mm_3dprint(n) for i, n in enumerate(
+    ["rail13_mag_module", "rail13_waste_module", "rail13_elution_plate_module", "rail13_wash1_module", "rail13_wash2_module"])}
+  lh.deck.assign_child_resource(MFX_CAR_L5_base("car_13", modules=m13), rails=13)
+  r13 = [Alpaqua_96_magnum_flx("mag_plate"), AGenBio_1_troughplate_100000uL_Fl("waste_trough"),
+         BioER_96_wellplate_Vb_2200uL("elution_plate"), AGenBio_1_troughplate_100000uL_Fl("wash1_trough"),
+         AGenBio_1_troughplate_100000uL_Fl("wash2_trough")]
+  for i, r in enumerate(r13):
+    m13[i].assign_child_resource(r)
+  m19 = {i: Hamilton_MFX_plateholder_DWP_metal_tapped(n) for i, n in enumerate(
+    ["rail19_binding_module", "rail19_source_module", "rail19_flowthrough_module", "rail19_binding_trough_module", "rail19_elution_trough_module"])}
+  lh.deck.assign_child_resource(MFX_CAR_L5_base("car_19", modules=m19), rails=19)
+  r19 = [BioER_96_wellplate_Vb_2200uL("binding_plate"), BioER_96_wellplate_Vb_2200uL("source_plate"),
+         BioER_96_wellplate_Vb_2200uL("flowthrough_plate"), AGenBio_1_troughplate_100000uL_Fl("binding_trough"),
+         AGenBio_1_troughplate_100000uL_Fl("elution_trough")]
+  for i, r in enumerate(r19):
+    m19[i].assign_child_resource(r)
+  res = {r.name: r for r in r13 + r19}
+  return res, racks
+
+
+def record(lh):
+  def wrap(method, kind, fn):
+    orig = getattr(lh, method)
+
+    async def inner(*a, **k):
+      EVENTS.append(fn(*a, **k) | {"kind": kind})
+      try:
+        return await orig(*a, **k)
+      except ValueError as e:  # PLR 0.2.2 cannot spread 8 channels in one trough; newer PLR can. Logged, not simulated.
+        if "space channels" not in str(e):
+          raise
+    setattr(lh, method, inner)
+
+  arg = lambda a, k, i, *keys: a[i] if len(a) > i else next((k[x] for x in keys if x in k), None)
+  wrap("pick_up_tips", "pick", lambda *a, **k: {"spots": names(arg(a, k, 0, "tip_spots") or [])})
+  wrap("drop_tips", "drop", lambda *a, **k: {"spots": names(arg(a, k, 0, "tip_spots") or [])})
+  wrap("return_tips", "drop", lambda *a, **k: {"spots": []})
+  wrap("aspirate", "aspirate", lambda *a, **k: {"wells": names(arg(a, k, 0, "resources") or []), "vols": list(arg(a, k, 1, "vols") or [])})
+  wrap("dispense", "dispense", lambda *a, **k: {"wells": names(arg(a, k, 0, "resources") or []), "vols": list(arg(a, k, 1, "vols") or [])})
+  wrap("move_plate", "move", lambda *a, **k: {"plate": getattr(arg(a, k, 0, "plate"), "name", "?"), "to": getattr(arg(a, k, 1, "to"), "name", "?")})
+  wrap("move_resource", "move", lambda *a, **k: {"plate": getattr(arg(a, k, 0, "resource"), "name", "?"), "to": getattr(arg(a, k, 1, "to"), "name", "?")})
+
+
+async def main(proto_dir, out):
+  sys.path.insert(0, proto_dir)
+  set_tip_tracking(False)
+  set_volume_tracking(False)
+  _print = builtins.print
+  builtins.input = lambda *a, **k: EVENTS.append({"kind": "comment", "text": "Operator pause: " + " ".join(map(str, a))[:140]}) or ""
+  _sleep = asyncio.sleep
+
+  async def no_wait(s, *a, **k):
+    if s >= 1:
+      EVENTS.append({"kind": "delay", "seconds": float(s)})
+    await _sleep(0)
+  asyncio.sleep = no_wait
+  time.sleep = lambda s: EVENTS.append({"kind": "delay", "seconds": float(s)})
+
+  backend = SimulatedSTAR(num_channels=8)
+  lh = LiquidHandler(backend=backend, deck=STARLetDeck(core_grippers="1000uL-at-waste"))
+  await lh.setup()
+  res, racks = build(lh)
+  record(lh)
+  import ape_96w_protocol as ape
+  ape.print = lambda *a, **k: EVENTS.append({"kind": "comment", "text": " ".join(map(str, a))[:160]})  # protocol's own messages only
+  ok, err = True, None
+  try:
+    await ape.run_ape_96w(lh, binding_plate=res["binding_plate"], source_plate=res["source_plate"],
+      flowthrough_plate=res["flowthrough_plate"], elution_plate=res["elution_plate"], mag_plate=res["mag_plate"],
+      binding_trough=res["binding_trough"], wash1_trough=res["wash1_trough"], wash2_trough=res["wash2_trough"],
+      waste_trough=res["waste_trough"], elution_trough=res["elution_trough"], tip_racks=racks,
+      lysate_volume=300, binding_buffer_volume=1500, bead_volume_note=50, wash_volume=900, elution_volume=120,
+      binding_mix_rounds=6, binding_incubation_seconds=30, wash_mix_rounds=1, wash1_cycles=1, magnet_settle_seconds=20,
+      wash_incubation_seconds=2, elution_incubation_seconds=120)
+  except Exception as e:
+    ok, err = False, f"{type(e).__name__}: {e}"[:400]
+  json.dump({"ok": ok, "error": err, "events": EVENTS}, open(out, "w"))
+  print("ok" if ok else "FAIL " + err, len(EVENTS), "events")
+
+
+if __name__ == "__main__":
+  asyncio.run(main(sys.argv[1], sys.argv[2]))

docker/sim/mylog.py

@@ -0,0 +1,13 @@
+"""eval/runlog.py, but keeping protocol comments as caption events."""
+import json, sys
+sys.path.insert(0, sys.argv[4])
+import runlog
+_parse = runlog.parse
+def parse(text, payload):
+    if text.startswith("COMMENT: "):
+        return {"kind": "comment", "text": text[9:], "instrument": "", "channels": 1}
+    return _parse(text, payload)
+runlog.parse = parse
+out = runlog.main(sys.argv[1], sys.argv[2])
+open(sys.argv[3], "w").write(json.dumps(out))
+print(sys.argv[1].split("results/")[-1], "ok" if out["ok"] else out["error"][:200], len(out.get("events", [])))

docker/sim/runlog.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""Simulate an OT-2 protocol and print its robot actions as JSON events."""
+import json
+import re
+import sys
+import traceback
+
+from opentrons import protocol_api, simulate
+
+WELL = r"([A-P]\d{1,2}) (?:of|in) (.+?)(?: at [\d.]+ uL/sec)?$"
+PATTERNS = [
+    ("aspirate", re.compile(r"^Aspirating ([\d.]+) uL from " + WELL)),
+    ("dispense", re.compile(r"^Dispensing ([\d.]+) uL into " + WELL)),
+    ("pick", re.compile(r"^Picking up tip")),
+    ("drop", re.compile(r"^(?:Dropping|Returning) tip")),
+    ("delay", re.compile(r"^Delaying for (\d+) minutes and ([\d.]+) seconds")),
+    ("engage", re.compile(r"^Engaging Magnetic Module")),
+    ("disengage", re.compile(r"^Disengaging Magnetic Module")),
+    ("temp", re.compile(r"^Setting Temperature Module temperature to ([\d.]+)")),
+    # Thermocycler module actions (lid, block temperature and hold times, profiles); kept as the simulator words them.
+    ("thermocycler", re.compile(r"^((?:Opening|Closing|Setting|Deactivating|Executing|Thermocycler).*[Tt]hermocycler.*)$")),
+]
+INSTRUMENT_KINDS = {"aspirate", "dispense", "pick", "drop"}
+
+_comment = protocol_api.ProtocolContext.comment
+LOADNAMES: dict[str, str] = {}   # str(labware) as it appears in the run log -> the load name it was created from
+
+
+def _key(labware):
+    """'<label or display name> on <slot>', whatever the API level: str(labware) drops the slot from 2.14 on, and the
+    run log says 'on slot 2' there and 'on 2' before."""
+    slot = re.sub(r"^slot ", "", str(labware.parent))
+    name = labware.name if labware.name != labware.load_name else re.sub(r" on .*$", "", str(labware))
+    return f"{name} on {slot}"
+
+
+def _recording(original):
+    def load_labware(self, load_name, *args, **kwargs):
+        labware = original(self, load_name, *args, **kwargs)
+        LOADNAMES[_key(labware)] = labware.load_name
+        return labware
+    return load_labware
+
+
+for _cls in (protocol_api.ProtocolContext, protocol_api.ModuleContext):
+    _cls.load_labware = _recording(_cls.load_labware)
+
+
+def tagged_comment(self, msg):
+    return _comment(self, f"COMMENT: {msg}")
+
+
+def parse(text: str, payload: dict) -> dict | None:
+    if text.startswith("COMMENT: "):
+        return None
+    for kind, pattern in PATTERNS:
+        match = pattern.search(text)
+        if not match:
+            continue
+        instrument = str(payload.get("instrument", ""))
+        if kind in INSTRUMENT_KINDS and not instrument:
+            return None
+        event = {"kind": kind, "instrument": instrument}
+        event["channels"] = 8 if "8-Channel" in instrument else 1
+        if kind in {"aspirate", "dispense"}:
+            event.update(volume=float(match[1]), well=match[2], labware=re.sub(r" on slot ", " on ", match[3]))
+        elif kind == "delay":
+            event["seconds"] = int(match[1]) * 60 + float(match[2])
+        elif kind == "temp":
+            event["celsius"] = float(match[1])
+        elif kind == "thermocycler":
+            event["text"] = match[1].strip()
+        return event
+    return None
+
+
+def main(protocol: str, labware_dir: str) -> dict:
+    protocol_api.ProtocolContext.comment = tagged_comment
+    try:
+        with open(protocol) as handle:
+            runlog, _ = simulate.simulate(
+                handle, protocol, custom_labware_paths=[labware_dir]
+            )
+    except Exception as exc:
+        return {"ok": False, "error": f"{type(exc).__name__}: {exc}"[-3000:],
+                "traceback": traceback.format_exc()[-3000:]}
+    events: list[dict] = []
+    pending_air = False
+
+    def walk(entries):
+        nonlocal pending_air
+        for entry in entries:
+            payload = entry.get("payload", {})
+            text = str(payload.get("text", ""))
+            if text.startswith("Air gap"):
+                pending_air = True
+            event = parse(text, payload)
+            if event:
+                if pending_air and event["kind"] == "aspirate":
+                    event["kind"] = "air"
+                    pending_air = False
+                events.append(event)
+            walk(entry.get("subcommands", []))
+
+    walk(runlog)
+    return {"ok": True, "events": events, "n_commands": len(runlog), "labware": dict(LOADNAMES)}
+
+
+if __name__ == "__main__":
+    result = json.dumps(main(sys.argv[1], sys.argv[2]))
+    if len(sys.argv) > 3:       # the grader names a file for the result; stdout is not trusted (a protocol can print)
+        with open(sys.argv[3], "w") as handle:
+            handle.write(result)
+    print(result)

docker/sim/simulate_all.py

@@ -0,0 +1,68 @@
+"""Inside the container: simulate every Opentrons protocol in /work/<candidate>/... and write run logs to /out.
+
+usage (entrypoint): simulate_all.py <candidate>[,<candidate>...] | all
+Each candidate folder is a cloned repo mounted read-only. For every .py with `def run` and an Opentrons
+marker, writes /out/<candidate>/<file-slug>.json (events + labware, or the error), using eval/runlog.py
+plus comment events (mylog.py). Custom labware JSON anywhere in the repo is passed to the simulator.
+"""
+import json, os, re, subprocess, sys
+
+WORK, OUT = "/work", "/out"
+MARK = re.compile(r"apiLevel|from opentrons|import opentrons")
+
+
+def labware_dirs(repo):
+    dirs = set()
+    for root, _, files in os.walk(repo):
+        if ".git" in root:
+            continue
+        for f in files:
+            if f.endswith(".json"):
+                try:
+                    if '"wellOrdering"' in open(os.path.join(root, f), errors="replace").read(200000):
+                        dirs.add(root)
+                except OSError:
+                    pass
+    return sorted(dirs)
+
+
+def main(sel):
+    names = sorted(os.listdir(WORK)) if sel == "all" else sel.split(",")
+    for name in names:
+        repo = os.path.join(WORK, name)
+        lw = labware_dirs(repo)
+        lwdir = "/tmp/lw_" + name
+        os.makedirs(lwdir, exist_ok=True)
+        for d in lw:  # runlog.py takes one labware dir: merge them
+            for f in os.listdir(d):
+                if f.endswith(".json"):
+                    src = os.path.join(d, f)
+                    try:
+                        open(os.path.join(lwdir, f), "w").write(open(src).read())
+                    except OSError:
+                        pass
+        for root, dirs, files in os.walk(repo):
+            dirs[:] = [d for d in dirs if d not in (".git", "node_modules", ".venv", "venv")]
+            for f in sorted(files):
+                if not f.endswith(".py"):
+                    continue
+                path = os.path.join(root, f)
+                text = open(path, errors="replace").read()
+                if "def run" not in text or not MARK.search(text):
+                    continue
+                rel = os.path.relpath(path, repo)
+                slug = re.sub(r"[^A-Za-z0-9]+", "_", rel)[:120]
+                os.makedirs(os.path.join(OUT, name), exist_ok=True)
+                out = os.path.join(OUT, name, slug + ".json")
+                try:
+                    p = subprocess.run([sys.executable, "/t2w/mylog.py", path, lwdir, out, "/t2w"],
+                                       capture_output=True, text=True, timeout=180, cwd="/tmp")
+                    ok = os.path.exists(out) and json.load(open(out)).get("ok")
+                except subprocess.TimeoutExpired:
+                    json.dump({"ok": False, "error": "timed out after 180 s"}, open(out, "w"))
+                    ok = False
+                print(f"{name}/{rel}: {'ok' if ok else 'FAIL'}", flush=True)
+
+
+if __name__ == "__main__":
+    main(sys.argv[1] if len(sys.argv) > 1 else "all")