Stacker: sequential ("stacking access") storage capability

CHANGELOG.md

@@ -12,6 +12,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 - In-process `MicroSpinMockServer` (`pylabrobot.centrifuge.highres.mock_server`) that faithfully emulates the MicroSpin's wire protocol -- including the firmware's "`status` blocks until the spindle has stopped" semantics and the low-G spin-down-detection hang -- usable as a Python async context manager or runnable as a script (`python -m pylabrobot.centrifuge.highres.mock_server`) for `nc`/`telnet` debugging.
 - `MicroSpinBackend.reset()` recovery helper that issues `abort` -> `clearbuttonabort` -> `status`, using the last as the gate that genuinely confirms the rotor has stopped.
 - User guide notebook for the MicroSpin (`docs/user_guide/01_material-handling/centrifuge/highres_microspin.ipynb`).
+- `Stacker` capability (`pylabrobot.storage.Stacker`) for sequential ("stacking access") plate storage: one or more single-ended LIFO `ResourceStack` stacks plus a transfer position ("loading tray"), with `downstack`/`upstack` operations and a `StackerBackend` interface (plus `StackerChatterboxBackend`). Intended for devices like the Agilent BenchCel and HighRes MicroServe (#1113).
 
 ### Fixed
 

docs/user_guide/01_material-handling/storage/storage.rst

@@ -133,6 +133,62 @@ Combined Retrieval & Access Summary
        LiCONiC STX Series
 
 
+------------------------------------------
+
+In PyLabRobot, these two retrieval patterns map to two capabilities:
+
+* **Random access** -> the ``Incubator`` frontend, which holds addressable
+  ``PlateHolder`` sites in ``PlateCarrier`` racks (any plate is directly
+  reachable).
+* **Stacking access (sequential)** -> the ``Stacker`` capability
+  (``pylabrobot.storage.Stacker``), described below.
+
+The ``Stacker`` capability
+--------------------------------------------------
+
+A ``Stacker`` models one or more single-ended LIFO stacks -- each a
+``ResourceStack`` with ``direction="z"`` -- plus a single transfer position, the
+*loading tray* (the same term incubators use). Only the **accessible** (top)
+plate of a stack can be moved without first moving the plates above it, and
+plates nest by their ``stacking_z_height`` so the stack height is computed
+correctly.
+
+Two primitives move plates between a stack and the loading tray:
+
+* ``downstack(stack)`` -- move the accessible plate of ``stack`` onto the loading
+  tray (and return it).
+* ``upstack(stack, plate=None)`` -- move a plate from the loading tray onto
+  ``stack`` (defaults to whatever is currently on the tray).
+
+``Stacker`` is a *capability*, not a device-specific frontend: a machine that is
+a stacker (e.g. the Agilent BenchCel or HighRes MicroServe) composes it and
+provides a ``StackerBackend`` that implements the device-specific
+``downstack``/``upstack`` transfers. ``StackerChatterboxBackend`` is a no-op
+backend useful for trying the API out without hardware:
+
+.. code-block:: python
+
+   from pylabrobot.resources import Coordinate
+   from pylabrobot.resources.resource_stack import ResourceStack
+   from pylabrobot.storage import Stacker, StackerChatterboxBackend
+
+   stacker = Stacker(
+     backend=StackerChatterboxBackend(),
+     name="stacker",
+     size_x=200,
+     size_y=200,
+     size_z=300,
+     stacks=[ResourceStack(f"stack_{i}", "z") for i in range(4)],
+     loading_tray_location=Coordinate(0, 0, 0),
+   )
+   await stacker.setup()
+
+   # Move the accessible plate of stack 0 onto the loading tray:
+   plate = await stacker.downstack(0)
+   # ... hand it to a robot arm / reader, then return a plate from the tray:
+   await stacker.upstack(1)
+
+
 ------------------------------------------
 
 .. toctree::

pylabrobot/storage/__init__.py

@@ -4,3 +4,6 @@
 from .incubator import Incubator
 from .inheco.scila import SCILABackend
 from .liconic import ExperimentalLiconicBackend
+from .stacker import EmptyStackError, LoadingTrayOccupiedError, Stacker
+from .stacker_backend import StackerBackend
+from .stacker_chatterbox import StackerChatterboxBackend

pylabrobot/storage/stacker.py

@@ -0,0 +1,167 @@
+from typing import List, Optional, Union
+
+from pylabrobot.machines import Machine
+from pylabrobot.resources import (
+  Coordinate,
+  Plate,
+  PlateHolder,
+  Resource,
+  ResourceNotFoundError,
+  Rotation,
+)
+from pylabrobot.resources.resource_stack import ResourceStack
+from pylabrobot.serializer import serialize
+
+from .stacker_backend import StackerBackend
+
+
+class EmptyStackError(Exception):
+  """Raised when downstacking from a stack that has no plates."""
+
+
+class LoadingTrayOccupiedError(Exception):
+  """Raised when a transfer would collide with a plate already on the loading tray."""
+
+
+class Stacker(Machine, Resource):
+  """Sequential ("stacking access") plate-storage capability.
+
+  Models one or more single-ended LIFO stacks of (nesting) plates plus a single transfer
+  position -- the "loading tray", borrowing the incubator's term. Each stack is a
+  :class:`~pylabrobot.resources.resource_stack.ResourceStack` (``direction="z"``), which enforces
+  LIFO access (only the top plate can be removed) and computes the stack height from each plate's
+  ``stacking_z_height``.
+
+  This is a *capability*: it is meant to be composed onto a machine (e.g.
+  ``self.stacker = Stacker(backend=...)``) rather than subclassed into a device-specific frontend.
+  Devices that are stackers include the Agilent BenchCel and the HighRes MicroServe.
+  """
+
+  def __init__(
+    self,
+    backend: StackerBackend,
+    name: str,
+    size_x: float,
+    size_y: float,
+    size_z: float,
+    stacks: List[ResourceStack],
+    loading_tray_location: Coordinate,
+    rotation: Optional[Rotation] = None,
+    category: Optional[str] = None,
+    model: Optional[str] = None,
+  ):
+    Machine.__init__(self, backend=backend)
+    self.backend: StackerBackend = backend  # fix type
+    Resource.__init__(
+      self,
+      name=name,
+      size_x=size_x,
+      size_y=size_y,
+      size_z=size_z,
+      rotation=rotation,
+      category=category,
+      model=model,
+    )
+
+    self.loading_tray = PlateHolder(
+      name=self.name + "_tray", size_x=127.76, size_y=85.48, size_z=0, pedestal_size_z=0
+    )
+    self.assign_child_resource(self.loading_tray, location=loading_tray_location)
+
+    self._stacks = stacks
+    for stack in self._stacks:
+      self.assign_child_resource(stack, location=None)
+
+  @property
+  def stacks(self) -> List[ResourceStack]:
+    return self._stacks
+
+  async def setup(self, **backend_kwargs):
+    await super().setup(**backend_kwargs)
+    await self.backend.set_stacks(self._stacks)
+
+  def _resolve_stack(self, stack: Union[ResourceStack, int]) -> ResourceStack:
+    if isinstance(stack, int):
+      return self._stacks[stack]
+    if stack not in self._stacks:
+      raise ValueError(f"Stack {stack.name!r} is not part of stacker '{self.name}'")
+    return stack
+
+  def get_accessible_plate(self, stack: Union[ResourceStack, int]) -> Optional[Plate]:
+    """The only plate that can be downstacked without moving others (the top of the stack)."""
+    stack = self._resolve_stack(stack)
+    if len(stack.children) == 0:
+      return None
+    top = stack.get_top_item()
+    return top if isinstance(top, Plate) else None
+
+  def get_stack_by_plate_name(self, plate_name: str) -> ResourceStack:
+    for stack in self._stacks:
+      for child in stack.children:
+        if child.name == plate_name:
+          return stack
+    raise ResourceNotFoundError(f"Plate {plate_name} not found in stacker '{self.name}'")
+
+  async def downstack(self, stack: Union[ResourceStack, int], **backend_kwargs) -> Plate:
+    """Move the accessible (top) plate of ``stack`` onto the loading tray and return it."""
+    stack = self._resolve_stack(stack)
+    plate = self.get_accessible_plate(stack)
+    if plate is None:
+      raise EmptyStackError(f"Stack {stack.name!r} of stacker '{self.name}' is empty")
+    if self.loading_tray.resource is not None:
+      raise LoadingTrayOccupiedError(
+        f"Loading tray of stacker '{self.name}' already holds '{self.loading_tray.resource.name}'"
+      )
+    await self.backend.downstack(stack, **backend_kwargs)
+    plate.unassign()
+    self.loading_tray.assign_child_resource(plate)
+    return plate
+
+  async def upstack(
+    self,
+    stack: Union[ResourceStack, int],
+    plate: Optional[Plate] = None,
+    **backend_kwargs,
+  ) -> None:
+    """Move a plate from the loading tray onto ``stack`` (its new accessible plate).
+
+    ``plate`` defaults to whatever is on the loading tray.
+    """
+    stack = self._resolve_stack(stack)
+    if plate is None:
+      tray_resource = self.loading_tray.resource
+      if not isinstance(tray_resource, Plate):
+        raise ResourceNotFoundError(f"No plate on the loading tray of stacker '{self.name}'")
+      plate = tray_resource
+    await self.backend.upstack(stack, plate, **backend_kwargs)
+    plate.unassign()
+    stack.assign_child_resource(plate)
+
+  def summary(self) -> str:
+    lines = [f"Stacker '{self.name}' ({len(self._stacks)} stacks)"]
+    for i, stack in enumerate(self._stacks):
+      # bottom -> top; the accessible plate is last.
+      contents = [child.name for child in stack.children] or ["<empty>"]
+      lines.append(f"  stack {i}: " + " -> ".join(contents) + "  (top = accessible)")
+    tray = self.loading_tray.resource
+    lines.append(f"  loading tray: {tray.name if tray is not None else '<empty>'}")
+    return "\n".join(lines)
+
+  def serialize(self) -> dict:
+    return {
+      **Machine.serialize(self),
+      **Resource.serialize(self),
+      "backend": self.backend.serialize(),
+      "stacks": [stack.serialize() for stack in self._stacks],
+      "loading_tray_location": serialize(self.loading_tray.location),
+    }
+
+  @classmethod
+  def deserialize(cls, data: dict, allow_marshal: bool = False) -> "Stacker":
+    # Deserialization is not supported yet: it needs ResourceStack serialization support
+    # (ResourceStack.__init__ takes ``direction`` rather than ``size_*``, so it does not round-trip
+    # through the generic Resource.(de)serialize path). Tracked as a follow-up. This override also
+    # resolves the otherwise-ambiguous ``deserialize`` inherited from both Machine and Resource.
+    raise NotImplementedError(
+      "Stacker.deserialize is not implemented yet (pending ResourceStack serialization support)."
+    )

pylabrobot/storage/stacker_backend.py

@@ -0,0 +1,42 @@
+from abc import ABCMeta, abstractmethod
+from typing import List, Optional
+
+from pylabrobot.machines.backend import MachineBackend
+from pylabrobot.resources.plate import Plate
+from pylabrobot.resources.resource_stack import ResourceStack
+
+
+class StackerBackend(MachineBackend, metaclass=ABCMeta):
+  """Backend interface for the sequential ("stacking access") :class:`Stacker` capability.
+
+  A stacker stores plates in one or more single-ended LIFO stacks. Unlike an
+  incubator's random-access racks, only the accessible (top) plate of each stack can be moved
+  without first moving the plates above it. The device exposes two primitive transfers between a
+  stack and the stacker's transfer position (the "loading tray"): ``downstack`` (stack ->
+  transfer position) and ``upstack`` (transfer position -> stack).
+
+  Backends are not incubators: there is deliberately no door/temperature/shaking here. A device
+  that both stores sequentially and, say, controls temperature would compose this capability with
+  a separate temperature-control capability.
+  """
+
+  def __init__(self) -> None:
+    super().__init__()
+    self._stacks: Optional[List[ResourceStack]] = None
+
+  @property
+  def stacks(self) -> List[ResourceStack]:
+    assert self._stacks is not None, "Backend not set up?"
+    return self._stacks
+
+  async def set_stacks(self, stacks: List[ResourceStack]) -> None:
+    """Configure the stacks the device manages. Called by :meth:`Stacker.setup`."""
+    self._stacks = stacks
+
+  @abstractmethod
+  async def downstack(self, stack: ResourceStack, **backend_kwargs) -> None:
+    """Move the accessible plate from ``stack`` to the stacker's transfer position."""
+
+  @abstractmethod
+  async def upstack(self, stack: ResourceStack, plate: Plate, **backend_kwargs) -> None:
+    """Move ``plate`` from the stacker's transfer position onto ``stack``."""

pylabrobot/storage/stacker_chatterbox.py

@@ -0,0 +1,19 @@
+from pylabrobot.resources.plate import Plate
+from pylabrobot.resources.resource_stack import ResourceStack
+from pylabrobot.storage.stacker_backend import StackerBackend
+
+
+class StackerChatterboxBackend(StackerBackend):
+  """A no-op :class:`StackerBackend` that prints each operation; for tests and demos."""
+
+  async def setup(self):
+    print("Setting up stacker backend")
+
+  async def stop(self):
+    print("Stopping stacker backend")
+
+  async def downstack(self, stack: ResourceStack, **backend_kwargs):
+    print(f"Downstacking accessible plate from stack '{stack.name}'")
+
+  async def upstack(self, stack: ResourceStack, plate: Plate, **backend_kwargs):
+    print(f"Upstacking plate '{plate.name}' onto stack '{stack.name}'")