STAR: act firmware commands out in the viewer (motion playback)

pylabrobot/hamilton/star/driver/features/core_grippers.py

@@ -77,6 +77,10 @@ def __init__(
     self._holding_resource_width: Optional[float] = None
     self._held_resource: Optional[Resource] = None
     self._taken_from: Optional[Tuple[Resource, Optional[Coordinate]]] = None
+    # For whoever acts a command out (the viewer), set before it is sent: what a grip takes and
+    # where it will hang from the front tool, or what a release puts down and where. Cleared by
+    # nothing in particular: each grip or release overwrites it.
+    self._handover: Optional[Tuple[Resource, Optional[Resource], Optional[Coordinate]]] = None
 
   # -- what carries them ---------------------------------------------------------------------------
 
@@ -168,28 +172,40 @@ def _front_tool(self) -> Optional[Resource]:
     shaft = self._pipettes.shaft(self._front_channel)
     return shaft.tip if shaft is not None and shaft.has_tip() else None
 
-  def _hang_held_resource_on_the_front_tool(self) -> None:
-    """Hang the held resource from the front tool where the jaws hold it, so it rides with them.
+  def _hang_location(self, held: Resource, from_top: Optional[float]) -> Optional[Coordinate]:
+    """Where `held` hangs from the front tool, as its child location, or None without the model.
 
-    Its centre at the jaws' centre, its top `pickup_distance_from_top` above the grip line - the
-    front channel's stop disc less the tool's overhang. Nothing happens while nothing models them.
+    Its centre at the jaws' centre, its top `from_top` below the grip line - the front channel's
+    stop disc less the tool's overhang.
     """
-    held, from_top, tool = self._held_resource, self._pickup_distance_from_top, self._front_tool()
+    tool = self._front_tool()
     if held is None or from_top is None or not isinstance(tool, HamiltonCoreGripperTool):
-      return
+      return None
     back = self._pipettes.get_reference_point_location(cast(int, self._back_channel))
     front = self._pipettes.get_reference_point_location(cast(int, self._front_channel))
     if back is None or front is None:
-      return
+      return None
     grip_line = front.z - (tool.get_size_z() - tool.fitting_depth - tool.grip_line_height)
     center = held.center().rotated(held.get_absolute_rotation())
     lfb = Coordinate(
       front.x - center.x,
       (back.y + front.y) / 2 - center.y,
       grip_line + from_top - held.get_absolute_size_z(),
     )
+    return lfb - tool.get_location_wrt(self._deck)
+
+  def _hang_held_resource_on_the_front_tool(self) -> None:
+    """Hang the held resource from the front tool where the jaws hold it, so it rides with them.
+
+    Nothing happens while nothing models them.
+    """
+    held, from_top = self._held_resource, self._pickup_distance_from_top
+    location = self._hang_location(held, from_top) if held is not None else None
+    tool = self._front_tool()
+    if held is None or location is None or not isinstance(tool, HamiltonCoreGripperTool):
+      return
     held.unassign()
-    tool.assign_child_resource(held, location=lfb - tool.get_location_wrt(self._deck))
+    tool.assign_child_resource(held, location=location)
 
   def _put_held_resource_on_the_deck(self) -> None:
     """Put a resource hanging from the front tool on the deck, where it is now."""
@@ -856,6 +872,9 @@ async def pick_up_resource(
       raise ValueError(f"the jaws would close to {closed} mm; squeeze_mm is too large")
     await pipettes._require_iswap_parked()
 
+    # For whoever acts the command out: what is taken, and where it will hang.
+    self._handover = (resource, self._front_tool(), self._hang_location(resource, from_top))
+
     source = (resource.parent, resource.location)
     try:
       async with pipettes._temporary_z_drive_profile(
@@ -966,6 +985,9 @@ async def drop_resource(
       raise ValueError(f"y_clearance must be 0 or more, is {y_clearance}")
     await pipettes._require_iswap_parked()
 
+    # For whoever acts the command out: what is put down, and where it lands.
+    self._handover = (held, destination, child)
+
     try:
       async with pipettes._temporary_z_drive_profile(
         acceleration=z_acceleration, channels=[back, front]

pylabrobot/hamilton/star/driver/features/head96.py

@@ -751,6 +751,27 @@ async def pick_up_tips(
       await self._record_after_tip_command(command_error)
     await self.dispensing_drive_request_uL_position()
 
+  def _spots_under_shafts(self, offset: Optional[Coordinate]) -> List[Optional[int]]:
+    """For each shaft, in spot order (A1, B1, ..., H12), the index of the rack spot it stands over
+    when head channel A1 is sent to spot A1 plus `offset`, or None where it is past the rack.
+
+    Raises:
+      ValueError: If the offset leaves the channels between spots rather than over them.
+    """
+    pitch = self.configuration.channel_pitch
+    offset = offset or Coordinate.zero()
+    columns, rows = round(offset.x / pitch), round(-offset.y / pitch)
+    for residual in (offset.x - columns * pitch, -offset.y - rows * pitch):
+      if abs(residual) > 1.0:
+        raise ValueError(
+          f"an offset of ({offset.x}, {offset.y}) puts the channels between the rack's spots"
+        )
+    under: List[Optional[int]] = []
+    for shaft in range(96):
+      column, row = shaft // 8 + columns, shaft % 8 + rows
+      under.append(column * 8 + row if 0 <= column < 12 and 0 <= row < 8 else None)
+    return under
+
   # -- tip drop ------------------------------------------------------------------------------------
 
   async def drop_tips(

pylabrobot/hamilton/star/driver/features/pipettes.py

@@ -66,6 +66,7 @@
 from pylabrobot.resources.container import Container
 from pylabrobot.resources.coordinate import Coordinate
 from pylabrobot.resources.errors import HasTipError, NoTipError
+from pylabrobot.resources.hamilton.core_gripper_tools import HamiltonCoreGripperTool
 from pylabrobot.resources.hamilton.tip_creators import HamiltonTip, TipDropMethod, TipPickupMethod
 from pylabrobot.resources.n_channel_pipettes import NChannelPipette, TipMountingShaft
 from pylabrobot.resources.resource import Resource
@@ -84,6 +85,20 @@
 """An X tolerance wider than any deck: X alone never splits a tip command into batches, so spots
 spread across columns go out in one, as legacy sends them."""
 
+
+def core_tool_face_distance(tool: HamiltonCoreGripperTool) -> float:
+  """How far a CO-RE grip tool's face stands from its channel's axis, in Y."""
+  pick_up = tool.pick_up_location or tool.get_anchor("c", "c", "t")
+  return float(tool.get_size_y() - pick_up.y)
+
+
+def core_tool_grip_line_overhang(tool: HamiltonCoreGripperTool) -> float:
+  """How far a mounted CO-RE grip tool's grip line hangs below its channel's stop disc, in mm: the
+  height the CO-RE plate commands are given in."""
+  pick_up = tool.pick_up_location or tool.get_anchor("c", "c", "t")
+  return float(pick_up.z - tool.fitting_depth - tool.grip_line_height)
+
+
 T = TypeVar("T")
 
 ChannelType = Literal["ML_STAR", "ML_STAR_RPC"]

pylabrobot/hamilton/star/driver/simulator.py

@@ -15,7 +15,7 @@
 import datetime
 import logging
 import math
-from typing import Any, Dict, List, Literal, Optional, Tuple, cast
+from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Tuple, cast
 
 from pylabrobot.hamilton.protocol.text.framing import (
   assemble_channel_command,
@@ -1892,6 +1892,11 @@ def __init__(
     # What the drives would still be doing, in seconds: the longest move recorded since the last
     # command, waited out before the next one goes.
     self._motion_owed = 0.0
+    # Who acts out a command's motion before the device answers it, or None: called with the
+    # module, the command and its parameters, and awaited. A viewer that animates what the drives
+    # do between the positions the model records sets this, so the command takes as long as the
+    # animation and the model moves only once it has played.
+    self.motion_listener: Optional[Callable[[str, str, Dict[str, Any]], Awaitable[None]]] = None
     # How far a tip of each defined type stands below the stop disc, by tip type index, as
     # `define_tip_needle` was told. A tip command names one of these, and what it collects hangs
     # that far down: the traverse height it ends at is the tip's, so the stop disc ends higher.
@@ -2098,6 +2103,8 @@ async def _send(
       **kwargs,
     )
     await self.pay_motion_time()
+    if self.motion_listener is not None:
+      await self.motion_listener(module, command, kwargs)
     answered = await self._answer(module, command, **kwargs)
     if answered is None:
       self._log_exchange(cmd, None)