fix(azenta): handle IntelliXcap V49 command behavior

Author: rickwierenga

pylabrobot/azenta/fluidx/intellixcap96.py

@@ -31,8 +31,9 @@
 READY = "k"
 CARTRIDGE_EJECT = "c"
 CARTRIDGE_LOAD = "C"
-TRAY_EXTEND = "t"
-TRAY_RETRACT = "T"
+# Newer-firmware tray opcodes, intentionally not sent until verified there:
+# TRAY_EXTEND = "t"
+# TRAY_RETRACT = "T"
 TRAY_STEP_OUT = "s"
 TRAY_STEP_IN = "S"
 CARTRIDGE_COUNTER_RESET = "X"
@@ -457,22 +458,35 @@ class FluidXIntelliXcap96:
   to select a compatible cartridge.
 
   Single-character commands, each written followed by ETX:
-    a   request status                     e   extended status bitmask
-    h   start decapping                    E   cartridge profile settings
-    i   start recapping                    V   firmware versions
-    b   release held caps into a carrier   N   cartridge profile/cycles/serial
-    f   open the tray                      8   latched error code
-    g   close the tray                     c   eject the cartridge onto the tray
-    Z   home all axes                      C   load the cartridge from the tray
-    j   enter standby                      X   reset the cartridge cycle counter
-    k   leave standby (ready)              t   extend the tray to S127
-    l   light curtain detection off        T   retract the tray to S3
-    L   light curtain detection on         s   step the tray out by S88
-    d   dry-run mode on                    S   step the tray in by S88
-    D   dry-run mode off                   Q   force a decap retry
-    +   safety door on                     5   eject held caps (manual mode)
-    -   safety door off                    6   home the cap head (manual mode)
-    z   initialize keeping caps on pins    7   open the safety door (manual mode)
+    a   request status
+    b   release held caps into a carrier
+    c   eject the cartridge onto the tray
+    d   dry-run mode on
+    D   dry-run mode off
+    e   extended status bitmask
+    E   cartridge profile settings
+    f   open the tray
+    g   close the tray
+    h   start decapping
+    i   start recapping
+    j   enter standby
+    k   leave standby (ready)
+    l   light curtain detection off
+    L   light curtain detection on
+    N   cartridge profile/cycles/serial
+    Q   force a decap retry
+    s   step the tray out by S88
+    S   step the tray in by S88
+    V   firmware versions
+    X   reset the cartridge cycle counter
+    Z   home all axes
+    z   initialize keeping caps on pins
+    +   safety door on
+    -   safety door off
+    5   eject held caps (manual mode)
+    6   home the cap head (manual mode)
+    7   open the safety door (manual mode)
+    8   latched error code
 
   A command is answered with an ACK frame (0x06), a ``<cmd>OK`` echo frame, and a
   result frame. The status word, in the priority the firmware reports it, is
@@ -495,11 +509,17 @@ class FluidXIntelliXcap96:
   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``. The remaining
-  commands are implemented from the RS232 command list (Azenta part 386063 Rev.
-  A) and have not been exercised on an instrument.
+  Newer firmware documents ``t``/``T`` commands for absolute tray travel
+  between S3 and S127. Unit firmware V49 ignored both commands completely, so
+  :meth:`extend_tray` and :meth:`retract_tray` raise
+  :class:`NotImplementedError` without writing to the instrument.
+
+  Verified against hardware: connection, queries, tray open/close, home,
+  standby/ready, settings, cartridge eject/load, the decap error/recovery path,
+  forced decap retry, and decap/recap with a loaded 0.5 mL rack, including
+  release of held caps with ``waste``. Unit firmware V49 did not acknowledge
+  the four tray travel commands. The cartridge counter reset and manual
+  recovery commands have not been exercised on an instrument.
 
   See the Azenta IntelliXcap user manual for the required carrier and physical
   setup:
