address non-legacy VW error by resetting num_arms to main behavior

Author: har1eyk

pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py

@@ -1441,9 +1441,7 @@ def core96_head_installed(self) -> bool:
 
   @property
   def num_arms(self) -> int:
-    has_iswap = self.extended_conf.left_x_drive.iswap_installed
-    has_core_grippers = self._deck is not None and self._deck.has_resource("core_grippers")
-    return 1 if (has_iswap or has_core_grippers) else 0
+    return 1 if self.extended_conf.left_x_drive.iswap_installed else 0
 
   @property
   def head96_installed(self) -> Optional[bool]:
@@ -1551,6 +1549,16 @@ async def _pip_channel_request_configuration(self, channel: int) -> PipChannelIn
       pressure_adc="Analog_Devices_AD5263" if hw_tokens[3] == "1" else "Renesas_X9268",
     )
 
+  def _is_unsupported_pip_channel_configuration_error(self, error: STARFirmwareError) -> bool:
+    """Return whether a `VW` query failed because the firmware does not support that command."""
+
+    return (
+      len(error.errors) == 1
+      and "Pipetting channel 1" in error.errors
+      and isinstance(error.errors["Pipetting channel 1"], UnknownHamiltonError)
+      and error.errors["Pipetting channel 1"].message == "Unknown command"
+    )
+
   def get_id_from_fw_response(self, resp: str) -> Optional[int]:
     """Get the id from a firmware response."""
     parsed = parse_star_fw_string(resp, "id####")
@@ -1792,9 +1800,14 @@ async def set_up_pip():
       self._channels_minimum_y_spacing = await self.channels_request_y_minimum_spacing()
 
       # Cache per-channel hardware configuration for version-specific behavior
-      self._pip_channel_information = [
-        await self._pip_channel_request_configuration(ch) for ch in range(self.num_channels)
-      ]
+      try:
+        self._pip_channel_information = [
+          await self._pip_channel_request_configuration(ch) for ch in range(self.num_channels)
+        ]
+      except STARFirmwareError as error:
+        if not self._is_unsupported_pip_channel_configuration_error(error):
+          raise
+        self._pip_channel_information = []
 
     async def set_up_autoload():
       if self.machine_conf.auto_load_installed and not skip_autoload:

pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py

@@ -200,6 +200,81 @@ def test_valid_y_bounds_are_kept(self):
     self.assertEqual(star.extended_conf.pip_maximal_y_position, 606.5)
     self.assertEqual(star.extended_conf.left_arm_min_y_position, 6.0)
 
+  def test_num_arms_depends_only_on_iswap(self):
+    star = STARBackend()
+    star._extended_conf = replace(
+      _DEFAULT_EXTENDED_CONFIGURATION,
+      left_x_drive=replace(_DEFAULT_EXTENDED_CONFIGURATION.left_x_drive, iswap_installed=False),
+    )
+    star._deck = unittest.mock.Mock()
+    star._deck.has_resource.return_value = True
+
+    self.assertEqual(star.num_arms, 0)
+
+
+class TestSTARSetup(unittest.IsolatedAsyncioTestCase):
+  def _make_backend_for_setup(self) -> STARBackend:
+    backend = STARBackend()
+    backend.request_machine_configuration = unittest.mock.AsyncMock(  # type: ignore
+      return_value=_DEFAULT_MACHINE_CONFIGURATION
+    )
+    backend.request_extended_configuration = unittest.mock.AsyncMock(  # type: ignore
+      return_value=_DEFAULT_EXTENDED_CONFIGURATION
+    )
+    backend.request_instrument_initialization_status = unittest.mock.AsyncMock(return_value=True)
+    backend.move_all_channels_in_z_safety = unittest.mock.AsyncMock()
+    backend.move_core_96_to_safe_position = unittest.mock.AsyncMock()
+    backend.request_tip_presence = unittest.mock.AsyncMock(return_value=[False] * 8)
+    backend.request_firmware_version = unittest.mock.AsyncMock(return_value="C0RFid0001rf2024.01.01")
+    backend.channels_request_y_minimum_spacing = unittest.mock.AsyncMock(return_value=[9.0] * 8)
+    return backend
+
+  def _firmware_error(self, response: str) -> STARFirmwareError:
+    backend = STARBackend()
+    with self.assertRaises(STARFirmwareError) as ctx:
+      backend.check_fw_string_error(response)
+    return ctx.exception
+
+  async def test_setup_ignores_unsupported_pip_channel_configuration_query(self):
+    backend = self._make_backend_for_setup()
+    backend._pip_channel_request_configuration = unittest.mock.AsyncMock(  # type: ignore
+      side_effect=self._firmware_error("P1VWid1111er30")
+    )
+
+    with unittest.mock.patch(
+      "pylabrobot.liquid_handling.backends.hamilton.STAR_backend.HamiltonLiquidHandler.setup",
+      new=unittest.mock.AsyncMock(),
+    ):
+      await backend.setup(skip_autoload=True, skip_iswap=True, skip_core96_head=True)
+
+    self.assertEqual(backend._pip_channel_information, [])
+
+  async def test_setup_reraises_other_pip_channel_configuration_firmware_errors(self):
+    backend = self._make_backend_for_setup()
+    backend._pip_channel_request_configuration = unittest.mock.AsyncMock(  # type: ignore
+      side_effect=self._firmware_error("P1VWid1111er31")
+    )
+
+    with unittest.mock.patch(
+      "pylabrobot.liquid_handling.backends.hamilton.STAR_backend.HamiltonLiquidHandler.setup",
+      new=unittest.mock.AsyncMock(),
+    ):
+      with self.assertRaises(STARFirmwareError):
+        await backend.setup(skip_autoload=True, skip_iswap=True, skip_core96_head=True)
+
+  async def test_setup_reraises_malformed_pip_channel_configuration_responses(self):
+    backend = self._make_backend_for_setup()
+    backend._pip_channel_request_configuration = unittest.mock.AsyncMock(  # type: ignore
+      side_effect=ValueError("bad VW response")
+    )
+
+    with unittest.mock.patch(
+      "pylabrobot.liquid_handling.backends.hamilton.STAR_backend.HamiltonLiquidHandler.setup",
+      new=unittest.mock.AsyncMock(),
+    ):
+      with self.assertRaises(ValueError):
+        await backend.setup(skip_autoload=True, skip_iswap=True, skip_core96_head=True)
+
 
 class STARCommandCatcher(STARBackend):
   """Mock backend for star that catches commands and saves them instead of sending them to the