Port Vantage to new Device/Driver/CapabilityBackend architecture

pylabrobot/capabilities/led_control/__init__.py

@@ -0,0 +1,2 @@
+from .backend import LEDBackend
+from .led_control import LEDControlCapability

pylabrobot/capabilities/led_control/backend.py

@@ -0,0 +1,35 @@
+from abc import ABCMeta, abstractmethod
+from typing import Optional
+
+from pylabrobot.capabilities.capability import BackendParams, CapabilityBackend
+
+
+class LEDBackend(CapabilityBackend, metaclass=ABCMeta):
+  """Abstract backend for LED control."""
+
+  @abstractmethod
+  async def set_color(
+    self,
+    mode: str,
+    intensity: int,
+    white: int,
+    red: int,
+    green: int,
+    blue: int,
+    backend_params: Optional[BackendParams] = None,
+  ) -> None:
+    """Set the LED color.
+
+    Args:
+      mode: "on", "off", or "blink".
+      intensity: Brightness 0-100.
+      white: White channel 0-100.
+      red: Red channel 0-100.
+      green: Green channel 0-100.
+      blue: Blue channel 0-100.
+      backend_params: Vendor-specific parameters (e.g. UV, blink interval).
+    """
+
+  @abstractmethod
+  async def turn_off(self) -> None:
+    """Turn the LED off."""

pylabrobot/capabilities/led_control/led_control.py

@@ -0,0 +1,60 @@
+import asyncio
+import random
+from typing import Optional
+
+from pylabrobot.capabilities.capability import BackendParams, Capability
+
+from .backend import LEDBackend
+
+
+class LEDControlCapability(Capability):
+  """LED control capability with convenience methods."""
+
+  def __init__(self, backend: LEDBackend):
+    super().__init__(backend=backend)
+    self.backend: LEDBackend = backend
+
+  async def set_color(
+    self,
+    mode: str = "on",
+    intensity: int = 100,
+    white: int = 0,
+    red: int = 0,
+    green: int = 0,
+    blue: int = 0,
+    backend_params: Optional[BackendParams] = None,
+  ) -> None:
+    """Set the LED color.
+
+    Args:
+      mode: "on", "off", or "blink".
+      intensity: Brightness 0-100.
+      white: White channel 0-100.
+      red: Red channel 0-100.
+      green: Green channel 0-100.
+      blue: Blue channel 0-100.
+      backend_params: Vendor-specific parameters.
+    """
+    await self.backend.set_color(
+      mode=mode, intensity=intensity,
+      white=white, red=red, green=green, blue=blue,
+      backend_params=backend_params,
+    )
+
+  async def turn_off(self) -> None:
+    """Turn the LED off."""
+    await self.backend.turn_off()
+
+  async def disco_mode(self, cycles: int = 69, delay: float = 0.1) -> None:
+    """Cycle through random colors.
+
+    Args:
+      cycles: Number of color changes.
+      delay: Seconds between color changes.
+    """
+    for _ in range(cycles):
+      r = random.randint(30, 100)
+      g = random.randint(30, 100)
+      b = random.randint(30, 100)
+      await self.set_color(mode="on", intensity=100, red=r, green=g, blue=b)
+      await asyncio.sleep(delay)

pylabrobot/hamilton/liquid_handlers/vantage/__init__.py

@@ -0,0 +1,14 @@
+from .chatterbox import VantageChatterboxDriver
+from .driver import (
+  VantageDriver,
+  VantageFirmwareError,
+  parse_vantage_fw_string,
+  vantage_response_string_to_error,
+)
+from .head96_backend import VantageHead96Backend
+from .ipg import VantageIPG
+from .led_backend import VantageLEDBackend, VantageLEDParams
+from .loading_cover import VantageLoadingCover
+from .pip_backend import VantagePIPBackend
+from .vantage import Vantage
+from .x_arm import VantageXArm

pylabrobot/hamilton/liquid_handlers/vantage/chatterbox.py

@@ -0,0 +1,79 @@
+"""VantageChatterboxDriver: prints commands instead of sending them over USB."""
+
+from .driver import VantageDriver
+
+
+class VantageChatterboxDriver(VantageDriver):
+  """Chatterbox driver for Vantage. Prints firmware commands instead of sending them over USB."""
+
+  def __init__(self, num_channels: int = 8):
+    super().__init__()
+    self._num_channels_override = num_channels
+
+  @property
+  def num_channels(self) -> int:
+    return self._num_channels_override
+
+  # -- lifecycle: skip USB, use canned config --------------------------------
+
+  async def setup(
+    self,
+    skip_loading_cover: bool = False,
+    skip_core96: bool = False,
+    skip_ipg: bool = False,
+  ):
+    # No USB — just set up backends directly.
+    self.id_ = 0
+    self._num_channels = self._num_channels_override
+
+    from .pip_backend import VantagePIPBackend
+
+    self.pip = VantagePIPBackend(self, tip_presences=[False] * self._num_channels_override)
+
+    if not skip_core96:
+      from .head96_backend import VantageHead96Backend
+
+      self.head96 = VantageHead96Backend(self)
+    else:
+      self.head96 = None
+
+    if not skip_ipg:
+      from .ipg import VantageIPG
+
+      self.ipg = VantageIPG(driver=self)
+      self.ipg._parked = True
+    else:
+      self.ipg = None
+
+    from .led_backend import VantageLEDBackend
+    from .loading_cover import VantageLoadingCover
+    from .x_arm import VantageXArm
+
+    self.led = VantageLEDBackend(self)
+    self.loading_cover = VantageLoadingCover(driver=self)
+    self.x_arm = VantageXArm(driver=self)
+
+  async def stop(self):
+    self._num_channels = None
+    self.head96 = None
+    self.ipg = None
+    self.led = None
+    self.loading_cover = None
+    self.x_arm = None
+
+  # -- I/O: print instead of USB --------------------------------------------
+
+  async def send_command(self, module, command, auto_id=True, tip_pattern=None,
+                         write_timeout=None, read_timeout=None, wait=True,
+                         fmt=None, **kwargs):
+    cmd, _ = self._assemble_command(
+      module=module, command=command, auto_id=auto_id,
+      tip_pattern=tip_pattern, **kwargs,
+    )
+    print(cmd)
+    return None
+
+  async def send_raw_command(self, command, write_timeout=None, read_timeout=None,
+                             wait=True):
+    print(command)
+    return None