From abb168f9da243cd492f760e19a6e40a471033a73 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Mon, 20 Jul 2026 18:33:56 -0700 Subject: [PATCH 1/8] feat(azenta): add FluidX IntelliXcap 96 decapper driver FluidXIntelliXcap96 drives the FluidX IntelliXcap 96 automated screw-cap decapper over serial. It decaps, holds, and recaps a 96-format rack of screw-cap tubes, drops held caps to waste, opens and closes the tray, homes, resets, and enters/leaves standby. Serial 9600 8N1, no handshake, ETX (0x03) line terminator; motions are acknowledged with a single 0x06 byte and progress is tracked by polling the status command. Single plain class under pylabrobot/azenta/fluidx/, with docs and wire-protocol tests against a stubbed serial. Not hardware-tested: setup() emits a not-verified warning. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/api/pylabrobot.azenta.rst | 14 + pylabrobot/azenta/__init__.py | 1 + pylabrobot/azenta/fluidx/__init__.py | 1 + pylabrobot/azenta/fluidx/intellixcap96.py | 370 ++++++++++++++++++ .../azenta/fluidx/intellixcap96_tests.py | 178 +++++++++ 5 files changed, 564 insertions(+) create mode 100644 pylabrobot/azenta/fluidx/__init__.py create mode 100644 pylabrobot/azenta/fluidx/intellixcap96.py create mode 100644 pylabrobot/azenta/fluidx/intellixcap96_tests.py diff --git a/docs/api/pylabrobot.azenta.rst b/docs/api/pylabrobot.azenta.rst index 9012ce85723..acd657f06d8 100644 --- a/docs/api/pylabrobot.azenta.rst +++ b/docs/api/pylabrobot.azenta.rst @@ -45,3 +45,17 @@ XPeel Peeler .. autoclass:: pylabrobot.azenta.xpeel.XPeelPeelerBackend.PeelParams :members: + + +FluidX IntelliXcap 96 +--------------------- + +.. currentmodule:: pylabrobot.azenta.fluidx.intellixcap96 + +.. autosummary:: + :toctree: _autosummary + :nosignatures: + :recursive: + + FluidXIntelliXcap96 + FluidXError diff --git a/pylabrobot/azenta/__init__.py b/pylabrobot/azenta/__init__.py index b00149e5035..169014bd5ac 100644 --- a/pylabrobot/azenta/__init__.py +++ b/pylabrobot/azenta/__init__.py @@ -1,2 +1,3 @@ from .a4s import A4S, A4SDriver, A4SSealerBackend, A4SStatus, A4STemperatureBackend +from .fluidx import FluidXError, FluidXIntelliXcap96 from .xpeel import XPeel, XPeelDriver, XPeelPeelerBackend diff --git a/pylabrobot/azenta/fluidx/__init__.py b/pylabrobot/azenta/fluidx/__init__.py new file mode 100644 index 00000000000..4099b5a19ef --- /dev/null +++ b/pylabrobot/azenta/fluidx/__init__.py @@ -0,0 +1 @@ +from .intellixcap96 import FluidXError, FluidXIntelliXcap96 diff --git a/pylabrobot/azenta/fluidx/intellixcap96.py b/pylabrobot/azenta/fluidx/intellixcap96.py new file mode 100644 index 00000000000..8c428dade3b --- /dev/null +++ b/pylabrobot/azenta/fluidx/intellixcap96.py @@ -0,0 +1,370 @@ +import asyncio +import logging +from typing import Optional, Tuple + +from pylabrobot.io.serial import Serial + +logger = logging.getLogger(__name__) + +# Commands and replies are terminated with ETX (0x03), not CR/LF. +ETX = b"\x03" +# Single-byte acknowledgement the firmware returns before a long-running motion. +ACK = "\x06" + +# Single-character commands the decapper understands. +STATUS = "a" +DECAP_START = "h" +RECAP_START = "i" +WASTE = "b" +OPEN_TRAY = "f" +CLOSE_TRAY = "g" +HOME_ALL = "Z" +RESET = "z" +STANDBY = "j" +READY = "k" + +# Descriptions for the fault conditions the firmware reports. +ERROR_MESSAGES = { + "StatusNotOK": ( + "Status is not ok. The device is in an error state. Check whether the plate is already " + "decapped (if so, recap and initialize again); otherwise clear the error on the device and " + "cycle the power." + ), + "NeedToRecap": "Decap operation is already done. Select the recap operation on the device.", + "NeedToDecap": "Recap operation is already executed.", + "DecapWasNotSuccesful": ( + "Decapping was not successful. Fix the error on the device manually and check whether any " + "tubes are still on the head." + ), + "RecapWasNotSuccesful": "Recapping was not successful. Fix the error on the device manually.", + "StoreWasNotSuccesful": "Storing was not successful. Fix the error on the device manually.", + "OpenTrayWasNotSuccesful": ( + "Opening the tray was not successful. Fix the error on the device manually." + ), + "CloseTrayWasNotSuccesful": ( + "Closing the tray was not successful. Fix the error on the device manually." + ), + "HomeNotSuccesful": ( + "Device was not able to reach the home position. The device is in an error state. " + "Restart the device." + ), + "CannotReset": ( + "Cannot send the reset command to the device. The device is in an error state. " + "Reset the device and try again." + ), + "ResetNotSuccesful": ( + "Reset did not complete. The device is in an error state. Reset the device and try again." + ), + "CannotGoInStandbyMode": "Cannot go to standby mode. Check the errors on the device.", + "NoAck": "Device did not acknowledge the command.", + "CommandIgnore": "Command was ignored by the device.", +} + + +class FluidXError(Exception): + """Exceptions raised by a FluidX IntelliXcap 96 decapper.""" + + def __init__(self, title: str, message: Optional[str] = None) -> None: + self.title = title + self.message = message + + def __str__(self) -> str: + return f"{self.title}: {self.message}" if self.message else self.title + + +def _fault(key: str) -> FluidXError: + return FluidXError(title=ERROR_MESSAGES.get(key, key)) + + +class FluidXIntelliXcap96: + """FluidX IntelliXcap 96 automated screw-cap decapper. + + A benchtop instrument that decaps and recaps a 96-format rack of screw-cap + tubes in a single stroke. It holds one nest; a plate mover loads the rack, the + decapper unscrews all 96 caps (``decap``), holds them, and screws them back on + (``recap``). Held caps can also be dropped to the waste bin (``waste``), and + the loading tray opened and closed. + + Serial settings: + 9600 baud, 8 data bits, no parity, 1 stop bit, no handshake, ETX (0x03) + terminator. + + Single-character commands (each is written followed by ETX; the reply is read + back up to the next ETX): + a request status + h start decapping + i start recapping + b drop held caps to waste + f open the tray + g close the tray + Z home all axes + z reset + j enter standby + k leave standby (ready) + A status reply contains ``OK``/``StatusOK`` when idle, ``StatusSleep`` in + standby, ``Recap``/``Decap`` for the pending inverse operation, and + ``StatusError``/``Error`` on a fault. Operation replies report ``...DONE`` on + success and ``...ERROR`` on failure. Motion commands are acknowledged with a + single 0x06 byte. + + Not verified: has NOT been tested against hardware in PyLabRobot. A warning is + emitted at setup. + """ + + def __init__( + self, + port: str, + timeout: float = 5.0, + command_delay: float = 1.0, + read_delay: float = 0.5, + poll_interval: float = 1.0, + ) -> None: + """ + Args: + port: serial port the decapper is connected to. + timeout: serial read/write timeout in seconds. + command_delay: pause after writing a command before reading its reply. + read_delay: pause after reading a reply. + poll_interval: pause between status polls while an operation runs. + """ + self.command_delay = command_delay + self.read_delay = read_delay + self.poll_interval = poll_interval + self.io = Serial( + human_readable_device_name="FluidX IntelliXcap 96", + port=port, + baudrate=9600, + bytesize=8, + parity="N", + stopbits=1, + timeout=timeout, + ) + + async def setup(self) -> None: + logger.warning( + "FluidXIntelliXcap96 has NOT been tested against hardware in PyLabRobot. " + "Please make a PR to remove this message if you have verified it on your hardware." + ) + await self.io.setup() + # Confirm the decapper is reachable and reports a healthy status. + await self._send(STATUS) + await self._read_line() + if "OK" not in await self._read_line(): + raise _fault("StatusNotOK") + logger.info("[IntelliXcap96 %s] connected", self.io.port) + + async def stop(self) -> None: + """Close the serial connection.""" + await self.io.stop() + + # === Command layer === + + async def _send(self, command: str) -> None: + """Discard any pending input, write a command with its terminator, then pace.""" + await self.io.reset_input_buffer() + await self.io.write(command.encode("ascii") + ETX) + logger.debug("[IntelliXcap96] send: %r", command) + await asyncio.sleep(self.command_delay) + + async def _read_line(self) -> str: + """Read one ETX-terminated reply and return it stripped.""" + buf = bytearray() + while True: + char = await self.io.read(1) + if char in (b"", ETX): # terminator or read timeout + break + buf += char + reply = buf.decode("ascii", errors="replace").strip() + logger.debug("[IntelliXcap96] recv: %r", reply) + await asyncio.sleep(self.read_delay) + return reply + + async def request_status(self) -> str: + """Poll the device and return its raw status reply.""" + await self._send(STATUS) + return await self._read_line() + + async def send_command(self, command: str) -> str: + """Send a single raw command and return its reply. Escape hatch for bring-up.""" + await self._send(command) + return await self._read_line() + + # === Operations === + + async def decap(self, timeout: float = 60.0) -> None: + """Unscrew and hold all 96 caps. + + Args: + timeout: maximum time in seconds to wait for the stroke to finish. + """ + status = await self.request_status() + if "Recap" in status: + raise _fault("NeedToRecap") + self._raise_on_error(status, "StatusNotOK") + + await self._send(DECAP_START) + await self._read_line() + await self._poll_until( + timeout=timeout, + done=("RECAP", "DONE"), + error=("DecapERROR", "StatusError", "Error"), + fault="DecapWasNotSuccesful", + ) + logger.info("[IntelliXcap96 %s] decap complete", self.io.port) + + async def recap(self, timeout: float = 50.0) -> None: + """Screw the held caps back on. + + Args: + timeout: maximum time in seconds to wait for the stroke to finish. + """ + status = await self.request_status() + if "Decap" in status: + raise _fault("NeedToDecap") + self._raise_on_error(status, "StatusNotOK") + + await self._send(RECAP_START) + if ACK not in await self._read_line(): + raise _fault("NoAck") + await self._poll_until( + timeout=timeout, + done=("RecapDONE", "StatusOK"), + error=("RecapERROR", "StatusError", "Error"), + fault="RecapWasNotSuccesful", + ignore=("CommandIgnore",), + ) + logger.info("[IntelliXcap96 %s] recap complete", self.io.port) + + async def waste(self, timeout: float = 60.0) -> None: + """Drop the currently held caps into the waste bin. + + Args: + timeout: maximum time in seconds to wait for the stroke to finish. + """ + status = await self.request_status() + self._raise_on_error(status, "StatusNotOK") + + await self._send(WASTE) + if ACK not in await self._read_line(): + raise _fault("NoAck") + await self._poll_until( + timeout=timeout, + done=("StoreDONE", "StatusOK"), + error=("StoreERROR", "StatusError", "Error"), + fault="StoreWasNotSuccesful", + ) + logger.info("[IntelliXcap96 %s] waste complete", self.io.port) + + async def open_tray(self, timeout: float = 10.0) -> None: + """Open the loading tray.""" + with self.io.temporary_timeout(timeout): + await self._move_tray( + OPEN_TRAY, ("OpenERROR", "StatusError", "Error"), "OpenTrayWasNotSuccesful" + ) + logger.info("[IntelliXcap96 %s] tray open", self.io.port) + + async def close_tray(self, timeout: float = 10.0) -> None: + """Close the loading tray.""" + with self.io.temporary_timeout(timeout): + await self._move_tray( + CLOSE_TRAY, ("CloseERROR", "StatusError", "Error"), "CloseTrayWasNotSuccesful" + ) + logger.info("[IntelliXcap96 %s] tray closed", self.io.port) + + async def home(self, timeout: float = 20.0) -> None: + """Home all axes.""" + with self.io.temporary_timeout(timeout): + await self._send(STATUS) + if ACK not in await self._read_line(): + raise _fault("NoAck") + if "OK" not in await self._read_line(): + raise _fault("StatusNotOK") + await self._send(HOME_ALL) + await self._read_line() + if "Error" in await self._read_line(): + raise _fault("HomeNotSuccesful") + logger.info("[IntelliXcap96 %s] homed", self.io.port) + + async def reset(self, settle_time: float = 5.0) -> None: + """Reset the device and wait for it to settle. + + Args: + settle_time: time in seconds to wait after the reset command. + """ + if "Error" in await self.request_status(): + raise _fault("CannotReset") + await self._send(RESET) + await asyncio.sleep(settle_time) + await self._read_line() + if "Error" in await self.request_status(): + raise _fault("ResetNotSuccesful") + logger.info("[IntelliXcap96 %s] reset complete", self.io.port) + + async def standby(self) -> None: + """Put the decapper into standby (sleep) mode.""" + status = await self.request_status() + if "Recap" not in status and ("OK" not in status or "Error" in status): + raise _fault("StatusNotOK") + await self._send(STANDBY) + if "Standby" not in await self._read_line(): + raise _fault("CannotGoInStandbyMode") + logger.info("[IntelliXcap96 %s] standby", self.io.port) + + async def ready(self) -> None: + """Wake the decapper from standby if it is asleep.""" + if "StatusSleep" not in await self.request_status(): + return + await self._send(READY) + if "Ready" not in await self._read_line(): + raise _fault("StatusNotOK") + logger.info("[IntelliXcap96 %s] ready", self.io.port) + + # === Operation helpers === + + async def _move_tray(self, command: str, error: Tuple[str, ...], fault: str) -> None: + status = await self.request_status() + self._raise_on_error(status, "StatusNotOK") + await self._send(command) + if ACK not in await self._read_line(): + raise _fault("NoAck") + # The tray move reports two follow-up lines; either may carry the error. + for _ in range(2): + reply = await self._read_line() + if any(token in reply for token in error): + raise _fault(fault) + if "CommandIgnore" in reply: + raise _fault("CommandIgnore") + + async def _poll_until( + self, + timeout: float, + done: Tuple[str, ...], + error: Tuple[str, ...], + fault: str, + ignore: Tuple[str, ...] = (), + ) -> None: + """Poll status until a done/error token appears or the timeout elapses. + + Each poll reads three status lines and inspects the last, matching the way + the firmware streams its progress. + """ + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while loop.time() < deadline: + await self._send(STATUS) + reply = "" + for _ in range(3): + reply = await self._read_line() + if any(token in reply for token in error): + raise _fault(fault) + if any(token in reply for token in ignore): + raise _fault("CommandIgnore") + if any(token in reply for token in done): + return + await asyncio.sleep(self.poll_interval) + raise FluidXError(title="Operation timed out", message=f"no completion within {timeout}s") + + @staticmethod + def _raise_on_error(status: str, fault: str) -> None: + if "StatusError" in status or "Error" in status: + raise _fault(fault) diff --git a/pylabrobot/azenta/fluidx/intellixcap96_tests.py b/pylabrobot/azenta/fluidx/intellixcap96_tests.py new file mode 100644 index 00000000000..cf1023cd21e --- /dev/null +++ b/pylabrobot/azenta/fluidx/intellixcap96_tests.py @@ -0,0 +1,178 @@ +import contextlib +import unittest +from typing import Iterator, List +from unittest.mock import AsyncMock, patch + +from pylabrobot.azenta.fluidx import FluidXError, FluidXIntelliXcap96 + + +class FakeXcapSerial: + """In-memory serial stand-in for the ETX-framed decapper protocol. + + ``script`` is a list of turns, one per ``write``. Each turn is the list of + reply lines the device emits in response to that write; every line is queued + ETX-terminated, exactly as the firmware frames its replies. + """ + + def __init__(self, script: List[List[str]]) -> None: + self.port = "FAKE" + self.written: List[str] = [] + self._script = list(script) + self._rx = bytearray() + + async def setup(self) -> None: + pass + + async def stop(self) -> None: + pass + + async def reset_input_buffer(self) -> None: + self._rx.clear() + + def get_read_timeout(self) -> float: + return 5.0 + + def set_read_timeout(self, timeout: float) -> None: + pass + + @contextlib.contextmanager + def temporary_timeout(self, timeout: float) -> Iterator[None]: + yield + + async def write(self, data: bytes) -> None: + self.written.append(data.decode("ascii").rstrip("\x03")) + lines = self._script.pop(0) if self._script else [] + for line in lines: + self._rx += line.encode("ascii") + b"\x03" + + async def read(self, num_bytes: int = 1) -> bytes: + if not self._rx: + return b"" + out = bytes(self._rx[:num_bytes]) + del self._rx[:num_bytes] + return out + + +class TestIntelliXcap96(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + patcher = patch("pylabrobot.azenta.fluidx.intellixcap96.asyncio.sleep", new_callable=AsyncMock) + patcher.start() + self.addCleanup(patcher.stop) + + def _make(self, script: List[List[str]]) -> FluidXIntelliXcap96: + device = FluidXIntelliXcap96(port="FAKE") + device.io = FakeXcapSerial(script) # type: ignore[assignment] + return device + + async def test_setup_checks_status_ok(self): + device = self._make([["boot", "StatusOK"]]) + await device.setup() + self.assertEqual(device.io.written, ["a"]) # type: ignore[attr-defined] + + async def test_setup_rejects_bad_status(self): + device = self._make([["boot", "StatusError"]]) + with self.assertRaises(FluidXError): + await device.setup() + + async def test_decap_success(self): + device = self._make( + [ + ["StatusOK"], # status precheck + ["ack"], # h (decap start) + ["l1", "l2", "DecapDONE"], # status poll -> 3 lines, done on the last + ] + ) + await device.decap() + self.assertEqual(device.io.written, ["a", "h", "a"]) # type: ignore[attr-defined] + + async def test_decap_blocked_when_recap_pending(self): + device = self._make([["StatusRecap"]]) + with self.assertRaises(FluidXError): + await device.decap() + + async def test_decap_reports_failure(self): + device = self._make( + [ + ["StatusOK"], + ["ack"], + ["l1", "l2", "DecapERROR"], + ] + ) + with self.assertRaises(FluidXError): + await device.decap() + + async def test_recap_requires_ack(self): + device = self._make( + [ + ["StatusOK"], # status precheck (no "Decap" pending) + ["no-ack-here"], # i reply lacks 0x06 + ] + ) + with self.assertRaises(FluidXError): + await device.recap() + + async def test_recap_success(self): + device = self._make( + [ + ["StatusOK"], + ["\x06"], # i acknowledged + ["l1", "l2", "RecapDONE"], + ] + ) + await device.recap() + self.assertEqual(device.io.written, ["a", "i", "a"]) # type: ignore[attr-defined] + + async def test_waste_success(self): + device = self._make( + [ + ["StatusOK"], + ["\x06"], # b acknowledged + ["l1", "l2", "StoreDONE"], + ] + ) + await device.waste() + self.assertEqual(device.io.written, ["a", "b", "a"]) # type: ignore[attr-defined] + + async def test_home_success(self): + device = self._make( + [ + ["\x06", "StatusOK"], # a: ack then OK + ["homing", "done"], # Z: discard then no "Error" + ] + ) + await device.home() + self.assertEqual(device.io.written, ["a", "Z"]) # type: ignore[attr-defined] + + async def test_open_tray_success(self): + device = self._make( + [ + ["StatusOK"], # status precheck + ["\x06", "OpenDONE", "StatusOK"], # f: ack + two follow-up lines + ] + ) + await device.open_tray() + self.assertEqual(device.io.written, ["a", "f"]) # type: ignore[attr-defined] + + async def test_reset_success(self): + device = self._make( + [ + ["StatusOK"], # a precheck (no error) + ["resetting"], # z + ["StatusOK"], # a postcheck (no error) + ] + ) + await device.reset() + self.assertEqual(device.io.written, ["a", "z", "a"]) # type: ignore[attr-defined] + + async def test_standby_and_ready(self): + device = self._make([["StatusOK"], ["StandbyDONE"]]) + await device.standby() + self.assertEqual(device.io.written, ["a", "j"]) # type: ignore[attr-defined] + + device = self._make([["StatusSleep"], ["ReadyDONE"]]) + await device.ready() + self.assertEqual(device.io.written, ["a", "k"]) # type: ignore[attr-defined] + + +if __name__ == "__main__": + unittest.main() From 4f981c8271144bbc838c1382047f485ac7a5f09f Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Mon, 20 Jul 2026 18:57:00 -0700 Subject: [PATCH 2/8] feat(azenta): hardware-verify IntelliXcap 96 protocol and add tutorial Rework the IntelliXcap 96 I/O to the protocol confirmed on hardware: replies are STX/ETX framed, and a command is answered with an ACK frame, a "OK" echo frame, and a result frame. A motion completes when the status word returns from StatusBUSY to StatusOK; a refused or no-op command answers CommandIgnore. setup() now reads the real status and, when the device reports StatusBUSY at connect, raises with an e-stop / safety-guard / interlock hint (the instrument locks out and ignores commands until it is cleared). Status words are matched case-insensitively (the firmware reports StatusSLEEP, StatusBUSY). Verified against hardware: connection, status, tray open/close, home, and standby/ready. Decap, recap, and waste are not yet verified; setup() warns. Add a hello-world tutorial notebook under docs/user_guide/azenta/fluidx. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../fluidx/intellixcap96/hello-world.ipynb | 200 ++++++++ docs/user_guide/azenta/index.md | 1 + pylabrobot/azenta/fluidx/intellixcap96.py | 429 ++++++++---------- .../azenta/fluidx/intellixcap96_tests.py | 168 ++++--- 4 files changed, 482 insertions(+), 316 deletions(-) create mode 100644 docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb diff --git a/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb b/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb new file mode 100644 index 00000000000..1bf37079dd8 --- /dev/null +++ b/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb @@ -0,0 +1,200 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# FluidX IntelliXcap 96\n", + "\n", + "The FluidX IntelliXcap 96 (an Azenta brand) is an automated screw-cap **decapper**. It unscrews, holds, and screws back on all 96 caps of a screw-cap tube rack in a single stroke. A plate mover loads the rack onto its one nest.\n", + "\n", + "| Property | Value |\n", + "|---|---|\n", + "| Communication | Serial, STX/ETX-framed ASCII |\n", + "| Serial settings | 9600 baud, 8 data bits, no parity, 1 stop bit, no handshake |\n", + "| Framing | each reply is wrapped in STX (`0x02`) .. ETX (`0x03`) |\n", + "| Reply pattern | ACK (`0x06`), then `OK` echo, then a result frame |\n", + "\n", + "Every command is a single character written followed by ETX. A motion command is accepted with an ACK and a `OK` echo; the status word then reads `StatusBUSY` while it moves and returns to `StatusOK` when done. A refused or no-op command answers `CommandIgnore`.\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "```{note}\n", + "Connection, status, tray open/close, home, and standby/ready are verified against\n", + "hardware. **Decap, recap, and waste engage the cap head and are not yet\n", + "hardware-verified** -- validate them on your instrument before relying on them.\n", + "`setup()` emits a warning to that effect.\n", + "```\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Physical setup\n", + "\n", + "Connect the decapper's serial port to your computer through its USB-to-serial adapter and use the stable `by-id` path so the port survives re-enumeration, e.g. `/dev/serial/by-id/usb-FTDI_USB_Serial_Converter_XXXXXXXX-if00-port0`.\n", + "\n", + "```{important}\n", + "If the instrument comes up reporting `StatusBUSY` and ignores commands, it is\n", + "locked out. The most common cause is an **engaged e-stop**; the safety guard/hood\n", + "and other interlocks do the same. There is no command to read the e-stop\n", + "directly, so `setup()` raises with this hint when it sees `StatusBUSY` at connect.\n", + "Clear the e-stop and retry.\n", + "```\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Connect\n", + "\n", + "`setup()` opens the serial port and reads the status. It raises if the device is not ready (e.g. e-stop), and logs the not-fully-verified warning.\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from pylabrobot.azenta.fluidx import FluidXIntelliXcap96\n", + "\n", + "decapper = FluidXIntelliXcap96(port=\"/dev/ttyUSB0\") # replace with your port\n", + "await decapper.setup()\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Status\n", + "\n", + "`request_status()` returns the current status word: `StatusOK` (idle), `StatusBUSY` (moving), or `StatusSLEEP` (standby). Note the status word does **not** encode the tray position.\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "print(\"status:\", await decapper.request_status())\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Tray\n", + "\n", + "Open and close the loading tray. Each call returns once the tray has finished moving (status back to `StatusOK`). Asking for a position the tray is already in is a no-op and raises `FluidXError` (`CommandIgnore`).\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "await decapper.open_tray()\n", + "await decapper.close_tray()\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Home\n", + "\n", + "Home all axes.\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "await decapper.home()\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Decap, recap, waste\n", + "\n", + "With a rack of screw-cap tubes loaded on the nest, `decap()` unscrews and holds all 96 caps, `recap()` screws them back on, and `waste()` drops the held caps into the waste bin. `decap()` refuses if caps are already held (recap first); `recap()` refuses if none are held.\n", + "\n", + "```{warning}\n", + "These operations are not yet hardware-verified. Test carefully.\n", + "```\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# await decapper.decap() # unscrew and hold all 96 caps\n", + "# await decapper.recap() # screw the held caps back on\n", + "# await decapper.waste() # drop held caps to the waste bin\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Standby\n", + "\n", + "Put the decapper to sleep with `standby()` and wake it with `ready()`.\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "await decapper.standby()\n", + "await decapper.ready()\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Teardown\n", + "\n", + "`stop()` closes the serial connection.\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "await decapper.stop()\n" + ] + } + ], + "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/azenta/index.md b/docs/user_guide/azenta/index.md index 72f87300a09..75d024de891 100644 --- a/docs/user_guide/azenta/index.md +++ b/docs/user_guide/azenta/index.md @@ -5,4 +5,5 @@ a4s/hello-world xpeel/hello-world +fluidx/intellixcap96/hello-world ``` diff --git a/pylabrobot/azenta/fluidx/intellixcap96.py b/pylabrobot/azenta/fluidx/intellixcap96.py index 8c428dade3b..db4e9253510 100644 --- a/pylabrobot/azenta/fluidx/intellixcap96.py +++ b/pylabrobot/azenta/fluidx/intellixcap96.py @@ -1,14 +1,18 @@ import asyncio import logging -from typing import Optional, Tuple +from typing import List, Optional from pylabrobot.io.serial import Serial logger = logging.getLogger(__name__) -# Commands and replies are terminated with ETX (0x03), not CR/LF. +# Every reply is framed between STX (0x02) and ETX (0x03). A command is answered +# with an ACK frame (a lone 0x06), a command-echo frame ("OK"), and then a +# result frame (a "Status..." word, or "CommandIgnore" when the command is a +# no-op or refused). Motions run asynchronously: the status word is "StatusBUSY" +# while moving and returns to "StatusOK" when done. +STX = b"\x02" ETX = b"\x03" -# Single-byte acknowledgement the firmware returns before a long-running motion. ACK = "\x06" # Single-character commands the decapper understands. @@ -23,42 +27,7 @@ STANDBY = "j" READY = "k" -# Descriptions for the fault conditions the firmware reports. -ERROR_MESSAGES = { - "StatusNotOK": ( - "Status is not ok. The device is in an error state. Check whether the plate is already " - "decapped (if so, recap and initialize again); otherwise clear the error on the device and " - "cycle the power." - ), - "NeedToRecap": "Decap operation is already done. Select the recap operation on the device.", - "NeedToDecap": "Recap operation is already executed.", - "DecapWasNotSuccesful": ( - "Decapping was not successful. Fix the error on the device manually and check whether any " - "tubes are still on the head." - ), - "RecapWasNotSuccesful": "Recapping was not successful. Fix the error on the device manually.", - "StoreWasNotSuccesful": "Storing was not successful. Fix the error on the device manually.", - "OpenTrayWasNotSuccesful": ( - "Opening the tray was not successful. Fix the error on the device manually." - ), - "CloseTrayWasNotSuccesful": ( - "Closing the tray was not successful. Fix the error on the device manually." - ), - "HomeNotSuccesful": ( - "Device was not able to reach the home position. The device is in an error state. " - "Restart the device." - ), - "CannotReset": ( - "Cannot send the reset command to the device. The device is in an error state. " - "Reset the device and try again." - ), - "ResetNotSuccesful": ( - "Reset did not complete. The device is in an error state. Reset the device and try again." - ), - "CannotGoInStandbyMode": "Cannot go to standby mode. Check the errors on the device.", - "NoAck": "Device did not acknowledge the command.", - "CommandIgnore": "Command was ignored by the device.", -} +COMMAND_IGNORE = "CommandIgnore" class FluidXError(Exception): @@ -72,10 +41,6 @@ def __str__(self) -> str: return f"{self.title}: {self.message}" if self.message else self.title -def _fault(key: str) -> FluidXError: - return FluidXError(title=ERROR_MESSAGES.get(key, key)) - - class FluidXIntelliXcap96: """FluidX IntelliXcap 96 automated screw-cap decapper. @@ -86,11 +51,10 @@ class FluidXIntelliXcap96: the loading tray opened and closed. Serial settings: - 9600 baud, 8 data bits, no parity, 1 stop bit, no handshake, ETX (0x03) - terminator. + 9600 baud, 8 data bits, no parity, 1 stop bit, no handshake. Replies are + framed between STX (0x02) and ETX (0x03). - Single-character commands (each is written followed by ETX; the reply is read - back up to the next ETX): + Single-character commands, each written followed by ETX: a request status h start decapping i start recapping @@ -101,22 +65,24 @@ class FluidXIntelliXcap96: z reset j enter standby k leave standby (ready) - A status reply contains ``OK``/``StatusOK`` when idle, ``StatusSleep`` in - standby, ``Recap``/``Decap`` for the pending inverse operation, and - ``StatusError``/``Error`` on a fault. Operation replies report ``...DONE`` on - success and ``...ERROR`` on failure. Motion commands are acknowledged with a - single 0x06 byte. - - Not verified: has NOT been tested against hardware in PyLabRobot. A warning is - emitted at setup. + A command is answered with an ACK frame (0x06), a ``OK`` echo frame, and a + result frame. The status word is ``StatusOK`` when idle, ``StatusBUSY`` while a + motion runs, ``StatusSLEEP`` in standby, and carries ``ERROR``/``Recap``/ + ``Decap`` for a fault or a pending inverse operation. A refused or no-op + command answers with ``CommandIgnore``. Motions complete when the status word + returns from ``StatusBUSY`` to ``StatusOK``. + + Verified against hardware: connection, status, tray open/close, home, and + standby/ready. Decap, recap, and waste engage the cap head and are NOT yet + hardware-verified; a warning is emitted at setup. """ def __init__( self, port: str, timeout: float = 5.0, - command_delay: float = 1.0, - read_delay: float = 0.5, + command_delay: float = 0.3, + frame_gap: float = 0.5, poll_interval: float = 1.0, ) -> None: """ @@ -124,11 +90,12 @@ def __init__( port: serial port the decapper is connected to. timeout: serial read/write timeout in seconds. command_delay: pause after writing a command before reading its reply. - read_delay: pause after reading a reply. - poll_interval: pause between status polls while an operation runs. + frame_gap: how long to wait for another reply frame before concluding the + reply is complete. + poll_interval: pause between status polls while a motion runs. """ self.command_delay = command_delay - self.read_delay = read_delay + self.frame_gap = frame_gap self.poll_interval = poll_interval self.io = Serial( human_readable_device_name="FluidX IntelliXcap 96", @@ -142,97 +109,179 @@ def __init__( async def setup(self) -> None: logger.warning( - "FluidXIntelliXcap96 has NOT been tested against hardware in PyLabRobot. " - "Please make a PR to remove this message if you have verified it on your hardware." + "FluidXIntelliXcap96: connection, status, tray, home, and standby/ready are " + "hardware-verified, but decap, recap, and waste are NOT yet verified in PyLabRobot. " + "Please make a PR to remove this message once you have verified them on your hardware." ) await self.io.setup() - # Confirm the decapper is reachable and reports a healthy status. - await self._send(STATUS) - await self._read_line() - if "OK" not in await self._read_line(): - raise _fault("StatusNotOK") - logger.info("[IntelliXcap96 %s] connected", self.io.port) + status = await self.request_status() + up = status.upper() + if "BUSY" in up: + # At connect there is no motion in flight, so a persistent StatusBUSY means + # the instrument is locked out. By far the most common cause is an engaged + # e-stop; the safety guard/hood and other interlocks do the same. There is + # no command to read the e-stop directly, so surface the likely cause here. + logger.error( + "[IntelliXcap96 %s] reports StatusBUSY at connect and will ignore commands. " + "Check the E-STOP first (most common cause), then the safety guard/hood and " + "interlocks, and retry.", + self.io.port, + ) + raise FluidXError( + title="Decapper is not ready (StatusBUSY)", + message=( + "The device reports BUSY and ignores commands. This is almost always an engaged " + "E-STOP; also check the safety guard/hood and interlocks, then retry." + ), + ) + if "ERROR" in up: + raise FluidXError(title="Decapper reports an error", message=status) + logger.info("[IntelliXcap96 %s] connected: %s", self.io.port, status) async def stop(self) -> None: """Close the serial connection.""" await self.io.stop() - # === Command layer === + # === Framed command layer === async def _send(self, command: str) -> None: - """Discard any pending input, write a command with its terminator, then pace.""" + """Discard pending input, write a command with its terminator, then pace.""" await self.io.reset_input_buffer() await self.io.write(command.encode("ascii") + ETX) - logger.debug("[IntelliXcap96] send: %r", command) await asyncio.sleep(self.command_delay) - async def _read_line(self) -> str: - """Read one ETX-terminated reply and return it stripped.""" + async def _read_frame(self) -> Optional[str]: + """Read one STX..ETX frame and return its payload, or None if none arrives.""" + while True: + byte = await self.io.read(1) + if byte == b"": + return None + if byte == STX: + break buf = bytearray() while True: - char = await self.io.read(1) - if char in (b"", ETX): # terminator or read timeout + byte = await self.io.read(1) + if byte in (b"", ETX): break - buf += char - reply = buf.decode("ascii", errors="replace").strip() - logger.debug("[IntelliXcap96] recv: %r", reply) - await asyncio.sleep(self.read_delay) - return reply + buf += byte + return buf.decode("ascii", errors="replace") + + async def _command(self, command: str) -> List[str]: + """Send a command and collect every reply frame until the reply goes quiet.""" + await self._send(command) + frames: List[str] = [] + with self.io.temporary_timeout(self.frame_gap): + while True: + frame = await self._read_frame() + if frame is None: + break + frames.append(frame) + logger.debug("[IntelliXcap96] %r -> %r", command, frames) + return frames + + @staticmethod + def _status_frame(frames: List[str]) -> Optional[str]: + return next((f for f in frames if f.upper().startswith("STATUS")), None) + + def _require_accepted(self, command: str, frames: List[str], name: str) -> None: + """Raise unless the device acked and echoed the command without ignoring it.""" + if ACK not in frames: + raise FluidXError(title=f"{name} was not acknowledged", message=repr(frames)) + if any(COMMAND_IGNORE in f for f in frames): + raise FluidXError( + title=f"{name} was ignored", + message="The device is already in that state or is not ready.", + ) + if f"{command}OK" not in frames: + raise FluidXError(title=f"{name} was not accepted", message=repr(frames)) + + async def _wait_for_idle(self, timeout: float, name: str) -> None: + """Poll status until it returns to StatusOK, raising on an error or timeout.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while loop.time() < deadline: + frames = await self._command(STATUS) + if any("ERROR" in f.upper() for f in frames): + raise FluidXError(title=f"{name} failed", message=repr(frames)) + status = self._status_frame(frames) + if status is not None and "OK" in status.upper(): + return + await asyncio.sleep(self.poll_interval) + raise FluidXError(title=f"{name} timed out", message=f"not idle within {timeout}s") + + # === Status === async def request_status(self) -> str: - """Poll the device and return its raw status reply.""" - await self._send(STATUS) - return await self._read_line() + """Poll the device and return its status word (e.g. ``StatusOK``).""" + frames = await self._command(STATUS) + status = self._status_frame(frames) + if status is None: + raise FluidXError(title="No status reply", message=repr(frames)) + return status - async def send_command(self, command: str) -> str: - """Send a single raw command and return its reply. Escape hatch for bring-up.""" - await self._send(command) - return await self._read_line() + async def send_command(self, command: str) -> List[str]: + """Send a raw single command and return its reply frames. Escape hatch.""" + return await self._command(command) # === Operations === + async def open_tray(self, timeout: float = 15.0) -> None: + """Open the loading tray.""" + frames = await self._command(OPEN_TRAY) + self._require_accepted(OPEN_TRAY, frames, "Opening the tray") + await self._wait_for_idle(timeout, "Opening the tray") + logger.info("[IntelliXcap96 %s] tray open", self.io.port) + + async def close_tray(self, timeout: float = 15.0) -> None: + """Close the loading tray.""" + frames = await self._command(CLOSE_TRAY) + self._require_accepted(CLOSE_TRAY, frames, "Closing the tray") + await self._wait_for_idle(timeout, "Closing the tray") + logger.info("[IntelliXcap96 %s] tray closed", self.io.port) + + async def home(self, timeout: float = 30.0) -> None: + """Home all axes.""" + frames = await self._command(HOME_ALL) + self._require_accepted(HOME_ALL, frames, "Homing") + await self._wait_for_idle(timeout, "Homing") + logger.info("[IntelliXcap96 %s] homed", self.io.port) + async def decap(self, timeout: float = 60.0) -> None: """Unscrew and hold all 96 caps. Args: timeout: maximum time in seconds to wait for the stroke to finish. """ - status = await self.request_status() - if "Recap" in status: - raise _fault("NeedToRecap") - self._raise_on_error(status, "StatusNotOK") - - await self._send(DECAP_START) - await self._read_line() - await self._poll_until( - timeout=timeout, - done=("RECAP", "DONE"), - error=("DecapERROR", "StatusError", "Error"), - fault="DecapWasNotSuccesful", - ) + status = (await self.request_status()).upper() + if "RECAP" in status: + raise FluidXError( + title="Already decapped", + message="Caps are held; recap before decapping again.", + ) + if "ERROR" in status: + raise FluidXError(title="Decapper is not ready", message=status) + frames = await self._command(DECAP_START) + self._require_accepted(DECAP_START, frames, "Decapping") + await self._wait_for_idle(timeout, "Decapping") logger.info("[IntelliXcap96 %s] decap complete", self.io.port) - async def recap(self, timeout: float = 50.0) -> None: + async def recap(self, timeout: float = 60.0) -> None: """Screw the held caps back on. Args: timeout: maximum time in seconds to wait for the stroke to finish. """ - status = await self.request_status() - if "Decap" in status: - raise _fault("NeedToDecap") - self._raise_on_error(status, "StatusNotOK") - - await self._send(RECAP_START) - if ACK not in await self._read_line(): - raise _fault("NoAck") - await self._poll_until( - timeout=timeout, - done=("RecapDONE", "StatusOK"), - error=("RecapERROR", "StatusError", "Error"), - fault="RecapWasNotSuccesful", - ignore=("CommandIgnore",), - ) + status = (await self.request_status()).upper() + if "DECAP" in status: + raise FluidXError( + title="Nothing to recap", + message="No caps are held; decap first.", + ) + if "ERROR" in status: + raise FluidXError(title="Decapper is not ready", message=status) + frames = await self._command(RECAP_START) + self._require_accepted(RECAP_START, frames, "Recapping") + await self._wait_for_idle(timeout, "Recapping") logger.info("[IntelliXcap96 %s] recap complete", self.io.port) async def waste(self, timeout: float = 60.0) -> None: @@ -241,130 +290,48 @@ async def waste(self, timeout: float = 60.0) -> None: Args: timeout: maximum time in seconds to wait for the stroke to finish. """ - status = await self.request_status() - self._raise_on_error(status, "StatusNotOK") - - await self._send(WASTE) - if ACK not in await self._read_line(): - raise _fault("NoAck") - await self._poll_until( - timeout=timeout, - done=("StoreDONE", "StatusOK"), - error=("StoreERROR", "StatusError", "Error"), - fault="StoreWasNotSuccesful", - ) + status = (await self.request_status()).upper() + if "ERROR" in status: + raise FluidXError(title="Decapper is not ready", message=status) + frames = await self._command(WASTE) + self._require_accepted(WASTE, frames, "Wasting caps") + await self._wait_for_idle(timeout, "Wasting caps") logger.info("[IntelliXcap96 %s] waste complete", self.io.port) - async def open_tray(self, timeout: float = 10.0) -> None: - """Open the loading tray.""" - with self.io.temporary_timeout(timeout): - await self._move_tray( - OPEN_TRAY, ("OpenERROR", "StatusError", "Error"), "OpenTrayWasNotSuccesful" - ) - logger.info("[IntelliXcap96 %s] tray open", self.io.port) - - async def close_tray(self, timeout: float = 10.0) -> None: - """Close the loading tray.""" - with self.io.temporary_timeout(timeout): - await self._move_tray( - CLOSE_TRAY, ("CloseERROR", "StatusError", "Error"), "CloseTrayWasNotSuccesful" - ) - logger.info("[IntelliXcap96 %s] tray closed", self.io.port) + async def standby(self, timeout: float = 15.0) -> None: + """Put the decapper into standby (sleep) mode.""" + frames = await self._command(STANDBY) + self._require_accepted(STANDBY, frames, "Entering standby") + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while loop.time() < deadline: + if "SLEEP" in (await self.request_status()).upper(): + logger.info("[IntelliXcap96 %s] standby", self.io.port) + return + await asyncio.sleep(self.poll_interval) + raise FluidXError(title="Entering standby timed out") - async def home(self, timeout: float = 20.0) -> None: - """Home all axes.""" - with self.io.temporary_timeout(timeout): - await self._send(STATUS) - if ACK not in await self._read_line(): - raise _fault("NoAck") - if "OK" not in await self._read_line(): - raise _fault("StatusNotOK") - await self._send(HOME_ALL) - await self._read_line() - if "Error" in await self._read_line(): - raise _fault("HomeNotSuccesful") - logger.info("[IntelliXcap96 %s] homed", self.io.port) + async def ready(self, timeout: float = 30.0) -> None: + """Wake the decapper from standby if it is asleep.""" + if "SLEEP" not in (await self.request_status()).upper(): + return + frames = await self._command(READY) + self._require_accepted(READY, frames, "Waking from standby") + await self._wait_for_idle(timeout, "Waking from standby") + logger.info("[IntelliXcap96 %s] ready", self.io.port) async def reset(self, settle_time: float = 5.0) -> None: """Reset the device and wait for it to settle. + A reset is a no-op when the device is already idle (the firmware answers + ``CommandIgnore``); it clears a recoverable error state. + Args: settle_time: time in seconds to wait after the reset command. """ - if "Error" in await self.request_status(): - raise _fault("CannotReset") - await self._send(RESET) + await self._command(RESET) await asyncio.sleep(settle_time) - await self._read_line() - if "Error" in await self.request_status(): - raise _fault("ResetNotSuccesful") + status = (await self.request_status()).upper() + if "ERROR" in status: + raise FluidXError(title="Reset did not clear the error", message=status) logger.info("[IntelliXcap96 %s] reset complete", self.io.port) - - async def standby(self) -> None: - """Put the decapper into standby (sleep) mode.""" - status = await self.request_status() - if "Recap" not in status and ("OK" not in status or "Error" in status): - raise _fault("StatusNotOK") - await self._send(STANDBY) - if "Standby" not in await self._read_line(): - raise _fault("CannotGoInStandbyMode") - logger.info("[IntelliXcap96 %s] standby", self.io.port) - - async def ready(self) -> None: - """Wake the decapper from standby if it is asleep.""" - if "StatusSleep" not in await self.request_status(): - return - await self._send(READY) - if "Ready" not in await self._read_line(): - raise _fault("StatusNotOK") - logger.info("[IntelliXcap96 %s] ready", self.io.port) - - # === Operation helpers === - - async def _move_tray(self, command: str, error: Tuple[str, ...], fault: str) -> None: - status = await self.request_status() - self._raise_on_error(status, "StatusNotOK") - await self._send(command) - if ACK not in await self._read_line(): - raise _fault("NoAck") - # The tray move reports two follow-up lines; either may carry the error. - for _ in range(2): - reply = await self._read_line() - if any(token in reply for token in error): - raise _fault(fault) - if "CommandIgnore" in reply: - raise _fault("CommandIgnore") - - async def _poll_until( - self, - timeout: float, - done: Tuple[str, ...], - error: Tuple[str, ...], - fault: str, - ignore: Tuple[str, ...] = (), - ) -> None: - """Poll status until a done/error token appears or the timeout elapses. - - Each poll reads three status lines and inspects the last, matching the way - the firmware streams its progress. - """ - loop = asyncio.get_running_loop() - deadline = loop.time() + timeout - while loop.time() < deadline: - await self._send(STATUS) - reply = "" - for _ in range(3): - reply = await self._read_line() - if any(token in reply for token in error): - raise _fault(fault) - if any(token in reply for token in ignore): - raise _fault("CommandIgnore") - if any(token in reply for token in done): - return - await asyncio.sleep(self.poll_interval) - raise FluidXError(title="Operation timed out", message=f"no completion within {timeout}s") - - @staticmethod - def _raise_on_error(status: str, fault: str) -> None: - if "StatusError" in status or "Error" in status: - raise _fault(fault) diff --git a/pylabrobot/azenta/fluidx/intellixcap96_tests.py b/pylabrobot/azenta/fluidx/intellixcap96_tests.py index cf1023cd21e..5487bcc2aba 100644 --- a/pylabrobot/azenta/fluidx/intellixcap96_tests.py +++ b/pylabrobot/azenta/fluidx/intellixcap96_tests.py @@ -5,13 +5,15 @@ from pylabrobot.azenta.fluidx import FluidXError, FluidXIntelliXcap96 +ACK = "\x06" + class FakeXcapSerial: - """In-memory serial stand-in for the ETX-framed decapper protocol. + """In-memory serial stand-in for the STX/ETX framed decapper protocol. ``script`` is a list of turns, one per ``write``. Each turn is the list of - reply lines the device emits in response to that write; every line is queued - ETX-terminated, exactly as the firmware frames its replies. + frame payloads the device emits in response to that write; every payload is + queued wrapped in STX (0x02) .. ETX (0x03), exactly as the firmware frames it. """ def __init__(self, script: List[List[str]]) -> None: @@ -41,9 +43,9 @@ def temporary_timeout(self, timeout: float) -> Iterator[None]: async def write(self, data: bytes) -> None: self.written.append(data.decode("ascii").rstrip("\x03")) - lines = self._script.pop(0) if self._script else [] - for line in lines: - self._rx += line.encode("ascii") + b"\x03" + turn = self._script.pop(0) if self._script else [] + for payload in turn: + self._rx += b"\x02" + payload.encode("ascii") + b"\x03" async def read(self, num_bytes: int = 1) -> bytes: if not self._rx: @@ -53,6 +55,11 @@ async def read(self, num_bytes: int = 1) -> bytes: return out +def status(word: str, echo: str = "a") -> List[str]: + """A full status reply: ACK, command echo, and the status word.""" + return [ACK, f"{echo}OK", word] + + class TestIntelliXcap96(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: patcher = patch("pylabrobot.azenta.fluidx.intellixcap96.asyncio.sleep", new_callable=AsyncMock) @@ -64,114 +71,105 @@ def _make(self, script: List[List[str]]) -> FluidXIntelliXcap96: device.io = FakeXcapSerial(script) # type: ignore[assignment] return device - async def test_setup_checks_status_ok(self): - device = self._make([["boot", "StatusOK"]]) + async def test_setup_ok(self): + device = self._make([status("StatusOK")]) await device.setup() self.assertEqual(device.io.written, ["a"]) # type: ignore[attr-defined] - async def test_setup_rejects_bad_status(self): - device = self._make([["boot", "StatusError"]]) + async def test_setup_busy_raises_estop_hint(self): + device = self._make([status("StatusBUSY")]) + with self.assertRaises(FluidXError) as ctx: + await device.setup() + self.assertIn("E-STOP", str(ctx.exception)) + + async def test_setup_error_raises(self): + device = self._make([status("StatusError")]) with self.assertRaises(FluidXError): await device.setup() - async def test_decap_success(self): + async def test_request_status_returns_status_word(self): + device = self._make([status("StatusOK")]) + self.assertEqual(await device.request_status(), "StatusOK") + + async def test_open_tray_moves_then_idle(self): device = self._make( [ - ["StatusOK"], # status precheck - ["ack"], # h (decap start) - ["l1", "l2", "DecapDONE"], # status poll -> 3 lines, done on the last + [ACK, "fOK"], # f accepted + status("StatusBUSY"), # still moving + status("StatusOK"), # settled ] ) - await device.decap() - self.assertEqual(device.io.written, ["a", "h", "a"]) # type: ignore[attr-defined] + await device.open_tray() + self.assertEqual(device.io.written, ["f", "a", "a"]) # type: ignore[attr-defined] - async def test_decap_blocked_when_recap_pending(self): - device = self._make([["StatusRecap"]]) + async def test_open_tray_ignored_raises(self): + device = self._make([[ACK, "fOK", "CommandIgnore"]]) with self.assertRaises(FluidXError): - await device.decap() + await device.open_tray() - async def test_decap_reports_failure(self): - device = self._make( - [ - ["StatusOK"], - ["ack"], - ["l1", "l2", "DecapERROR"], - ] - ) - with self.assertRaises(FluidXError): - await device.decap() + async def test_close_tray_moves_then_idle(self): + device = self._make([[ACK, "gOK"], status("StatusBUSY"), status("StatusOK")]) + await device.close_tray() + self.assertEqual(device.io.written, ["g", "a", "a"]) # type: ignore[attr-defined] - async def test_recap_requires_ack(self): - device = self._make( - [ - ["StatusOK"], # status precheck (no "Decap" pending) - ["no-ack-here"], # i reply lacks 0x06 - ] - ) - with self.assertRaises(FluidXError): - await device.recap() + async def test_home_moves_then_idle(self): + device = self._make([[ACK, "ZOK"], status("StatusBUSY"), status("StatusOK")]) + await device.home() + self.assertEqual(device.io.written, ["Z", "a", "a"]) # type: ignore[attr-defined] - async def test_recap_success(self): - device = self._make( - [ - ["StatusOK"], - ["\x06"], # i acknowledged - ["l1", "l2", "RecapDONE"], - ] - ) - await device.recap() - self.assertEqual(device.io.written, ["a", "i", "a"]) # type: ignore[attr-defined] + async def test_standby_reaches_sleep(self): + device = self._make([[ACK, "jOK"], status("StatusSLEEP")]) + await device.standby() + self.assertEqual(device.io.written, ["j", "a"]) # type: ignore[attr-defined] - async def test_waste_success(self): - device = self._make( - [ - ["StatusOK"], - ["\x06"], # b acknowledged - ["l1", "l2", "StoreDONE"], - ] - ) - await device.waste() - self.assertEqual(device.io.written, ["a", "b", "a"]) # type: ignore[attr-defined] + async def test_ready_noop_when_awake(self): + device = self._make([status("StatusOK")]) + await device.ready() + self.assertEqual(device.io.written, ["a"]) # type: ignore[attr-defined] - async def test_home_success(self): + async def test_ready_wakes_from_sleep(self): device = self._make( [ - ["\x06", "StatusOK"], # a: ack then OK - ["homing", "done"], # Z: discard then no "Error" + status("StatusSLEEP"), # request_status: asleep + [ACK, "kOK"], # k accepted + status("StatusBUSY"), # waking + status("StatusOK"), # awake ] ) - await device.home() - self.assertEqual(device.io.written, ["a", "Z"]) # type: ignore[attr-defined] + await device.ready() + self.assertEqual(device.io.written, ["a", "k", "a", "a"]) # type: ignore[attr-defined] - async def test_open_tray_success(self): + async def test_decap_success(self): device = self._make( [ - ["StatusOK"], # status precheck - ["\x06", "OpenDONE", "StatusOK"], # f: ack + two follow-up lines + status("StatusOK"), # precheck: not recapped, no error + [ACK, "hOK"], # h accepted + status("StatusBUSY"), + status("StatusOK"), ] ) - await device.open_tray() - self.assertEqual(device.io.written, ["a", "f"]) # type: ignore[attr-defined] + await device.decap() + self.assertEqual(device.io.written, ["a", "h", "a", "a"]) # type: ignore[attr-defined] - async def test_reset_success(self): - device = self._make( - [ - ["StatusOK"], # a precheck (no error) - ["resetting"], # z - ["StatusOK"], # a postcheck (no error) - ] - ) - await device.reset() - self.assertEqual(device.io.written, ["a", "z", "a"]) # type: ignore[attr-defined] + async def test_decap_blocked_when_already_decapped(self): + device = self._make([status("StatusRecap")]) + with self.assertRaises(FluidXError): + await device.decap() - async def test_standby_and_ready(self): - device = self._make([["StatusOK"], ["StandbyDONE"]]) - await device.standby() - self.assertEqual(device.io.written, ["a", "j"]) # type: ignore[attr-defined] + async def test_decap_reports_error_during_motion(self): + device = self._make([status("StatusOK"), [ACK, "hOK"], [ACK, "aOK", "DecapERROR"]]) + with self.assertRaises(FluidXError): + await device.decap() - device = self._make([["StatusSleep"], ["ReadyDONE"]]) - await device.ready() - self.assertEqual(device.io.written, ["a", "k"]) # type: ignore[attr-defined] + async def test_recap_blocked_when_not_decapped(self): + device = self._make([status("StatusDecap")]) + with self.assertRaises(FluidXError): + await device.recap() + + async def test_reset_ignored_is_not_fatal(self): + device = self._make([[ACK, "zOK", "CommandIgnore"], status("StatusOK")]) + await device.reset() + self.assertEqual(device.io.written, ["z", "a"]) # type: ignore[attr-defined] if __name__ == "__main__": From 53dce55f0c928110695ae67c8dd02055cb24ddb9 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Mon, 20 Jul 2026 19:02:35 -0700 Subject: [PATCH 3/8] fix(azenta): restore firmware error messages; runnable tutorial cells Bring back the ERROR_MESSAGES table (the fault text read out of the instrument firmware) that was dropped during the framing rework, and raise every fault through it again via _fault() so exceptions carry the device's own wording instead of paraphrases. Uncomment decap/recap/waste in the tutorial and split each operation into its own cell. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../fluidx/intellixcap96/hello-world.ipynb | 168 +++++++++++------- pylabrobot/azenta/fluidx/intellixcap96.py | 95 ++++++---- 2 files changed, 163 insertions(+), 100 deletions(-) diff --git a/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb b/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb index 1bf37079dd8..713499f4c0e 100644 --- a/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb +++ b/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb @@ -4,56 +4,56 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "# FluidX IntelliXcap 96\n", - "\n", - "The FluidX IntelliXcap 96 (an Azenta brand) is an automated screw-cap **decapper**. It unscrews, holds, and screws back on all 96 caps of a screw-cap tube rack in a single stroke. A plate mover loads the rack onto its one nest.\n", - "\n", - "| Property | Value |\n", - "|---|---|\n", - "| Communication | Serial, STX/ETX-framed ASCII |\n", - "| Serial settings | 9600 baud, 8 data bits, no parity, 1 stop bit, no handshake |\n", - "| Framing | each reply is wrapped in STX (`0x02`) .. ETX (`0x03`) |\n", - "| Reply pattern | ACK (`0x06`), then `OK` echo, then a result frame |\n", - "\n", - "Every command is a single character written followed by ETX. A motion command is accepted with an ACK and a `OK` echo; the status word then reads `StatusBUSY` while it moves and returns to `StatusOK` when done. A refused or no-op command answers `CommandIgnore`.\n" + "# FluidX IntelliXcap 96\n\n", + "\n\n", + "The FluidX IntelliXcap 96 (an Azenta brand) is an automated screw-cap **decapper**. It unscrews, holds, and screws back on all 96 caps of a screw-cap tube rack in a single stroke. A plate mover loads the rack onto its one nest.\n\n", + "\n\n", + "| Property | Value |\n\n", + "|---|---|\n\n", + "| Communication | Serial, STX/ETX-framed ASCII |\n\n", + "| Serial settings | 9600 baud, 8 data bits, no parity, 1 stop bit, no handshake |\n\n", + "| Framing | each reply is wrapped in STX (`0x02`) .. ETX (`0x03`) |\n\n", + "| Reply pattern | ACK (`0x06`), then `OK` echo, then a result frame |\n\n", + "\n\n", + "Every command is a single character written followed by ETX. A motion command is accepted with an ACK and a `OK` echo; the status word then reads `StatusBUSY` while it moves and returns to `StatusOK` when done. A refused or no-op command answers `CommandIgnore`." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "```{note}\n", - "Connection, status, tray open/close, home, and standby/ready are verified against\n", - "hardware. **Decap, recap, and waste engage the cap head and are not yet\n", - "hardware-verified** -- validate them on your instrument before relying on them.\n", - "`setup()` emits a warning to that effect.\n", - "```\n" + "```{note}\n\n", + "Connection, status, tray open/close, home, and standby/ready are verified against\n\n", + "hardware. **Decap, recap, and waste engage the cap head and are not yet\n\n", + "hardware-verified** -- validate them on your instrument before relying on them.\n\n", + "`setup()` emits a warning to that effect.\n\n", + "```" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "## Physical setup\n", - "\n", - "Connect the decapper's serial port to your computer through its USB-to-serial adapter and use the stable `by-id` path so the port survives re-enumeration, e.g. `/dev/serial/by-id/usb-FTDI_USB_Serial_Converter_XXXXXXXX-if00-port0`.\n", - "\n", - "```{important}\n", - "If the instrument comes up reporting `StatusBUSY` and ignores commands, it is\n", - "locked out. The most common cause is an **engaged e-stop**; the safety guard/hood\n", - "and other interlocks do the same. There is no command to read the e-stop\n", - "directly, so `setup()` raises with this hint when it sees `StatusBUSY` at connect.\n", - "Clear the e-stop and retry.\n", - "```\n" + "## Physical setup\n\n", + "\n\n", + "Connect the decapper's serial port to your computer through its USB-to-serial adapter and use the stable `by-id` path so the port survives re-enumeration, e.g. `/dev/serial/by-id/usb-FTDI_USB_Serial_Converter_XXXXXXXX-if00-port0`.\n\n", + "\n\n", + "```{important}\n\n", + "If the instrument comes up reporting `StatusBUSY` and ignores commands, it is\n\n", + "locked out. The most common cause is an **engaged e-stop**; the safety guard/hood\n\n", + "and other interlocks do the same. There is no command to read the e-stop\n\n", + "directly, so `setup()` raises with this hint when it sees `StatusBUSY` at connect.\n\n", + "Clear the e-stop and retry.\n\n", + "```" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "## Connect\n", - "\n", - "`setup()` opens the serial port and reads the status. It raises if the device is not ready (e.g. e-stop), and logs the not-fully-verified warning.\n" + "## Connect\n\n", + "\n\n", + "`setup()` opens the serial port and reads the status. It raises if the device is not ready (e.g. e-stop), and logs the not-fully-verified warning." ] }, { @@ -62,19 +62,19 @@ "metadata": {}, "outputs": [], "source": [ - "from pylabrobot.azenta.fluidx import FluidXIntelliXcap96\n", - "\n", - "decapper = FluidXIntelliXcap96(port=\"/dev/ttyUSB0\") # replace with your port\n", - "await decapper.setup()\n" + "from pylabrobot.azenta.fluidx import FluidXIntelliXcap96\n\n", + "\n\n", + "decapper = FluidXIntelliXcap96(port=\"/dev/ttyUSB0\") # replace with your port\n\n", + "await decapper.setup()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "## Status\n", - "\n", - "`request_status()` returns the current status word: `StatusOK` (idle), `StatusBUSY` (moving), or `StatusSLEEP` (standby). Note the status word does **not** encode the tray position.\n" + "## Status\n\n", + "\n\n", + "`request_status()` returns the current status word: `StatusOK` (idle), `StatusBUSY` (moving), or `StatusSLEEP` (standby). Note the status word does **not** encode the tray position." ] }, { @@ -83,16 +83,16 @@ "metadata": {}, "outputs": [], "source": [ - "print(\"status:\", await decapper.request_status())\n" + "print(\"status:\", await decapper.request_status())" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "## Tray\n", - "\n", - "Open and close the loading tray. Each call returns once the tray has finished moving (status back to `StatusOK`). Asking for a position the tray is already in is a no-op and raises `FluidXError` (`CommandIgnore`).\n" + "## Tray\n\n", + "\n\n", + "Open and close the loading tray. Each call returns once the tray has finished moving (status back to `StatusOK`). Asking for a position the tray is already in is a no-op and raises `FluidXError` (`CommandIgnore`)." ] }, { @@ -101,17 +101,25 @@ "metadata": {}, "outputs": [], "source": [ - "await decapper.open_tray()\n", - "await decapper.close_tray()\n" + "await decapper.open_tray()" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "await decapper.close_tray()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "## Home\n", - "\n", - "Home all axes.\n" + "## Home\n\n", + "\n\n", + "Home all axes." ] }, { @@ -120,20 +128,38 @@ "metadata": {}, "outputs": [], "source": [ - "await decapper.home()\n" + "await decapper.home()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "## Decap, recap, waste\n", - "\n", - "With a rack of screw-cap tubes loaded on the nest, `decap()` unscrews and holds all 96 caps, `recap()` screws them back on, and `waste()` drops the held caps into the waste bin. `decap()` refuses if caps are already held (recap first); `recap()` refuses if none are held.\n", - "\n", - "```{warning}\n", - "These operations are not yet hardware-verified. Test carefully.\n", - "```\n" + "## Decap, recap, waste\n\n", + "\n\n", + "With a rack of screw-cap tubes loaded on the nest, `decap()` unscrews and holds all 96 caps, `recap()` screws them back on, and `waste()` drops the held caps into the waste bin. `decap()` refuses if caps are already held (recap first); `recap()` refuses if none are held.\n\n", + "\n\n", + "```{warning}\n\n", + "These operations are not yet hardware-verified. Test carefully.\n\n", + "```" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "await decapper.decap() # unscrew and hold all 96 caps" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "await decapper.recap() # screw the held caps back on" ] }, { @@ -142,18 +168,25 @@ "metadata": {}, "outputs": [], "source": [ - "# await decapper.decap() # unscrew and hold all 96 caps\n", - "# await decapper.recap() # screw the held caps back on\n", - "# await decapper.waste() # drop held caps to the waste bin\n" + "await decapper.waste() # drop held caps to the waste bin" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "## Standby\n", - "\n", - "Put the decapper to sleep with `standby()` and wake it with `ready()`.\n" + "## Standby\n\n", + "\n\n", + "Put the decapper to sleep with `standby()` and wake it with `ready()`." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "await decapper.standby()" ] }, { @@ -162,17 +195,16 @@ "metadata": {}, "outputs": [], "source": [ - "await decapper.standby()\n", - "await decapper.ready()\n" + "await decapper.ready()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "## Teardown\n", - "\n", - "`stop()` closes the serial connection.\n" + "## Teardown\n\n", + "\n\n", + "`stop()` closes the serial connection." ] }, { @@ -181,7 +213,7 @@ "metadata": {}, "outputs": [], "source": [ - "await decapper.stop()\n" + "await decapper.stop()" ] } ], diff --git a/pylabrobot/azenta/fluidx/intellixcap96.py b/pylabrobot/azenta/fluidx/intellixcap96.py index db4e9253510..f4f48729521 100644 --- a/pylabrobot/azenta/fluidx/intellixcap96.py +++ b/pylabrobot/azenta/fluidx/intellixcap96.py @@ -29,6 +29,39 @@ COMMAND_IGNORE = "CommandIgnore" +# Fault descriptions as reported by the instrument firmware. +ERROR_MESSAGES = { + "StatusNotOK": ( + "Status is not ok. The device is in an error state. Check whether the plate is already " + "decapped (if so, recap and initialize again); otherwise clear the error on the device and " + "cycle the power." + ), + "NeedToRecap": "Decap operation is already done. Select the recap operation on the device.", + "NeedToDecap": "Recap operation is already executed.", + "DecapWasNotSuccesful": ( + "Decapping was not successful. Fix the error on the device manually and check whether any " + "tubes are still on the head." + ), + "RecapWasNotSuccesful": "Recapping was not successful. Fix the error on the device manually.", + "StoreWasNotSuccesful": "Storing was not successful. Fix the error on the device manually.", + "OpenTrayWasNotSuccesful": ( + "Opening the tray was not successful. Fix the error on the device manually." + ), + "CloseTrayWasNotSuccesful": ( + "Closing the tray was not successful. Fix the error on the device manually." + ), + "HomeNotSuccesful": ( + "Device was not able to reach the home position. The device is in an error state. " + "Restart the device." + ), + "ResetNotSuccesful": ( + "Reset did not complete. The device is in an error state. Reset the device and try again." + ), + "CannotGoInStandbyMode": "Cannot go to standby mode. Check the errors on the device.", + "CommandIgnore": "Command was ignored by the device.", + "NoAck": "Device did not acknowledge the command.", +} + class FluidXError(Exception): """Exceptions raised by a FluidX IntelliXcap 96 decapper.""" @@ -41,6 +74,11 @@ def __str__(self) -> str: return f"{self.title}: {self.message}" if self.message else self.title +def _fault(key: str, detail: Optional[str] = None) -> FluidXError: + """Build a FluidXError carrying the firmware's own description for ``key``.""" + return FluidXError(title=ERROR_MESSAGES.get(key, key), message=detail) + + class FluidXIntelliXcap96: """FluidX IntelliXcap 96 automated screw-cap decapper. @@ -135,7 +173,7 @@ async def setup(self) -> None: ), ) if "ERROR" in up: - raise FluidXError(title="Decapper reports an error", message=status) + raise _fault("StatusNotOK", status) logger.info("[IntelliXcap96 %s] connected: %s", self.io.port, status) async def stop(self) -> None: @@ -185,24 +223,23 @@ def _status_frame(frames: List[str]) -> Optional[str]: def _require_accepted(self, command: str, frames: List[str], name: str) -> None: """Raise unless the device acked and echoed the command without ignoring it.""" - if ACK not in frames: - raise FluidXError(title=f"{name} was not acknowledged", message=repr(frames)) + if ACK not in frames or f"{command}OK" not in frames: + raise _fault("NoAck", f"{name}: {frames!r}") if any(COMMAND_IGNORE in f for f in frames): - raise FluidXError( - title=f"{name} was ignored", - message="The device is already in that state or is not ready.", - ) - if f"{command}OK" not in frames: - raise FluidXError(title=f"{name} was not accepted", message=repr(frames)) + raise _fault("CommandIgnore", f"{name}: device already in that state or not ready") + + async def _wait_for_idle(self, timeout: float, name: str, fail_key: str) -> None: + """Poll status until it returns to StatusOK, raising on an error or timeout. - async def _wait_for_idle(self, timeout: float, name: str) -> None: - """Poll status until it returns to StatusOK, raising on an error or timeout.""" + ``fail_key`` names the firmware error message to raise if the status word + reports an error while waiting. + """ loop = asyncio.get_running_loop() deadline = loop.time() + timeout while loop.time() < deadline: frames = await self._command(STATUS) if any("ERROR" in f.upper() for f in frames): - raise FluidXError(title=f"{name} failed", message=repr(frames)) + raise _fault(fail_key, f"{name}: {frames!r}") status = self._status_frame(frames) if status is not None and "OK" in status.upper(): return @@ -229,21 +266,21 @@ async def open_tray(self, timeout: float = 15.0) -> None: """Open the loading tray.""" frames = await self._command(OPEN_TRAY) self._require_accepted(OPEN_TRAY, frames, "Opening the tray") - await self._wait_for_idle(timeout, "Opening the tray") + await self._wait_for_idle(timeout, "Opening the tray", "OpenTrayWasNotSuccesful") logger.info("[IntelliXcap96 %s] tray open", self.io.port) async def close_tray(self, timeout: float = 15.0) -> None: """Close the loading tray.""" frames = await self._command(CLOSE_TRAY) self._require_accepted(CLOSE_TRAY, frames, "Closing the tray") - await self._wait_for_idle(timeout, "Closing the tray") + await self._wait_for_idle(timeout, "Closing the tray", "CloseTrayWasNotSuccesful") logger.info("[IntelliXcap96 %s] tray closed", self.io.port) async def home(self, timeout: float = 30.0) -> None: """Home all axes.""" frames = await self._command(HOME_ALL) self._require_accepted(HOME_ALL, frames, "Homing") - await self._wait_for_idle(timeout, "Homing") + await self._wait_for_idle(timeout, "Homing", "HomeNotSuccesful") logger.info("[IntelliXcap96 %s] homed", self.io.port) async def decap(self, timeout: float = 60.0) -> None: @@ -254,15 +291,12 @@ async def decap(self, timeout: float = 60.0) -> None: """ status = (await self.request_status()).upper() if "RECAP" in status: - raise FluidXError( - title="Already decapped", - message="Caps are held; recap before decapping again.", - ) + raise _fault("NeedToRecap") if "ERROR" in status: - raise FluidXError(title="Decapper is not ready", message=status) + raise _fault("StatusNotOK", status) frames = await self._command(DECAP_START) self._require_accepted(DECAP_START, frames, "Decapping") - await self._wait_for_idle(timeout, "Decapping") + await self._wait_for_idle(timeout, "Decapping", "DecapWasNotSuccesful") logger.info("[IntelliXcap96 %s] decap complete", self.io.port) async def recap(self, timeout: float = 60.0) -> None: @@ -273,15 +307,12 @@ async def recap(self, timeout: float = 60.0) -> None: """ status = (await self.request_status()).upper() if "DECAP" in status: - raise FluidXError( - title="Nothing to recap", - message="No caps are held; decap first.", - ) + raise _fault("NeedToDecap") if "ERROR" in status: - raise FluidXError(title="Decapper is not ready", message=status) + raise _fault("StatusNotOK", status) frames = await self._command(RECAP_START) self._require_accepted(RECAP_START, frames, "Recapping") - await self._wait_for_idle(timeout, "Recapping") + await self._wait_for_idle(timeout, "Recapping", "RecapWasNotSuccesful") logger.info("[IntelliXcap96 %s] recap complete", self.io.port) async def waste(self, timeout: float = 60.0) -> None: @@ -292,10 +323,10 @@ async def waste(self, timeout: float = 60.0) -> None: """ status = (await self.request_status()).upper() if "ERROR" in status: - raise FluidXError(title="Decapper is not ready", message=status) + raise _fault("StatusNotOK", status) frames = await self._command(WASTE) self._require_accepted(WASTE, frames, "Wasting caps") - await self._wait_for_idle(timeout, "Wasting caps") + await self._wait_for_idle(timeout, "Wasting caps", "StoreWasNotSuccesful") logger.info("[IntelliXcap96 %s] waste complete", self.io.port) async def standby(self, timeout: float = 15.0) -> None: @@ -309,7 +340,7 @@ async def standby(self, timeout: float = 15.0) -> None: logger.info("[IntelliXcap96 %s] standby", self.io.port) return await asyncio.sleep(self.poll_interval) - raise FluidXError(title="Entering standby timed out") + raise _fault("CannotGoInStandbyMode", "standby timed out") async def ready(self, timeout: float = 30.0) -> None: """Wake the decapper from standby if it is asleep.""" @@ -317,7 +348,7 @@ async def ready(self, timeout: float = 30.0) -> None: return frames = await self._command(READY) self._require_accepted(READY, frames, "Waking from standby") - await self._wait_for_idle(timeout, "Waking from standby") + await self._wait_for_idle(timeout, "Waking from standby", "StatusNotOK") logger.info("[IntelliXcap96 %s] ready", self.io.port) async def reset(self, settle_time: float = 5.0) -> None: @@ -333,5 +364,5 @@ async def reset(self, settle_time: float = 5.0) -> None: await asyncio.sleep(settle_time) status = (await self.request_status()).upper() if "ERROR" in status: - raise FluidXError(title="Reset did not clear the error", message=status) + raise _fault("ResetNotSuccesful", status) logger.info("[IntelliXcap96 %s] reset complete", self.io.port) From 0c69ae026f25ed912e5e20388d8773ad8505b073 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Mon, 20 Jul 2026 19:07:12 -0700 Subject: [PATCH 4/8] fix(azenta): restore vendor-exact reset precheck and CannotReset message reset() reads status first and raises CannotReset (the firmware's own wording) when the device is already in an error state, matching the vendor sequence, before issuing the reset and re-checking for ResetNotSuccesful. Co-Authored-By: Claude Opus 4.8 (1M context) --- pylabrobot/azenta/fluidx/intellixcap96.py | 11 ++++++++--- pylabrobot/azenta/fluidx/intellixcap96_tests.py | 10 ++++++++-- 2 files changed, 16 insertions(+), 5 deletions(-) diff --git a/pylabrobot/azenta/fluidx/intellixcap96.py b/pylabrobot/azenta/fluidx/intellixcap96.py index f4f48729521..4f0be6e77fb 100644 --- a/pylabrobot/azenta/fluidx/intellixcap96.py +++ b/pylabrobot/azenta/fluidx/intellixcap96.py @@ -54,6 +54,10 @@ "Device was not able to reach the home position. The device is in an error state. " "Restart the device." ), + "CannotReset": ( + "Cannot send the reset command to the device. The device is in an error state. " + "Reset the device and try again." + ), "ResetNotSuccesful": ( "Reset did not complete. The device is in an error state. Reset the device and try again." ), @@ -360,9 +364,10 @@ async def reset(self, settle_time: float = 5.0) -> None: Args: settle_time: time in seconds to wait after the reset command. """ + if "ERROR" in (await self.request_status()).upper(): + raise _fault("CannotReset") await self._command(RESET) await asyncio.sleep(settle_time) - status = (await self.request_status()).upper() - if "ERROR" in status: - raise _fault("ResetNotSuccesful", status) + if "ERROR" in (await self.request_status()).upper(): + raise _fault("ResetNotSuccesful") logger.info("[IntelliXcap96 %s] reset complete", self.io.port) diff --git a/pylabrobot/azenta/fluidx/intellixcap96_tests.py b/pylabrobot/azenta/fluidx/intellixcap96_tests.py index 5487bcc2aba..470f32e672d 100644 --- a/pylabrobot/azenta/fluidx/intellixcap96_tests.py +++ b/pylabrobot/azenta/fluidx/intellixcap96_tests.py @@ -167,9 +167,15 @@ async def test_recap_blocked_when_not_decapped(self): await device.recap() async def test_reset_ignored_is_not_fatal(self): - device = self._make([[ACK, "zOK", "CommandIgnore"], status("StatusOK")]) + # Precheck OK, z is a no-op (CommandIgnore) at idle, postcheck OK. + device = self._make([status("StatusOK"), [ACK, "zOK", "CommandIgnore"], status("StatusOK")]) await device.reset() - self.assertEqual(device.io.written, ["z", "a"]) # type: ignore[attr-defined] + self.assertEqual(device.io.written, ["a", "z", "a"]) # type: ignore[attr-defined] + + async def test_reset_refused_when_errored(self): + device = self._make([status("StatusError")]) + with self.assertRaises(FluidXError): + await device.reset() if __name__ == "__main__": From 2b46868735a46aea6a862cc31441af922b9d5874 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Tue, 21 Jul 2026 17:35:09 -0700 Subject: [PATCH 5/8] feat(azenta): auto-recover from latched error; idempotent tray A fault latches the IntelliXcap 96 in StatusError (e.g. a decap with no rack). On this firmware the reset command is ignored and only homing clears the error, so add auto_recover (default on): an operation issued while the device is latched in error homes to recover and then proceeds. Disable to raise instead. Make tray open/close idempotent: a CommandIgnore reply means the tray is already in the requested position, which is success (no motion to wait for) rather than an error -- this also lets a tray move succeed right after a recovery home. Hardware-validated end to end: a no-rack decap raises DecapWasNotSuccesful and latches StatusError, and the next tray op auto-homes to recover and completes. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../fluidx/intellixcap96/hello-world.ipynb | 20 ++-- pylabrobot/azenta/fluidx/intellixcap96.py | 98 ++++++++++++++----- .../azenta/fluidx/intellixcap96_tests.py | 47 +++++++-- 3 files changed, 123 insertions(+), 42 deletions(-) diff --git a/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb b/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb index 713499f4c0e..25bf51dd819 100644 --- a/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb +++ b/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb @@ -90,9 +90,9 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "## Tray\n\n", - "\n\n", - "Open and close the loading tray. Each call returns once the tray has finished moving (status back to `StatusOK`). Asking for a position the tray is already in is a no-op and raises `FluidXError` (`CommandIgnore`)." + "## Tray\n", + "\n", + "Open and close the loading tray. Each call returns once the tray has finished moving (status back to `StatusOK`). Asking for a position the tray is already in is a harmless no-op." ] }, { @@ -135,12 +135,14 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "## Decap, recap, waste\n\n", - "\n\n", - "With a rack of screw-cap tubes loaded on the nest, `decap()` unscrews and holds all 96 caps, `recap()` screws them back on, and `waste()` drops the held caps into the waste bin. `decap()` refuses if caps are already held (recap first); `recap()` refuses if none are held.\n\n", - "\n\n", - "```{warning}\n\n", - "These operations are not yet hardware-verified. Test carefully.\n\n", + "## Decap, recap, waste\n", + "\n", + "With a rack of screw-cap tubes loaded on the nest, `decap()` unscrews and holds all 96 caps, `recap()` screws them back on, and `waste()` drops the held caps into the waste bin. `decap()` refuses if caps are already held (recap first); `recap()` refuses if none are held.\n", + "\n", + "A fault (e.g. decapping with no rack loaded) latches the device in `StatusError`, and the failing call raises a `FluidXError` carrying the firmware's message. A latched error is cleared **only by homing** -- the reset command is ignored on this firmware. By default (`auto_recover=True`) the next operation you issue homes to clear the error and then proceeds; pass `auto_recover=False` to have a latched error raise instead.\n", + "\n", + "```{warning}\n", + "These operations are not yet hardware-verified on a loaded rack. Test carefully.\n", "```" ] }, diff --git a/pylabrobot/azenta/fluidx/intellixcap96.py b/pylabrobot/azenta/fluidx/intellixcap96.py index 4f0be6e77fb..3c1a4cbdcc0 100644 --- a/pylabrobot/azenta/fluidx/intellixcap96.py +++ b/pylabrobot/azenta/fluidx/intellixcap96.py @@ -114,9 +114,14 @@ class FluidXIntelliXcap96: command answers with ``CommandIgnore``. Motions complete when the status word returns from ``StatusBUSY`` to ``StatusOK``. - Verified against hardware: connection, status, tray open/close, home, and - standby/ready. Decap, recap, and waste engage the cap head and are NOT yet - hardware-verified; a warning is emitted at setup. + A fault latches the device in ``StatusError`` (e.g. running a decap with no + rack loaded). This is cleared only by homing -- the reset command is ignored. + With ``auto_recover`` enabled (the default), an operation issued while the + device is latched in error homes to recover and then proceeds. + + Verified against hardware: connection, status, tray open/close, home, + standby/ready, and the decap error/recovery path. Decap, recap, and waste on a + loaded rack are NOT yet hardware-verified; a warning is emitted at setup. """ def __init__( @@ -126,6 +131,8 @@ def __init__( command_delay: float = 0.3, frame_gap: float = 0.5, poll_interval: float = 1.0, + auto_recover: bool = True, + recover_timeout: float = 30.0, ) -> None: """ Args: @@ -135,10 +142,17 @@ def __init__( frame_gap: how long to wait for another reply frame before concluding the reply is complete. poll_interval: pause between status polls while a motion runs. + auto_recover: when an operation finds the device latched in StatusError, + home it to clear the error and continue. A latched error is only cleared + by homing (reset is a no-op on this firmware). Disable to make a latched + error raise instead. + recover_timeout: timeout in seconds for the recovery home. """ self.command_delay = command_delay self.frame_gap = frame_gap self.poll_interval = poll_interval + self.auto_recover = auto_recover + self.recover_timeout = recover_timeout self.io = Serial( human_readable_device_name="FluidX IntelliXcap 96", port=port, @@ -225,12 +239,23 @@ async def _command(self, command: str) -> List[str]: def _status_frame(frames: List[str]) -> Optional[str]: return next((f for f in frames if f.upper().startswith("STATUS")), None) - def _require_accepted(self, command: str, frames: List[str], name: str) -> None: - """Raise unless the device acked and echoed the command without ignoring it.""" + def _require_accepted( + self, command: str, frames: List[str], name: str, idempotent: bool = False + ) -> bool: + """Check a command's reply. Return True if it started a motion. + + Raises if the device did not ack and echo the command. A ``CommandIgnore`` + reply means the command was a no-op (the device is already in the requested + state): for an ``idempotent`` command that is success and returns False (no + motion to wait for); otherwise it is raised. + """ if ACK not in frames or f"{command}OK" not in frames: raise _fault("NoAck", f"{name}: {frames!r}") if any(COMMAND_IGNORE in f for f in frames): + if idempotent: + return False raise _fault("CommandIgnore", f"{name}: device already in that state or not ready") + return True async def _wait_for_idle(self, timeout: float, name: str, fail_key: str) -> None: """Poll status until it returns to StatusOK, raising on an error or timeout. @@ -268,23 +293,52 @@ async def send_command(self, command: str) -> List[str]: async def open_tray(self, timeout: float = 15.0) -> None: """Open the loading tray.""" + await self._ensure_ready() frames = await self._command(OPEN_TRAY) - self._require_accepted(OPEN_TRAY, frames, "Opening the tray") - await self._wait_for_idle(timeout, "Opening the tray", "OpenTrayWasNotSuccesful") + if self._require_accepted(OPEN_TRAY, frames, "Opening the tray", idempotent=True): + await self._wait_for_idle(timeout, "Opening the tray", "OpenTrayWasNotSuccesful") logger.info("[IntelliXcap96 %s] tray open", self.io.port) async def close_tray(self, timeout: float = 15.0) -> None: """Close the loading tray.""" + await self._ensure_ready() frames = await self._command(CLOSE_TRAY) - self._require_accepted(CLOSE_TRAY, frames, "Closing the tray") - await self._wait_for_idle(timeout, "Closing the tray", "CloseTrayWasNotSuccesful") + if self._require_accepted(CLOSE_TRAY, frames, "Closing the tray", idempotent=True): + await self._wait_for_idle(timeout, "Closing the tray", "CloseTrayWasNotSuccesful") logger.info("[IntelliXcap96 %s] tray closed", self.io.port) - async def home(self, timeout: float = 30.0) -> None: - """Home all axes.""" + async def _home_sequence(self, timeout: float, name: str) -> None: + """Send the home command and wait for it to finish. Also clears a latched error.""" frames = await self._command(HOME_ALL) - self._require_accepted(HOME_ALL, frames, "Homing") - await self._wait_for_idle(timeout, "Homing", "HomeNotSuccesful") + self._require_accepted(HOME_ALL, frames, name) + await self._wait_for_idle(timeout, name, "HomeNotSuccesful") + + async def _ensure_ready(self) -> str: + """Return the current status, first clearing a latched error by homing. + + A ``StatusError`` is only cleared by homing (the reset command is ignored on + this firmware). With ``auto_recover`` enabled, an operation that finds the + device latched in error homes to recover and then proceeds; otherwise the + latched error is raised. + """ + status = await self.request_status() + if "ERROR" not in status.upper(): + return status + if not self.auto_recover: + raise _fault("StatusNotOK", status) + logger.warning( + "[IntelliXcap96 %s] latched in StatusError; homing to recover before continuing.", + self.io.port, + ) + await self._home_sequence(self.recover_timeout, "Homing (error recovery)") + status = await self.request_status() + if "ERROR" in status.upper(): + raise _fault("StatusNotOK", "error persisted after recovery home") + return status + + async def home(self, timeout: float = 30.0) -> None: + """Home all axes. Also clears a latched StatusError.""" + await self._home_sequence(timeout, "Homing") logger.info("[IntelliXcap96 %s] homed", self.io.port) async def decap(self, timeout: float = 60.0) -> None: @@ -293,11 +347,9 @@ async def decap(self, timeout: float = 60.0) -> None: Args: timeout: maximum time in seconds to wait for the stroke to finish. """ - status = (await self.request_status()).upper() + status = (await self._ensure_ready()).upper() if "RECAP" in status: raise _fault("NeedToRecap") - if "ERROR" in status: - raise _fault("StatusNotOK", status) frames = await self._command(DECAP_START) self._require_accepted(DECAP_START, frames, "Decapping") await self._wait_for_idle(timeout, "Decapping", "DecapWasNotSuccesful") @@ -309,11 +361,9 @@ async def recap(self, timeout: float = 60.0) -> None: Args: timeout: maximum time in seconds to wait for the stroke to finish. """ - status = (await self.request_status()).upper() + status = (await self._ensure_ready()).upper() if "DECAP" in status: raise _fault("NeedToDecap") - if "ERROR" in status: - raise _fault("StatusNotOK", status) frames = await self._command(RECAP_START) self._require_accepted(RECAP_START, frames, "Recapping") await self._wait_for_idle(timeout, "Recapping", "RecapWasNotSuccesful") @@ -325,9 +375,7 @@ async def waste(self, timeout: float = 60.0) -> None: Args: timeout: maximum time in seconds to wait for the stroke to finish. """ - status = (await self.request_status()).upper() - if "ERROR" in status: - raise _fault("StatusNotOK", status) + await self._ensure_ready() frames = await self._command(WASTE) self._require_accepted(WASTE, frames, "Wasting caps") await self._wait_for_idle(timeout, "Wasting caps", "StoreWasNotSuccesful") @@ -358,8 +406,10 @@ async def ready(self, timeout: float = 30.0) -> None: async def reset(self, settle_time: float = 5.0) -> None: """Reset the device and wait for it to settle. - A reset is a no-op when the device is already idle (the firmware answers - ``CommandIgnore``); it clears a recoverable error state. + The reset command is a no-op on this firmware (it answers ``CommandIgnore``) + and does not clear a latched error. Use :meth:`home` to recover from a + ``StatusError``; operations do this automatically when ``auto_recover`` is + enabled. Args: settle_time: time in seconds to wait after the reset command. diff --git a/pylabrobot/azenta/fluidx/intellixcap96_tests.py b/pylabrobot/azenta/fluidx/intellixcap96_tests.py index 470f32e672d..58f6279c6c8 100644 --- a/pylabrobot/azenta/fluidx/intellixcap96_tests.py +++ b/pylabrobot/azenta/fluidx/intellixcap96_tests.py @@ -66,8 +66,8 @@ async def asyncSetUp(self) -> None: patcher.start() self.addCleanup(patcher.stop) - def _make(self, script: List[List[str]]) -> FluidXIntelliXcap96: - device = FluidXIntelliXcap96(port="FAKE") + def _make(self, script: List[List[str]], auto_recover: bool = True) -> FluidXIntelliXcap96: + device = FluidXIntelliXcap96(port="FAKE", auto_recover=auto_recover) device.io = FakeXcapSerial(script) # type: ignore[assignment] return device @@ -94,23 +94,27 @@ async def test_request_status_returns_status_word(self): async def test_open_tray_moves_then_idle(self): device = self._make( [ + status("StatusOK"), # _ensure_ready: not in error [ACK, "fOK"], # f accepted status("StatusBUSY"), # still moving status("StatusOK"), # settled ] ) await device.open_tray() - self.assertEqual(device.io.written, ["f", "a", "a"]) # type: ignore[attr-defined] + self.assertEqual(device.io.written, ["a", "f", "a", "a"]) # type: ignore[attr-defined] - async def test_open_tray_ignored_raises(self): - device = self._make([[ACK, "fOK", "CommandIgnore"]]) - with self.assertRaises(FluidXError): - await device.open_tray() + async def test_open_tray_already_open_is_noop(self): + # CommandIgnore = tray already open = success; no status wait afterwards. + device = self._make([status("StatusOK"), [ACK, "fOK", "CommandIgnore"]]) + await device.open_tray() + self.assertEqual(device.io.written, ["a", "f"]) # type: ignore[attr-defined] async def test_close_tray_moves_then_idle(self): - device = self._make([[ACK, "gOK"], status("StatusBUSY"), status("StatusOK")]) + device = self._make( + [status("StatusOK"), [ACK, "gOK"], status("StatusBUSY"), status("StatusOK")] + ) await device.close_tray() - self.assertEqual(device.io.written, ["g", "a", "a"]) # type: ignore[attr-defined] + self.assertEqual(device.io.written, ["a", "g", "a", "a"]) # type: ignore[attr-defined] async def test_home_moves_then_idle(self): device = self._make([[ACK, "ZOK"], status("StatusBUSY"), status("StatusOK")]) @@ -177,6 +181,31 @@ async def test_reset_refused_when_errored(self): with self.assertRaises(FluidXError): await device.reset() + async def test_operation_auto_recovers_from_latched_error(self): + device = self._make( + [ + status("StatusError"), # _ensure_ready: latched in error + [ACK, "ZOK"], # recovery home accepted + status("StatusBUSY"), # homing + status("StatusOK"), # homed + status("StatusOK"), # _ensure_ready re-check: clear + [ACK, "fOK"], # open accepted + status("StatusBUSY"), + status("StatusOK"), + ] + ) + await device.open_tray() + self.assertEqual( + device.io.written, # type: ignore[attr-defined] + ["a", "Z", "a", "a", "a", "f", "a", "a"], + ) + + async def test_latched_error_raises_when_auto_recover_disabled(self): + device = self._make([status("StatusError")], auto_recover=False) + with self.assertRaises(FluidXError): + await device.open_tray() + self.assertEqual(device.io.written, ["a"]) # type: ignore[attr-defined] + if __name__ == "__main__": unittest.main() From 3a877be073baafdaee3503ea43da545980827c2c Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Thu, 23 Jul 2026 19:08:51 -0700 Subject: [PATCH 6/8] fix(azenta): align IntelliXcap behavior with hardware --- docs/api/pylabrobot.azenta.rst | 1 + .../fluidx/intellixcap96/hello-world.ipynb | 61 +++- pylabrobot/azenta/__init__.py | 2 +- pylabrobot/azenta/fluidx/__init__.py | 7 +- pylabrobot/azenta/fluidx/intellixcap96.py | 289 ++++++++++++++++-- .../azenta/fluidx/intellixcap96_tests.py | 105 ++++++- 6 files changed, 420 insertions(+), 45 deletions(-) diff --git a/docs/api/pylabrobot.azenta.rst b/docs/api/pylabrobot.azenta.rst index acd657f06d8..064609171d9 100644 --- a/docs/api/pylabrobot.azenta.rst +++ b/docs/api/pylabrobot.azenta.rst @@ -59,3 +59,4 @@ FluidX IntelliXcap 96 FluidXIntelliXcap96 FluidXError + get_error_message diff --git a/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb b/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb index 25bf51dd819..e14acbbb36c 100644 --- a/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb +++ b/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb @@ -15,7 +15,7 @@ "| Framing | each reply is wrapped in STX (`0x02`) .. ETX (`0x03`) |\n\n", "| Reply pattern | ACK (`0x06`), then `OK` echo, then a result frame |\n\n", "\n\n", - "Every command is a single character written followed by ETX. A motion command is accepted with an ACK and a `OK` echo; the status word then reads `StatusBUSY` while it moves and returns to `StatusOK` when done. A refused or no-op command answers `CommandIgnore`." + "Every command is a single character written followed by ETX. A motion command is accepted with an ACK and a `OK` echo; the status word then reads `StatusBUSY` while it moves. The terminal state depends on the operation: decap finishes at `StatusRECAP` while the head holds caps, whereas recap and waste finish at `StatusOK`. A refused or no-op command answers `CommandIgnore`." ] }, { @@ -23,10 +23,8 @@ "metadata": {}, "source": [ "```{note}\n\n", - "Connection, status, tray open/close, home, and standby/ready are verified against\n\n", - "hardware. **Decap, recap, and waste engage the cap head and are not yet\n\n", - "hardware-verified** -- validate them on your instrument before relying on them.\n\n", - "`setup()` emits a warning to that effect.\n\n", + "Connection, status, tray open/close, home, standby/ready, decap, recap, and\n\n", + "waste have been verified against hardware with a loaded 0.5 mL rack.\n\n", "```" ] }, @@ -38,6 +36,8 @@ "\n\n", "Connect the decapper's serial port to your computer through its USB-to-serial adapter and use the stable `by-id` path so the port survives re-enumeration, e.g. `/dev/serial/by-id/usb-FTDI_USB_Serial_Converter_XXXXXXXX-if00-port0`.\n\n", "\n\n", + "Tube volume is not a driver setting. The installed IntelliCartridge and its firmware profile define the supported tube height, cap geometry/thread, and motion settings. Use the cartridge specified for the exact tube family: two nominally 0.5 mL tube types may require different cartridges. Cartridge IDs 1–14 load their matching profiles automatically; extended cartridges require the corresponding stored profile to be selected on the instrument. The Python commands are otherwise identical for every supported tube type. See the [official Azenta IntelliXcap user manual](https://web.azenta.com/hubfs/azenta-files/resources/manuals-guides/319430-IXC-User-Manual.pdf) for cartridge/profile setup and compatibility guidance.\n\n", + "\n\n", "```{important}\n\n", "If the instrument comes up reporting `StatusBUSY` and ignores commands, it is\n\n", "locked out. The most common cause is an **engaged e-stop**; the safety guard/hood\n\n", @@ -53,7 +53,7 @@ "source": [ "## Connect\n\n", "\n\n", - "`setup()` opens the serial port and reads the status. It raises if the device is not ready (e.g. e-stop), and logs the not-fully-verified warning." + "`setup()` opens the serial port and reads the status. It raises if the device is not ready (e.g. e-stop)." ] }, { @@ -74,7 +74,7 @@ "source": [ "## Status\n\n", "\n\n", - "`request_status()` returns the current status word: `StatusOK` (idle), `StatusBUSY` (moving), or `StatusSLEEP` (standby). Note the status word does **not** encode the tray position." + "`request_status()` returns the current status word: `StatusOK` (idle with no caps held), `StatusBUSY` (moving), `StatusRECAP` (decapped; caps held and ready to recap or waste), `StatusSLEEP` (standby), or `StatusMANUAL` (the instrument requires inspection and touchscreen recovery). Note the status word does **not** encode the tray position." ] }, { @@ -86,13 +86,35 @@ "print(\"status:\", await decapper.request_status())" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Error codes\n", + "\n", + "The legacy serial protocol may report a generic operation failure such as `DecapERROR` without including the instrument's numeric error code. When a reply does include a known code, `FluidXError` decodes it automatically and exposes it as `error_code`. If the code is available separately from the instrument error log, look up its documented meaning with `get_error_message()` or construct the same exception with `FluidXError.from_error_code()`.\n", + "\n", + "```python\n", + "from pylabrobot.azenta.fluidx import FluidXError, get_error_message\n", + "\n", + "print(get_error_message(145))\n", + "# Light curtain calibration max retries exceeded.\n", + "\n", + "error = FluidXError.from_error_code(145)\n", + "print(error)\n", + "# IntelliXcap error 145: Light curtain calibration max retries exceeded.\n", + "```\n", + "\n", + "The complete local mapping comes from the error table in the [official Azenta IntelliXcap user manual](https://web.azenta.com/hubfs/azenta-files/resources/manuals-guides/319430-IXC-User-Manual.pdf)." + ] + }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Tray\n", "\n", - "Open and close the loading tray. Each call returns once the tray has finished moving (status back to `StatusOK`). Asking for a position the tray is already in is a harmless no-op." + "Open and close the loading tray. Asking for a position the tray is already in is a harmless no-op. Tray motion does not change whether caps are held, so after decapping the terminal state remains `StatusRECAP` rather than returning to `StatusOK`." ] }, { @@ -137,15 +159,25 @@ "source": [ "## Decap, recap, waste\n", "\n", - "With a rack of screw-cap tubes loaded on the nest, `decap()` unscrews and holds all 96 caps, `recap()` screws them back on, and `waste()` drops the held caps into the waste bin. `decap()` refuses if caps are already held (recap first); `recap()` refuses if none are held.\n", + "With a rack of screw-cap tubes loaded on the nest, `decap()` unscrews and holds all 96 caps. After decapping, choose **one** terminal operation: `recap()` screws those caps back onto the tubes, while `waste()` releases them into a separately positioned cap carrier. `decap()` refuses if caps are already held (recap or waste first); `recap()` refuses if none are held.\n", "\n", - "A fault (e.g. decapping with no rack loaded) latches the device in `StatusError`, and the failing call raises a `FluidXError` carrying the firmware's message. A latched error is cleared **only by homing** -- the reset command is ignored on this firmware. By default (`auto_recover=True`) the next operation you issue homes to clear the error and then proceeds; pass `auto_recover=False` to have a latched error raise instead.\n", + "A fault (e.g. decapping with no rack loaded) can latch the device in `StatusError`, and the failing call raises a `FluidXError` carrying the firmware's message. A latched error is cleared **only by homing** -- the reset command is ignored on this firmware. By default (`auto_recover=True`) the next operation you issue homes to clear `StatusError` and then proceeds; pass `auto_recover=False` to have it raise instead. `StatusMANUAL` requires an explicit safety decision: inspect the rack and cap head, confirm that axis motion is safe, then call `reset_error()`. Hardware testing confirmed that it homes the machine from `StatusMANUAL` through `StatusBUSY` to `StatusOK`. Normal motion commands remain blocked until recovery completes.\n", "\n", "```{warning}\n", - "These operations are not yet hardware-verified on a loaded rack. Test carefully.\n", + "`waste()` is irreversible and does not detect whether a cap carrier is present. Before calling it, open the tray, remove the sample-tube rack, and position the correct cap carrier or collection vessel according to your instrument's procedure. On hardware, leaving the tube rack beneath the head caused the released caps to fall back onto the tube openings. They looked recapped but were not guaranteed to be threaded or torqued. See the [official Azenta IntelliXcap user manual](https://web.azenta.com/hubfs/azenta-files/resources/manuals-guides/319430-IXC-User-Manual.pdf), which defines waste as dropping caps into a carrier.\n", "```" ] }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# After inspecting the rack/head and confirming motion is safe:\n", + "await decapper.reset_error()" + ] + }, { "cell_type": "code", "execution_count": null, @@ -161,7 +193,8 @@ "metadata": {}, "outputs": [], "source": [ - "await decapper.recap() # screw the held caps back on" + "# Choose this path to retain the caps.\n", + "await decapper.recap() # screw and torque the held caps back on" ] }, { @@ -170,7 +203,9 @@ "metadata": {}, "outputs": [], "source": [ - "await decapper.waste() # drop held caps to the waste bin" + "# Alternative to recap: with caps held, remove the tube rack and position\n", + "# the correct cap carrier first. Do not run this after recap.\n", + "await decapper.waste() # irreversibly release held caps into the carrier" ] }, { diff --git a/pylabrobot/azenta/__init__.py b/pylabrobot/azenta/__init__.py index 169014bd5ac..8f32b8a13b6 100644 --- a/pylabrobot/azenta/__init__.py +++ b/pylabrobot/azenta/__init__.py @@ -1,3 +1,3 @@ from .a4s import A4S, A4SDriver, A4SSealerBackend, A4SStatus, A4STemperatureBackend -from .fluidx import FluidXError, FluidXIntelliXcap96 +from .fluidx import ERROR_CODE_MESSAGES, FluidXError, FluidXIntelliXcap96, get_error_message from .xpeel import XPeel, XPeelDriver, XPeelPeelerBackend diff --git a/pylabrobot/azenta/fluidx/__init__.py b/pylabrobot/azenta/fluidx/__init__.py index 4099b5a19ef..6f398a5a2ef 100644 --- a/pylabrobot/azenta/fluidx/__init__.py +++ b/pylabrobot/azenta/fluidx/__init__.py @@ -1 +1,6 @@ -from .intellixcap96 import FluidXError, FluidXIntelliXcap96 +from .intellixcap96 import ( + ERROR_CODE_MESSAGES, + FluidXError, + FluidXIntelliXcap96, + get_error_message, +) diff --git a/pylabrobot/azenta/fluidx/intellixcap96.py b/pylabrobot/azenta/fluidx/intellixcap96.py index 3c1a4cbdcc0..e249bf10f6b 100644 --- a/pylabrobot/azenta/fluidx/intellixcap96.py +++ b/pylabrobot/azenta/fluidx/intellixcap96.py @@ -1,6 +1,7 @@ import asyncio import logging -from typing import List, Optional +import re +from typing import Dict, List, Optional, Tuple from pylabrobot.io.serial import Serial @@ -10,7 +11,7 @@ # with an ACK frame (a lone 0x06), a command-echo frame ("OK"), and then a # result frame (a "Status..." word, or "CommandIgnore" when the command is a # no-op or refused). Motions run asynchronously: the status word is "StatusBUSY" -# while moving and returns to "StatusOK" when done. +# while moving, then changes to the operation-specific terminal state. STX = b"\x02" ETX = b"\x03" ACK = "\x06" @@ -62,17 +63,146 @@ "Reset did not complete. The device is in an error state. Reset the device and try again." ), "CannotGoInStandbyMode": "Cannot go to standby mode. Check the errors on the device.", + "StatusManual": ( + "The device is in manual recovery mode. Inspect the rack and cap head, then complete the " + "appropriate recovery from the instrument touchscreen before sending another motion command." + ), "CommandIgnore": "Command was ignored by the device.", "NoAck": "Device did not acknowledge the command.", } +# Legacy serial firmware generally reports only a generic frame such as +# "DecapERROR", but some firmware may include a numeric error code in a reply. +# +# Source: Azenta IntelliXcap User Manual, part 319430 Rev. E, pp. 88-91: +# https://web.azenta.com/hubfs/azenta-files/resources/manuals-guides/319430-IXC-User-Manual.pdf +ERROR_CODE_MESSAGES: Dict[int, str] = { + 100: ( + "M1 top switch not detected during homing sequence. Could get overwritten by other error " + "codes within higher level sequencing logic, so it is most likely during startup." + ), + 101: "M2 initial homing failure. Likely to override other M1 homing error codes.", + 102: "M1 top switch stuck closed during homing sequence.", + 103: "M1 top switch second trigger not detected during homing sequence.", + 104: "M4 homing error: top switch not detected.", + 105: ( + "M3 top switch not detected during homing sequence. Could get overwritten by other error " + "codes within higher level sequencing logic, so it is most likely during startup." + ), + 106: "M3 stop switch stuck closed during homing sequence.", + 107: "M3 top switch second trigger not detected during homing sequence.", + 108: "M3 initial homing failure. Likely to override other M3 homing error codes.", + 109: ( + "M2 top switch not detected during homing sequence. Could get overwritten by other error " + "codes within higher level sequencing logic, so it is most likely during startup." + ), + 110: "M2 top switch stuck closed during homing sequence.", + 111: "M2 top switch second trigger not detected during homing sequence.", + 112: "Door close failure.", + 113: ( + "M1 moved to M1_SAFETY_LOW_POS (S33): no light curtain trigger was detected while " + "scanning for caps." + ), + 114: "Invalid tube height detected.", + 115: "Door open failure.", + 116: "Door close failure at start of sequence.", + 117: ( + "M1 moved to M1_SAFETY_LOW_POS (S33): no light curtain trigger was detected while " + "scanning for caps." + ), + 118: "Invalid tube height detected.", + 119: "Open door failure.", + 120: "Open door failure on entry to manual mode.", + 121: "Door close failure.", + 122: "M3 limit switch timeout on cartridge eject.", + 123: "Door open failure at end of cartridge eject sequence.", + 124: "Door close failure at end of cartridge eject sequence.", + 125: "M1 failed to reach the waste position within S4 during the auto-waste sequence.", + 133: "M1 homing error.", + 134: "Open door failure.", + 135: ("Cap detected at valid height. This may be an informational device-state code."), + 136: "Maximum decap attempts exceeded (S46).", + 137: "Maximum recap attempts exceeded (S45).", + 138: ( + "M3 bottom switch closed while the motor was still moving; extended-stage lead-screw " + "protection activated." + ), + 139: "Open tray failure; no cartridge detected after initial homing.", + 140: ( + "Cartridge-ejected notification; or the door should be up but the top switch was not detected." + ), + 141: "The door should be down but the bottom switch was not detected.", + 142: "Unexpected object on tray during cartridge eject.", + 143: "Cartridge not detected during cartridge load sequence.", + 144: ( + "Cartridge detection height incorrect during cartridge load sequence: detected height " + "was less than S73 - S59." + ), + 145: "Light curtain calibration max retries exceeded.", + 146: "Light curtain calibration max retries exceeded.", + 147: "Light curtain calibration max retries exceeded.", + 148: "Tray open failure.", + 150: "M3 homing error during auto-waste sequence.", + 151: "Tray close failure.", + 152: "Tube detected after decap retry; caps were screwed back on.", + 153: "Close tray failure; M3 homing error.", + 154: "Close tray failure.", + 155: "Open door failure.", + 156: "M1 homing error.", + 157: "M2 homing error.", + 158: "M3 homing error.", + 159: "M2 homing error.", + 160: "Door close failure at end of sequence; tray open failure.", + 161: "M4 homing error.", + 164: "Tray open failure.", + 165: "Sequence-state error; the same firmware logic may report error 167.", + 166: "M2 homing error during tray decap-quit.", + 167: "Door-open or tray-close failure during decap-quit.", + 200: "Light curtain communications failure: no Modbus data received.", + 201: "Light curtain signal failure; check wiring between controller and light curtain.", + 202: ( + "Conflicting limit switches: top and bottom switches both appear closed. This usually " + "indicates a power-supply failure or a faulty switch." + ), + 238: "Emergency stop engaged or motor voltage low.", +} + + +def get_error_message(code: int) -> str: + """Return the documented meaning of an IntelliXcap error code. + + Some codes have multiple meanings because their interpretation depends on + the firmware sequence that reported them. + """ + message = ERROR_CODE_MESSAGES.get(code) + if message is None: + return "Unknown IntelliXcap error code." + return message + class FluidXError(Exception): """Exceptions raised by a FluidX IntelliXcap 96 decapper.""" - def __init__(self, title: str, message: Optional[str] = None) -> None: + def __init__( + self, + title: str, + message: Optional[str] = None, + error_code: Optional[int] = None, + ) -> None: self.title = title self.message = message + self.error_code = error_code + + @classmethod + def from_error_code(cls, code: int, detail: Optional[str] = None) -> "FluidXError": + """Build an exception from a numeric IntelliXcap error code.""" + meaning = get_error_message(code) + message = f"{meaning} {detail}" if detail else meaning + return cls( + title=f"IntelliXcap error {code}", + message=message, + error_code=code, + ) def __str__(self) -> str: return f"{self.title}: {self.message}" if self.message else self.title @@ -83,24 +213,43 @@ def _fault(key: str, detail: Optional[str] = None) -> FluidXError: return FluidXError(title=ERROR_MESSAGES.get(key, key), message=detail) +def _error_code(frames: List[str]) -> Optional[int]: + """Extract a known three-digit error code from serial reply frames.""" + for frame in frames: + for value in re.findall(r"(? None: - logger.warning( - "FluidXIntelliXcap96: connection, status, tray, home, and standby/ready are " - "hardware-verified, but decap, recap, and waste are NOT yet verified in PyLabRobot. " - "Please make a PR to remove this message once you have verified them on your hardware." - ) await self.io.setup() status = await self.request_status() up = status.upper() @@ -190,6 +338,8 @@ async def setup(self) -> None: "E-STOP; also check the safety guard/hood and interlocks, then retry." ), ) + if "MANUAL" in up: + raise _fault("StatusManual", status) if "ERROR" in up: raise _fault("StatusNotOK", status) logger.info("[IntelliXcap96 %s] connected: %s", self.io.port, status) @@ -257,23 +407,41 @@ def _require_accepted( raise _fault("CommandIgnore", f"{name}: device already in that state or not ready") return True - async def _wait_for_idle(self, timeout: float, name: str, fail_key: str) -> None: - """Poll status until it returns to StatusOK, raising on an error or timeout. + async def _wait_for_idle( + self, + timeout: float, + name: str, + fail_key: str, + terminal_statuses: Tuple[str, ...] = ("StatusOK",), + ) -> None: + """Poll status until it reaches an expected idle state. ``fail_key`` names the firmware error message to raise if the status word - reports an error while waiting. + reports an error while waiting. ``terminal_statuses`` accounts for + operation state retained while the instrument is idle: hardware reports + ``StatusRECAP`` after decapping and after tray motion with caps held. """ loop = asyncio.get_running_loop() deadline = loop.time() + timeout + expected = {status.upper() for status in terminal_statuses} while loop.time() < deadline: frames = await self._command(STATUS) if any("ERROR" in f.upper() for f in frames): - raise _fault(fail_key, f"{name}: {frames!r}") + code = _error_code(frames) + if code is not None: + raise FluidXError.from_error_code(code, detail=f"{name}: {frames!r}") + raise _fault( + fail_key, + f"{name}: {frames!r}. Check the instrument error log for the numeric error code.", + ) status = self._status_frame(frames) - if status is not None and "OK" in status.upper(): + if status is not None and status.upper() in expected: return await asyncio.sleep(self.poll_interval) - raise FluidXError(title=f"{name} timed out", message=f"not idle within {timeout}s") + raise FluidXError( + title=f"{name} timed out", + message=f"did not reach {terminal_statuses!r} within {timeout}s", + ) # === Status === @@ -296,7 +464,12 @@ async def open_tray(self, timeout: float = 15.0) -> None: await self._ensure_ready() frames = await self._command(OPEN_TRAY) if self._require_accepted(OPEN_TRAY, frames, "Opening the tray", idempotent=True): - await self._wait_for_idle(timeout, "Opening the tray", "OpenTrayWasNotSuccesful") + await self._wait_for_idle( + timeout, + "Opening the tray", + "OpenTrayWasNotSuccesful", + ("StatusOK", "StatusRECAP", "StatusDECAP"), + ) logger.info("[IntelliXcap96 %s] tray open", self.io.port) async def close_tray(self, timeout: float = 15.0) -> None: @@ -304,7 +477,12 @@ async def close_tray(self, timeout: float = 15.0) -> None: await self._ensure_ready() frames = await self._command(CLOSE_TRAY) if self._require_accepted(CLOSE_TRAY, frames, "Closing the tray", idempotent=True): - await self._wait_for_idle(timeout, "Closing the tray", "CloseTrayWasNotSuccesful") + await self._wait_for_idle( + timeout, + "Closing the tray", + "CloseTrayWasNotSuccesful", + ("StatusOK", "StatusRECAP", "StatusDECAP"), + ) logger.info("[IntelliXcap96 %s] tray closed", self.io.port) async def _home_sequence(self, timeout: float, name: str) -> None: @@ -322,7 +500,10 @@ async def _ensure_ready(self) -> str: latched error is raised. """ status = await self.request_status() - if "ERROR" not in status.upper(): + up = status.upper() + if "MANUAL" in up: + raise _fault("StatusManual", status) + if "ERROR" not in up: return status if not self.auto_recover: raise _fault("StatusNotOK", status) @@ -336,6 +517,31 @@ async def _ensure_ready(self) -> str: raise _fault("StatusNotOK", "error persisted after recovery home") return status + async def reset_error(self, timeout: Optional[float] = None) -> None: + """Recover from ``StatusError`` or ``StatusMANUAL`` by homing all axes. + + The firmware reset command does not clear these states. Hardware testing + confirmed that the home-all command transitions ``StatusMANUAL`` through + ``StatusBUSY`` to ``StatusOK``. Call this only after inspecting the rack and + cap head and confirming that axis motion is safe. + + This method is a no-op when the instrument is not in an error or manual + recovery state. + + Args: + timeout: maximum recovery time in seconds. Defaults to + ``recover_timeout`` configured on this instance. + """ + status = await self.request_status() + up = status.upper() + if "ERROR" not in up and "MANUAL" not in up: + return + await self._home_sequence( + self.recover_timeout if timeout is None else timeout, + "Resetting error", + ) + logger.info("[IntelliXcap96 %s] error reset by homing", self.io.port) + async def home(self, timeout: float = 30.0) -> None: """Home all axes. Also clears a latched StatusError.""" await self._home_sequence(timeout, "Homing") @@ -352,7 +558,12 @@ async def decap(self, timeout: float = 60.0) -> None: raise _fault("NeedToRecap") frames = await self._command(DECAP_START) self._require_accepted(DECAP_START, frames, "Decapping") - await self._wait_for_idle(timeout, "Decapping", "DecapWasNotSuccesful") + await self._wait_for_idle( + timeout, + "Decapping", + "DecapWasNotSuccesful", + ("StatusRECAP",), + ) logger.info("[IntelliXcap96 %s] decap complete", self.io.port) async def recap(self, timeout: float = 60.0) -> None: @@ -362,23 +573,45 @@ async def recap(self, timeout: float = 60.0) -> None: timeout: maximum time in seconds to wait for the stroke to finish. """ status = (await self._ensure_ready()).upper() - if "DECAP" in status: + if "RECAP" not in status: raise _fault("NeedToDecap") frames = await self._command(RECAP_START) self._require_accepted(RECAP_START, frames, "Recapping") - await self._wait_for_idle(timeout, "Recapping", "RecapWasNotSuccesful") + await self._wait_for_idle( + timeout, + "Recapping", + "RecapWasNotSuccesful", + ("StatusOK", "StatusDECAP"), + ) logger.info("[IntelliXcap96 %s] recap complete", self.io.port) async def waste(self, timeout: float = 60.0) -> None: - """Drop the currently held caps into the waste bin. + """Release the currently held caps into a separately positioned cap carrier. + + This is irreversible. Before calling, remove the sample-tube rack and + position the correct cap carrier/collection vessel as specified in the + Azenta IntelliXcap user manual. The instrument does not detect or verify the + carrier. If the tube rack remains beneath the head, released caps can fall + back onto the tube openings and look recapped even though they may not be + threaded or torqued. + + User manual: + https://web.azenta.com/hubfs/azenta-files/resources/manuals-guides/319430-IXC-User-Manual.pdf Args: timeout: maximum time in seconds to wait for the stroke to finish. """ - await self._ensure_ready() + status = (await self._ensure_ready()).upper() + if "RECAP" not in status: + raise _fault("NeedToDecap", "waste requires caps held after decapping") frames = await self._command(WASTE) self._require_accepted(WASTE, frames, "Wasting caps") - await self._wait_for_idle(timeout, "Wasting caps", "StoreWasNotSuccesful") + await self._wait_for_idle( + timeout, + "Wasting caps", + "StoreWasNotSuccesful", + ("StatusOK", "StatusDECAP"), + ) logger.info("[IntelliXcap96 %s] waste complete", self.io.port) async def standby(self, timeout: float = 15.0) -> None: diff --git a/pylabrobot/azenta/fluidx/intellixcap96_tests.py b/pylabrobot/azenta/fluidx/intellixcap96_tests.py index 58f6279c6c8..3955509bbef 100644 --- a/pylabrobot/azenta/fluidx/intellixcap96_tests.py +++ b/pylabrobot/azenta/fluidx/intellixcap96_tests.py @@ -3,7 +3,7 @@ from typing import Iterator, List from unittest.mock import AsyncMock, patch -from pylabrobot.azenta.fluidx import FluidXError, FluidXIntelliXcap96 +from pylabrobot.azenta.fluidx import FluidXError, FluidXIntelliXcap96, get_error_message ACK = "\x06" @@ -87,10 +87,31 @@ async def test_setup_error_raises(self): with self.assertRaises(FluidXError): await device.setup() + async def test_setup_manual_recovery_raises(self): + device = self._make([status("StatusMANUAL")]) + with self.assertRaises(FluidXError) as ctx: + await device.setup() + self.assertIn("manual recovery", str(ctx.exception)) + async def test_request_status_returns_status_word(self): device = self._make([status("StatusOK")]) self.assertEqual(await device.request_status(), "StatusOK") + def test_get_error_message(self): + self.assertEqual( + get_error_message(145), + "Light curtain calibration max retries exceeded.", + ) + self.assertEqual(get_error_message(999), "Unknown IntelliXcap error code.") + + def test_fluidx_error_from_error_code(self): + error = FluidXError.from_error_code(145) + self.assertEqual(error.error_code, 145) + self.assertEqual( + str(error), + "IntelliXcap error 145: Light curtain calibration max retries exceeded.", + ) + async def test_open_tray_moves_then_idle(self): device = self._make( [ @@ -116,6 +137,13 @@ async def test_close_tray_moves_then_idle(self): await device.close_tray() self.assertEqual(device.io.written, ["a", "g", "a", "a"]) # type: ignore[attr-defined] + async def test_close_tray_preserves_caps_held_state(self): + device = self._make( + [status("StatusRECAP"), [ACK, "gOK"], status("StatusBUSY"), status("StatusRECAP")] + ) + await device.close_tray() + self.assertEqual(device.io.written, ["a", "g", "a", "a"]) # type: ignore[attr-defined] + async def test_home_moves_then_idle(self): device = self._make([[ACK, "ZOK"], status("StatusBUSY"), status("StatusOK")]) await device.home() @@ -149,7 +177,7 @@ async def test_decap_success(self): status("StatusOK"), # precheck: not recapped, no error [ACK, "hOK"], # h accepted status("StatusBUSY"), - status("StatusOK"), + status("StatusRECAP"), # hardware terminal state: caps held ] ) await device.decap() @@ -165,11 +193,48 @@ async def test_decap_reports_error_during_motion(self): with self.assertRaises(FluidXError): await device.decap() + async def test_decap_decodes_error_code_when_firmware_includes_it(self): + device = self._make([status("StatusOK"), [ACK, "hOK"], [ACK, "aOK", "DecapERROR 145"]]) + with self.assertRaises(FluidXError) as ctx: + await device.decap() + self.assertEqual(ctx.exception.error_code, 145) + self.assertIn("Light curtain calibration max retries exceeded", str(ctx.exception)) + async def test_recap_blocked_when_not_decapped(self): device = self._make([status("StatusDecap")]) with self.assertRaises(FluidXError): await device.recap() + async def test_recap_success(self): + device = self._make( + [ + status("StatusRECAP"), # precheck: caps held + [ACK, "iOK"], # i accepted + status("StatusBUSY"), + status("StatusOK"), + ] + ) + await device.recap() + self.assertEqual(device.io.written, ["a", "i", "a", "a"]) # type: ignore[attr-defined] + + async def test_waste_blocked_when_no_caps_are_held(self): + device = self._make([status("StatusOK")]) + with self.assertRaises(FluidXError): + await device.waste() + self.assertEqual(device.io.written, ["a"]) # type: ignore[attr-defined] + + async def test_waste_success(self): + device = self._make( + [ + status("StatusRECAP"), # precheck: caps held + [ACK, "bOK"], # b accepted + status("StatusBUSY"), + status("StatusOK"), + ] + ) + await device.waste() + self.assertEqual(device.io.written, ["a", "b", "a", "a"]) # type: ignore[attr-defined] + async def test_reset_ignored_is_not_fatal(self): # Precheck OK, z is a no-op (CommandIgnore) at idle, postcheck OK. device = self._make([status("StatusOK"), [ACK, "zOK", "CommandIgnore"], status("StatusOK")]) @@ -181,6 +246,35 @@ async def test_reset_refused_when_errored(self): with self.assertRaises(FluidXError): await device.reset() + async def test_reset_error_recovers_manual_state_by_homing(self): + device = self._make( + [ + status("StatusMANUAL"), + [ACK, "ZOK"], + status("StatusBUSY"), + status("StatusOK"), + ] + ) + await device.reset_error() + self.assertEqual(device.io.written, ["a", "Z", "a", "a"]) # type: ignore[attr-defined] + + async def test_reset_error_recovers_error_state_by_homing(self): + device = self._make( + [ + status("StatusError"), + [ACK, "ZOK"], + status("StatusBUSY"), + status("StatusOK"), + ] + ) + await device.reset_error() + self.assertEqual(device.io.written, ["a", "Z", "a", "a"]) # type: ignore[attr-defined] + + async def test_reset_error_is_noop_when_healthy(self): + device = self._make([status("StatusOK")]) + await device.reset_error() + self.assertEqual(device.io.written, ["a"]) # type: ignore[attr-defined] + async def test_operation_auto_recovers_from_latched_error(self): device = self._make( [ @@ -206,6 +300,13 @@ async def test_latched_error_raises_when_auto_recover_disabled(self): await device.open_tray() self.assertEqual(device.io.written, ["a"]) # type: ignore[attr-defined] + async def test_operation_blocked_during_manual_recovery(self): + device = self._make([status("StatusMANUAL")]) + with self.assertRaises(FluidXError) as ctx: + await device.open_tray() + self.assertIn("manual recovery", str(ctx.exception)) + self.assertEqual(device.io.written, ["a"]) # type: ignore[attr-defined] + if __name__ == "__main__": unittest.main() From 4626c8334e2e8fe03001fa006d4a8c01b5ce1912 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Thu, 23 Jul 2026 19:11:35 -0700 Subject: [PATCH 7/8] docs(azenta): fix IntelliXcap tutorial header --- .../fluidx/intellixcap96/hello-world.ipynb | 37 ++++++++----------- 1 file changed, 16 insertions(+), 21 deletions(-) diff --git a/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb b/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb index e14acbbb36c..40c9848310b 100644 --- a/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb +++ b/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb @@ -4,30 +4,25 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "# FluidX IntelliXcap 96\n\n", - "\n\n", - "The FluidX IntelliXcap 96 (an Azenta brand) is an automated screw-cap **decapper**. It unscrews, holds, and screws back on all 96 caps of a screw-cap tube rack in a single stroke. A plate mover loads the rack onto its one nest.\n\n", - "\n\n", - "| Property | Value |\n\n", - "|---|---|\n\n", - "| Communication | Serial, STX/ETX-framed ASCII |\n\n", - "| Serial settings | 9600 baud, 8 data bits, no parity, 1 stop bit, no handshake |\n\n", - "| Framing | each reply is wrapped in STX (`0x02`) .. ETX (`0x03`) |\n\n", - "| Reply pattern | ACK (`0x06`), then `OK` echo, then a result frame |\n\n", - "\n\n", + "# FluidX IntelliXcap 96\n", + "\n", + "The FluidX IntelliXcap 96 (an Azenta brand) is an automated screw-cap **decapper**. It unscrews, holds, and screws back on all 96 caps of a screw-cap tube rack in a single stroke. A plate mover loads the rack onto its one nest.\n", + "\n", + "## Resources\n", + "\n", + "- [Azenta IntelliXcap user manual](https://web.azenta.com/hubfs/azenta-files/resources/manuals-guides/319430-IXC-User-Manual.pdf)\n", + "- [Azenta IntelliXcap 96 product page](https://www.azenta.com/products/intellixcap-automated-screw-cap-decapper-recapper-96-format)\n", + "\n", + "| Property | Value |\n", + "| --- | --- |\n", + "| Communication | Serial, STX/ETX-framed ASCII |\n", + "| Serial settings | 9600 baud, 8 data bits, no parity, 1 stop bit, no handshake |\n", + "| Framing | each reply is wrapped in STX (`0x02`) .. ETX (`0x03`) |\n", + "| Reply pattern | ACK (`0x06`), then `OK` echo, then a result frame |\n", + "\n", "Every command is a single character written followed by ETX. A motion command is accepted with an ACK and a `OK` echo; the status word then reads `StatusBUSY` while it moves. The terminal state depends on the operation: decap finishes at `StatusRECAP` while the head holds caps, whereas recap and waste finish at `StatusOK`. A refused or no-op command answers `CommandIgnore`." ] }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "```{note}\n\n", - "Connection, status, tray open/close, home, standby/ready, decap, recap, and\n\n", - "waste have been verified against hardware with a loaded 0.5 mL rack.\n\n", - "```" - ] - }, { "cell_type": "markdown", "metadata": {}, From 40ee686bac07b9dccd3899231d7fff4b2b753d80 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Thu, 23 Jul 2026 19:12:52 -0700 Subject: [PATCH 8/8] refactor(azenta): expose IntelliXcap command transport --- pylabrobot/azenta/fluidx/intellixcap96.py | 30 ++++++++++------------- 1 file changed, 13 insertions(+), 17 deletions(-) diff --git a/pylabrobot/azenta/fluidx/intellixcap96.py b/pylabrobot/azenta/fluidx/intellixcap96.py index e249bf10f6b..21edf45c986 100644 --- a/pylabrobot/azenta/fluidx/intellixcap96.py +++ b/pylabrobot/azenta/fluidx/intellixcap96.py @@ -372,8 +372,8 @@ async def _read_frame(self) -> Optional[str]: buf += byte return buf.decode("ascii", errors="replace") - async def _command(self, command: str) -> List[str]: - """Send a command and collect every reply frame until the reply goes quiet.""" + async def send_command(self, command: str) -> List[str]: + """Send a raw command and collect every reply frame until the reply goes quiet.""" await self._send(command) frames: List[str] = [] with self.io.temporary_timeout(self.frame_gap): @@ -425,7 +425,7 @@ async def _wait_for_idle( deadline = loop.time() + timeout expected = {status.upper() for status in terminal_statuses} while loop.time() < deadline: - frames = await self._command(STATUS) + frames = await self.send_command(STATUS) if any("ERROR" in f.upper() for f in frames): code = _error_code(frames) if code is not None: @@ -447,22 +447,18 @@ async def _wait_for_idle( async def request_status(self) -> str: """Poll the device and return its status word (e.g. ``StatusOK``).""" - frames = await self._command(STATUS) + frames = await self.send_command(STATUS) status = self._status_frame(frames) if status is None: raise FluidXError(title="No status reply", message=repr(frames)) return status - async def send_command(self, command: str) -> List[str]: - """Send a raw single command and return its reply frames. Escape hatch.""" - return await self._command(command) - # === Operations === async def open_tray(self, timeout: float = 15.0) -> None: """Open the loading tray.""" await self._ensure_ready() - frames = await self._command(OPEN_TRAY) + frames = await self.send_command(OPEN_TRAY) if self._require_accepted(OPEN_TRAY, frames, "Opening the tray", idempotent=True): await self._wait_for_idle( timeout, @@ -475,7 +471,7 @@ async def open_tray(self, timeout: float = 15.0) -> None: async def close_tray(self, timeout: float = 15.0) -> None: """Close the loading tray.""" await self._ensure_ready() - frames = await self._command(CLOSE_TRAY) + frames = await self.send_command(CLOSE_TRAY) if self._require_accepted(CLOSE_TRAY, frames, "Closing the tray", idempotent=True): await self._wait_for_idle( timeout, @@ -487,7 +483,7 @@ async def close_tray(self, timeout: float = 15.0) -> None: async def _home_sequence(self, timeout: float, name: str) -> None: """Send the home command and wait for it to finish. Also clears a latched error.""" - frames = await self._command(HOME_ALL) + frames = await self.send_command(HOME_ALL) self._require_accepted(HOME_ALL, frames, name) await self._wait_for_idle(timeout, name, "HomeNotSuccesful") @@ -556,7 +552,7 @@ async def decap(self, timeout: float = 60.0) -> None: status = (await self._ensure_ready()).upper() if "RECAP" in status: raise _fault("NeedToRecap") - frames = await self._command(DECAP_START) + frames = await self.send_command(DECAP_START) self._require_accepted(DECAP_START, frames, "Decapping") await self._wait_for_idle( timeout, @@ -575,7 +571,7 @@ async def recap(self, timeout: float = 60.0) -> None: status = (await self._ensure_ready()).upper() if "RECAP" not in status: raise _fault("NeedToDecap") - frames = await self._command(RECAP_START) + frames = await self.send_command(RECAP_START) self._require_accepted(RECAP_START, frames, "Recapping") await self._wait_for_idle( timeout, @@ -604,7 +600,7 @@ async def waste(self, timeout: float = 60.0) -> None: status = (await self._ensure_ready()).upper() if "RECAP" not in status: raise _fault("NeedToDecap", "waste requires caps held after decapping") - frames = await self._command(WASTE) + frames = await self.send_command(WASTE) self._require_accepted(WASTE, frames, "Wasting caps") await self._wait_for_idle( timeout, @@ -616,7 +612,7 @@ async def waste(self, timeout: float = 60.0) -> None: async def standby(self, timeout: float = 15.0) -> None: """Put the decapper into standby (sleep) mode.""" - frames = await self._command(STANDBY) + frames = await self.send_command(STANDBY) self._require_accepted(STANDBY, frames, "Entering standby") loop = asyncio.get_running_loop() deadline = loop.time() + timeout @@ -631,7 +627,7 @@ async def ready(self, timeout: float = 30.0) -> None: """Wake the decapper from standby if it is asleep.""" if "SLEEP" not in (await self.request_status()).upper(): return - frames = await self._command(READY) + frames = await self.send_command(READY) self._require_accepted(READY, frames, "Waking from standby") await self._wait_for_idle(timeout, "Waking from standby", "StatusNotOK") logger.info("[IntelliXcap96 %s] ready", self.io.port) @@ -649,7 +645,7 @@ async def reset(self, settle_time: float = 5.0) -> None: """ if "ERROR" in (await self.request_status()).upper(): raise _fault("CannotReset") - await self._command(RESET) + await self.send_command(RESET) await asyncio.sleep(settle_time) if "ERROR" in (await self.request_status()).upper(): raise _fault("ResetNotSuccesful")