Add thermocycling capability and Opentrons vendor module

pylabrobot/capabilities/thermocycling/__init__.py

@@ -0,0 +1,3 @@
+from .backend import ThermocyclingBackend
+from .standard import Protocol, Stage, Step
+from .thermocycling import ThermocyclingCapability

pylabrobot/capabilities/thermocycling/backend.py

@@ -0,0 +1,55 @@
+from abc import ABCMeta, abstractmethod
+
+from pylabrobot.device import DeviceBackend
+
+from .standard import Protocol
+
+
+class ThermocyclingBackend(DeviceBackend, metaclass=ABCMeta):
+  """Abstract backend for thermocyclers.
+
+  Only thermocycling-specific operations live here: lid control, protocol
+  execution, and profile progress queries. Block/lid temperature control
+  is handled by separate TemperatureControllerBackend instances.
+  """
+
+  @abstractmethod
+  async def open_lid(self) -> None:
+    """Open the thermocycler lid."""
+
+  @abstractmethod
+  async def close_lid(self) -> None:
+    """Close the thermocycler lid."""
+
+  @abstractmethod
+  async def get_lid_open(self) -> bool:
+    """Return True if the lid is open."""
+
+  @abstractmethod
+  async def run_protocol(self, protocol: Protocol, block_max_volume: float) -> None:
+    """Execute a thermocycler protocol.
+
+    Args:
+      protocol: Protocol containing stages with steps and repeats.
+      block_max_volume: Maximum block volume in uL.
+    """
+
+  @abstractmethod
+  async def get_hold_time(self) -> float:
+    """Get remaining hold time in seconds."""
+
+  @abstractmethod
+  async def get_current_cycle_index(self) -> int:
+    """Get the zero-based index of the current cycle."""
+
+  @abstractmethod
+  async def get_total_cycle_count(self) -> int:
+    """Get the total cycle count."""
+
+  @abstractmethod
+  async def get_current_step_index(self) -> int:
+    """Get the zero-based index of the current step within the cycle."""
+
+  @abstractmethod
+  async def get_total_step_count(self) -> int:
+    """Get the total number of steps in the current cycle."""

pylabrobot/capabilities/thermocycling/standard.py

@@ -0,0 +1,65 @@
+import enum
+from dataclasses import dataclass
+from typing import List, Optional
+
+from pylabrobot.serializer import SerializableMixin
+
+
+@dataclass
+class Step(SerializableMixin):
+  """A single step in a thermocycler profile."""
+
+  temperature: List[float]
+  hold_seconds: float
+  rate: Optional[float] = None  # degrees Celsius per second
+
+  def serialize(self) -> dict:
+    return {
+      "temperature": self.temperature,
+      "hold_seconds": self.hold_seconds,
+      "rate": self.rate,
+    }
+
+
+@dataclass
+class Stage(SerializableMixin):
+  """A stage in a thermocycler protocol: a list of steps repeated N times."""
+
+  steps: List[Step]
+  repeats: int
+
+  def serialize(self) -> dict:
+    return {
+      "steps": [step.serialize() for step in self.steps],
+      "repeats": self.repeats,
+    }
+
+
+@dataclass
+class Protocol(SerializableMixin):
+  """A thermocycler protocol: a list of stages."""
+
+  stages: List[Stage]
+
+  def serialize(self) -> dict:
+    return {
+      "stages": [stage.serialize() for stage in self.stages],
+    }
+
+  @classmethod
+  def deserialize(cls, data: dict) -> "Protocol":
+    stages = []
+    for stage_data in data.get("stages", []):
+      steps = [Step(**step) for step in stage_data["steps"]]
+      stages.append(Stage(steps=steps, repeats=stage_data.get("repeats", 1)))
+    return cls(stages=stages)
+
+
+class LidStatus(enum.Enum):
+  IDLE = "idle"
+  HOLDING_AT_TARGET = "holding at target"
+
+
+class BlockStatus(enum.Enum):
+  IDLE = "idle"
+  HOLDING_AT_TARGET = "holding at target"

pylabrobot/capabilities/thermocycling/thermocycling.py

