Control Opentrons temperature module directly with USB (#735)

Author: nedru004

docs/user_guide/01_material-handling/temperature-controllers/ot-temperature-controller.ipynb

@@ -8,7 +8,7 @@
     "\n",
     "| Summary                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       | Photo                                            |\n",
     "|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------------|\n",
-    "| - [OEM Link](https://opentrons.com/products/temperature-module-gen2?sku=991-00350-0)<br>- **Communication Protocol / Hardware**: ? / USB-A (currently only supported via direct connection into OT-2)<br>- **Communication Level**: ?<br><br>- **OEM version**: GEN2<br>- **Temperature range**: 4 to 95°C<br> | ![quadrants](img/ot_temperature_module_gen2.webp) |\n",
+    "| - [OEM Link](https://opentrons.com/products/temperature-module-gen2?sku=991-00350-0)<br>- **Communication Protocol / Hardware**: ? / USB-A <br>- **Communication Level**: ?<br><br>- **OEM version**: GEN2<br>- **Temperature range**: 4 to 95°C<br> | ![quadrants](img/ot_temperature_module_gen2.webp) |\n",
     "\n"
    ]
   },
@@ -19,7 +19,7 @@
     "---\n",
     "## Setup Instructions (Physical)\n",
     "\n",
-    "WIP"
+    "Connect with USB to opentrons or to computer running pylabrobot"
    ]
   },
   {
@@ -28,7 +28,9 @@
    "source": [
     "---\n",
     "## Setup Instructions (Programmatic)\n",
-    "\n"
+    "\n",
+    "\n",
+    "### Setup with Opentrons"
    ]
   },
   {
@@ -164,6 +166,34 @@
     "lh.deck.assign_child_at_slot(t, slot=3)"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "---\n",
+    "\n",
+    "### Setup with Com Port"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "To communicate with the temperature module, find the COM port for the USB connection. Use that to replace `COM#` when setting up the temperature module in your code."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from pylabrobot.temperature_controlling.opentrons import OpentronsTemperatureModuleV2\n",
+    "\n",
+    "tc = OpentronsTemperatureModuleV2(name='tc', opentrons_id=None, serial_port=\"OM#\")\n",
+    "await tc.setup()"
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {},
@@ -329,7 +359,7 @@
  ],
  "metadata": {
   "kernelspec": {
-   "display_name": "env",
+   "display_name": "Python 3 (ipykernel)",
    "language": "python",
    "name": "python3"
   },
@@ -343,9 +373,9 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.10.12"
+   "version": "3.12.9"
   }
  },
  "nbformat": 4,
- "nbformat_minor": 2
+ "nbformat_minor": 4
 }

pylabrobot/temperature_controlling/opentrons.py

@@ -2,9 +2,13 @@
 
 from pylabrobot.resources import Coordinate, ItemizedResource
 from pylabrobot.resources.opentrons.module import OTModule
+from pylabrobot.temperature_controlling.backend import TemperatureControllerBackend
 from pylabrobot.temperature_controlling.opentrons_backend import (
   OpentronsTemperatureModuleBackend,
 )
+from pylabrobot.temperature_controlling.opentrons_backend_usb import (
+  OpentronsTemperatureModuleUSBBackend,
+)
 from pylabrobot.temperature_controlling.temperature_controller import (
   TemperatureController,
 )
@@ -20,32 +24,47 @@ class OpentronsTemperatureModuleV2(TemperatureController, OTModule):
   def __init__(
     self,
     name: str,
-    opentrons_id: str,
-    child_location: Coordinate = Coordinate.zero(),
+    opentrons_id: Optional[str] = None,
+    serial_port: Optional[str] = None,
+    child_location: Coordinate = Coordinate(
+      0, 0, 80.1
+    ),  # dimensional drawing from OT (x and y are not changed wrt parent)
     child: Optional[ItemizedResource] = None,
   ):
     """Create a new Opentrons temperature module v2.
 
     Args:
       name: Name of the temperature module.
       opentrons_id: Opentrons ID of the temperature module. Get it from
-        `OpentronsBackend(host="x.x.x.x", port=31950).list_connected_modules()`.
+        `OpentronsBackend(host="x.x.x.x", port=31950).list_connected_modules()`. Exactly one of `opentrons_id` or `serial_port` must be provided.
+      serial_port: Serial port for USB communication. Exactly one of `opentrons_id` or `serial_port` must be provided.
       child: Optional child resource like a tube rack or well plate to use on the
         temperature controller module.
     """
 
