diff --git a/docs/api/pylabrobot.generic.rst b/docs/api/pylabrobot.generic.rst new file mode 100644 index 00000000000..0baff896920 --- /dev/null +++ b/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 diff --git a/docs/api/pylabrobot.rst b/docs/api/pylabrobot.rst index 206ff03dd1e..ad4eedd5c1f 100644 --- a/docs/api/pylabrobot.rst +++ b/docs/api/pylabrobot.rst @@ -10,6 +10,7 @@ Subpackages :maxdepth: 1 pylabrobot.config + pylabrobot.generic pylabrobot.resources pylabrobot.utils diff --git a/docs/user_guide/generic/index.md b/docs/user_guide/generic/index.md new file mode 100644 index 00000000000..6d3ab3bb02d --- /dev/null +++ b/docs/user_guide/generic/index.md @@ -0,0 +1,7 @@ +# Generic devices + +```{toctree} +:maxdepth: 1 + +line-barcode-scanner/hello-world +``` diff --git a/docs/user_guide/generic/line-barcode-scanner/hello-world.ipynb b/docs/user_guide/generic/line-barcode-scanner/hello-world.ipynb new file mode 100644 index 00000000000..d60f95448a8 --- /dev/null +++ b/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 +} diff --git a/docs/user_guide/index.md b/docs/user_guide/index.md index 0d07f7280bb..8aeff3139df 100644 --- a/docs/user_guide/index.md +++ b/docs/user_guide/index.md @@ -19,6 +19,7 @@ getting-started/units machines definitions +generic/index 00_liquid-handling/_liquid-handling ``` diff --git a/pylabrobot/generic/__init__.py b/pylabrobot/generic/__init__.py new file mode 100644 index 00000000000..8b044b642fd --- /dev/null +++ b/pylabrobot/generic/__init__.py @@ -0,0 +1 @@ +from .line_barcode_scanner import SerialBarcodeScanner diff --git a/pylabrobot/generic/line_barcode_scanner/__init__.py b/pylabrobot/generic/line_barcode_scanner/__init__.py new file mode 100644 index 00000000000..d9f4dcbc44a --- /dev/null +++ b/pylabrobot/generic/line_barcode_scanner/__init__.py @@ -0,0 +1 @@ +from .serial import SerialBarcodeScanner diff --git a/pylabrobot/generic/line_barcode_scanner/serial.py b/pylabrobot/generic/line_barcode_scanner/serial.py new file mode 100644 index 00000000000..c157d324a91 --- /dev/null +++ b/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, + ) diff --git a/pylabrobot/generic/line_barcode_scanner/serial_tests.py b/pylabrobot/generic/line_barcode_scanner/serial_tests.py new file mode 100644 index 00000000000..a783d0d736d --- /dev/null +++ b/pylabrobot/generic/line_barcode_scanner/serial_tests.py @@ -0,0 +1,164 @@ +import unittest +from typing import List + +from pylabrobot.generic import SerialBarcodeScanner + + +class FakeSerialIO: + def __init__(self, chunks: List[bytes]): + self.chunks = chunks + self.writes: List[bytes] = [] + self.port = "COM_TEST" + self.timeout: float = 1 + self.setup_called = False + self.stop_called = False + self.reset_input_buffer_called = False + + async def setup(self): + self.setup_called = True + + async def stop(self): + self.stop_called = True + + async def read(self, num_bytes: int = 1) -> bytes: + del num_bytes + if len(self.chunks) == 0: + return b"" + return self.chunks.pop(0) + + async def write(self, data: bytes): + self.writes.append(data) + + async def reset_input_buffer(self): + self.reset_input_buffer_called = True + + def get_read_timeout(self) -> float: + return self.timeout + + def set_read_timeout(self, timeout: float) -> None: + self.timeout = timeout + + def temporary_timeout(self, timeout: float): + fake = self + + class TemporaryTimeout: + def __enter__(self): + self.original_timeout = fake.timeout + fake.timeout = timeout + + def __exit__(self, exc_type, exc_value, traceback): + fake.timeout = self.original_timeout + + return TemporaryTimeout() + + +def make_scanner(chunks: List[bytes]) -> SerialBarcodeScanner: + scanner = SerialBarcodeScanner(port="COM_TEST") + scanner.io = FakeSerialIO(chunks) # type: ignore[assignment] + return scanner + + +class TestSerialBarcodeScanner(unittest.IsolatedAsyncioTestCase): + async def test_read_line_carriage_return(self): + scanner = make_scanner([b"1", b"2", b"3", b"\r"]) + + self.assertEqual(await scanner.read_line(timeout=1), "123") + + async def test_read_line_newline(self): + scanner = make_scanner([b"A", b"B", b"C", b"\n"]) + + self.assertEqual(await scanner.read_line(timeout=1), "ABC") + + async def test_read_line_timeout_before_data(self): + scanner = make_scanner([]) + + self.assertEqual(await scanner.read_line(timeout=0), "") + + async def test_read_line_rejects_negative_timeout(self): + scanner = make_scanner([]) + + with self.assertRaises(ValueError): + await scanner.read_line(timeout=-1) + + async def test_reset_input_buffer(self): + scanner = make_scanner([]) + + await scanner.reset_input_buffer() + + fake_io = scanner.io + assert isinstance(fake_io, FakeSerialIO) + self.assertTrue(fake_io.reset_input_buffer_called) + + def test_rejects_empty_terminators(self): + with self.assertRaises(ValueError): + SerialBarcodeScanner(port="COM_TEST", terminators=[]) + + def test_rejects_multi_byte_terminators(self): + with self.assertRaises(ValueError): + SerialBarcodeScanner(port="COM_TEST", terminators=[b"\r\n"]) + + def test_rejects_non_positive_max_line_length(self): + with self.assertRaises(ValueError): + SerialBarcodeScanner(port="COM_TEST", max_line_length=0) + + async def test_scan_barcode(self): + scanner = SerialBarcodeScanner(port="COM_TEST") + scanner.io = FakeSerialIO([b"2", b"2", b"6", b"\r"]) # type: ignore[assignment] + + barcode = await scanner.scan_barcode( + read_time=1, + symbology="Code 128 (Subset B and C)", + position_on_resource="right", + ) + + assert barcode is not None + self.assertEqual(barcode.data, "226") + self.assertEqual(barcode.symbology, "Code 128 (Subset B and C)") + self.assertEqual(barcode.position_on_resource, "right") + + async def test_scan_barcode_returns_none_on_timeout(self): + scanner = make_scanner([]) + + self.assertIsNone(await scanner.scan_barcode(read_time=0)) + + async def test_scan_barcode_with_trigger_command(self): + scanner = SerialBarcodeScanner( + port="COM_TEST", + trigger_command=b"TRIGGER\r", + untrigger_command=b"UNTRIGGER\r", + ) + scanner.io = FakeSerialIO([b"1", b"2", b"3", b"\r"]) # type: ignore[assignment] + + barcode = await scanner.scan_barcode(read_time=1) + + assert barcode is not None + self.assertEqual(barcode.data, "123") + fake_io = scanner.io + assert isinstance(fake_io, FakeSerialIO) + self.assertEqual(fake_io.writes, [b"TRIGGER\r", b"UNTRIGGER\r"]) + + async def test_scan_barcode_rejects_negative_read_time(self): + scanner = make_scanner([]) + + with self.assertRaises(ValueError): + await scanner.scan_barcode(read_time=-1) + + async def test_setup_scan_stop(self): + scanner = SerialBarcodeScanner(port="COM_TEST") + fake_io = FakeSerialIO([b"X", b"Y", b"Z", b"\r"]) + scanner.io = fake_io # type: ignore[assignment] + + await scanner.setup() + barcode = await scanner.scan_barcode(read_time=1, symbology="Code 39") + await scanner.stop() + + assert barcode is not None + self.assertEqual(barcode.data, "XYZ") + self.assertEqual(barcode.symbology, "Code 39") + self.assertEqual(barcode.position_on_resource, "bottom") + self.assertTrue(fake_io.setup_called) + self.assertTrue(fake_io.stop_called) + + +if __name__ == "__main__": + unittest.main()