add io layer (#392)

Author: rickwierenga

docs/user_guide/index.md

@@ -20,6 +20,7 @@ hamilton-star/hamilton-star
 moving-channels-around
 tip-spot-generators
 96head
+validation
 ```
 
 ```{toctree}

docs/user_guide/validation.ipynb

@@ -1,12 +1,11 @@
 import datetime
 import logging
-import sys
-import warnings
 from pathlib import Path
 from typing import Optional, Union
 
 from pylabrobot.__version__ import __version__
 from pylabrobot.config import Config, load_config
+from pylabrobot.io import end_validation, start_capture, stop_capture, validate
 
 CONFIG_FILE_NAME = "pylabrobot"
 

pylabrobot/__init__.py

@@ -2,7 +2,7 @@
 
 from abc import ABCMeta, abstractmethod
 
-from pylabrobot.machines.backends import MachineBackend
+from pylabrobot.machines.backend import MachineBackend
 
 
 class CentrifugeBackend(MachineBackend, metaclass=ABCMeta):

pylabrobot/centrifuge/backend.py

@@ -3,18 +3,12 @@
 import time
 from typing import Optional, Union
 
+from pylabrobot.io.ftdi import FTDI
+
 from .backend import CentrifugeBackend, LoaderBackend
 from .standard import LoaderNoPlateError
 
-try:
-  from pylibftdi import Device
-
-  USE_FTDI = True
-except ImportError:
-  USE_FTDI = False
-
-
-logger = logging.getLogger("pylabrobot.centrifuge.vspin")
+logger = logging.getLogger(__name__)
 
 
 class Access2Backend(LoaderBackend):
@@ -28,15 +22,15 @@ def __init__(
       device_id: The libftdi id for the loader. Find using
         `python3 -m pylibftdi.examples.list_devices`
     """
-    self.dev = Device(lazy_open=True, device_id=device_id)
+    self.io = FTDI(device_id=device_id)
     self.timeout = timeout
 
   async def _read(self) -> bytes:
     x = b""
     r = None
     start = time.time()
     while r != b"" or x == b"":
-      r = self.dev.read(1)
+      r = self.io.read(1)
       x += r
       if r == b"":
         await asyncio.sleep(0.1)
@@ -46,17 +40,16 @@ async def _read(self) -> bytes:
 
   async def send_command(self, command: bytes) -> bytes:
     logger.debug("[loader] Sending %s", command.hex())
-    self.dev.write(command)
+    self.io.write(command)
     return await self._read()
 
   async def setup(self):
     logger.debug("[loader] setup")
 
-    self.dev.open()
-    self.dev.baudrate = 115384
+    await self.io.setup()
+    self.io.set_baudrate(115384)
 
     status = await self.get_status()
-    print("status", status)
     if not status.startswith(bytes.fromhex("1105")):
       raise RuntimeError("Failed to get status")
 
@@ -75,10 +68,10 @@ async def setup(self):
 
   async def stop(self):
     logger.debug("[loader] stop")
-    self.dev.close()
+    await self.io.stop()
 
   def serialize(self):
-    return {"device_id": self.dev.device_id, "timeout": self.timeout}
+    return {"io": self.io.serialize(), "timeout": self.timeout}
 
   async def get_status(self) -> bytes:
     logger.debug("[loader] get_status")
@@ -145,16 +138,12 @@ def __init__(self, bucket_1_position: int, device_id: Optional[str] = None):
         an arbitrary value, move to the bucket, and call get_position() to get the position. Then
         use this value for future runs.
     """
-    if not USE_FTDI:
-      raise RuntimeError("pylibftdi is not installed.")
-    self.dev = Device(lazy_open=True, device_id=device_id)
+    self.io = FTDI(device_id=device_id)
     self.bucket_1_position = bucket_1_position
     self.homing_position = 0
-    self.device_id = device_id
 
   async def setup(self):
-    self.dev.open()
-    logger.debug("open")
+    await self.io.setup()
     # TODO: add functionality where if robot has been intialized before nothing needs to happen
     for _ in range(3):
       await self.configure_and_initialize()
@@ -166,9 +155,9 @@ async def setup(self):
     await self.send(b"\xaa\x00\x21\x03\xff\x23")
     await self.send(b"\xaa\xff\x1a\x14\x2d")
 
-    self.dev.baudrate = 57600
-    self.dev.ftdi_fn.ftdi_setrts(1)
-    self.dev.ftdi_fn.ftdi_setdtr(1)
+    self.io.set_baudrate(57600)
+    self.io.set_rts(True)
+    self.io.set_dtr(True)
 
     await self.send(b"\xaa\x01\x0e\x0f")
     await self.send(b"\xaa\x01\x12\x1f\x32")
@@ -264,7 +253,7 @@ async def setup(self):
   async def stop(self):
     await self.send(b"\xaa\x02\x0e\x10")
     await self.configure_and_initialize()
-    self.dev.close()
+    await self.io.stop()
 
   async def get_status(self):
     """Returns 14 bytes
@@ -296,15 +285,12 @@ async def get_position(self):
   async def read_resp(self, timeout=20) -> bytes:
     """Read a response from the centrifuge. If the timeout is reached, return the data that has
     been read so far."""
-    if not self.dev:
-      raise RuntimeError("Device not initialized")
-
     data = b""
     end_byte_found = False
     start_time = time.time()
 
     while True:
-      chunk = self.dev.read(25)
+      chunk = self.io.read(25)
       if chunk:
         data += chunk
         end_byte_found = data[-1] == 0x0D
@@ -320,9 +306,7 @@ async def read_resp(self, timeout=20) -> bytes:
     return data
 
   async def send(self, cmd: Union[bytearray, bytes], read_timeout=0.2) -> bytes:
-    logger.debug("Sending %s", cmd.hex())
-    written = self.dev.write(cmd.decode("latin-1"))
-    logger.debug("Wrote %s bytes", written)
+    written = self.io.write(bytes(cmd))  # TODO: why decode? .decode("latin-1")
 
     if written != len(cmd):
       raise RuntimeError("Failed to write all bytes")
@@ -343,18 +327,17 @@ async def configure_and_initialize(self):
 
   async def set_configuration_data(self):
     """Set the device configuration data."""
-    self.dev.ftdi_fn.ftdi_set_latency_timer(16)
-    self.dev.ftdi_fn.ftdi_set_line_property(8, 1, 0)
-    self.dev.ftdi_fn.ftdi_setflowctrl(0)
-    self.dev.baudrate = 19200
+    self.io.set_latency_timer(16)
+    self.io.set_line_property(bits=8, stopbits=1, parity=0)
+    self.io.set_flowctrl(0)
+    self.io.set_baudrate(19200)
 
   async def initialize(self):
-    if self.dev:
-      self.dev.write(b"\x00" * 20)
-      for i in range(33):
-        packet = b"\xaa" + bytes([i & 0xFF, 0x0E, 0x0E + (i & 0xFF)]) + b"\x00" * 8
-        self.dev.write(packet)
-      await self.send(b"\xaa\xff\x0f\x0e")
+    self.io.write(b"\x00" * 20)
+    for i in range(33):
+      packet = b"\xaa" + bytes([i & 0xFF, 0x0E, 0x0E + (i & 0xFF)]) + b"\x00" * 8
+      self.io.write(packet)
+    await self.send(b"\xaa\xff\x0f\x0e")
 
   # Centrifuge operations
 

pylabrobot/centrifuge/vspin.py

@@ -2,18 +2,17 @@
 from typing import Literal
 
 from pylabrobot.heating_shaking.backend import HeaterShakerBackend
-from pylabrobot.machines.backends.usb import USBBackend
+from pylabrobot.io.usb import USB
 
 
 class PlateLockPosition(Enum):
   LOCKED = 1
   UNLOCKED = 0
 
 
-class HamiltonHeatShaker(HeaterShakerBackend, USBBackend):
+class HamiltonHeatShaker(HeaterShakerBackend):
   """
-  Backend for Hamilton Heater Shaker devices connected through
-  an Heat Shaker Box
+  Backend for Hamilton Heater Shaker devices connected through an Heater Shaker Box
   """
 
   def __init__(
@@ -30,39 +29,35 @@ def __init__(
     self.shaker_index = shaker_index
     self.command_id = 0
 
-    HeaterShakerBackend.__init__(self)
-    USBBackend.__init__(self, id_vendor, id_product)
+    super().__init__()
+    self.io = USB(id_vendor=id_vendor, id_product=id_product)
 
   async def setup(self):
     """
-    If USBBackend.setup() fails, ensure that libusb drivers were installed
-    for the HHS as per PyLabRobot documentation.
+    If io.setup() fails, ensure that libusb drivers were installed for the HHS as per docs.
     """
-    await USBBackend.setup(self)
+    await self.io.setup()
     await self._initialize_lock()
 
   async def stop(self):
-    await USBBackend.stop(self)
+    await self.io.stop()
 
   def serialize(self) -> dict:
-    usb_backend_serialized = USBBackend.serialize(self)
+    usb_serialized = self.io.serialize()
     heater_shaker_serialized = HeaterShakerBackend.serialize(self)
     return {
-      **usb_backend_serialized,
+      **usb_serialized,
       **heater_shaker_serialized,
       "shaker_index": self.shaker_index,
     }
 
   def _send_command(self, command: str, **kwargs):
     assert len(command) == 2, "Command must be 2 characters long"
     args = "".join([f"{key}{value}" for key, value in kwargs.items()])
-    USBBackend.write(
-      self,
-      f"T{self.shaker_index}{command}id{str(self.command_id).zfill(4)}{args}",
-    )
+    self.io.write(f"T{self.shaker_index}{command}id{str(self.command_id).zfill(4)}{args}".encode())
 
     self.command_id = (self.command_id + 1) % 10_000
-    return USBBackend.read(self)
+    return self.io.read()
 
   async def shake(
     self,

pylabrobot/heating_shaking/hamilton.py

@@ -2,13 +2,7 @@
 import typing
 
 from pylabrobot.heating_shaking.backend import HeaterShakerBackend
-
-try:
-  import hid  # type: ignore
-
-  USE_IDE = True
-except ImportError:
-  USE_IDE = False
+from pylabrobot.io.hid import HID
 
 
 class InhecoThermoShake(HeaterShakerBackend):
@@ -18,26 +12,20 @@ class InhecoThermoShake(HeaterShakerBackend):
   """
 
   def __init__(self, vid=0x03EB, pid=0x2023, serial_number=None):
-    self.vid = vid
-    self.pid = pid
-    self.serial_number = serial_number
+    self.io = HID(vid=vid, pid=pid, serial_number=serial_number)
 
   async def setup(self):
-    if not USE_IDE:
-      raise RuntimeError("This backend requires the `hid` package to be installed")
-    self.device = hid.Device(vid=self.vid, pid=self.pid, serial=self.serial_number)
+    await self.io.setup()
 
   async def stop(self):
     await self.stop_shaking()
     await self.stop_temperature_control()
-    self.device.close()
+    await self.io.stop()
 
   def serialize(self) -> dict:
     return {
       **super().serialize(),
-      "vid": self.vid,
-      "pid": self.pid,
-      "serial_number": self.serial_number,
+      **self.io.serialize(),
     }
 
   @typing.no_type_check
@@ -102,7 +90,7 @@ def _read_until_end(self, timeout: int) -> str:
     start = time.time()
     response = b""
     while time.time() - start < timeout:
-      packet = self.device.read(64, timeout=timeout)
+      packet = self.io.read(64, timeout=timeout)
       if packet is not None and packet != b"":
         if packet.endswith(b"\x00"):
           response += packet.rstrip(b"\x00")  # strip trailing \x00's
@@ -139,7 +127,7 @@ async def send_command(self, command: str, timeout: int = 3):
     """Send a command to the device and return the response"""
     packets = self._generate_packets(command)
     for packet in packets:
-      self.device.write(bytes(packet))
+      self.io.write(bytes(packet))
 
     response = self._read_response(command, timeout=timeout)
 

pylabrobot/heating_shaking/inheco.py

@@ -1,7 +1,7 @@
 from abc import ABCMeta, abstractmethod
 from typing import List, Optional
 
-from pylabrobot.machines.backends import MachineBackend
+from pylabrobot.machines.backend import MachineBackend
 from pylabrobot.resources import Plate, PlateCarrier, PlateHolder