+    if opentrons_id is None and serial_port is None:
+      raise ValueError("Exactly one of `opentrons_id` or `serial_port` must be provided.")
+    if opentrons_id is not None and serial_port is not None:
+      raise ValueError("Exactly one of `opentrons_id` or `serial_port` must be provided.")
+
+    backend: TemperatureControllerBackend
+    if serial_port is not None:
+      backend = OpentronsTemperatureModuleUSBBackend(port=serial_port)
+    else:
+      assert opentrons_id is not None
+      backend = OpentronsTemperatureModuleBackend(opentrons_id=opentrons_id)
+
     super().__init__(
       name=name,
       size_x=193.5,
       size_y=89.2,
       size_z=84.0,  # height without any aluminum block
       child_location=child_location,
-      backend=OpentronsTemperatureModuleBackend(opentrons_id=opentrons_id),
+      backend=backend,
       category="temperature_controller",
       model="temperatureModuleV2",  # Must match OT moduleModel in list_connected_modules()
     )
 
-    self.backend = OpentronsTemperatureModuleBackend(opentrons_id=opentrons_id)
     self.child = child
 
     if child is not None:

pylabrobot/temperature_controlling/opentrons_backend_usb.py

@@ -0,0 +1,84 @@
+from typing import Optional
+
+from pylabrobot.io.serial import Serial
+from pylabrobot.temperature_controlling.backend import (
+  TemperatureControllerBackend,
+)
+
+
+class OpentronsTemperatureModuleUSBBackend(TemperatureControllerBackend):
+  """Opentrons temperature module backend."""
+
+  @property
+  def supports_active_cooling(self) -> bool:
+    return True
+
+  def __init__(self, port: str):
+    """Create a new Opentrons temperature module backend.
+
+    Args:
+      port: Serial port for USB communication.
+    """
+
+    self.port = port
+    self._serial: Optional["Serial"] = None
+
+  @property
+  def serial(self) -> "Serial":
+    if self._serial is None:
+      raise RuntimeError("Serial device not initialized. Call setup() first.")
+    return self._serial
+
+  async def setup(self):
+    # Setup serial communication for USB
+    self._serial = Serial(port=self.port, baudrate=115200, timeout=3)
+    await self._serial.setup()
+
+  async def stop(self):
+    await self.deactivate()
+    if self._serial is not None:
+      await self._serial.stop()
+      self._serial = None
+
+  def serialize(self) -> dict:
+    return {**super().serialize(), "port": self.port}
+
+  async def set_temperature(self, temperature: float):
+    tmp_message = f"M104 S{temperature}\r\n"
+    await self.serial.write(tmp_message.encode("utf-8"))
+    # Read response (should be "ok\r\nok\r\n")
+    response1 = await self.serial.readline()
+    response2 = await self.serial.readline()
+    # Verify we got the expected response
+    if b"ok" not in response1 or b"ok" not in response2:
+      raise RuntimeError(
+        f"Unexpected response from device: {response1.decode(encoding='utf-8')} {response2.decode(encoding='utf-8')}"
+      )
+
+  async def deactivate(self):
+    await self.serial.write(b"M18\r\n")
+    # Read response (should be "ok\r\nok\r\n")
+    response1 = await self.serial.readline()
+    response2 = await self.serial.readline()
+    # Verify we got the expected response
+    if b"ok" not in response1 or b"ok" not in response2:
+      raise RuntimeError(
+        f"Unexpected response from device: {response1.decode(encoding='utf-8')} {response2.decode(encoding='utf-8')}"
+      )
+
+  async def get_current_temperature(self) -> float:
+    await self.serial.write(b"M105\r\n")
+    # Read response (should be "T:XX.XXX C:XX.XXX\r\nok\r\nok\r\n")
+    response = await self.serial.readline()
+    # Verify we got the expected response
+    # Read response (should be "ok\r\nok\r\n")
+    if b"C" not in response:
+      raise ValueError(f"Unexpected response from device: {response.decode(encoding='utf-8')}")
+
+    response1 = await self.serial.readline()
+    response2 = await self.serial.readline()
+    if b"ok" not in response1 or b"ok" not in response2:
+      raise RuntimeError(
+        f"Unexpected response from device: {response1.decode(encoding='utf-8')} {response2.decode(encoding='utf-8')}"
+      )
+    return float(response.strip().split(b"C:")[-1])