Liconic backend (Liconic STX series automated incubator support)

docs/user_guide/01_material-handling/storage/liconic.ipynb

@@ -0,0 +1,241 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "id": "b63b4656",
+   "metadata": {},
+   "source": [
+    "<h2>Liconic STX Series</h2>\n",
+    "\n",
+    "The Liconic STX line of automated incubators come in a variety of sizes including STX 1000, STX 500, STX 280, STX 220, STX 110, STX 44. Which corresponds to the number of plates each size can store using the standard 22 plate capacity cassettes/cartridges (plate height 17mm, 505mm total height.) There are other cassette size for plates height ranging from 5 to 104mm in height (higher plates = less number of plates storage capacity.)\n",
+    "\n",
+    "The Liconic STX line comes in a variety of climate control options including Ultra High Temp. (HTT), Incubator (IC), Dry Storage (DC2), Humid Cooler (HC), Humid Wide Range (HR), Dry Wide Range (DR2), Humidity Controlled (AR), Deep Freezer (DF) and Ultra Deep Freezer (UDF). Each have different ranges of temperatures and humidity control ability.\n",
+    "\n",
+    "Other accessories that can be included with the STX and can be utilized with this driver include N2 gassing, CO2 gassing, a Turn Station (rotation of plates 90 degrees on the transfer station), internal barcode scanners, a swap station (two transfer plate positions that can be rotated 180 degrees) and internal shaking. \n",
+    "\n",
+    "This tutorial shows how to\n",
+    " - Connect the Liconic incubator\n",
+    " - Configure racks\n",
+    " - Move plates in and out\n",
+    " - Set and monitor temperature and humidity values"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "fcd75e15",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "\n",
+    "from pylabrobot.resources.coordinate import Coordinate\n",
+    "from pylabrobot.storage import LiconicBackend\n",
+    "from pylabrobot.storage.incubator import Incubator\n",
+    "from pylabrobot.storage.liconic.racks import liconic_rack_17mm_22, liconic_rack_44mm_10\n",
+    "\n",
+    "backend = LiconicBackend(port=\"COM3\", model=\"STX220_HC\", barcode_installed=True, barcode_port=\"COM4\")\n",
+    "\n",
+    "rack = [\n",
+    "        liconic_rack_44mm_10(\"cassette_0\"),\n",
+    "        liconic_rack_44mm_10(\"cassette_1\"),\n",
+    "        liconic_rack_44mm_10(\"cassette_2\"),\n",
+    "        liconic_rack_17mm_22(\"cassette_3\"),\n",
+    "        liconic_rack_17mm_22(\"cassette_4\"),\n",
+    "        liconic_rack_17mm_22(\"cassette_5\"),\n",
+    "        liconic_rack_17mm_22(\"cassette_6\"),\n",
+    "        liconic_rack_17mm_22(\"cassette_7\"),\n",
+    "        liconic_rack_17mm_22(\"cassette_8\"),\n",
+    "        liconic_rack_17mm_22(\"cassette_9\")\n",
+    "        ]\n",
+    "\n",
+    "incubator = Incubator(\n",
+    "  backend=backend,\n",
+    "  name=\"My Incubator\",\n",
+    "  size_x=100,\n",
+    "  size_y=100,\n",
+    "  size_z=100,\n",
+    "  racks=rack,\n",
+    "  loading_tray_location=Coordinate(x=0, y=0, z=0),\n",
+    ")\n",
+    "\n",
+    "await incubator.setup()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "19b3a6cc",
+   "metadata": {
+    "vscode": {
+     "languageId": "plaintext"
+    }
+   },
+   "source": [
+    "Setup\n",
+    "\n",
+    "To setup the incubator and start sending commands first the backed needs to be declared. For the Liconic the LiconcBackend class is used with the COM port used for connection (in this case COM3) and the model needs to specified (in this case the STX 220 Humid Cooler, STX220_HC). If an internal barcode is installed the barcode_installed parameter is set to True and its COM port is also specified. These two parameters are optional so can be omitted for Liconics without an internal barcode scanner. \n",
+    "\n",
+    "Given a STX 220 (220 plate position / 22 plates per rack = 10 racks) can hold 10 racks the list of racks is built and includes mixing and matching different plate height racks. The differences in racks are handled prior to plate retrieval and storage. \n",
+    "\n",
+    "Once the these are built the base Incubator class is created and the connection to the incubator is initialized using:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "9d7a4f49",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "await incubator.setup()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "52f79811",
+   "metadata": {},
+   "source": [
+    "To store a plate first a plate resource is initialized and then assigned to the loading tray. The method take_in_plate is then called on the incubator object.\n",
+    "\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "d26e039d",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from pylabrobot.resources import Azenta4titudeFrameStar_96_wellplate_200ul_Vb\n",
+    "\n",
+    "new_plate = Azenta4titudeFrameStar_96_wellplate_200ul_Vb(name=\"TEST\")\n",
+    "incubator.loading_tray.assign_child_resource(new_plate)\n",
+    "await incubator.take_in_plate(\"smallest\")  # choose the smallest free site\n",
+    "\n",
+    "# other options:\n",
+    "# await incubator.take_in_plate(\"random\")  # random free site\n",
+    "# await incubator.take_in_plate(rack[3])  # store at rack position 3"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "85dcddb7",
+   "metadata": {},
+   "source": [
+    "To retrieve a plate the plate name can used"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "00308838",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "await incubator.fetch_plate_to_loading_tray(plate=\"TEST\")\n",
+    "retrieved = incubator.loading_tray.resource"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "0045e703",
+   "metadata": {},
+   "source": [
+    "You can also print a barcode from this call (if barcode is installed per the backend insatiation). Returning of the barcode as a return object still needs to be implemented. Currently the barcode is just printed to the terminal.\n",
+    "\n",
+    "Barcode can returned by setting the read_barcode to True for \n",
+    "- take_in_plate\n",
+    "- fetch_plate_to_loading_tray\n",
+    "- move_position_to_position"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "c8730560",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "position = rack[9][0] # rack number 9 position 1\n",
+    "\n",
+    "await incubator.fetch_plate_to_loading_tray(plate_name=\"TEST\",read_barcode=True)\n",
+    "\n",
+    "await incubator.take_in_plate(position,read_barcode=True)\n",
+    "\n",
+    "await incubator.move_position_to_position(plate_name=\"TEST\",dest_site=position,read_barcode=True)\n",
+    "# will print the barcode to the terminal"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "14efdf69",
+   "metadata": {},
+   "source": [
+    "The humdity, temperature, N2 gas and CO2 gas levels can all be controlled and queried. For temperature for example:\n",
+    "\n",
+    "- To get the current temperature"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "73e38f2a",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "temperature = await incubator.get_temperature() # returns temperature as float in Celsius to the 10th place\n",
+    "print(str(temperature))"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "c7383277",
+   "metadata": {},
+   "source": [
+    "- To set the temperature of the Liconic"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "2c51c385",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "await incubator.set_temperature(8.0) # set the temperature to 8 degrees Celsius"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "4f07f349",
+   "metadata": {},
+   "source": [
+    "- You can also retrieve the set value (the value sent for set_temperature) using:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "288dea91",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "set_temperature = await incubator.get_set_temperature() # will return a float for the set temperature in degrees Celsius"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "3a1d9ef3",
+   "metadata": {},
+   "source": [
+    "This pattern is the same for CO2, N2 and Humidity control of the Liconic. "
+   ]
+  }
+ ],
+ "metadata": {
+  "language_info": {
+   "name": "python"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

pylabrobot/barcode_scanners/__init__.py

@@ -0,0 +1,3 @@
+from .backend import BarcodeScannerBackend, BarcodeScannerError
+from .barcode_scanner import BarcodeScanner
+from .keyence import KeyenceBarcodeScannerBackend

pylabrobot/barcode_scanners/backend.py

@@ -0,0 +1,18 @@
+from abc import ABCMeta, abstractmethod
+
+from pylabrobot.machines.backend import MachineBackend
+from pylabrobot.resources.barcode import Barcode
+
+
+class BarcodeScannerError(Exception):
+  """Error raised by a barcode scanner backend."""
+
+
+class BarcodeScannerBackend(MachineBackend, metaclass=ABCMeta):
+  def __init__(self):
+    super().__init__()
+
+  @abstractmethod
+  async def scan_barcode(self) -> Barcode:
+    """Scan a barcode and return its value."""
+    pass

pylabrobot/barcode_scanners/barcode_scanner.py

@@ -0,0 +1,15 @@
+from pylabrobot.barcode_scanners.backend import BarcodeScannerBackend
+from pylabrobot.machines.machine import Machine
+from pylabrobot.resources.barcode import Barcode
+
+
+class BarcodeScanner(Machine):
+  """Frontend for barcode scanners."""
+
+  def __init__(self, backend: BarcodeScannerBackend):
+    super().__init__(backend=backend)
+    self.backend: BarcodeScannerBackend = backend
+
+  async def scan(self) -> Barcode:
+    """Scan a barcode and return its value."""
+    return await self.backend.scan_barcode()

pylabrobot/barcode_scanners/keyence/__init__.py

@@ -0,0 +1 @@
+from .barcode_scanner_backend import KeyenceBarcodeScannerBackend

pylabrobot/barcode_scanners/keyence/barcode_scanner_backend.py

@@ -0,0 +1,111 @@
+import asyncio
+import time
+from typing import Awaitable, Callable, Optional
+
+import serial
+
+from pylabrobot.barcode_scanners.backend import (
+  BarcodeScannerBackend,
+  BarcodeScannerError,
+)
+from pylabrobot.io.serial import Serial
+from pylabrobot.resources.barcode import Barcode
+
+
+class KeyenceBarcodeScannerBackend(BarcodeScannerBackend):
+  default_baudrate = 9600
+  serial_messaging_encoding = "ascii"
+  init_timeout = 1.0  # seconds
+  poll_interval = 0.2  # seconds
+
+  def __init__(
+    self,
+    serial_port: str,
+  ):
+    super().__init__()
+
+    # BL-1300 Barcode reader factory default serial communication settings
+    # should be the same factory default for the BL-600HA and BL-1300 models
+    self.io = Serial(
+      port=serial_port,
+      baudrate=self.default_baudrate,
+      bytesize=serial.SEVENBITS,
+      parity=serial.PARITY_EVEN,
+      stopbits=serial.STOPBITS_ONE,
+      write_timeout=1,
+      timeout=1,
+      rtscts=False,
+    )
+
+  async def setup(self):
+    await self.io.setup()
+    await self.initialize_scanner()
+
+  async def initialize_scanner(self):
+    """Initialize the Keyence barcode scanner."""
+
+    response = await self.send_command("RMOTOR")
+
+    deadline = time.time() + self.init_timeout
+    while time.time() < deadline:
+      response = await self.send_command("RMOTOR")
+      if response.strip() == "MOTORON":
+        print("Barcode scanner motor is ON.")
+        break
+      elif response.strip() == "MOTOROFF":
+        raise BarcodeScannerError("Failed to initialize Keyence barcode scanner: Motor is off.")
+      await asyncio.sleep(self.poll_interval)
+    else:
+      raise BarcodeScannerError(
+        "Failed to initialize Keyence barcode scanner: " "Timeout waiting for motor to turn on."
+      )
+
+  async def send_command(self, command: str) -> str:
+    """Send a command to the barcode scanner and return the response.
+    Keyence uses carriage return \r as the line ending by default."""
+
+    await self.io.write((command + "\r").encode(self.serial_messaging_encoding))
+    response = await self.io.read()
+    return response.decode(self.serial_messaging_encoding).strip()
+
+  async def send_command_and_stream(
+    self,
+    command: str,
+    on_response: Callable[[str], Awaitable[None]],
+    timeout: float = 5.0,
+    stop_condition: Optional[Callable[[str], bool]] = None,
+  ):
+    """Send a command and call on_response for each barcode response."""
+    await self.io.write((command + "\r").encode(self.serial_messaging_encoding))
+    deadline = time.time() + timeout
+
+    while time.time() < deadline:
+      try:
+        response = await asyncio.wait_for(self.io.readline(), timeout=1.0)
+        if response:
+          decoded = response.decode(self.serial_messaging_encoding).strip()
+          print(f"Received from barcode scanner: {decoded}")
+          if decoded:
+            try:
+              await on_response(decoded)  # Call the callback
+            except Exception as e:
+              print(f"Error in callback: {e}")
+            if stop_condition and stop_condition(decoded):
+              break
+      except asyncio.TimeoutError:
+        print("Barcode scanner timeout, continuing...")
+        continue
+      except Exception as e:
+        print(f"Error reading from barcode scanner: {e}")
+        continue
+
+  async def stop(self):
+    await self.io.stop()
+
+  async def scan_barcode(self) -> Barcode:
+    data = await self.send_command("LON")
+    if data.startswith("NG"):
+      raise BarcodeScannerError("Barcode reader is off: cannot read barcode")
+    if data.startswith("ERR99"):
+      raise BarcodeScannerError(f"Error response from barcode reader: {data}")
+    return Barcode(data=data, symbology="unknown", position_on_resource="front")