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..a7791f14bfd --- /dev/null +++ b/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() 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..565b156aed6 --- /dev/null +++ b/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()