@@ -547,10 +567,10 @@ def __init__(
   async def setup(self) -> None:
     await self.io.setup()
     logger.warning(
-      "[IntelliXcap96 %s] most of this driver's commands are implemented from the RS232 command "
-      "list and have not been run on an instrument. The queries, tray open/close, home, "
-      "standby/ready and decap/recap/waste are hardware-verified; the tray travel, cartridge, "
-      "settings and manual recovery commands are not. Please report back so this can be updated.",
+      "[IntelliXcap96 %s] connection, queries, tray open/close, home, standby/ready, settings, "
+      "cartridge eject/load, forced decap retry and decap/recap/waste are hardware-verified. "
+      "Unit firmware V49 does not acknowledge the tray travel commands. The cartridge counter "
+      "reset and manual recovery commands are not hardware-verified.",
       self.io.port,
     )
     status = await self.request_status()
@@ -920,6 +940,8 @@ async def _tray_move(self, command: str, name: str, timeout: float) -> None:
 
     These acknowledge and answer with the same ``<cmd>OK`` frame, so a completed
     move is a second echo and a failed one is ``<cmd>ERROR``.
+
+    Unit firmware V49 does not acknowledge these commands.
     """
     await self._ensure_ready()
     echo = f"{command}OK"
@@ -934,14 +956,26 @@ async def _tray_move(self, command: str, name: str, timeout: float) -> None:
   async def extend_tray(self, timeout: float = 15.0) -> None:
     """Move the tray from the load position (S3) out to the extended position (S127).
 
-    Presents decapped tubes further out of the instrument. Fails unless the tray
-    is at the load position, so open the tray first.
+    This command is documented for newer firmware but is not implemented here:
+    unit firmware V49 ignored it completely. The method raises without writing
+    to the instrument.
     """
-    await self._tray_move(TRAY_EXTEND, "Extending the tray", timeout)
+    raise NotImplementedError(
+      "extend_tray() uses the newer-firmware 't' command, which is not supported by "
+      "IntelliXcap unit firmware V49"
+    )
 
   async def retract_tray(self, timeout: float = 15.0) -> None:
-    """Move the tray from the extended position (S127) back to the load position (S3)."""
-    await self._tray_move(TRAY_RETRACT, "Retracting the tray", timeout)
+    """Move the tray from the extended position (S127) back to the load position (S3).
+
+    This command is documented for newer firmware but is not implemented here:
+    unit firmware V49 ignored it completely. The method raises without writing
+    to the instrument.
+    """
+    raise NotImplementedError(
+      "retract_tray() uses the newer-firmware 'T' command, which is not supported by "
+      "IntelliXcap unit firmware V49"
+    )
 
   async def step_tray_out(self, timeout: float = 15.0) -> None:
     """Move the tray further out by the step distance in setpoint 88.
@@ -1100,7 +1134,9 @@ async def retry_decap(self, timeout: float = 60.0) -> None:
     The command list documents only that this command is used "after a
     successful Decap" and "requires lightcurtain OFF". That those are the same
     condition is inferred from the automatic-retry and error-detection commands
-    rather than stated by the vendor.
+    rather than stated by the vendor. The instrument finishes in
+    ``StatusRECAP`` because it is still holding the caps for a subsequent
+    :meth:`recap`.
     """
     await self._ensure_ready()
     frames = await self.send_command(RETRY_DECAP)
@@ -1110,7 +1146,7 @@ async def retry_decap(self, timeout: float = 60.0) -> None:
       timeout,
       "Retrying the decap",
       "RetryDecapWasNotSuccesful",
-      ("StatusOK",),
+      ("StatusRECAP",),
       done_frames=("DecapDONE",),
     )
     logger.info("[IntelliXcap96 %s] decap retry complete", self.io.port)
@@ -1262,8 +1298,8 @@ async def reset_cartridge_counter(self) -> None:
     """Reset the installed cartridge's cycle counter to zero."""
     name = "Resetting the cartridge counter"
     frames = await self.send_command(CARTRIDGE_COUNTER_RESET)
-    self._require_accepted(CARTRIDGE_COUNTER_RESET, frames, name, idempotent=True)
-    self._require_answer(frames, "CarResetDONE", name)
+    if self._require_accepted(CARTRIDGE_COUNTER_RESET, frames, name, idempotent=True):
+      self._require_answer(frames, "CarResetDONE", name)
     logger.info("[IntelliXcap96 %s] cartridge counter reset", self.io.port)
 
   # === Settings ===
@@ -1278,8 +1314,8 @@ async def set_error_detection_enabled(self, enabled: bool) -> None:
     command = ERROR_DETECTION_ON if enabled else ERROR_DETECTION_OFF
     name = "Enabling error detection" if enabled else "Disabling error detection"
     frames = await self.send_command(command)
-    self._require_accepted(command, frames, name, idempotent=True)
-    self._require_answer(frames, "ErrorDetectON" if enabled else "ErrorDetectOFF", name)
+    if self._require_accepted(command, frames, name, idempotent=True):
+      self._require_answer(frames, "ErrorDetectON" if enabled else "ErrorDetectOFF", name)
     logger.info("[IntelliXcap96 %s] error detection %s", self.io.port, "on" if enabled else "off")
 
   async def set_dry_run_enabled(self, enabled: bool) -> None:
@@ -1293,8 +1329,8 @@ async def set_dry_run_enabled(self, enabled: bool) -> None:
     command = DRY_RUN_ON if enabled else DRY_RUN_OFF
     name = "Enabling dry-run mode" if enabled else "Disabling dry-run mode"
     frames = await self.send_command(command)
-    self._require_accepted(command, frames, name, idempotent=True)
-    self._require_answer(frames, "DryRunON" if enabled else "DryRunOFF", name)
+    if self._require_accepted(command, frames, name, idempotent=True):
+      self._require_answer(frames, "DryRunON" if enabled else "DryRunOFF", name)
     logger.info("[IntelliXcap96 %s] dry-run mode %s", self.io.port, "on" if enabled else "off")
 
   async def set_safety_door_enabled(self, enabled: bool) -> None:
@@ -1306,8 +1342,8 @@ async def set_safety_door_enabled(self, enabled: bool) -> None:
     command = SAFETY_DOOR_ENABLE if enabled else SAFETY_DOOR_DISABLE
     name = "Enabling the safety door" if enabled else "Disabling the safety door"
     frames = await self.send_command(command)
-    self._require_accepted(command, frames, name, idempotent=True)
-    self._require_answer(frames, "DoorONDONE" if enabled else "DoorOFFDONE", name)
+    if self._require_accepted(command, frames, name, idempotent=True):
+      self._require_answer(frames, "DoorONDONE" if enabled else "DoorOFFDONE", name)
     logger.info("[IntelliXcap96 %s] safety door %s", self.io.port, "on" if enabled else "off")
 
   # === Manual recovery ===

pylabrobot/azenta/fluidx/intellixcap96_tests.py

@@ -270,17 +270,6 @@ async def test_close_tray_preserves_caps_held_state(self):
     await device.close_tray()
     self.assertEqual(self._written(device), ["a", "g", "a", "a"])
 
-  async def test_extend_tray(self):
-    # A tray move acknowledges and answers with the same echo, so success is two.
-    device = self._make([status("StatusOK"), [ACK, "tOK", "tOK"]])
-    await device.extend_tray()
-    self.assertEqual(self._written(device), ["a", "t"])
-
-  async def test_retract_tray(self):
-    device = self._make([status("StatusOK"), [ACK, "TOK", "TOK"]])
-    await device.retract_tray()
-    self.assertEqual(self._written(device), ["a", "T"])
-
   async def test_step_tray_out_and_in(self):
     device = self._make(
       [
@@ -449,10 +438,17 @@ async def test_manual_halt_during_motion_fails_immediately(self):
     self.assertEqual(ctx.exception.error_code, 167)
     self.assertFalse(ctx.exception.recoverable)
 
-  async def test_retry_decap_is_acknowledged_as_a_decap(self):
-    device = self._make([status("StatusOK"), [ACK, "hOK"], [ACK, "aOK", "DecapDONE"]])
+  async def test_retry_decap_finishes_with_caps_held(self):
+    device = self._make(
+      [
+        status("StatusRECAP"),
+        [ACK, "hOK"],
+        status("StatusBUSY"),
+        status("StatusRECAP"),
+      ]
+    )
     await device.retry_decap()
-    self.assertEqual(self._written(device), ["a", "Q", "a"])
+    self.assertEqual(self._written(device), ["a", "Q", "a", "a"])
 
   async def test_recap_blocked_when_not_decapped(self):
     device = self._make([status("StatusDecap")])
@@ -565,6 +561,11 @@ async def test_reset_cartridge_counter_requires_its_answer(self):
     with self.assertRaises(FluidXError):
       await device.reset_cartridge_counter()
 
+  async def test_reset_cartridge_counter_is_idempotent(self):
+    device = self._make([[ACK, "XOK", "CommandIgnore"]])
+    await device.reset_cartridge_counter()
+    self.assertEqual(self._written(device), ["X"])
+
   async def test_load_cartridge_reads_the_profile_from_the_end_of_the_motion(self):
     # onnOK is documented as an end-of-motion answer, not part of the ack.
     device = self._make(
@@ -596,6 +597,11 @@ async def test_set_error_detection_on(self):
     await device.set_error_detection_enabled(True)
     self.assertEqual(self._written(device), ["L"])
 
+  async def test_set_error_detection_is_idempotent(self):
+    device = self._make([[ACK, "LOK", "CommandIgnore"]])
+    await device.set_error_detection_enabled(True)
+    self.assertEqual(self._written(device), ["L"])
+
   async def test_set_dry_run_enabled(self):
     device = self._make([[ACK, "dOK", "DryRunON"]])
     await device.set_dry_run_enabled(True)
@@ -606,6 +612,11 @@ async def test_set_dry_run_disabled(self):
     await device.set_dry_run_enabled(False)
     self.assertEqual(self._written(device), ["D"])
 
+  async def test_set_dry_run_is_idempotent(self):
+    device = self._make([[ACK, "DOK", "CommandIgnore"]])
+    await device.set_dry_run_enabled(False)
+    self.assertEqual(self._written(device), ["D"])
+
   async def test_setting_raises_when_the_device_does_not_confirm(self):
     device = self._make([[ACK, "dOK"]])
     with self.assertRaises(FluidXError):
@@ -621,6 +632,11 @@ async def test_set_safety_door_disabled(self):
     await device.set_safety_door_enabled(False)
     self.assertEqual(self._written(device), ["-"])
 
+  async def test_set_safety_door_is_idempotent(self):
+    device = self._make([[ACK, "+OK", "CommandIgnore"]])
+    await device.set_safety_door_enabled(True)
+    self.assertEqual(self._written(device), ["+"])
+
   # === Manual recovery commands ===
 
   async def test_eject_caps_requires_manual_mode(self):