diff --git a/docs/_static/devices.json b/docs/_static/devices.json index 56cb4cba7c6..c6ec378ff72 100644 --- a/docs/_static/devices.json +++ b/docs/_static/devices.json @@ -805,6 +805,21 @@ "manager": "https://discuss.pylabrobot.org/u/rickwierenga", "oem": "https://www.mt.com/us/en/home/products/Industrial_Weighing_Solutions/high-precision-weigh-sensors/weigh-module-wxs205sdu-15-11121008.html" }, + { + "id": "micronic-code-reader", + "vendor": "Micronic", + "name": "RD235", + "kind": "barcode scanner", + "capabilities": [ + "barcode reading" + ], + "status": "full", + "api": "pylabrobot.micronic.MicronicRD235", + "api_version": "v1", + "code_slug": "micronic/code_reader", + "doc_slug": "micronic/code_reader/hello-world", + "oem": "https://www.micronic.com/products/code-reader/" + }, { "id": "molecular-devices-imagexpress-micro", "vendor": "Molecular Devices", diff --git a/docs/api/pylabrobot.micronic.rst b/docs/api/pylabrobot.micronic.rst new file mode 100644 index 00000000000..62cc8f70dc7 --- /dev/null +++ b/docs/api/pylabrobot.micronic.rst @@ -0,0 +1,44 @@ +.. currentmodule:: pylabrobot.micronic + +pylabrobot.micronic package +=========================== + +Direct Micronic RD235 code-reader integration. + +Code reader +----------- + +.. currentmodule:: pylabrobot.micronic.code_reader.driver + +.. autosummary:: + :toctree: _autosummary + :nosignatures: + :recursive: + + MicronicRD235 + RackScanEntry + RackScanResult + +.. currentmodule:: pylabrobot.micronic.code_reader.errors + +.. autosummary:: + :toctree: _autosummary + :nosignatures: + :recursive: + + MicronicError + + +Scanners +-------- + +.. currentmodule:: pylabrobot.micronic.code_reader.scanner + +.. autosummary:: + :toctree: _autosummary + :nosignatures: + :recursive: + + Scanner + TwainScanner + SaneScanner diff --git a/docs/api/pylabrobot.rst b/docs/api/pylabrobot.rst index ad4eedd5c1f..67dd588713c 100644 --- a/docs/api/pylabrobot.rst +++ b/docs/api/pylabrobot.rst @@ -33,6 +33,7 @@ Manufacturers pylabrobot.kbioscience pylabrobot.kbiosystems pylabrobot.mettler_toledo + pylabrobot.micronic pylabrobot.molecular_devices pylabrobot.qinstruments pylabrobot.sartorius diff --git a/docs/user_guide/index.md b/docs/user_guide/index.md index 8aeff3139df..b77381922a3 100644 --- a/docs/user_guide/index.md +++ b/docs/user_guide/index.md @@ -41,6 +41,7 @@ inheco/index kbioscience/index kbiosystems/index mettler_toledo/index +micronic/index molecular_devices/index qinstruments/index sartorius/index diff --git a/docs/user_guide/micronic/code_reader/hello-world.ipynb b/docs/user_guide/micronic/code_reader/hello-world.ipynb new file mode 100644 index 00000000000..ca361bd570b --- /dev/null +++ b/docs/user_guide/micronic/code_reader/hello-world.ipynb @@ -0,0 +1,221 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "micronic-intro", + "metadata": {}, + "source": [ + "# Micronic RD235 Code Reader\n", + "\n", + "The Micronic RD235 Code Reader scans the side barcode of an 8x12 tube rack, acquires an image of\n", + "the rack, and decodes the 96 tube DataMatrix codes locally. It does not require Micronic Code\n", + "Reader or IO Monitor.\n", + "\n", + "| Property | Value |\n", + "|---|---|\n", + "| Model | RD235 |\n", + "| Rack-ID communication | Serial trigger and response |\n", + "| Serial settings | 9600 baud, 7 data bits, even parity, 1 stop bit |\n", + "| Rack image | TWAIN helper on Windows, SANE `scanimage` on Linux, or a custom `Scanner` |\n", + "| Rack layout | 8 rows by 12 columns |\n", + "| Tube barcode | Ten-digit DataMatrix |\n", + "| Rack barcode | Code 128 |\n", + "\n", + "[OEM link](https://www.micronic.com/products/code-reader/)" + ] + }, + { + "cell_type": "markdown", + "id": "device-card", + "metadata": {}, + "source": [ + "```{device-card} micronic-code-reader\n", + "```" + ] + }, + { + "cell_type": "markdown", + "id": "micronic-communication", + "metadata": {}, + "source": [ + "## How it talks\n", + "\n", + "The side reader uses `pylabrobot.io.Serial`. A rack-ID scan clears pending input, sends\n", + "`` followed by CRLF, reads through the response terminator, and returns the first sequence\n", + "of at least six digits. It returns `NOREAD` when the response contains no rack identifier.\n", + "\n", + "Rack-image acquisition runs through PyLabRobot's command-line transport. `TwainScanner` calls\n", + "an operator-installed TWAIN helper; `SaneScanner` calls `scanimage`. PyLabRobot then performs\n", + "grid fitting and DataMatrix decoding locally." + ] + }, + { + "cell_type": "markdown", + "id": "micronic-physical-setup", + "metadata": {}, + "source": [ + "## Physical setup\n", + "\n", + "1. Connect the rack scanner to the computer and install its operating-system driver.\n", + "2. Connect the side rack-ID reader and identify its serial port.\n", + "3. On Windows, install the local TWAIN helper and note its path and TWAIN source name. On\n", + " Linux, install SANE and confirm that `scanimage --list-devices` lists the scanner.\n", + "4. Install the serial and decoding dependencies: `pylabrobot[serial]`, `pillow`,\n", + " `opencv-python-headless`, `numpy`, and `zxing-cpp`.\n", + "5. Place an 8x12 Micronic tube rack in the scanner with the orientation expected by the\n", + " instrument." + ] + }, + { + "cell_type": "markdown", + "id": "micronic-connect", + "metadata": {}, + "source": [ + "## Connect\n", + "\n", + "Choose the scanner implementation for the host. This example uses a Windows TWAIN helper. On\n", + "Linux, replace it with `SaneScanner(sane_device=\"\")`.\n", + "`setup()` resolves the scanner executable, creates the image directory, and opens the side\n", + "reader's serial port." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "micronic-connect-code", + "metadata": {}, + "outputs": [], + "source": [ + "from pylabrobot.micronic import MicronicRD235, TwainScanner\n", + "\n", + "reader = MicronicRD235(\n", + " scanner=TwainScanner(\n", + " twain_scanner_path=r\"C:\\Tools\\twain_scan.exe\",\n", + " twain_source=\"AVA6PlusG\",\n", + " ),\n", + " serial_port=\"COM4\",\n", + " image_dir=r\"C:\\ProgramData\\PyLabRobot\\micronic-images\",\n", + ")\n", + "await reader.setup()" + ] + }, + { + "cell_type": "markdown", + "id": "micronic-rack-id", + "metadata": {}, + "source": [ + "## Read the rack ID\n", + "\n", + "`scan_rack_id()` triggers only the side barcode reader. It returns the decoded identifier or\n", + "`NOREAD`." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "micronic-rack-id-code", + "metadata": {}, + "outputs": [], + "source": [ + "rack_id = await reader.scan_rack_id(timeout=5.0)\n", + "print(rack_id)" + ] + }, + { + "cell_type": "markdown", + "id": "micronic-rack-resource", + "metadata": {}, + "source": [ + "## Represent the rack\n", + "\n", + "`scan_rack()` accepts a `TubeRack` with exactly 8 rows and 12 columns. Define the rack layout\n", + "before scanning it." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "micronic-rack-resource-code", + "metadata": {}, + "outputs": [], + "source": [ + "from pylabrobot.resources import ResourceHolder, TubeRack, create_ordered_items_2d\n", + "\n", + "rack = TubeRack(\n", + " name=\"micronic_96_tube_rack\",\n", + " size_x=85.0,\n", + " size_y=127.0,\n", + " size_z=20.0,\n", + " ordered_items=create_ordered_items_2d(\n", + " ResourceHolder,\n", + " num_items_x=12,\n", + " num_items_y=8,\n", + " dx=0,\n", + " dy=0,\n", + " dz=0,\n", + " item_dx=9.0,\n", + " item_dy=9.0,\n", + " size_x=9.0,\n", + " size_y=9.0,\n", + " size_z=20.0,\n", + " ),\n", + ")" + ] + }, + { + "cell_type": "markdown", + "id": "micronic-scan-rack", + "metadata": {}, + "source": [ + "## Scan the rack\n", + "\n", + "`scan_rack()` reads the side rack barcode, acquires the rack image, and returns one entry for\n", + "each tube position. A position reports `NOREAD` when no tube barcode was decoded." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "micronic-scan-rack-code", + "metadata": {}, + "outputs": [], + "source": [ + "result = await reader.scan_rack(rack=rack, timeout=90.0)\n", + "print(result.rack_id)\n", + "print(len([entry for entry in result.entries if entry.status == \"OK\"]))" + ] + }, + { + "cell_type": "markdown", + "id": "micronic-disconnect", + "metadata": {}, + "source": [ + "## Disconnect\n", + "\n", + "`stop()` closes the side reader's serial connection." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "micronic-disconnect-code", + "metadata": {}, + "outputs": [], + "source": [ + "await reader.stop()" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/docs/user_guide/micronic/index.md b/docs/user_guide/micronic/index.md new file mode 100644 index 00000000000..b1a7a4a55c7 --- /dev/null +++ b/docs/user_guide/micronic/index.md @@ -0,0 +1,7 @@ +# Micronic + +```{toctree} +:maxdepth: 1 + +code_reader/hello-world +``` diff --git a/pylabrobot/micronic/__init__.py b/pylabrobot/micronic/__init__.py new file mode 100644 index 00000000000..c5fa4371e2b --- /dev/null +++ b/pylabrobot/micronic/__init__.py @@ -0,0 +1,9 @@ +from pylabrobot.micronic.code_reader import ( + MicronicError, + MicronicRD235, + RackScanEntry, + RackScanResult, + SaneScanner, + Scanner, + TwainScanner, +) diff --git a/pylabrobot/micronic/code_reader/__init__.py b/pylabrobot/micronic/code_reader/__init__.py new file mode 100644 index 00000000000..055eb0cc1e7 --- /dev/null +++ b/pylabrobot/micronic/code_reader/__init__.py @@ -0,0 +1,7 @@ +from pylabrobot.micronic.code_reader.driver import ( + MicronicRD235, + RackScanEntry, + RackScanResult, +) +from pylabrobot.micronic.code_reader.errors import MicronicError +from pylabrobot.micronic.code_reader.scanner import SaneScanner, Scanner, TwainScanner diff --git a/pylabrobot/micronic/code_reader/driver.py b/pylabrobot/micronic/code_reader/driver.py new file mode 100644 index 00000000000..4b6f76000fa --- /dev/null +++ b/pylabrobot/micronic/code_reader/driver.py @@ -0,0 +1,735 @@ +"""Direct integration for the Micronic RD235 rack scanner. + +This driver does not call Micronic Code Reader or IO Monitor. It owns the local +scanner path directly: + +- acquire a rack image through a caller-supplied :class:`Scanner`, +- read barcodes through the side serial barcode reader, and +- decode tube barcodes and return position-indexed rack results. +""" + +from __future__ import annotations + +import asyncio +import logging +import re +import tempfile +import time +from dataclasses import dataclass +from datetime import datetime +from functools import partial +from pathlib import Path +from typing import Any, Iterable, Iterator, Literal, Optional + +from pylabrobot.io.serial import Serial +from pylabrobot.resources.barcode import Barcode +from pylabrobot.resources.tube_rack import TubeRack + +from .errors import MicronicError +from .scanner import Scanner + +logger = logging.getLogger(__name__) + +ROWS = "ABCDEFGH" +COLS = 12 +RACK_ROWS = 8 +RACK_COLS = 12 + + +@dataclass(frozen=True) +class DecodeResult: + """A tube barcode decoded from a rack image. + + Attributes: + tube_id: Ten-digit tube identifier. + method: Image-decoding strategy that produced the result. + """ + + tube_id: str + method: str + + +@dataclass +class RackScanEntry: + """One decoded rack position. + + Attributes: + position: Rack position such as ``"A1"``. + tube_id: Decoded tube identifier, or ``None`` when no code was read. + status: ``"OK"`` when a tube code was read, otherwise ``"NOREAD"``. + barcode: Structured representation of the decoded tube barcode. + """ + + position: str + tube_id: Optional[str] + status: Literal["OK", "NOREAD"] + barcode: Optional[Barcode] = None + + +@dataclass +class RackScanResult: + """The rack identifier and position-indexed tube scan results. + + Attributes: + rack_id: Side barcode value, or ``"NOREAD"`` when no rack code was read. + entries: Results for all 96 positions in row-major order. + rack_barcode: Structured representation of the rack barcode. + """ + + rack_id: str + entries: list[RackScanEntry] + rack_barcode: Optional[Barcode] = None + + +class MicronicRD235: + """Control a Micronic RD235 rack scanner without the OEM application. + + Args: + scanner: Image acquisition implementation for the flatbed scanner. + serial_port: Port for the side rack-barcode reader. + image_dir: Directory for temporary or retained rack images. + scanner_timeout: Image acquisition timeout in seconds. + serial_timeout: Side barcode read timeout in seconds. + keep_images: Preserve acquired images after decoding when ``True``. + """ + + def __init__( + self, + scanner: Scanner, + serial_port: str, + image_dir: Optional[str] = None, + scanner_timeout: float = 90.0, + serial_timeout: float = 2.5, + keep_images: bool = False, + ) -> None: + """Initialize the code reader and its serial transport. + + Raises: + ValueError: If either device timeout is not positive. + """ + if scanner_timeout <= 0: + raise ValueError("scanner_timeout must be positive") + if serial_timeout <= 0: + raise ValueError("serial_timeout must be positive") + self.scanner = scanner + self.image_dir = ( + Path(image_dir) if image_dir else Path(tempfile.gettempdir()) / "pylabrobot-micronic" + ) + self.scanner_timeout = scanner_timeout + self.serial_timeout = serial_timeout + self.keep_images = keep_images + self.io = Serial( + human_readable_device_name="Micronic rack ID reader", + port=serial_port, + baudrate=9600, + bytesize=7, + parity="E", + stopbits=1, + timeout=0.1, + write_timeout=1.0, + ) + self.last_image_path: Optional[Path] = None + self.last_scan_metadata: dict[str, object] = {} + self.last_decode_metadata: dict[str, object] = {} + self._scan_lock = asyncio.Lock() + + async def setup(self) -> None: + """Create the image directory and connect to the side barcode reader.""" + self.image_dir.mkdir(parents=True, exist_ok=True) + await self.scanner.setup() + try: + await self.io.setup() + except BaseException: + try: + await self.scanner.stop() + except Exception: + logger.warning("Failed to stop Micronic scanner after setup failure", exc_info=True) + raise + logger.info("Set up Micronic code reader") + + async def stop(self) -> None: + """Disconnect from the side barcode reader and image scanner.""" + try: + await self.io.stop() + finally: + await self.scanner.stop() + logger.info("Stopped Micronic code reader") + + async def _read_barcode(self) -> str: + """Trigger and parse one side rack-barcode read. + + Returns: + The first sequence of at least six digits, or ``"NOREAD"``. + + Raises: + MicronicError: If serial communication fails. + """ + deadline = time.monotonic() + self.serial_timeout + chunks: list[bytes] = [] + try: + await self.io.reset_input_buffer() + await self.io.write(b"\r\n") + while time.monotonic() < deadline: + value = await self.io.read(1) + if value: + chunks.append(value) + if value in {b"\r", b"\n"}: + break + except Exception as exc: + logger.exception("Micronic rack ID serial read failed") + raise MicronicError( + "Rack ID serial read failed. Install the PLR serial extra with " + "`pip install pylabrobot[serial]` and verify the serial port: " + f"{exc}" + ) from exc + text = b"".join(chunks).decode("utf-8", errors="ignore") + match = re.search(r"\d{6,}", text) + return match.group(0) if match else "NOREAD" + + async def _acquire_image(self) -> Path: + """Acquire one rack image and retain its scanner metadata.""" + self.image_dir.mkdir(parents=True, exist_ok=True) + image_path = ( + self.image_dir + / f"micronic_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}.{self.scanner.image_extension}" + ) + try: + self.last_scan_metadata = await self.scanner.acquire(image_path, self.scanner_timeout) + except asyncio.CancelledError: + self._release_image(image_path) + raise + except Exception: + self._release_image(image_path) + raise + self.last_image_path = image_path + return image_path + + def _release_image(self, image_path: Path) -> None: + """Delete an acquired image unless image retention is enabled.""" + if not self.keep_images: + try: + image_path.unlink() + self.last_image_path = None + except FileNotFoundError: + self.last_image_path = None + except OSError: + logger.warning("Could not remove Micronic rack image %s", image_path, exc_info=True) + + @staticmethod + def _validate_rack(rack: TubeRack) -> None: + """Reject rack resources that do not represent an 8x12 layout.""" + if rack.num_items_x != RACK_COLS or rack.num_items_y != RACK_ROWS: + raise MicronicError( + f"Micronic code reader only supports {RACK_ROWS}x{RACK_COLS} racks; " + f"got {rack.num_items_y}x{rack.num_items_x}." + ) + + async def scan_rack(self, rack: TubeRack, timeout: float = 90.0) -> RackScanResult: + """Scan the rack ID and all tube positions in an 8x12 rack. + + Args: + rack: Rack resource whose 96 positions correspond to the scanner layout. + timeout: Overall operation timeout in seconds. + + Returns: + Rack and tube barcode results. + + Raises: + MicronicError: If the rack shape is unsupported, a scan is already running, image + acquisition fails, or decoding fails. + asyncio.TimeoutError: If the operation exceeds ``timeout``. + """ + logger.info("Starting Micronic rack scan") + try: + result = await asyncio.wait_for(self._scan_rack(rack), timeout=timeout) + except Exception: + logger.exception("Micronic rack scan failed") + raise + logger.info( + "Completed Micronic rack scan: rack_id=%s decoded=%d", + result.rack_id, + sum(entry.status == "OK" for entry in result.entries), + ) + return result + + async def scan_rack_id(self, timeout: float = 5.0) -> str: + """Read the side rack barcode. + + Args: + timeout: Overall operation timeout in seconds. + + Returns: + The decoded rack identifier, or ``"NOREAD"``. + + Raises: + MicronicError: If serial communication fails. + asyncio.TimeoutError: If the operation exceeds ``timeout``. + """ + try: + rack_id = await asyncio.wait_for(self._read_barcode(), timeout=timeout) + except Exception: + logger.exception("Micronic rack ID scan failed") + raise + logger.info("Completed Micronic rack ID scan: rack_id=%s", rack_id) + return rack_id + + async def _scan_rack(self, rack: TubeRack) -> RackScanResult: + """Coordinate one cancellation-safe rack scan.""" + self._validate_rack(rack) + if self._scan_lock.locked(): + raise MicronicError("Micronic rack scan is already in progress.") + await self._scan_lock.acquire() + release_lock = True + image_path: Optional[Path] = None + try: + rack_id = await self._read_barcode() + image_path = await self._acquire_image() + loop = asyncio.get_running_loop() + scan_future = loop.run_in_executor( + None, + self._decode_rack_image, + image_path, + rack_id, + ) + try: + return await asyncio.shield(scan_future) + except asyncio.CancelledError: + release_lock = False + scan_future.add_done_callback(partial(self._finish_cancelled_scan, image_path=image_path)) + image_path = None + raise + finally: + if image_path is not None: + self._release_image(image_path) + if release_lock: + self._release_scan_lock() + + def _finish_cancelled_scan( + self, + future: asyncio.Future[RackScanResult], + *, + image_path: Path, + ) -> None: + """Release scan resources after a cancelled caller's decode finishes.""" + try: + exception = future.exception() + except asyncio.CancelledError: + exception = None + if exception is not None: + logger.error("Cancelled Micronic rack scan later failed: %s", exception) + self._release_image(image_path) + self._release_scan_lock() + + def _release_scan_lock(self) -> None: + """Release the scan lock if it is currently held.""" + if self._scan_lock.locked(): + self._scan_lock.release() + + def _decode_rack_image( + self, + image_path: Path, + rack_id: str, + ) -> RackScanResult: + """Decode an acquired rack image in a worker thread.""" + decoded, self.last_decode_metadata = decode_image(image_path) + + for position, result in decoded.items(): + logger.debug("Micronic decoded %s via %s", position, result.method) + + entries = [ + RackScanEntry( + position=position, + tube_id=decoded[position].tube_id if position in decoded else None, + status="OK" if position in decoded else "NOREAD", + barcode=( + Barcode( + data=decoded[position].tube_id, + symbology="DataMatrix", + position_on_resource="bottom", + ) + if position in decoded + else None + ), + ) + for position in iter_positions() + ] + + return RackScanResult( + rack_id=rack_id, + entries=entries, + rack_barcode=Barcode( + data=rack_id, + symbology="Code 128 (Subset B and C)", + position_on_resource="right", + ) + if rack_id != "NOREAD" + else None, + ) + + +def decode_image(image_path: Path) -> tuple[dict[str, DecodeResult], dict[str, object]]: + """Decode all tube barcodes in a Micronic rack image. + + Args: + image_path: Path to the acquired rack image. + + Returns: + Position-indexed decode results and diagnostic metadata. + + Raises: + MicronicError: If dependencies are missing, the grid cannot be calibrated, or duplicate tube + identifiers are found. + """ + cv2, np, zxingcpp, Image, ImageOps = import_decode_dependencies() + with Image.open(image_path) as loaded_image: + image = loaded_image.convert("L") + full_results = zxingcpp.read_barcodes( + image, + formats=zxingcpp.BarcodeFormat.DataMatrix, + try_rotate=True, + try_downscale=True, + try_invert=True, + ) + + detected: list[tuple[float, float, str]] = [] + for result in full_results: + if not is_tube_id(result.text): + continue + corners = [ + result.position.top_left, + result.position.top_right, + result.position.bottom_right, + result.position.bottom_left, + ] + detected.append( + ( + sum(corner.x for corner in corners) / 4, + sum(corner.y for corner in corners) / 4, + result.text, + ) + ) + + if len(detected) < 24: + raise MicronicError(f"Only {len(detected)} DataMatrix codes were found in the full image.") + + xs = fitted_axis(cluster_axis([item[0] for item in detected], RACK_ROWS, 90), RACK_ROWS) + ys = fitted_axis(cluster_axis([item[1] for item in detected], RACK_COLS, 90), RACK_COLS) + x_pitch = abs(xs[-1] - xs[0]) / (RACK_ROWS - 1) + y_pitch = abs(ys[-1] - ys[0]) / (RACK_COLS - 1) + + decoded: dict[str, DecodeResult] = {} + for x, y, tube_id in detected: + scan_col = min(range(RACK_ROWS), key=lambda index: abs(xs[index] - x)) + scan_row = min(range(RACK_COLS), key=lambda index: abs(ys[index] - y)) + if abs(xs[scan_col] - x) > x_pitch * 0.45 or abs(ys[scan_row] - y) > y_pitch * 0.45: + continue + decoded[rack_position(scan_row, scan_col)] = DecodeResult(tube_id=tube_id, method="full-image") + + for scan_row in range(RACK_COLS): + for scan_col in range(RACK_ROWS): + position = rack_position(scan_row, scan_col) + if position in decoded: + continue + crop_result = decode_well_crop( + image, + xs[scan_col], + ys[scan_row], + cv2, + np, + zxingcpp, + Image, + ImageOps, + ) + if crop_result: + decoded[position] = crop_result + + duplicate_ids = find_duplicate_ids(decoded) + if duplicate_ids: + raise MicronicError( + f"Duplicate tube IDs decoded from more than one well: {', '.join(duplicate_ids)}" + ) + + metadata = { + "imageSize": image.size, + "fullImageDecoded": len(detected), + "gridX": [round(value, 1) for value in xs], + "gridY": [round(value, 1) for value in ys], + "decodedWells": len(decoded), + "missing": [position for position in iter_positions() if position not in decoded], + } + return decoded, metadata + + +def import_decode_dependencies() -> tuple[Any, Any, Any, Any, Any]: + """Import optional image-decoding dependencies on demand. + + Returns: + The OpenCV, NumPy, zxing-cpp, Pillow Image, and Pillow ImageOps modules. + + Raises: + MicronicError: If any decoding dependency is unavailable. + """ + try: + import cv2 # type: ignore + import numpy as np # type: ignore + import zxingcpp # type: ignore + from PIL import Image, ImageOps # type: ignore + except ImportError as exc: + raise MicronicError( + "Micronic decode dependencies are missing. Install pillow, " + "opencv-python-headless, numpy, and zxing-cpp." + ) from exc + return cv2, np, zxingcpp, Image, ImageOps + + +def cluster_axis(values: list[float], expected_count: int, tolerance: float) -> list[float]: + """Cluster detected coordinates along one scanner axis. + + Args: + values: Detected barcode coordinates. + expected_count: Number of rack rows or columns expected on the axis. + tolerance: Maximum gap within one cluster, in image pixels. + + Returns: + Cluster centers, interpolated to ``expected_count`` when necessary. + + Raises: + MicronicError: If fewer than two usable clusters can be found. + """ + if not values: + raise MicronicError("No decoded barcode positions are available for grid calibration.") + + clusters: list[list[float]] = [] + for value in sorted(values): + if not clusters: + clusters.append([value]) + continue + mean = sum(clusters[-1]) / len(clusters[-1]) + if abs(value - mean) > tolerance: + clusters.append([value]) + else: + clusters[-1].append(value) + + means = [sum(cluster) / len(cluster) for cluster in clusters] + if len(means) == expected_count: + return means + if len(means) >= 2: + return fitted_axis(means, expected_count) + raise MicronicError( + f"Could not fit {expected_count} grid clusters from {len(values)} decoded positions." + ) + + +def fitted_axis(means: list[float], expected_count: int) -> list[float]: + """Fit evenly spaced coordinates between the first and last cluster centers.""" + return [ + means[0] + index * (means[-1] - means[0]) / (expected_count - 1) + for index in range(expected_count) + ] + + +def rack_position(scan_row: int, scan_col: int) -> str: + """Map scanner-oriented grid coordinates to a rack position name.""" + return f"{ROWS[RACK_ROWS - 1 - scan_col]}{RACK_COLS - scan_row}" + + +def iter_positions() -> Iterable[str]: + """Yield every 8x12 rack position in row-major order.""" + for row in ROWS: + for column in range(1, COLS + 1): + yield f"{row}{column}" + + +def is_tube_id(value: object) -> bool: + """Return whether a decoded value is a ten-digit Micronic tube identifier.""" + return isinstance(value, str) and value.isdigit() and len(value) == 10 + + +def decode_well_crop( + image: Any, + center_x: float, + center_y: float, + cv2: Any, + np: Any, + zxingcpp: Any, + Image: Any, + ImageOps: Any, +) -> Optional[DecodeResult]: + """Decode one well using progressively larger direct and perspective-corrected crops.""" + for size in [150, 160, 180, 200, 220, 240]: + crop = centered_crop(image, center_x, center_y, size) + decoded = decode_pil_variants(crop, zxingcpp, ImageOps) + if decoded: + return DecodeResult(tube_id=decoded, method=f"crop-{size}") + + for size in [100, 120, 140, 160]: + crop = centered_crop(image, center_x, center_y, size) + decoded = decode_perspective_crop(crop, cv2, np, zxingcpp, Image, ImageOps) + if decoded: + return DecodeResult(tube_id=decoded, method=f"perspective-{size}") + + return None + + +def centered_crop(image: Any, center_x: float, center_y: float, size: int) -> Any: + """Return a square image crop centered on scanner coordinates.""" + half = size / 2 + return image.crop( + ( + int(round(center_x - half)), + int(round(center_y - half)), + int(round(center_x + half)), + int(round(center_y + half)), + ) + ) + + +def decode_pil_variants(crop: Any, zxingcpp: Any, ImageOps: Any) -> Optional[str]: + """Try original, autocontrasted, and equalized variants of an image crop.""" + for variant in [crop, ImageOps.autocontrast(crop), ImageOps.equalize(crop)]: + decoded = decode_with_zxing(variant, zxingcpp, ImageOps) + if decoded: + return decoded + return None + + +def decode_with_zxing(image: Any, zxingcpp: Any, ImageOps: Any) -> Optional[str]: + """Search image scales, polarity, borders, and binarizers for a tube identifier.""" + binarizers = [ + zxingcpp.Binarizer.LocalAverage, + zxingcpp.Binarizer.GlobalHistogram, + zxingcpp.Binarizer.FixedThreshold, + ] + for scale in [1, 2, 3, 4]: + scaled = image if scale == 1 else image.resize((image.width * scale, image.height * scale)) + for invert in [False, True]: + candidate = ImageOps.invert(scaled) if invert else scaled + for border in [0, 20, 50]: + padded = ImageOps.expand(candidate, border=border, fill=255) if border else candidate + for binarizer in binarizers: + for pure in [False, True]: + results = zxingcpp.read_barcodes( + padded, + formats=zxingcpp.BarcodeFormat.DataMatrix, + try_rotate=True, + try_downscale=False, + try_invert=True, + binarizer=binarizer, + is_pure=pure, + ) + for result in results: + if is_tube_id(result.text): + return str(result.text) + return None + + +def order_box(points: Any, np: Any) -> Any: + """Order four rectangle corners clockwise from the top-left corner.""" + points = np.array(points, dtype=np.float32) + sums = points.sum(axis=1) + diffs = np.diff(points, axis=1).ravel() + return np.array( + [ + points[np.argmin(sums)], + points[np.argmin(diffs)], + points[np.argmax(sums)], + points[np.argmax(diffs)], + ], + dtype=np.float32, + ) + + +def decode_perspective_crop( + crop: Any, + cv2: Any, + np: Any, + zxingcpp: Any, + Image: Any, + ImageOps: Any, +) -> Optional[str]: + """Locate, rectify, and decode a DataMatrix candidate within one well crop.""" + crop_array = np.array(crop) + for threshold in [30, 40, 50, 60, 70, 80, 90, 100, 120, 140]: + mask = (crop_array < threshold).astype(np.uint8) * 255 + for candidate_mask in candidate_masks(mask, cv2, np): + if not candidate_mask.any(): + continue + points = np.column_stack(np.where(candidate_mask > 0))[:, ::-1].astype(np.float32) + if len(points) < 40: + continue + rect = cv2.minAreaRect(points) + (rect_x, rect_y), (rect_w, rect_h), _angle = rect + if rect_w < 25 or rect_h < 25 or rect_w > crop.width * 0.9 or rect_h > crop.height * 0.9: + continue + if max(rect_w, rect_h) / max(1, min(rect_w, rect_h)) > 2: + continue + + box = cv2.boxPoints(rect) + center = np.array([rect_x, rect_y], dtype=np.float32) + for margin in [0.9, 1.0, 1.1, 1.2, 1.35]: + source = order_box((box - center) * margin + center, np) + for output_size in [60, 80, 100, 120, 160]: + destination = np.array( + [ + [0, 0], + [output_size - 1, 0], + [output_size - 1, output_size - 1], + [0, output_size - 1], + ], + dtype=np.float32, + ) + matrix = cv2.getPerspectiveTransform(source, destination) + warped = cv2.warpPerspective( + crop_array, matrix, (output_size, output_size), borderValue=255 + ) + for mode_array in perspective_variants(warped, threshold, cv2, Image, ImageOps): + decoded = decode_with_zxing(mode_array, zxingcpp, ImageOps) + if decoded: + return decoded + return None + + +def candidate_masks(mask: Any, cv2: Any, np: Any) -> Iterator[Any]: + """Yield the raw threshold mask and its centered connected components.""" + yield mask + number, labels, stats, centroids = cv2.connectedComponentsWithStats(mask, 8) + combined = np.zeros_like(mask) + size = mask.shape[0] + for index in range(1, number): + _x, _y, width, height, area = stats[index] + center_x, center_y = centroids[index] + if area < 15 or width < 8 or height < 8: + continue + if abs(center_x - size / 2) > size * 0.33 or abs(center_y - size / 2) > size * 0.33: + continue + if width > size * 0.85 or height > size * 0.85: + continue + combined[labels == index] = 255 + yield combined + + +def perspective_variants( + warped: Any, + threshold: int, + cv2: Any, + Image: Any, + ImageOps: Any, +) -> Iterator[Any]: + """Yield grayscale and binary variants of a rectified DataMatrix candidate.""" + yield Image.fromarray(warped) + yield ImageOps.autocontrast(Image.fromarray(warped)) + _, binary = cv2.threshold(warped, min(220, threshold + 70), 255, cv2.THRESH_BINARY) + yield Image.fromarray(binary) + yield Image.fromarray(255 - binary) + + +def find_duplicate_ids(decoded: dict[str, DecodeResult]) -> list[str]: + """Return tube identifiers assigned to more than one rack position.""" + seen: dict[str, str] = {} + duplicates: list[str] = [] + for position, result in decoded.items(): + previous = seen.get(result.tube_id) + if previous and previous != position: + duplicates.append(result.tube_id) + seen[result.tube_id] = position + return sorted(set(duplicates)) diff --git a/pylabrobot/micronic/code_reader/errors.py b/pylabrobot/micronic/code_reader/errors.py new file mode 100644 index 00000000000..ac8d9529598 --- /dev/null +++ b/pylabrobot/micronic/code_reader/errors.py @@ -0,0 +1,2 @@ +class MicronicError(Exception): + """Raised when Micronic driver or scanner operations fail.""" diff --git a/pylabrobot/micronic/code_reader/micronic_tests.py b/pylabrobot/micronic/code_reader/micronic_tests.py new file mode 100644 index 00000000000..fbdc647f3be --- /dev/null +++ b/pylabrobot/micronic/code_reader/micronic_tests.py @@ -0,0 +1,539 @@ +import asyncio +import os +import tempfile +import unittest +from pathlib import Path +from typing import cast +from unittest.mock import AsyncMock, MagicMock, patch + +from pylabrobot.io.command_line import CommandLineResult, CommandLineTransport +from pylabrobot.micronic import MicronicError, MicronicRD235, SaneScanner, TwainScanner +from pylabrobot.micronic.code_reader.driver import ( + DecodeResult, + cluster_axis, + decode_image, + find_duplicate_ids, + is_tube_id, + iter_positions, + rack_position, +) +from pylabrobot.resources.tube_rack import TubeRack + + +def _rack(num_items_x: int = 12, num_items_y: int = 8) -> TubeRack: + rack = MagicMock(spec=TubeRack) + rack.num_items_x = num_items_x + rack.num_items_y = num_items_y + return rack + + +def _mock_scanner(image_extension: str = "bmp") -> MagicMock: + """Return a scanner mock that succeeds without accessing hardware.""" + scanner = MagicMock() + scanner.image_extension = image_extension + scanner.setup = AsyncMock() + scanner.stop = AsyncMock() + scanner.acquire = AsyncMock(return_value={"source": "test"}) + return scanner + + +def _mock_command_line( + executable: str, + result: CommandLineResult = CommandLineResult(0, "", ""), +) -> MagicMock: + """Return a configured command-line I/O mock.""" + command_line = MagicMock(spec=CommandLineTransport) + command_line.executable = executable + command_line.setup = AsyncMock() + command_line.stop = AsyncMock() + command_line.run = AsyncMock(return_value=result) + return command_line + + +def _run_inline(_executor, function, *args): + """Execute an executor callback inline and return its result as a future.""" + future = asyncio.get_running_loop().create_future() + try: + future.set_result(function(*args)) + except Exception as exc: + future.set_exception(exc) + return future + + +class TestScannerClasses(unittest.IsolatedAsyncioTestCase): + """Tests for scanner command construction and error handling.""" + + async def test_sane_scanner_invokes_scanimage(self): + with tempfile.TemporaryDirectory() as image_dir: + output_path = Path(image_dir) / "rack.tiff" + output_path.touch() + command_line = _mock_command_line("/usr/bin/scanimage") + scanner = SaneScanner( + sane_device="avision:libusb:001:004", + command_line=command_line, + ) + await scanner.setup() + metadata = await scanner.acquire(output_path, timeout=1.0) + await scanner.stop() + + self.assertEqual(metadata["source"], "sane") + self.assertEqual(scanner.image_extension, "tiff") + command_line.setup.assert_awaited_once_with() + command_line.stop.assert_awaited_once_with() + command_line.run.assert_awaited_once_with( + [ + "--device-name", + "avision:libusb:001:004", + "--format=tiff", + "--output-file", + str(output_path), + ], + timeout=16.0, + ) + + def test_sane_scanner_raises_when_scanimage_missing(self): + with patch("pylabrobot.micronic.code_reader.scanner.shutil.which", return_value=None): + with self.assertRaises(MicronicError): + SaneScanner() + + def test_twain_scanner_resolves_path_from_env(self): + with ( + patch.dict(os.environ, {"MICRONIC_TWAIN_SCANNER_PATH": "/opt/twain_scan"}, clear=False), + patch("pylabrobot.micronic.code_reader.scanner.shutil.which", return_value=None), + ): + scanner = TwainScanner() + self.assertEqual(scanner.twain_scanner_path, "/opt/twain_scan") + + def test_twain_scanner_raises_when_helper_missing(self): + with ( + patch.dict(os.environ, {}, clear=True), + patch("pylabrobot.micronic.code_reader.scanner.shutil.which", return_value=None), + ): + with self.assertRaises(MicronicError): + TwainScanner() + + async def test_twain_scanner_acquire_runs_helper(self): + with tempfile.TemporaryDirectory() as image_dir: + output_path = Path(image_dir) / "rack.bmp" + output_path.touch() + command_line = _mock_command_line("/opt/twain_scan") + scanner = TwainScanner( + twain_source="AVA6PlusG", + command_line=command_line, + ) + await scanner.setup() + await scanner.acquire(output_path, timeout=1.25) + await scanner.stop() + + command_line.run.assert_awaited_once_with( + [str(output_path), "AVA6PlusG", "1250"], + timeout=16.25, + ) + + async def test_scanner_raises_on_helper_failure(self): + command_line = _mock_command_line( + "/opt/twain_scan", + result=CommandLineResult(2, "", "scanner fault"), + ) + scanner = TwainScanner(command_line=command_line) + + with self.assertRaisesRegex(MicronicError, "scanner fault"): + await scanner.acquire(Path("rack.bmp"), timeout=1.0) + + async def test_scanner_raises_when_helper_creates_no_image(self): + command_line = _mock_command_line("/opt/twain_scan") + scanner = TwainScanner(command_line=command_line) + + with tempfile.TemporaryDirectory() as image_dir: + with self.assertRaisesRegex(MicronicError, "did not create image"): + await scanner.acquire(Path(image_dir) / "rack.bmp", timeout=1.0) + + async def test_scanner_wraps_helper_timeout(self): + command_line = _mock_command_line("/opt/twain_scan") + command_line.run = AsyncMock(side_effect=asyncio.TimeoutError) + scanner = TwainScanner(command_line=command_line) + + with self.assertRaisesRegex(MicronicError, "timed out"): + await scanner.acquire(Path("rack.bmp"), timeout=1.0) + + async def test_scanner_wraps_missing_helper(self): + command_line = _mock_command_line("/missing/twain_scan") + command_line.run = AsyncMock(side_effect=FileNotFoundError) + scanner = TwainScanner(command_line=command_line) + + with self.assertRaisesRegex(MicronicError, "was not found"): + await scanner.acquire(Path("rack.bmp"), timeout=1.0) + + +class TestMicronicRD235(unittest.IsolatedAsyncioTestCase): + """Tests for the public Micronic code-reader operations.""" + + def test_rejects_non_positive_device_timeouts(self): + with self.assertRaisesRegex(ValueError, "scanner_timeout"): + MicronicRD235( + scanner=_mock_scanner(), + serial_port="/dev/ttyUSB0", + scanner_timeout=0, + ) + with self.assertRaisesRegex(ValueError, "serial_timeout"): + MicronicRD235( + scanner=_mock_scanner(), + serial_port="/dev/ttyUSB0", + serial_timeout=0, + ) + + async def test_acquire_image_runs_scanner_and_tracks_metadata(self): + with tempfile.TemporaryDirectory() as image_dir: + scanner = _mock_scanner() + reader = MicronicRD235( + scanner=scanner, + serial_port="/dev/ttyUSB0", + image_dir=image_dir, + scanner_timeout=1.25, + keep_images=True, + ) + image_path = await reader._acquire_image() + + self.assertEqual(reader.last_image_path, image_path) + self.assertEqual(reader.last_scan_metadata, {"source": "test"}) + scanner.acquire.assert_awaited_once_with(image_path, 1.25) + self.assertTrue(image_path.name.startswith("micronic_")) + self.assertEqual(image_path.suffix, ".bmp") + + async def test_acquire_image_removes_partial_image_after_failure(self): + async def fail_after_writing(output_path: Path, timeout: float) -> None: + """Create a partial output image before reporting acquisition failure.""" + del timeout + output_path.touch() + raise MicronicError("acquisition failed") + + with tempfile.TemporaryDirectory() as image_dir: + scanner = _mock_scanner() + scanner.acquire = AsyncMock(side_effect=fail_after_writing) + reader = MicronicRD235( + scanner=scanner, + serial_port="/dev/ttyUSB0", + image_dir=image_dir, + ) + + with self.assertRaisesRegex(MicronicError, "acquisition failed"): + await reader._acquire_image() + + self.assertEqual(list(Path(image_dir).iterdir()), []) + + async def test_scan_rack_id_uses_plr_serial(self): + instances: list[object] = [] + + class FakeSerial: + def __init__(self, **kwargs): + self.kwargs = kwargs + self.reads = iter([b"9", b"5", b"0", b"0", b"0", b"1", b"7", b"7", b"2", b"2", b"\r"]) + self.calls: list[str] = [] + instances.append(self) + + async def setup(self): + self.calls.append("setup") + + async def reset_input_buffer(self): + self.calls.append("reset_input_buffer") + + async def write(self, data: bytes): + self.calls.append(f"write:{data!r}") + + async def read(self, num_bytes: int = 1) -> bytes: + self.calls.append(f"read:{num_bytes}") + return next(self.reads) + + async def stop(self): + self.calls.append("stop") + + with patch("pylabrobot.micronic.code_reader.driver.Serial", FakeSerial): + scanner = _mock_scanner() + reader = MicronicRD235(scanner=scanner, serial_port="/dev/ttyUSB0") + await reader.setup() + try: + rack_id = await reader.scan_rack_id(timeout=1.0) + finally: + await reader.stop() + + self.assertEqual(len(instances), 1) + fake_serial = cast(FakeSerial, instances[0]) + self.assertEqual(rack_id, "9500017722") + self.assertEqual(fake_serial.kwargs["port"], "/dev/ttyUSB0") + self.assertEqual(fake_serial.kwargs["bytesize"], 7) + self.assertEqual(fake_serial.kwargs["parity"], "E") + self.assertIn("setup", fake_serial.calls) + self.assertIn("reset_input_buffer", fake_serial.calls) + self.assertIn("write:b'\\r\\n'", fake_serial.calls) + self.assertEqual(fake_serial.calls[-1], "stop") + scanner.setup.assert_awaited_once_with() + scanner.stop.assert_awaited_once_with() + + async def test_setup_stops_scanner_when_serial_setup_fails(self): + scanner = _mock_scanner() + reader = MicronicRD235(scanner=scanner, serial_port="/dev/ttyUSB0") + + with patch.object(reader.io, "setup", AsyncMock(side_effect=OSError("serial failed"))): + with self.assertRaisesRegex(OSError, "serial failed"): + await reader.setup() + + scanner.setup.assert_awaited_once_with() + scanner.stop.assert_awaited_once_with() + + async def test_stop_releases_scanner_when_serial_stop_fails(self): + scanner = _mock_scanner() + reader = MicronicRD235(scanner=scanner, serial_port="/dev/ttyUSB0") + + with patch.object(reader.io, "stop", AsyncMock(side_effect=OSError("serial failed"))): + with self.assertRaisesRegex(OSError, "serial failed"): + await reader.stop() + + scanner.stop.assert_awaited_once_with() + + async def test_scan_rack_populates_result(self): + with tempfile.TemporaryDirectory() as image_dir: + scanner = _mock_scanner() + reader = MicronicRD235( + scanner=scanner, + serial_port="/dev/ttyUSB0", + image_dir=image_dir, + keep_images=True, + ) + decoded = { + "A1": DecodeResult(tube_id="1111111111", method="test"), + "A2": DecodeResult(tube_id="2222222222", method="test"), + } + loop = asyncio.get_running_loop() + with ( + patch.object(reader, "_read_barcode", AsyncMock(return_value="9500017722")) as read_barcode, + patch.object(loop, "run_in_executor", side_effect=_run_inline), + patch( + "pylabrobot.micronic.code_reader.driver.decode_image", + return_value=(decoded, {"decodedWells": 2}), + ) as decode_image_mock, + ): + result = await reader.scan_rack(_rack(), timeout=1.0) + + self.assertEqual(result.rack_id, "9500017722") + rack_barcode = result.rack_barcode + assert rack_barcode is not None + self.assertEqual(rack_barcode.data, "9500017722") + self.assertEqual(rack_barcode.symbology, "Code 128 (Subset B and C)") + self.assertEqual(result.entries[0].position, "A1") + self.assertEqual(result.entries[0].tube_id, "1111111111") + tube_barcode = result.entries[0].barcode + assert tube_barcode is not None + self.assertEqual(tube_barcode.data, "1111111111") + self.assertEqual(tube_barcode.symbology, "DataMatrix") + self.assertEqual(result.entries[1].tube_id, "2222222222") + self.assertEqual(reader.last_scan_metadata, {"source": "test"}) + self.assertEqual(reader.last_decode_metadata, {"decodedWells": 2}) + scanner.acquire.assert_awaited_once() + read_barcode.assert_awaited_once() + decode_image_mock.assert_called_once() + + async def test_reader_can_scan_twice(self): + with tempfile.TemporaryDirectory() as image_dir: + scanner = _mock_scanner() + reader = MicronicRD235( + scanner=scanner, + serial_port="/dev/ttyUSB0", + image_dir=image_dir, + keep_images=True, + ) + decoded = {"A1": DecodeResult(tube_id="1111111111", method="test")} + loop = asyncio.get_running_loop() + with ( + patch.object(reader.io, "setup", AsyncMock()), + patch.object(reader.io, "stop", AsyncMock()), + patch.object(reader, "_read_barcode", AsyncMock(return_value="9500017722")), + patch.object(loop, "run_in_executor", side_effect=_run_inline), + patch( + "pylabrobot.micronic.code_reader.driver.decode_image", + return_value=(decoded, {"decodedWells": 1}), + ), + ): + await reader.setup() + try: + first = await reader.scan_rack(rack=_rack(), timeout=1.0) + second = await reader.scan_rack(rack=_rack(), timeout=1.0) + finally: + await reader.stop() + + self.assertEqual(first.rack_id, "9500017722") + self.assertEqual(second.rack_id, "9500017722") + self.assertEqual(scanner.acquire.await_count, 2) + + async def test_rejects_mismatched_rack_shape(self): + reader = MicronicRD235(scanner=_mock_scanner(), serial_port="/dev/ttyUSB0") + with self.assertRaises(MicronicError): + await reader.scan_rack(_rack(num_items_x=6, num_items_y=4), timeout=1.0) + + async def test_rejects_concurrent_scan(self): + reader = MicronicRD235(scanner=_mock_scanner(), serial_port="/dev/ttyUSB0") + await reader._scan_lock.acquire() + try: + with self.assertRaises(MicronicError): + await reader.scan_rack(_rack(), timeout=1.0) + finally: + reader._scan_lock.release() + + async def test_scan_rack_times_out(self): + reader = MicronicRD235(scanner=_mock_scanner(), serial_port="/dev/ttyUSB0") + + async def slow(rack): + del rack + await asyncio.sleep(1) + return MagicMock() + + with patch.object(reader, "_scan_rack", slow): + with self.assertRaises(asyncio.TimeoutError): + await reader.scan_rack(rack=_rack(), timeout=0.01) + + async def test_timeout_keeps_scan_lock_until_blocking_scan_finishes(self): + reader = MicronicRD235(scanner=_mock_scanner(), serial_port="/dev/ttyUSB0") + loop = asyncio.get_running_loop() + scan_future = loop.create_future() + loop.call_later(0.05, scan_future.set_result, MagicMock()) + + with ( + patch.object(reader, "_read_barcode", AsyncMock(return_value="9500017722")), + patch.object(loop, "run_in_executor", return_value=scan_future), + ): + with self.assertRaises(asyncio.TimeoutError): + await reader.scan_rack(rack=_rack(), timeout=0.01) + with self.assertRaises(MicronicError): + await reader.scan_rack(rack=_rack(), timeout=0.01) + + await asyncio.sleep(0.08) + self.assertFalse(reader._scan_lock.locked()) + + async def test_scan_rack_propagates_micronic_error(self): + reader = MicronicRD235(scanner=_mock_scanner(), serial_port="/dev/ttyUSB0") + with self.assertRaises(MicronicError): + await reader.scan_rack( + rack=_rack(num_items_x=6, num_items_y=4), + timeout=1.0, + ) + + async def test_scan_rack_id_reads_barcode(self): + reader = MicronicRD235(scanner=_mock_scanner(), serial_port="/dev/ttyUSB0") + with patch.object( + reader, "_read_barcode", AsyncMock(return_value="9500017722") + ) as read_barcode: + rack_id = await reader.scan_rack_id(timeout=5.0) + + self.assertEqual(rack_id, "9500017722") + read_barcode.assert_awaited_once_with() + + async def test_scan_rack_id_returns_noread_for_unrecognized_response(self): + reader = MicronicRD235(scanner=_mock_scanner(), serial_port="/dev/ttyUSB0") + with ( + patch.object(reader.io, "reset_input_buffer", AsyncMock()), + patch.object(reader.io, "write", AsyncMock()), + patch.object(reader.io, "read", AsyncMock(side_effect=[b"?", b"\r"])), + ): + rack_id = await reader.scan_rack_id(timeout=1.0) + + self.assertEqual(rack_id, "NOREAD") + + async def test_scan_rack_id_wraps_serial_error(self): + reader = MicronicRD235(scanner=_mock_scanner(), serial_port="/dev/ttyUSB0") + with patch.object( + reader.io, + "reset_input_buffer", + AsyncMock(side_effect=OSError("port disconnected")), + ): + with self.assertRaisesRegex(MicronicError, "port disconnected"): + await reader.scan_rack_id(timeout=1.0) + + def test_decode_rack_returns_noread_for_missing_wells(self): + reader = MicronicRD235(scanner=_mock_scanner(), serial_port="/dev/ttyUSB0") + decoded = {"A1": DecodeResult(tube_id="1111111111", method="test")} + + with patch( + "pylabrobot.micronic.code_reader.driver.decode_image", + return_value=(decoded, {"decodedWells": 1}), + ): + result = reader._decode_rack_image(Path("rack.bmp"), "9500017722") + + self.assertEqual(len(result.entries), 96) + self.assertEqual(result.entries[0].status, "OK") + self.assertEqual(result.entries[1].status, "NOREAD") + self.assertIsNone(result.entries[1].tube_id) + self.assertIsNone(result.entries[1].barcode) + + def test_decode_rack_represents_missing_rack_id(self): + reader = MicronicRD235(scanner=_mock_scanner(), serial_port="/dev/ttyUSB0") + decoded = {"A1": DecodeResult(tube_id="1111111111", method="test")} + + with patch( + "pylabrobot.micronic.code_reader.driver.decode_image", + return_value=(decoded, {"decodedWells": 1}), + ): + result = reader._decode_rack_image(Path("rack.bmp"), "NOREAD") + + self.assertIsNone(result.rack_barcode) + self.assertEqual(result.entries[0].status, "OK") + self.assertEqual(result.entries[1].status, "NOREAD") + + +class TestDecodeHelpers(unittest.TestCase): + """Tests for image-decoding validation and rack-coordinate parsing.""" + + def test_tube_id_validation(self): + self.assertTrue(is_tube_id("0123456789")) + self.assertFalse(is_tube_id("123456789")) + self.assertFalse(is_tube_id("12345A7890")) + self.assertFalse(is_tube_id(1234567890)) + + def test_rack_position_maps_scanner_orientation(self): + self.assertEqual(rack_position(scan_row=0, scan_col=0), "H12") + self.assertEqual(rack_position(scan_row=11, scan_col=7), "A1") + + def test_iter_positions_is_row_major(self): + positions = list(iter_positions()) + self.assertEqual(len(positions), 96) + self.assertEqual(positions[:2], ["A1", "A2"]) + self.assertEqual(positions[-2:], ["H11", "H12"]) + + def test_cluster_axis_uses_detected_centers(self): + self.assertEqual(cluster_axis([0.0, 2.0, 100.0, 102.0], 2, 10.0), [1.0, 101.0]) + + def test_cluster_axis_interpolates_missing_centers(self): + self.assertEqual(cluster_axis([0.0, 50.0, 100.0], 5, 10.0), [0.0, 25.0, 50.0, 75.0, 100.0]) + + def test_cluster_axis_rejects_insufficient_data(self): + with self.assertRaisesRegex(MicronicError, "Could not fit"): + cluster_axis([1.0, 2.0], 8, 10.0) + + def test_duplicate_ids_are_sorted_and_unique(self): + decoded = { + "A1": DecodeResult("2222222222", "test"), + "A2": DecodeResult("1111111111", "test"), + "A3": DecodeResult("2222222222", "test"), + "A4": DecodeResult("1111111111", "test"), + "A5": DecodeResult("1111111111", "test"), + } + self.assertEqual(find_duplicate_ids(decoded), ["1111111111", "2222222222"]) + + def test_decode_image_rejects_too_few_full_image_codes(self): + image = MagicMock() + image.size = (100, 100) + image_context = MagicMock() + image_context.__enter__.return_value.convert.return_value = image + image_module = MagicMock() + image_module.open.return_value = image_context + zxingcpp = MagicMock() + zxingcpp.read_barcodes.return_value = [] + + with patch( + "pylabrobot.micronic.code_reader.driver.import_decode_dependencies", + return_value=(MagicMock(), MagicMock(), zxingcpp, image_module, MagicMock()), + ): + with self.assertRaisesRegex(MicronicError, "Only 0 DataMatrix"): + decode_image(Path("rack.bmp")) + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/micronic/code_reader/scanner.py b/pylabrobot/micronic/code_reader/scanner.py new file mode 100644 index 00000000000..d52e256b1e0 --- /dev/null +++ b/pylabrobot/micronic/code_reader/scanner.py @@ -0,0 +1,238 @@ +"""Scanner classes that acquire a rack image for the Micronic driver.""" + +from __future__ import annotations + +import asyncio +import os +import shutil +from abc import ABC, abstractmethod +from pathlib import Path +from typing import Optional, Sequence + +from pylabrobot.io.command_line import CommandLineTransport + +from .errors import MicronicError + + +class Scanner(ABC): + """Abstract scanner that writes a rack image to disk on demand.""" + + image_extension: str + + async def setup(self) -> None: + """Prepare the scanner for image acquisition.""" + + async def stop(self) -> None: + """Release scanner acquisition resources.""" + + @abstractmethod + async def acquire(self, output_path: Path, timeout: float) -> dict[str, object]: + """Write a rack image to ``output_path``. + + Args: + output_path: Destination for the acquired image. + timeout: Scanner acquisition timeout in seconds. + + Returns: + Metadata describing the scanner command. + """ + + +class TwainScanner(Scanner): + """Windows TWAIN scanner driven by an operator-installed helper executable. + + Resolves the helper path from (in order): the ``twain_scanner_path`` argument, + the ``MICRONIC_TWAIN_SCANNER_PATH`` environment variable, or ``twain_scan`` / + ``twain_scan.exe`` on PATH. Raises ``MicronicError`` if none resolve. + """ + + image_extension = "bmp" + + def __init__( + self, + twain_scanner_path: Optional[str] = None, + twain_source: str = "AVA6PlusG", + command_line: Optional[CommandLineTransport] = None, + ) -> None: + """Initialize a TWAIN scanner. + + Args: + twain_scanner_path: Path to the operator-installed TWAIN helper. When omitted, resolve it + from ``MICRONIC_TWAIN_SCANNER_PATH`` or ``PATH``. + twain_source: TWAIN source name passed to the helper. + command_line: Configured command-line transport. When supplied, its executable takes + precedence over ``twain_scanner_path``. + + Raises: + MicronicError: If no TWAIN helper can be resolved. + """ + if command_line is None: + resolved = twain_scanner_path or _resolve_twain_scanner_path() + if resolved is None: + raise MicronicError( + "No TWAIN helper was found. Pass twain_scanner_path, set " + "MICRONIC_TWAIN_SCANNER_PATH, or put twain_scan on PATH." + ) + command_line = CommandLineTransport( + human_readable_device_name="Micronic TWAIN rack scanner", + executable=resolved, + ) + self.command_line = command_line + self.twain_scanner_path = command_line.executable + self.twain_source = twain_source + + async def setup(self) -> None: + """Resolve and prepare the TWAIN helper executable.""" + try: + await self.command_line.setup() + except FileNotFoundError as exc: + raise MicronicError(str(exc)) from exc + self.twain_scanner_path = self.command_line.executable + + async def stop(self) -> None: + """Stop the TWAIN helper transport.""" + await self.command_line.stop() + + async def acquire(self, output_path: Path, timeout: float) -> dict[str, object]: + """Acquire a BMP image through the configured TWAIN helper. + + Args: + output_path: Destination for the acquired image. + timeout: Scanner acquisition timeout in seconds. + + Returns: + Metadata describing the scanner command. + """ + timeout_ms = max(1, int(timeout * 1000)) + arguments = [str(output_path), self.twain_source, str(timeout_ms)] + return await _run_scan_command( + self.command_line, + arguments, + output_path, + timeout, + source="twain", + ) + + +class SaneScanner(Scanner): + """Linux SANE scanner driven through the ``scanimage`` CLI.""" + + image_extension = "tiff" + + def __init__( + self, + sane_device: Optional[str] = None, + scanimage_path: Optional[str] = None, + command_line: Optional[CommandLineTransport] = None, + ) -> None: + """Initialize a SANE scanner. + + Args: + sane_device: Optional SANE device identifier passed to ``scanimage``. + scanimage_path: Path to ``scanimage``. When omitted, resolve it from ``PATH``. + command_line: Configured command-line transport. When supplied, its executable takes + precedence over ``scanimage_path``. + + Raises: + MicronicError: If ``scanimage`` cannot be resolved. + """ + if command_line is None: + resolved = scanimage_path or shutil.which("scanimage") + if resolved is None: + raise MicronicError("scanimage was not found on PATH. Install SANE or pass scanimage_path.") + command_line = CommandLineTransport( + human_readable_device_name="Micronic SANE rack scanner", + executable=resolved, + ) + self.command_line = command_line + self.scanimage_path = command_line.executable + self.sane_device = sane_device + + async def setup(self) -> None: + """Resolve and prepare the ``scanimage`` executable.""" + try: + await self.command_line.setup() + except FileNotFoundError as exc: + raise MicronicError(str(exc)) from exc + self.scanimage_path = self.command_line.executable + + async def stop(self) -> None: + """Stop the SANE command-line transport.""" + await self.command_line.stop() + + async def acquire(self, output_path: Path, timeout: float) -> dict[str, object]: + """Acquire a TIFF image through ``scanimage``. + + Args: + output_path: Destination for the acquired image. + timeout: Scanner acquisition timeout in seconds. + + Returns: + Metadata describing the scanner command. + """ + arguments: list[str] = [] + if self.sane_device: + arguments.extend(["--device-name", self.sane_device]) + arguments.extend(["--format=tiff", "--output-file", str(output_path)]) + return await _run_scan_command( + self.command_line, + arguments, + output_path, + timeout, + source="sane", + ) + + +async def _run_scan_command( + command_line: CommandLineTransport, + arguments: Sequence[str], + output_path: Path, + timeout: float, + source: str, +) -> dict[str, object]: + """Run a scanner helper and validate its output image. + + Args: + command_line: Transport used to execute the scanner helper. + arguments: Arguments for the configured helper executable. + output_path: Image path the helper must create. + timeout: Scanner acquisition timeout in seconds. + source: Scanner backend name stored in the returned metadata. + + Returns: + Scanner command metadata. + + Raises: + MicronicError: If the helper is missing, times out, fails, or creates no image. + """ + try: + completed = await command_line.run(arguments, timeout=timeout + 15) + except FileNotFoundError as exc: + raise MicronicError(f"Scan command was not found: {command_line.executable}") from exc + except asyncio.TimeoutError as exc: + raise MicronicError( + f"Scan command timed out after {timeout:g} seconds: {command_line.executable}" + ) from exc + + if completed.returncode != 0: + raise MicronicError( + "Scan command failed with exit code " + f"{completed.returncode}: {completed.stderr.strip() or completed.stdout.strip()}" + ) + if not output_path.exists(): + raise MicronicError(f"Scan command did not create image: {output_path}") + return { + "stdout": completed.stdout.strip(), + "stderr": completed.stderr.strip(), + "source": source, + "command": [command_line.executable, *arguments], + } + + +def _resolve_twain_scanner_path() -> Optional[str]: + """Resolve the operator-installed TWAIN helper path, if available.""" + return ( + os.environ.get("MICRONIC_TWAIN_SCANNER_PATH") + or shutil.which("twain_scan.exe") + or shutil.which("twain_scan") + )