From 5ec79e557114d685b17506f6f8cd2e66f7065a26 Mon Sep 17 00:00:00 2001 From: cw_liao <48869711+Dondonn@users.noreply.github.com> Date: Wed, 7 Oct 2026 01:57:31 +0300 Subject: [PATCH 1/2] XPeel: read status codes past the error description, send adhere time as a code `send_command` returns each `*ready:` line with its error description appended (`*ready:00,00,00 [No error]`), and `request_status` (legacy: `get_status`) parsed the last field as `"00 [No error]"`, raising ValueError on every reply. The codes are now read from the first token after the colon. `peel` formatted `adhere_time` in seconds into the command (`*xpeel:42.5`), which the peeler rejects as error 05 "Illegal command". The peeler takes the adhere time as a code, 1-4 for 2.5, 5.0, 7.5 and 10.0 s (`*xpeel:41`). Both fixes apply to `pylabrobot.azenta.XPeel` and the legacy `XPeelBackend`, with tests for each. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + pylabrobot/azenta/xpeel.py | 10 ++-- pylabrobot/azenta/xpeel_tests.py | 49 +++++++++++++++++++ pylabrobot/legacy/peeling/xpeel_backend.py | 10 ++-- .../legacy/peeling/xpeel_backend_tests.py | 36 ++++++++++++++ 5 files changed, 100 insertions(+), 6 deletions(-) create mode 100644 pylabrobot/azenta/xpeel_tests.py create mode 100644 pylabrobot/legacy/peeling/xpeel_backend_tests.py diff --git a/CHANGELOG.md b/CHANGELOG.md index be0bf6c6b37..3a4cf67dbce 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/pylabrobot/azenta/xpeel.py b/pylabrobot/azenta/xpeel.py index 7ee309c4f09..3ef08e09db5 100644 --- a/pylabrobot/azenta/xpeel.py +++ b/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): diff --git a/pylabrobot/azenta/xpeel_tests.py b/pylabrobot/azenta/xpeel_tests.py new file mode 100644 index 00000000000..760949c4c44 --- /dev/null +++ b/pylabrobot/azenta/xpeel_tests.py @@ -0,0 +1,49 @@ +import unittest +from typing import Tuple +from unittest.mock import AsyncMock + +from pylabrobot.azenta.xpeel import 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 + + +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)) + + +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() diff --git a/pylabrobot/legacy/peeling/xpeel_backend.py b/pylabrobot/legacy/peeling/xpeel_backend.py index 934f6b0ebb6..13c0e5c1cf9 100644 --- a/pylabrobot/legacy/peeling/xpeel_backend.py +++ b/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, diff --git a/pylabrobot/legacy/peeling/xpeel_backend_tests.py b/pylabrobot/legacy/peeling/xpeel_backend_tests.py new file mode 100644 index 00000000000..5f5629face2 --- /dev/null +++ b/pylabrobot/legacy/peeling/xpeel_backend_tests.py @@ -0,0 +1,36 @@ +import unittest +from typing import Tuple +from unittest.mock import AsyncMock + +from pylabrobot.legacy.peeling.xpeel_backend import 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 + + +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)) + + +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() From 8c5c717da6577a21977f45fec9ed095b12683f1d Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Tue, 6 Oct 2026 16:51:59 -0700 Subject: [PATCH 2/2] Skip XPeel tests when pyserial is unavailable --- pylabrobot/azenta/xpeel_tests.py | 4 +++- pylabrobot/legacy/peeling/xpeel_backend_tests.py | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/pylabrobot/azenta/xpeel_tests.py b/pylabrobot/azenta/xpeel_tests.py index 760949c4c44..a7791f14bfd 100644 --- a/pylabrobot/azenta/xpeel_tests.py +++ b/pylabrobot/azenta/xpeel_tests.py @@ -2,7 +2,7 @@ from typing import Tuple from unittest.mock import AsyncMock -from pylabrobot.azenta.xpeel import XPeel +from pylabrobot.azenta.xpeel import HAS_SERIAL, XPeel def _xpeel(*lines: bytes) -> Tuple[XPeel, AsyncMock]: @@ -14,6 +14,7 @@ def _xpeel(*lines: bytes) -> Tuple[XPeel, AsyncMock]: 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") @@ -27,6 +28,7 @@ async def test_returns_nonzero_error_codes(self) -> None: 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)): diff --git a/pylabrobot/legacy/peeling/xpeel_backend_tests.py b/pylabrobot/legacy/peeling/xpeel_backend_tests.py index 5f5629face2..565b156aed6 100644 --- a/pylabrobot/legacy/peeling/xpeel_backend_tests.py +++ b/pylabrobot/legacy/peeling/xpeel_backend_tests.py @@ -2,7 +2,7 @@ from typing import Tuple from unittest.mock import AsyncMock -from pylabrobot.legacy.peeling.xpeel_backend import XPeelBackend +from pylabrobot.legacy.peeling.xpeel_backend import HAS_SERIAL, XPeelBackend def _backend(*lines: bytes) -> Tuple[XPeelBackend, AsyncMock]: @@ -14,6 +14,7 @@ def _backend(*lines: bytes) -> Tuple[XPeelBackend, AsyncMock]: 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") @@ -21,6 +22,7 @@ async def test_returns_the_three_error_codes(self) -> None: 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)):