different plate types on cytation5 (#403)

Author: rickwierenga

pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py

@@ -5,9 +5,7 @@
 from pylabrobot.liquid_handling import LiquidHandler
 from pylabrobot.liquid_handling.standard import GripDirection, Pickup
 from pylabrobot.plate_reading import PlateReader
-from pylabrobot.plate_reading.plate_reader_tests import (
-  MockPlateReaderBackend,
-)
+from pylabrobot.plate_reading.chatterbox import PlateReaderChatterboxBackend
 from pylabrobot.resources import (
   HT,
   HTF,
@@ -696,7 +694,7 @@ async def test_iswap(self):
   async def test_iswap_plate_reader(self):
     plate_reader = PlateReader(
       name="plate_reader",
-      backend=MockPlateReaderBackend(),
+      backend=PlateReaderChatterboxBackend(),
       size_x=0,
       size_y=0,
       size_z=0,

pylabrobot/plate_reading/backend.py

@@ -5,6 +5,7 @@
 
 from pylabrobot.machines.backends import MachineBackend
 from pylabrobot.plate_reading.standard import Exposure, FocalPosition, Gain, ImagingMode
+from pylabrobot.resources.plate import Plate
 
 
 class PlateReaderBackend(MachineBackend, metaclass=ABCMeta):
@@ -28,18 +29,19 @@ async def close(self) -> None:
     """Close the plate reader. Also known as plate in."""
 
   @abstractmethod
-  async def read_luminescence(self, focal_height: float) -> List[List[float]]:
+  async def read_luminescence(self, plate: Plate, focal_height: float) -> List[List[float]]:
     """Read the luminescence from the plate reader. This should return a list of lists, where the
     outer list is the columns of the plate and the inner list is the rows of the plate."""
 
   @abstractmethod
-  async def read_absorbance(self, wavelength: int) -> List[List[float]]:
+  async def read_absorbance(self, plate: Plate, wavelength: int) -> List[List[float]]:
     """Read the absorbance from the plate reader. This should return a list of lists, where the
     outer list is the columns of the plate and the inner list is the rows of the plate."""
 
   @abstractmethod
   async def read_fluorescence(
     self,
+    plate: Plate,
     excitation_wavelength: int,
     emission_wavelength: int,
     focal_height: float,
@@ -58,6 +60,7 @@ async def capture(
     exposure_time: Exposure,
     focal_height: FocalPosition,
     gain: Gain,
+    plate: Plate,
   ) -> List[List[float]]:
     """Capture an image of the plate in the specified mode."""
 

pylabrobot/plate_reading/biotek_backend.py

@@ -4,6 +4,8 @@
 import time
 from typing import List, Literal, Optional
 
+from pylabrobot.resources.plate import Plate
+
 try:
   from pylibftdi import Device
 
@@ -126,6 +128,7 @@ def __init__(self, timeout: float = 20, camera_serial_number: Optional[float] =
     self.camera_serial_number = camera_serial_number
     self.max_image_read_attempts = 8
 
+    self._plate: Optional[Plate] = None
     self._exposure: Optional[Exposure] = None
     self._focal_height: Optional[FocalPosition] = None
     self._gain: Optional[Gain] = None
@@ -137,10 +140,16 @@ async def setup(self, use_cam: bool = False) -> None:
     logger.info("[cytation5] setting up")
 
     self.dev.open()
-    # self.dev.baudrate = 9600 # worked in the past
-    self.dev.baudrate = 38400
+    self.dev.ftdi_fn.ftdi_usb_reset()
+    self.dev.ftdi_fn.ftdi_set_latency_timer(16)  # 0x10
+
+    self.dev.baudrate = 9600  # 0x38 0x41
+    # self.dev.baudrate = 38400
     self.dev.ftdi_fn.ftdi_set_line_property(8, 2, 0)  # 8 bits, 2 stop bits, no parity
     SIO_RTS_CTS_HS = 0x1 << 8
+    self.dev.ftdi_fn.ftdi_setdtr(1)
+    self.dev.ftdi_fn.ftdi_setrts(1)
+
     self.dev.ftdi_fn.ftdi_setflowctrl(SIO_RTS_CTS_HS)
     self.dev.ftdi_fn.ftdi_setrts(1)
 
@@ -265,31 +274,36 @@ async def _read_until(self, char: bytes, timeout: Optional[float] = None) -> byt
     return res
 
   async def send_command(
-    self, command: str, parameter: Optional[str] = None, wait_for_response=True
+    self,
+    command: str,
+    parameter: Optional[str] = None,
+    wait_for_response=True,
+    timeout: Optional[float] = None,
   ) -> Optional[bytes]:
     await self._purge_buffers()
     self.dev.write(command.encode())
     logger.debug("[cytation5] sent %s", command)
     response: Optional[bytes] = None
     if wait_for_response or parameter is not None:
-      # print("reading until", b"\x06" if parameter is not None else b"\x03")
-      response = await self._read_until(b"\x06" if parameter is not None else b"\x03")
+      response = await self._read_until(
+        b"\x06" if parameter is not None else b"\x03", timeout=timeout
+      )
 
     if parameter is not None:
       self.dev.write(parameter.encode())
       logger.debug("[cytation5] sent %s", parameter)
       if wait_for_response:
-        response = await self._read_until(b"\x03")
+        response = await self._read_until(b"\x03", timeout=timeout)
 
     return response
 
   async def get_serial_number(self) -> str:
-    resp = await self.send_command("C")
+    resp = await self.send_command("C", timeout=1)
     assert resp is not None
     return resp[1:].split(b" ")[0].decode()
 
   async def get_firmware_version(self) -> str:
-    resp = await self.send_command("e")
+    resp = await self.send_command("e", timeout=1)
     assert resp is not None
     return " ".join(resp[1:-1].decode().split(" ")[0:4])
 
@@ -323,11 +337,62 @@ def _parse_body(self, body: bytes) -> List[List[float]]:
         parsed_data[row_idx].append(value)
     return parsed_data
 
-  async def read_absorbance(self, wavelength: int) -> List[List[float]]:
+  async def set_plate(self, plate: Plate):
+    """
+    08120112207434014351135308559127881422
+                                      ^^^^ plate size z
+                                ^^^^^ plate size x
+                            ^^^^^ plate size y
+                      ^^^^^ bottom right x
+                  ^^^^^ top left x
+            ^^^^^ bottom right y
+        ^^^^^ top left y
+      ^^ columns
+    ^^ rows
+    """
+
+    if plate is self._plate:
+      return
+
+    rows = plate.num_items_y
+    columns = plate.num_items_x
+
+    bottom_right_well = plate.get_item(plate.num_items - 1)
+    assert bottom_right_well.location is not None
+    bottom_right_well_center = bottom_right_well.location + bottom_right_well.get_anchor(
+      x="c", y="c"
+    )
+    top_left_well = plate.get_item(0)
+    assert top_left_well.location is not None
+    top_left_well_center = top_left_well.location + top_left_well.get_anchor(x="c", y="c")
+
+    plate_size_y = plate.get_size_y()
+    plate_size_x = plate.get_size_x()
+    plate_size_z = plate.get_size_z()
+
+    top_left_well_center_y = plate.get_size_y() - top_left_well_center.y  # invert y axis
+    bottom_right_well_center_y = plate.get_size_y() - bottom_right_well_center.y  # invert y axis
+
+    cmd = (
+      f"{rows:02}"
+      f"{columns:02}"
+      f"{int(top_left_well_center_y*100):05}"
+      f"{int(bottom_right_well_center_y*100):05}"
+      f"{int(top_left_well_center.x*100):05}"
+      f"{int(bottom_right_well_center.x*100):05}"
+      f"{int(plate_size_y*100):05}"
+      f"{int(plate_size_x*100):05}"
+      f"{int(plate_size_z*100):04}"
+      "\x03"
+    )
+
+    return await self.send_command("y", cmd)
+
+  async def read_absorbance(self, plate: Plate, wavelength: int) -> List[List[float]]:
     if not 230 <= wavelength <= 999:
       raise ValueError("Wavelength must be between 230 and 999")
 
-    await self.send_command("y", "08120112207434014351135308559127881772\x03")
+    await self.set_plate(plate)
 
     wavelength_str = str(wavelength).zfill(4)
     cmd = f"00470101010812000120010000110010000010600008{wavelength_str}1"
@@ -343,14 +408,14 @@ async def read_absorbance(self, wavelength: int) -> List[List[float]]:
     assert resp is not None
     return self._parse_body(body)
 
-  async def read_luminescence(self, focal_height: float) -> List[List[float]]:
+  async def read_luminescence(self, plate: Plate, focal_height: float) -> List[List[float]]:
     if not 4.5 <= focal_height <= 13.88:
       raise ValueError("Focal height must be between 4.5 and 13.88")
 
     cmd = f"3{14220 + int(1000*focal_height)}\x03"
     await self.send_command("t", cmd)
 
-    await self.send_command("y", "08120112207434014351135308559127881772\x03")
+    await self.set_plate(plate)
 
     cmd = "008401010108120001200100001100100000123000500200200-001000-00300000000000000000001351092"
     await self.send_command("D", cmd)
@@ -364,6 +429,7 @@ async def read_luminescence(self, focal_height: float) -> List[List[float]]:
 
   async def read_fluorescence(
     self,
+    plate: Plate,
     excitation_wavelength: int,
     emission_wavelength: int,
     focal_height: float,
@@ -378,7 +444,7 @@ async def read_fluorescence(
     cmd = f"{614220 + int(1000*focal_height)}\x03"
     await self.send_command("t", cmd)
 
-    await self.send_command("y", "08120112207434014351135308559127881772\x03")
+    await self.set_plate(plate)
 
     excitation_wavelength_str = str(excitation_wavelength).zfill(4)
     emission_wavelength_str = str(emission_wavelength).zfill(4)
@@ -512,14 +578,14 @@ async def led_off(self):
   async def set_focus(self, focal_position: FocalPosition):
     """focus position in mm"""
 
-    if focal_position == self._focal_height:
-      logger.debug("Focus position is already set to %s", focal_position)
-      return
-
     if focal_position == "auto":
       await self.auto_focus()
       return
 
+    if focal_position == self._focal_height:
+      logger.debug("Focus position is already set to %s", focal_position)
+      return
+
     # There is a difference between the number in the program and the number sent to the machine,
     # which is modelled using the following linear relation. R^2=0.999999999
     # convert from mm to um
@@ -534,6 +600,9 @@ async def set_focus(self, focal_position: FocalPosition):
     self._focal_height = focal_position
 
   async def auto_focus(self, timeout: float = 30):
+    plate = self._plate
+    if plate is None:
+      raise RuntimeError("Plate not set. Run set_plate() first.")
     imaging_mode = self._imaging_mode
     if imaging_mode is None:
       raise RuntimeError("Imaging mode not set. Run set_imaging_mode() first.")
@@ -557,6 +626,7 @@ async def auto_focus(self, timeout: float = 30):
     # objective function: variance of laplacian
     async def evaluate_focus(focus_value):
       image = await self.capture(
+        plate=plate,
         row=row,
         column=column,
         mode=imaging_mode,
@@ -769,6 +839,7 @@ async def capture(
     exposure_time: Exposure,
     focal_height: FocalPosition,
     gain: Gain,
+    plate: Plate,
     color_processing_algorithm: int = SPINNAKER_COLOR_PROCESSING_ALGORITHM_HQ_LINEAR,
     pixel_format: int = PixelFormat_Mono8,
   ) -> List[List[float]]:
@@ -788,6 +859,8 @@ async def capture(
     if self.cam is None:
       raise ValueError("Camera not initialized. Run setup(use_cam=True) first.")
 
+    await self.set_plate(plate)
+
     await self.select(row, column)
     await self.set_imaging_mode(mode)
     await self.set_exposure(exposure_time)

pylabrobot/plate_reading/biotek_tests.py

@@ -3,6 +3,7 @@
 from typing import Iterator
 
 from pylabrobot.plate_reading.biotek_backend import Cytation5Backend
+from pylabrobot.resources import CellVis_24_wellplate_3600uL_Fb
 
 
 def _byte_iter(s: str) -> Iterator[bytes]:
@@ -18,11 +19,12 @@ async def asyncSetUp(self):
     self.backend.dev = unittest.mock.MagicMock()
     self.backend.dev.open.return_value = 0
     self.backend.dev.write.return_value = 0
+    self.plate = CellVis_24_wellplate_3600uL_Fb(name="plate")
 
   async def test_setup(self):
     await self.backend.setup()
     assert self.backend.dev.open.called
-    assert self.backend.dev.baudrate == 38400
+    assert self.backend.dev.baudrate == 9600
     self.backend.dev.ftdi_fn.ftdi_set_line_property.assert_called_with(8, 2, 0)
     self.backend.dev.ftdi_fn.ftdi_setflowctrl.assert_called_with(0x100)
     self.backend.dev.ftdi_fn.ftdi_setrts.assert_called_with(1)
@@ -77,10 +79,10 @@ async def test_read_absorbance(self):
       )
     )
 
-    resp = await self.backend.read_absorbance(wavelength=580)
+    resp = await self.backend.read_absorbance(plate=self.plate, wavelength=580)
 
     self.backend.dev.write.assert_any_call(b"y")
-    self.backend.dev.write.assert_any_call(b"08120112207434014351135308559127881772\x03")
+    self.backend.dev.write.assert_any_call(b"04060136807158017051135508525127501610\x03")
     self.backend.dev.write.assert_any_call(b"D")
     self.backend.dev.write.assert_any_call(
       b"004701010108120001200100001100100000106000080580113\x03"
@@ -233,6 +235,7 @@ async def test_read_fluorescence(self):
     )
 
     resp = await self.backend.read_fluorescence(
+      plate=self.plate,
       excitation_wavelength=485,
       emission_wavelength=528,
       focal_height=7.5,
@@ -241,7 +244,7 @@ async def test_read_fluorescence(self):
     self.backend.dev.write.assert_any_call(b"t")
     self.backend.dev.write.assert_any_call(b"621720\x03")
     self.backend.dev.write.assert_any_call(b"y")
-    self.backend.dev.write.assert_any_call(b"08120112207434014351135308559127881772\x03")
+    self.backend.dev.write.assert_any_call(b"04060136807158017051135508525127501610\x03")
     self.backend.dev.write.assert_any_call(b"D")
     self.backend.dev.write.assert_any_call(
       b"0084010101081200012001000011001000001350001002002000485000052800000000000000000021001119"

pylabrobot/plate_reading/chatterbox.py

@@ -1,6 +1,7 @@
 from typing import List
 
 from pylabrobot.plate_reading.backend import PlateReaderBackend
+from pylabrobot.resources.plate import Plate
 
 
 class PlateReaderChatterboxBackend(PlateReaderBackend):
@@ -24,16 +25,17 @@ async def open(self) -> None:
   async def close(self) -> None:
     print("Closing the plate reader.")
 
-  async def read_luminescence(self, focal_height: float) -> List[List[float]]:
+  async def read_luminescence(self, plate: Plate, focal_height: float) -> List[List[float]]:
     print(f"Reading luminescence at focal height {focal_height}.")
     return self.dummy_luminescence
 
-  async def read_absorbance(self, wavelength: int) -> List[List[float]]:
+  async def read_absorbance(self, plate: Plate, wavelength: int) -> List[List[float]]:
     print(f"Reading absorbance at wavelength {wavelength}.")
     return self.dummy_absorbance
 
   async def read_fluorescence(
     self,
+    plate: Plate,
     excitation_wavelength: int,
     emission_wavelength: int,
     focal_height: float,