Vspin driver improvements

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

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."
    ]
   },
   {

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,17 +518,15 @@ 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)
 
   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)

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)))