Add serial barcode scanner

docs/api/pylabrobot.generic.rst

@@ -0,0 +1,14 @@
+.. currentmodule:: pylabrobot.generic
+
+Generic devices
+===============
+
+Drivers for devices that use common, configurable protocols rather than a manufacturer-specific
+protocol.
+
+.. autosummary::
+  :toctree: _autosummary
+  :nosignatures:
+  :recursive:
+
+    SerialBarcodeScanner

docs/api/pylabrobot.rst

@@ -10,6 +10,7 @@ Subpackages
     :maxdepth: 1
 
     pylabrobot.config
+    pylabrobot.generic
     pylabrobot.resources
     pylabrobot.utils
 

docs/user_guide/generic/index.md

@@ -0,0 +1,7 @@
+# Generic devices
+
+```{toctree}
+:maxdepth: 1
+
+line-barcode-scanner/hello-world
+```

docs/user_guide/generic/line-barcode-scanner/hello-world.ipynb

@@ -0,0 +1,117 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "id": "serial-barcode-title",
+   "metadata": {},
+   "source": [
+    "# Line-oriented serial barcode scanners\n",
+    "\n",
+    "`SerialBarcodeScanner` supports barcode scanners that send one decoded barcode followed by a line terminator over RS-232 or a USB virtual COM port. One tested example is the [WONE NICE wireless scanner](https://www.amazon.com/dp/B00LE5VV1C), configured for USB virtual COM.\n",
+    "\n",
+    "| Property | Default |\n",
+    "|---|---|\n",
+    "| Serial settings | 9600 baud, 8 data bits, no parity, 1 stop bit |\n",
+    "| Line terminators | Carriage return or newline |\n",
+    "| Read mode | Passive, with optional trigger and untrigger commands |"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "serial-barcode-protocol",
+   "metadata": {},
+   "source": [
+    "## How it talks\n",
+    "\n",
+    "The scanner sends the decoded barcode as a line of text. The driver waits for a carriage return or newline, removes that terminator, and returns a `Barcode`. Scanners that require software triggering can be configured with their documented raw trigger commands."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "serial-barcode-physical-setup",
+   "metadata": {},
+   "source": [
+    "## Physical setup\n",
+    "\n",
+    "Connect the scanner through RS-232 or its USB virtual COM interface. Configure it to append a carriage return or newline to every scan, and note its port and serial settings."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "serial-barcode-setup-heading",
+   "metadata": {},
+   "source": [
+    "## Connect\n",
+    "\n",
+    "Create the scanner with the settings from its manual, then open the serial connection."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "serial-barcode-setup-code",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from pylabrobot.generic import SerialBarcodeScanner\n",
+    "\n",
+    "scanner = SerialBarcodeScanner(port=\"COM5\", baudrate=115200)\n",
+    "await scanner.setup()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "serial-barcode-scan-heading",
+   "metadata": {},
+   "source": [
+    "## Scan a barcode\n",
+    "\n",
+    "Scan a label within the read window. A timeout returns `None`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "serial-barcode-scan-code",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "barcode = await scanner.scan_barcode(read_time=5)\n",
+    "print(\"No barcode received\" if barcode is None else f\"Scanned: {barcode.data}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "serial-barcode-stop-heading",
+   "metadata": {},
+   "source": [
+    "## Disconnect\n",
+    "\n",
+    "Close the serial connection when scanning is complete."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "serial-barcode-stop-code",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "await scanner.stop()"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "name": "python",
+   "version": "3.12.0"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

docs/user_guide/index.md

@@ -19,6 +19,7 @@ getting-started/units
 
 machines
 definitions
+generic/index
 00_liquid-handling/_liquid-handling
 ```
 

pylabrobot/generic/__init__.py

@@ -0,0 +1 @@
+from .line_barcode_scanner import SerialBarcodeScanner

pylabrobot/generic/line_barcode_scanner/__init__.py

@@ -0,0 +1 @@
+from .serial import SerialBarcodeScanner

pylabrobot/generic/line_barcode_scanner/serial.py

@@ -0,0 +1,165 @@
+import asyncio
+import logging
+from typing import Literal, Optional, Sequence, Union
+
+from pylabrobot.io.serial import Serial
+from pylabrobot.resources.barcode import Barcode, Barcode1DSymbology, BarcodePosition
+
+logger = logging.getLogger(__name__)
+
+
+class SerialBarcodeScanner:
+  """Barcode scanner that returns one barcode per terminated serial line.
+
+  This class is intended for scanners configured as RS-232 or USB virtual COM devices. It reads
+  bytes from :class:`pylabrobot.io.Serial` until a configured line terminator is seen.
+  """
+
+  def __init__(
+    self,
+    port: Optional[str] = None,
+    vid: Optional[int] = None,
+    pid: Optional[int] = None,
+    baudrate: int = 9600,
+    bytesize: int = 8,
+    parity: str = "N",
+    stopbits: int = 1,
+    write_timeout: float = 1,
+    read_timeout: float = 1,
+    rtscts: bool = False,
+    dsrdtr: bool = False,
+    xonxoff: bool = False,
+    encoding: str = "utf-8",
+    terminators: Sequence[bytes] = (b"\r", b"\n"),
+    max_line_length: int = 4096,
+    trigger_command: Optional[bytes] = None,
+    untrigger_command: Optional[bytes] = None,
+  ) -> None:
+    if len(terminators) == 0:
+      raise ValueError("At least one line terminator must be configured.")
+    if any(len(t) != 1 for t in terminators):
+      raise ValueError("SerialBarcodeScanner only supports one-byte terminators.")
+    if max_line_length <= 0:
+      raise ValueError("max_line_length must be positive.")
+
+    self.io = Serial(
+      human_readable_device_name="Serial Barcode Scanner",
+      port=port,
+      vid=vid,
+      pid=pid,
+      baudrate=baudrate,
+      bytesize=bytesize,
+      parity=parity,
+      stopbits=stopbits,
+      write_timeout=write_timeout,
+      timeout=read_timeout,
+      rtscts=rtscts,
+      dsrdtr=dsrdtr,
+      xonxoff=xonxoff,
+    )
+    self.encoding = encoding
+    self.terminators = tuple(terminators)
+    self.max_line_length = max_line_length
+
+    self.trigger_command = trigger_command
+    self.untrigger_command = untrigger_command
+
+  async def setup(self) -> None:
+    """Connect to the scanner's serial port."""
+    await self.io.setup()
+    logger.info("[Serial barcode scanner %s] connected", self.io.port)
+
+  async def stop(self) -> None:
+    """Disconnect from the scanner's serial port."""
+    await self.io.stop()
+    logger.info("[Serial barcode scanner %s] disconnected", self.io.port)
+
+  async def read_line(self, timeout: Optional[float] = None) -> str:
+    """Read one barcode line from the serial stream.
+
+    Args:
+      timeout: Optional total read timeout in seconds. If omitted, the
+        underlying :class:`pylabrobot.io.Serial` timeout is used.
+
+    Returns:
+      The decoded line without the trailing line terminator. Returns an empty
+      string if the timeout elapses before any byte is read.
+    """
+    if timeout is not None and timeout < 0:
+      raise ValueError("timeout must be non-negative.")
+
+    raw = await self._read_until_terminator(timeout=timeout)
+    while any(raw.endswith(terminator) for terminator in self.terminators):
+      raw = raw[:-1]
+    return raw.decode(self.encoding, errors="replace")
+
+  async def write(self, data: bytes) -> None:
+    """Write raw bytes to the scanner."""
+    await self.io.write(data)
+
+  async def _read_until_terminator(self, timeout: Optional[float]) -> bytes:
+    """Read bytes until a configured terminator, timeout, or length limit is reached."""
+    loop = asyncio.get_running_loop()
+    deadline = None if timeout is None else loop.time() + timeout
+    buf = bytearray()
+
+    while len(buf) < self.max_line_length:
+      if deadline is None:
+        chunk = await self.io.read(1)
+      else:
+        remaining = deadline - loop.time()
+        if remaining <= 0:
+          break
+        with self.io.temporary_timeout(remaining):
+          chunk = await self.io.read(1)
+
+      if len(chunk) == 0:
+        break
+      buf.extend(chunk)
+      if bytes(chunk) in self.terminators:
+        break
+
+    return bytes(buf)
+
+  async def reset_input_buffer(self) -> None:
+    """Clear unread bytes buffered by the serial transport."""
+    await self.io.reset_input_buffer()
+
+  async def scan_barcode(
+    self,
+    read_time: Optional[float] = None,
+    symbology: Union[Barcode1DSymbology, Literal["unknown"]] = "unknown",
+    position_on_resource: BarcodePosition = "bottom",
+  ) -> Optional[Barcode]:
+    """Scan one barcode from the serial stream.
+
+    Args:
+      read_time: Maximum number of seconds to wait for a barcode. If omitted, the configured serial
+        read timeout is used.
+      symbology: Symbology assigned to the returned barcode.
+      position_on_resource: Position assigned to the returned barcode.
+
+    Returns:
+      The scanned barcode, or ``None`` if no data is received before the timeout.
+    """
+    if read_time is not None and read_time < 0:
+      raise ValueError("read_time must be non-negative.")
+
+    if self.trigger_command is not None:
+      await self.write(self.trigger_command)
+
+    try:
+      data = await self.read_line(timeout=read_time)
+    finally:
+      if self.untrigger_command is not None:
+        await self.write(self.untrigger_command)
+
+    if data == "":
+      return None
+
+    logger.info("[Serial barcode scanner %s] scanned barcode: %s", self.io.port, data)
+    return Barcode(
+      data=data,
+      symbology=symbology,
+      position_on_resource=position_on_resource,
+    )