@@ -0,0 +1,152 @@
+import asyncio
+from typing import List, Optional
+
+from pylabrobot.capabilities.capability import Capability
+from pylabrobot.capabilities.temperature_controlling import TemperatureControlCapability
+
+from .backend import ThermocyclingBackend
+from .standard import Protocol, Stage, Step
+
+
+class ThermocyclingCapability(Capability):
+  """Thermocycling capability.
+
+  Owns protocol execution (delegated to the backend) and lid control.
+  Block and lid temperature control for ad-hoc use (outside of protocol runs)
+  is accessed via the `block` and `lid` TemperatureControlCapability instances.
+  """
+
+  def __init__(
+    self,
+    backend: ThermocyclingBackend,
+    block: TemperatureControlCapability,
+    lid: TemperatureControlCapability,
+  ):
+    super().__init__(backend=backend)
+    self.backend: ThermocyclingBackend = backend
+    self.block = block
+    self.lid = lid
+
+  async def open_lid(self) -> None:
+    await self.backend.open_lid()
+
+  async def close_lid(self) -> None:
+    await self.backend.close_lid()
+
+  async def get_lid_open(self) -> bool:
+    return await self.backend.get_lid_open()
+
+  async def run_protocol(self, protocol: Protocol, block_max_volume: float) -> None:
+    """Execute a thermocycler protocol.
+
+    Args:
+      protocol: Protocol containing stages with steps and repeats.
+      block_max_volume: Maximum block volume in uL.
+    """
+    num_zones = len(protocol.stages[0].steps[0].temperature)
+    for stage in protocol.stages:
+      for i, step in enumerate(stage.steps):
+        if len(step.temperature) != num_zones:
+          raise ValueError(
+            f"All steps must have the same number of temperatures. "
+            f"Expected {num_zones}, got {len(step.temperature)} in step {i}."
+          )
+
+    await self.backend.run_protocol(protocol, block_max_volume)
+
+  async def run_pcr_profile(
+    self,
+    denaturation_temp: List[float],
+    denaturation_time: float,
+    annealing_temp: List[float],
+    annealing_time: float,
+    extension_temp: List[float],
+    extension_time: float,
+    num_cycles: int,
+    block_max_volume: float,
+    lid_temperature: float,
+    pre_denaturation_temp: Optional[List[float]] = None,
+    pre_denaturation_time: Optional[float] = None,
+    final_extension_temp: Optional[List[float]] = None,
+    final_extension_time: Optional[float] = None,
+    storage_temp: Optional[List[float]] = None,
+    storage_time: Optional[float] = None,
+  ) -> None:
+    """Build and run a standard PCR profile.
+
+    Sets the lid temperature first, waits for it, then runs the protocol.
+    """
+    await self.lid.set_temperature(lid_temperature)
+    await self.lid.wait_for_temperature()
+
+    stages: List[Stage] = []
+
+    if pre_denaturation_temp is not None and pre_denaturation_time is not None:
+      stages.append(
+        Stage(
+          steps=[Step(temperature=pre_denaturation_temp, hold_seconds=pre_denaturation_time)],
+          repeats=1,
+        )
+      )
+
+    stages.append(
+      Stage(
+        steps=[
+          Step(temperature=denaturation_temp, hold_seconds=denaturation_time),
+          Step(temperature=annealing_temp, hold_seconds=annealing_time),
+          Step(temperature=extension_temp, hold_seconds=extension_time),
+        ],
+        repeats=num_cycles,
+      )
+    )
+
+    if final_extension_temp is not None and final_extension_time is not None:
+      stages.append(
+        Stage(
+          steps=[Step(temperature=final_extension_temp, hold_seconds=final_extension_time)],
+          repeats=1,
+        )
+      )
+
+    if storage_temp is not None and storage_time is not None:
+      stages.append(
+        Stage(steps=[Step(temperature=storage_temp, hold_seconds=storage_time)], repeats=1)
+      )
+
+    await self.run_protocol(protocol=Protocol(stages=stages), block_max_volume=block_max_volume)
+
+  async def get_hold_time(self) -> float:
+    return await self.backend.get_hold_time()
+
+  async def get_current_cycle_index(self) -> int:
+    return await self.backend.get_current_cycle_index()
+
+  async def get_total_cycle_count(self) -> int:
+    return await self.backend.get_total_cycle_count()
+
+  async def get_current_step_index(self) -> int:
+    return await self.backend.get_current_step_index()
+
+  async def get_total_step_count(self) -> int:
+    return await self.backend.get_total_step_count()
+
+  async def is_profile_running(self) -> bool:
+    """Return True if a protocol is still in progress."""
+    hold = await self.backend.get_hold_time()
+    cycle = await self.backend.get_current_cycle_index()
+    total_cycles = await self.backend.get_total_cycle_count()
+    step = await self.backend.get_current_step_index()
+    total_steps = await self.backend.get_total_step_count()
+
+    if hold and hold > 0:
+      return True
+    if cycle < total_cycles - 1:
+      return True
+    if cycle == total_cycles - 1 and step < total_steps - 1:
+      return True
+    return False
+
+  async def wait_for_profile_completion(self, poll_interval: float = 60.0) -> None:
+    """Block until the protocol finishes."""
+    while await self.is_profile_running():
+      await asyncio.sleep(poll_interval)

pylabrobot/inheco/__init__.py

@@ -1,5 +1,6 @@
 from .control_box import InhecoTECControlBox
 from .cpac import InhecoCPAC, InhecoCPACBackend, inheco_cpac_ultraflat
+from .odtc import ODTC, ODTCBlockBackend, ODTCDriver, ODTCLidBackend, ODTCThermocyclingBackend
 from .thermoshake import (
   InhecoThermoShake,
   InhecoThermoshakeBackend,

pylabrobot/inheco/odtc/__init__.py

@@ -0,0 +1,7 @@
+from .odtc import (
+  ODTC,
+  ODTCBlockBackend,
+  ODTCDriver,
+  ODTCLidBackend,
+  ODTCThermocyclingBackend,
+)