diff --git a/docs/api/pylabrobot.azenta.rst b/docs/api/pylabrobot.azenta.rst index 9012ce85723..064609171d9 100644 --- a/docs/api/pylabrobot.azenta.rst +++ b/docs/api/pylabrobot.azenta.rst @@ -45,3 +45,18 @@ XPeel Peeler .. autoclass:: pylabrobot.azenta.xpeel.XPeelPeelerBackend.PeelParams :members: + + +FluidX IntelliXcap 96 +--------------------- + +.. currentmodule:: pylabrobot.azenta.fluidx.intellixcap96 + +.. autosummary:: + :toctree: _autosummary + :nosignatures: + :recursive: + + 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 new file mode 100644 index 00000000000..40c9848310b --- /dev/null +++ b/docs/user_guide/azenta/fluidx/intellixcap96/hello-world.ipynb @@ -0,0 +1,264 @@ +{ + "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", + "## 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": [ + "## 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", + "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", + "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", + "\n\n", + "`setup()` opens the serial port and reads the status. It raises if the device is not ready (e.g. e-stop)." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "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", + "\n\n", + "`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." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "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. 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`." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "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", + "\n\n", + "Home all axes." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "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. 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) 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", + "`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, + "metadata": {}, + "outputs": [], + "source": [ + "await decapper.decap() # unscrew and hold all 96 caps" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Choose this path to retain the caps.\n", + "await decapper.recap() # screw and torque the held caps back on" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# 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" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## 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()" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "await decapper.ready()" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Teardown\n\n", + "\n\n", + "`stop()` closes the serial connection." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "await decapper.stop()" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/docs/user_guide/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/__init__.py b/pylabrobot/azenta/__init__.py index b00149e5035..8f32b8a13b6 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 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 new file mode 100644 index 00000000000..6f398a5a2ef --- /dev/null +++ b/pylabrobot/azenta/fluidx/__init__.py @@ -0,0 +1,6 @@ +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 new file mode 100644 index 00000000000..21edf45c986 --- /dev/null +++ b/pylabrobot/azenta/fluidx/intellixcap96.py @@ -0,0 +1,652 @@ +import asyncio +import logging +import re +from typing import Dict, List, Optional, Tuple + +from pylabrobot.io.serial import Serial + +logger = logging.getLogger(__name__) + +# 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, then changes to the operation-specific terminal state. +STX = b"\x02" +ETX = b"\x03" +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" + +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." + ), + "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.", + "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, + 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 + + +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) + + +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"(?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``. + + 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, the decap error/recovery path, and decap/recap with a loaded + 0.5 mL rack, including release of held caps with ``waste``. + + See the Azenta IntelliXcap user manual for the required carrier and physical + setup: + https://web.azenta.com/hubfs/azenta-files/resources/manuals-guides/319430-IXC-User-Manual.pdf + """ + + def __init__( + self, + port: str, + timeout: float = 5.0, + 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: + 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. + 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, + baudrate=9600, + bytesize=8, + parity="N", + stopbits=1, + timeout=timeout, + ) + + async def setup(self) -> None: + await self.io.setup() + 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 "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) + + async def stop(self) -> None: + """Close the serial connection.""" + await self.io.stop() + + # === Framed command layer === + + async def _send(self, command: str) -> None: + """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) + await asyncio.sleep(self.command_delay) + + 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: + byte = await self.io.read(1) + if byte in (b"", ETX): + break + buf += byte + return buf.decode("ascii", errors="replace") + + 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): + 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, 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, + 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. ``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.send_command(STATUS) + if any("ERROR" in f.upper() for f in frames): + 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 status.upper() in expected: + return + await asyncio.sleep(self.poll_interval) + raise FluidXError( + title=f"{name} timed out", + message=f"did not reach {terminal_statuses!r} within {timeout}s", + ) + + # === Status === + + async def request_status(self) -> str: + """Poll the device and return its status word (e.g. ``StatusOK``).""" + 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 + + # === Operations === + + async def open_tray(self, timeout: float = 15.0) -> None: + """Open the loading tray.""" + await self._ensure_ready() + 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, + "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: + """Close the loading tray.""" + await self._ensure_ready() + 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, + "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: + """Send the home command and wait for it to finish. Also clears a latched error.""" + frames = await self.send_command(HOME_ALL) + 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() + 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) + 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 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") + 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._ensure_ready()).upper() + if "RECAP" in status: + raise _fault("NeedToRecap") + frames = await self.send_command(DECAP_START) + self._require_accepted(DECAP_START, frames, "Decapping") + 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: + """Screw the held caps back on. + + Args: + timeout: maximum time in seconds to wait for the stroke to finish. + """ + status = (await self._ensure_ready()).upper() + if "RECAP" not in status: + raise _fault("NeedToDecap") + frames = await self.send_command(RECAP_START) + self._require_accepted(RECAP_START, frames, "Recapping") + 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: + """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. + """ + status = (await self._ensure_ready()).upper() + if "RECAP" not in status: + raise _fault("NeedToDecap", "waste requires caps held after decapping") + frames = await self.send_command(WASTE) + self._require_accepted(WASTE, frames, "Wasting caps") + 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: + """Put the decapper into standby (sleep) mode.""" + frames = await self.send_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 _fault("CannotGoInStandbyMode", "standby timed out") + + 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.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) + + async def reset(self, settle_time: float = 5.0) -> None: + """Reset the device and wait for it to settle. + + 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. + """ + if "ERROR" in (await self.request_status()).upper(): + raise _fault("CannotReset") + await self.send_command(RESET) + await asyncio.sleep(settle_time) + 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 new file mode 100644 index 00000000000..3955509bbef --- /dev/null +++ b/pylabrobot/azenta/fluidx/intellixcap96_tests.py @@ -0,0 +1,312 @@ +import contextlib +import unittest +from typing import Iterator, List +from unittest.mock import AsyncMock, patch + +from pylabrobot.azenta.fluidx import FluidXError, FluidXIntelliXcap96, get_error_message + +ACK = "\x06" + + +class FakeXcapSerial: + """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 + 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: + 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")) + 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: + return b"" + out = bytes(self._rx[:num_bytes]) + del self._rx[:num_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) + patcher.start() + self.addCleanup(patcher.stop) + + 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 + + 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_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_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( + [ + 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, ["a", "f", "a", "a"]) # type: ignore[attr-defined] + + 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( + [status("StatusOK"), [ACK, "gOK"], status("StatusBUSY"), status("StatusOK")] + ) + 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() + self.assertEqual(device.io.written, ["Z", "a", "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_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_ready_wakes_from_sleep(self): + device = self._make( + [ + status("StatusSLEEP"), # request_status: asleep + [ACK, "kOK"], # k accepted + status("StatusBUSY"), # waking + status("StatusOK"), # awake + ] + ) + await device.ready() + self.assertEqual(device.io.written, ["a", "k", "a", "a"]) # type: ignore[attr-defined] + + async def test_decap_success(self): + device = self._make( + [ + status("StatusOK"), # precheck: not recapped, no error + [ACK, "hOK"], # h accepted + status("StatusBUSY"), + status("StatusRECAP"), # hardware terminal state: caps held + ] + ) + await device.decap() + self.assertEqual(device.io.written, ["a", "h", "a", "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_decap_reports_error_during_motion(self): + device = self._make([status("StatusOK"), [ACK, "hOK"], [ACK, "aOK", "DecapERROR"]]) + 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")]) + await device.reset() + 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() + + 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( + [ + 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] + + 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()