diff --git a/README.md b/README.md index 675f8b31cb7..f5a170c5f78 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,10 @@ await cf.setup() await cf.spin(g=800, duration=60) ``` +Use `VSpinBackend(command_set="old_firmware")` for VSpins that need the older +pneumatic command bytes. Some Velocity11-labeled units work with the default +command set. + For a HighRes Biosolutions MicroSpin, use the MicroSpin factory: ```python diff --git a/docs/user_guide/01_material-handling/centrifuge/agilent_vspin.ipynb b/docs/user_guide/01_material-handling/centrifuge/agilent_vspin.ipynb index e0ab2cc16b8..58575b47625 100644 --- a/docs/user_guide/01_material-handling/centrifuge/agilent_vspin.ipynb +++ b/docs/user_guide/01_material-handling/centrifuge/agilent_vspin.ipynb @@ -7,7 +7,9 @@ "source": [ "# Agilent VSpin\n", "\n", - "The VSpin centrifuge is controlled by the {class}`~pylabrobot.centrifuge.vspin_backend.VSpinBackend` class." + "The VSpin centrifuge is controlled by the {class}`~pylabrobot.centrifuge.vspin_backend.VSpinBackend` class. The default command set works for known Agilent units and some Velocity11-labeled units.\n", + "\n", + "Use `VSpinBackend(command_set=\\\"old_firmware\\\")` for units that need the older pneumatic command bytes." ] }, { diff --git a/pylabrobot/centrifuge/vspin_backend.py b/pylabrobot/centrifuge/vspin_backend.py index 38f4102ad0e..f28b1bf9ce5 100644 --- a/pylabrobot/centrifuge/vspin_backend.py +++ b/pylabrobot/centrifuge/vspin_backend.py @@ -180,6 +180,46 @@ def _save_vspin_calibrations(device_id, remainder: int): FULL_ROTATION: int = 8000 +_KNOWN_VSPIN_STATUSES = {0x08, 0x09, 0x0B, 0x11, 0x13, 0x18, 0x19, 0x88, 0x89, 0x91, 0x99} + +_VSPIN_COMMAND_SET_ALIASES = { + "agilent": "agilent", + "old_firmware": "old_firmware", +} + +_VSPIN_COMMANDS = { + "agilent": { + "open_door": bytes.fromhex("aa022600062e"), + "close_door": bytes.fromhex("aa022600042c"), + "lock_door": bytes.fromhex("aa0226000028"), + "unlock_door": bytes.fromhex("aa022600042c"), + "lock_bucket": bytes.fromhex("aa022600072f"), + "unlock_bucket": bytes.fromhex("aa022600062e"), + }, + "old_firmware": { + "open_door": bytes.fromhex("aa022600072f"), + "close_door": bytes.fromhex("aa022600052d"), + "lock_door": bytes.fromhex("aa0226000028"), + "unlock_door": bytes.fromhex("aa022600042c"), + "lock_bucket": bytes.fromhex("aa0226000129"), + "unlock_bucket": bytes.fromhex("aa0226200048"), + }, +} + + +def _normalize_vspin_command_set(command_set: str) -> str: + normalized = command_set.lower() + if normalized not in _VSPIN_COMMAND_SET_ALIASES: + raise ValueError("command_set must be 'agilent' or 'old_firmware'") + return _VSPIN_COMMAND_SET_ALIASES[normalized] + + +def _with_vspin_checksum(cmd: bytes) -> bytes: + """Return ``cmd`` with the final VSpin checksum byte recomputed.""" + if len(cmd) <= 2 or cmd[0] != 0xAA: + return cmd + payload = cmd[1:-1] + return b"\xaa" + payload + bytes([sum(payload) & 0xFF]) bucket_1_not_set_error = RuntimeError( @@ -193,11 +233,21 @@ class VSpinBackend(CentrifugeBackend): """Backend for the Agilent Centrifuge. Note that this is not a complete implementation.""" - def __init__(self, device_id: Optional[str] = None): + def __init__( + self, + device_id: Optional[str] = None, + command_set: str = "agilent", + ): """ Args: - device_id: The libftdi id for the centrifuge. Find using `python -m pylibftdi.examples.list_devices` + device_id: The libftdi id for the centrifuge. + Find using `python -m pylibftdi.examples.list_devices`. + command_set: VSpin firmware command set. ``"agilent"`` is the default + used by known Agilent units and some Velocity11 units. Use + ``"old_firmware"`` for older firmware that needs the legacy pneumatic + command bytes. """ + self._command_set = _normalize_vspin_command_set(command_set) self.io = FTDI(human_readable_device_name="Agilent VSpin Centrifuge", device_id=device_id) self._bucket_1_remainder: Optional[int] = None # only attempt loading calibration if device_id is not None @@ -320,6 +370,9 @@ async def stop(self): await self.configure_and_initialize() await self.io.stop() + def _get_command_bytes(self, name: str) -> bytes: + return _VSPIN_COMMANDS[self._command_set][name] + class _StatusPositionTachometer(ctypes.LittleEndianStructure): _pack_ = 1 _fields_ = [ @@ -332,6 +385,17 @@ class _StatusPositionTachometer(ctypes.LittleEndianStructure): ("checksum", ctypes.c_uint8), ] + @staticmethod + def _find_status_packet(resp: bytes) -> Optional[_StatusPositionTachometer]: + for start in range(max(0, len(resp) - 13)): + packet = resp[start : start + 14] + if len(packet) < 14 or packet[0] not in _KNOWN_VSPIN_STATUSES: + continue + if (sum(packet[:-1]) & 0xFF) != packet[-1]: + continue + return VSpinBackend._StatusPositionTachometer.from_buffer_copy(packet) + return None + async def _get_positions_and_tachometer(self) -> _StatusPositionTachometer: """Returns 14 bytes @@ -358,7 +422,10 @@ async def _get_positions_and_tachometer(self) -> _StatusPositionTachometer: resp = await self._send_command(bytes.fromhex("aa010e0f")) if len(resp) == 0: raise IOError("Empty status from centrifuge") - return VSpinBackend._StatusPositionTachometer.from_buffer_copy(resp) + status = self._find_status_packet(resp) + if status is None: + raise IOError(f"Invalid status from centrifuge: {resp.hex()}") + return status async def get_position(self) -> int: return (await self._get_positions_and_tachometer()).current_position # type: ignore @@ -421,7 +488,8 @@ async def _read_resp(self, timeout: float = 20) -> bytes: return data async def _send_command(self, cmd: bytes, read_timeout=0.2) -> bytes: - written = await self.io.write(bytes(cmd)) + cmd = _with_vspin_checksum(bytes(cmd)) + written = await self.io.write(cmd) if written != len(cmd): raise RuntimeError("Failed to write all bytes") @@ -450,8 +518,7 @@ async def initialize(self): async def open_door(self): if await self.get_door_open(): return - # used to be: aa022600072f - await self._send_command(bytes.fromhex("aa022600062e")) # same as unlock door + await self._send_command(self._get_command_bytes("open_door")) # same as unlock door on new firmware # we can't tell when the door is fully open, so we just wait a bit await asyncio.sleep(4) @@ -459,8 +526,7 @@ async def open_door(self): async def close_door(self): if not (await self.get_door_open()): return - # used to be: aa022600052d - await self._send_command(bytes.fromhex("aa022600042c")) # same as unlock door + await self._send_command(self._get_command_bytes("close_door")) # same as unlock door on new firmware # we can't tell when the door is fully closed, so we just wait a bit await asyncio.sleep(2) @@ -469,24 +535,22 @@ async def lock_door(self): raise RuntimeError("Cannot lock door while it is open.") if await self.get_door_locked(): return - # used to be aa0226000129 - await self._send_command(bytes.fromhex("aa0226000028")) + await self._send_command(self._get_command_bytes("lock_door")) async def unlock_door(self): if not await self.get_door_locked(): return - # used to be aa022600052d - await self._send_command(bytes.fromhex("aa022600042c")) # same as close door + await self._send_command(self._get_command_bytes("unlock_door")) # same as close door async def lock_bucket(self): if await self.get_bucket_locked(): return - await self._send_command(bytes.fromhex("aa022600072f")) + await self._send_command(self._get_command_bytes("lock_bucket")) async def unlock_bucket(self): if not await self.get_bucket_locked(): return - await self._send_command(bytes.fromhex("aa022600062e")) # same as open door + await self._send_command(self._get_command_bytes("unlock_bucket")) # same as open door on new firmware async def go_to_bucket1(self): await self.go_to_position(await self.get_bucket_1_position()) @@ -499,9 +563,9 @@ async def go_to_position(self, position: int): await self.lock_door() position_bytes = position.to_bytes(4, byteorder="little") - byte_string = bytes.fromhex("aa01d497") + position_bytes + bytes.fromhex("c3f52800d71a0000") - sum_byte = (sum(byte_string) - 0xAA) & 0xFF - byte_string += sum_byte.to_bytes(1, byteorder="little") + byte_string = _with_vspin_checksum( + bytes.fromhex("aa01d497") + position_bytes + bytes.fromhex("c3f52800d71a0000") + ) await self._send_command(bytes.fromhex("aa0226000028")) await self._send_command(bytes.fromhex("aa0117021a")) await self._send_command(bytes.fromhex("aa01e6c800b00496000f004b00a00f050007")) @@ -587,9 +651,9 @@ async def spin( rpm_b = int(rpm * 4473.925).to_bytes(4, byteorder="little") acceleration_b = int(9.15 * 100 * acceleration).to_bytes(4, byteorder="little") - byte_string = bytes.fromhex("aa01d497") + position_b + rpm_b + acceleration_b - checksum = (sum(byte_string) - 0xAA) & 0xFF - byte_string += checksum.to_bytes(1, byteorder="little") + byte_string = _with_vspin_checksum( + bytes.fromhex("aa01d497") + position_b + rpm_b + acceleration_b + b"\x00" + ) await self._send_command(bytes.fromhex("aa0226000028")) await self._send_command(bytes.fromhex("aa0117021a")) @@ -624,8 +688,9 @@ async def spin( # aa0194b6000000000a03000058: decel at 85 # aa0194b61283000012010000f3: used in setup (30%) decc = int(9.15 * 100 * deceleration).to_bytes(2, byteorder="little") - decel_command = bytes.fromhex("aa0194b600000000") + decc + bytes.fromhex("0000") - decel_command += ((sum(decel_command) - 0xAA) & 0xFF).to_bytes(1, byteorder="little") + decel_command = _with_vspin_checksum( + bytes.fromhex("aa0194b600000000") + decc + bytes.fromhex("000000") + ) await self._send_command(decel_command) await asyncio.sleep(2) diff --git a/pylabrobot/centrifuge/vspin_backend_tests.py b/pylabrobot/centrifuge/vspin_backend_tests.py new file mode 100644 index 00000000000..b8547c50a13 --- /dev/null +++ b/pylabrobot/centrifuge/vspin_backend_tests.py @@ -0,0 +1,77 @@ +import unittest +from unittest import mock + +from pylabrobot.centrifuge.vspin_backend import VSpinBackend, _with_vspin_checksum + + +# status=0x11, current_position=12070, tachometer=-10, home_position=6733, checksum=0x29 +_STATUS_PACKET = bytes.fromhex("11262f00004ff6ff184d1a000029") + + +def _make_backend(io: mock.Mock) -> VSpinBackend: + backend = object.__new__(VSpinBackend) + backend.io = io + backend._command_set = "agilent" + backend._bucket_1_remainder = None + return backend + + +class VSpinCommandSetTests(unittest.IsolatedAsyncioTestCase): + def test_default_command_set_is_agilent(self): + with mock.patch("pylabrobot.centrifuge.vspin_backend.FTDI"): + backend = VSpinBackend() + + self.assertEqual(backend._command_set, "agilent") + self.assertEqual(backend._get_command_bytes("open_door"), bytes.fromhex("aa022600062e")) + self.assertEqual(backend._get_command_bytes("lock_bucket"), bytes.fromhex("aa022600072f")) + + def test_old_firmware_command_set_uses_legacy_pneumatic_commands(self): + with mock.patch("pylabrobot.centrifuge.vspin_backend.FTDI"): + backend = VSpinBackend(command_set="old_firmware") + + self.assertEqual(backend._command_set, "old_firmware") + self.assertEqual(backend._get_command_bytes("open_door"), bytes.fromhex("aa022600072f")) + self.assertEqual(backend._get_command_bytes("close_door"), bytes.fromhex("aa022600052d")) + self.assertEqual(backend._get_command_bytes("lock_bucket"), bytes.fromhex("aa0226000129")) + self.assertEqual(backend._get_command_bytes("unlock_bucket"), bytes.fromhex("aa0226200048")) + + def test_velocity11_label_is_not_a_command_set(self): + with mock.patch("pylabrobot.centrifuge.vspin_backend.FTDI"): + with self.assertRaisesRegex(ValueError, "command_set"): + VSpinBackend(command_set="velocity11") + + def test_unknown_command_set_raises(self): + with mock.patch("pylabrobot.centrifuge.vspin_backend.FTDI"): + with self.assertRaisesRegex(ValueError, "command_set"): + VSpinBackend(command_set="velocity11-label") + + def test_with_vspin_checksum_repairs_final_byte(self): + self.assertEqual( + _with_vspin_checksum(bytes.fromhex("aa020e00")), + bytes.fromhex("aa020e10"), + ) + + async def test_send_command_repairs_checksum_before_write(self): + io = mock.Mock() + io.read = mock.AsyncMock(return_value=b"\r") + io.write = mock.AsyncMock(return_value=4) + backend = _make_backend(io) + + await backend._send_command(bytes.fromhex("aa020e00")) + + io.write.assert_awaited_once_with(bytes.fromhex("aa020e10")) + + def test_find_status_packet_parses_status_packet(self): + parsed = VSpinBackend._find_status_packet(_STATUS_PACKET) + + assert parsed is not None + self.assertEqual(parsed.status, 0x11) + self.assertEqual(parsed.current_position, 12070) + self.assertEqual(parsed.tachometer, -10) + self.assertEqual(parsed.home_position, 6733) + + def test_find_status_packet_rejects_bad_checksum(self): + packet = bytearray(_STATUS_PACKET) + packet[-1] ^= 0xFF + + self.assertIsNone(VSpinBackend._find_status_packet(bytes(packet)))