XPeel: fix status parsing and the adhere-time code in peel()

CHANGELOG.md

@@ -36,6 +36,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Fixed
 
+- Azenta XPeel (`pylabrobot.azenta.XPeel` and the legacy `XPeelBackend`): `request_status()` / `get_status()` read the three error codes from a ready line that carries its error description (`*ready:00,00,00 [No error]`), and `peel()` sends `adhere_time` as the peeler's code 1-4 (`*xpeel:41` for 2.5 s), which it accepts.
 - Imported `unittest.mock` in `pylabrobot/centrifuge/centrifuge_tests.py` (pre-existing bug that prevented the test class from running).
 - `HamiltonTCPClient` no longer retransmits a command after a failed read. A read timeout on a slow motion command previously re-sent it, which could execute the motion twice (#1195).
 - `HamiltonTCPClient.setup()` now resets all per-session state (client id, sequence numbers, instrument addresses, object registry) rather than carrying it into the new session, and refuses to run on an already-connected client instead of leaking the socket (#1195).

pylabrobot/azenta/xpeel.py

@@ -145,7 +145,9 @@ async def send_command(
   async def request_status(self) -> Tuple[int, int, int]:
     """Request instrument status; returns three error codes."""
     resp = await self.send_command("*stat")
-    return tuple([int(x) for x in resp[-1].split(":")[1].split(",")])  # type: ignore
+    # The ready line carries the description send_command appends: "*ready:00,00,00 [No error]".
+    first, second, third = (int(x) for x in resp[-1].split(":", 1)[1].split()[0].split(","))
+    return first, second, third
 
   async def request_version(self):
     """Request firmware version."""
@@ -236,7 +238,9 @@ async def peel(
       fast: If True, uses faster peel speed. Default False.
       adhere_time: Time in seconds for the roller to press on the seal before peeling.  Must be one of 2.5, 5.0, 7.5, or 10.0. Default 2.5.
     """
-    if adhere_time not in {2.5, 5.0, 7.5, 10.0}:
+    # The peeler takes the adhere time as a code: 1-4 for 2.5, 5.0, 7.5 and 10.0 seconds.
+    adhere_time_code = {2.5: 1, 5.0: 2, 7.5: 3, 10.0: 4}.get(adhere_time)
+    if adhere_time_code is None:
       raise ValueError("adhere_time must be one of: 2.5, 5.0, 7.5, 10.0")
     if begin_location not in {-2, 0, 2, 4}:
       raise ValueError("begin_location must be one of: -2, 0, 2, 4")
@@ -252,7 +256,7 @@ async def peel(
       (4, False): 8,
     }.get((begin_location, fast), 9)
 
-    cmd = f"*xpeel:{parameter_set}{adhere_time}"
+    cmd = f"*xpeel:{parameter_set}{adhere_time_code}"
     return await self.send_command(cmd, expect_ack=True, wait_for_ready=True)
 
   async def restart(self):

pylabrobot/azenta/xpeel_tests.py

@@ -0,0 +1,51 @@
+import unittest
+from typing import Tuple
+from unittest.mock import AsyncMock
+
+from pylabrobot.azenta.xpeel import HAS_SERIAL, XPeel
+
+
+def _xpeel(*lines: bytes) -> Tuple[XPeel, AsyncMock]:
+  """An XPeel and its serial port, which answers with `lines`, one per readline."""
+  xpeel = XPeel(port="/dev/null")
+  io = AsyncMock()
+  io.readline = AsyncMock(side_effect=list(lines))
+  xpeel.io = io
+  return xpeel, io
+
+
+@unittest.skipUnless(HAS_SERIAL, "pyserial is not installed")
+class TestXPeelRequestStatus(unittest.IsolatedAsyncioTestCase):
+  async def test_returns_the_three_error_codes(self) -> None:
+    xpeel, io = _xpeel(b"*ready:00,00,00\r\n")
+
+    self.assertEqual(await xpeel.request_status(), (0, 0, 0))
+    io.write.assert_awaited_once_with(b"*stat\r\n")
+
+  async def test_returns_nonzero_error_codes(self) -> None:
+    xpeel, _ = _xpeel(b"*ready:07,20,00\r\n")
+
+    self.assertEqual(await xpeel.request_status(), (7, 20, 0))
+
+
+@unittest.skipUnless(HAS_SERIAL, "pyserial is not installed")
+class TestXPeelPeel(unittest.IsolatedAsyncioTestCase):
+  async def test_sends_the_adhere_time_as_a_code(self) -> None:
+    for adhere_time, code in ((2.5, 1), (5.0, 2), (7.5, 3), (10.0, 4)):
+      with self.subTest(adhere_time=adhere_time):
+        xpeel, io = _xpeel(b"*ack\r\n", b"*ready:00,00,00\r\n")
+
+        await xpeel.peel(begin_location=0, fast=False, adhere_time=adhere_time)
+
+        io.write.assert_awaited_once_with(f"*xpeel:4{code}\r\n".encode("ascii"))
+
+  async def test_rejects_an_unsupported_adhere_time(self) -> None:
+    xpeel, io = _xpeel()
+
+    with self.assertRaises(ValueError):
+      await xpeel.peel(adhere_time=3.0)
+    io.write.assert_not_awaited()
+
+
+if __name__ == "__main__":
+  unittest.main()

pylabrobot/legacy/peeling/xpeel_backend.py

@@ -175,7 +175,9 @@ async def get_status(self) -> Tuple[int, int, int]:
     """
     self.logger.debug("Requesting status...")
     resp = await self._send_command("*stat")
-    return tuple([int(x) for x in resp[-1].split(":")[1].split(",")])  # type: ignore
+    # The ready line carries the description _send_command appends: "*ready:00,00,00 [No error]".
+    first, second, third = (int(x) for x in resp[-1].split(":", 1)[1].split()[0].split(","))
+    return first, second, third
 
   async def get_version(self):
     """Request firmware version."""
@@ -211,7 +213,9 @@ async def peel(
       f"Running peel with begin_location={begin_location}, fast={fast}, adhere_time={adhere_time}..."
     )
 
-    if adhere_time not in {2.5, 5.0, 7.5, 10.0}:
+    # The peeler takes the adhere time as a code: 1-4 for 2.5, 5.0, 7.5 and 10.0 seconds.
+    adhere_time_code = {2.5: 1, 5.0: 2, 7.5: 3, 10.0: 4}.get(adhere_time)
+    if adhere_time_code is None:
       raise ValueError("adhere_time must be one of: 2.5, 5.0, 7.5, 10.0")
     if begin_location not in {-2, 0, 2, 4}:
       raise ValueError("begin_location must be one of: -2, 0, 2, 4")
@@ -229,7 +233,7 @@ async def peel(
 
     if parameter_set not in range(1, 10):
       raise ValueError("parameter_set must be in 1-9")
-    cmd = f"*xpeel:{parameter_set}{adhere_time}"
+    cmd = f"*xpeel:{parameter_set}{adhere_time_code}"
     return await self._send_command(
       cmd,
       expect_ack=True,

pylabrobot/legacy/peeling/xpeel_backend_tests.py

@@ -0,0 +1,38 @@
+import unittest
+from typing import Tuple
+from unittest.mock import AsyncMock
+
+from pylabrobot.legacy.peeling.xpeel_backend import HAS_SERIAL, XPeelBackend
+
+
+def _backend(*lines: bytes) -> Tuple[XPeelBackend, AsyncMock]:
+  """An XPeelBackend and its serial port, which answers with `lines`, one per readline."""
+  backend = XPeelBackend(port="/dev/null")
+  io = AsyncMock()
+  io.readline = AsyncMock(side_effect=list(lines))
+  backend.io = io
+  return backend, io
+
+
+@unittest.skipUnless(HAS_SERIAL, "pyserial is not installed")
+class TestXPeelBackendGetStatus(unittest.IsolatedAsyncioTestCase):
+  async def test_returns_the_three_error_codes(self) -> None:
+    backend, _ = _backend(b"*ready:07,20,00\r\n")
+
+    self.assertEqual(await backend.get_status(), (7, 20, 0))
+
+
+@unittest.skipUnless(HAS_SERIAL, "pyserial is not installed")
+class TestXPeelBackendPeel(unittest.IsolatedAsyncioTestCase):
+  async def test_sends_the_adhere_time_as_a_code(self) -> None:
+    for adhere_time, code in ((2.5, 1), (5.0, 2), (7.5, 3), (10.0, 4)):
+      with self.subTest(adhere_time=adhere_time):
+        backend, io = _backend(b"*ack\r\n", b"*ready:00,00,00\r\n")
+
+        await backend.peel(begin_location=0, fast=False, adhere_time=adhere_time)
+
+        io.write.assert_awaited_once_with(f"*xpeel:4{code}\r\n".encode("ascii"))
+
+
+if __name__ == "__main__":
+  unittest.main()