StackerRetrieval: v1 sequential-storage capability (shares AutomatedRetrieval with RandomAccessRetrieval )

CHANGELOG.md

@@ -6,6 +6,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## Unreleased
 
+### Added
+
+- `StackerRetrieval` capability (`pylabrobot.capabilities.automated_retrieval.StackerRetrieval`) for sequential ("stacking access") plate storage: one or more single-ended LIFO `ResourceStack` stacks plus a loading tray, with `downstack`/`upstack` operations and a `StackerBackend` interface (plus `StackerChatterboxBackend`). Intended for devices like the Agilent BenchCel and HighRes MicroServe (#1113).
+- `AutomatedRetrieval` base capability (`pylabrobot.capabilities.automated_retrieval.AutomatedRetrieval`) that owns the loading tray and the plate-movement plumbing shared by the random-access `RandomAccessRetrieval` and the sequential `StackerRetrieval`. The former random-access `AutomatedRetrieval` is now `RandomAccessRetrieval` and extends this base.
+
 ## 0.2.1
 
 ### Added

docs/api/pylabrobot.capabilities.rst

@@ -209,15 +209,18 @@ Microscopy
 Automated Retrieval
 -------------------
 
-.. currentmodule:: pylabrobot.capabilities.automated_retrieval.automated_retrieval
+.. currentmodule:: pylabrobot.capabilities.automated_retrieval
 
 .. autosummary::
   :toctree: _autosummary
   :nosignatures:
   :recursive:
 
     AutomatedRetrieval
+    RandomAccessRetrieval
+    StackerRetrieval
     AutomatedRetrievalBackend
+    StackerBackend
 
 
 Plate Reading - Absorbance

docs/user_guide/capabilities/automated-retrieval.ipynb

@@ -6,7 +6,7 @@
    "source": [
     "# Automated Retrieval\n",
     "\n",
-    "{class}`~pylabrobot.capabilities.automated_retrieval.automated_retrieval.AutomatedRetrieval` controls automated storage systems (carousels, plate hotels, incubators with internal storage) that can fetch and store plates.\n",
+    "{class}`~pylabrobot.capabilities.automated_retrieval.RandomAccessRetrieval` controls automated storage systems (carousels, plate hotels, incubators with internal storage) that can fetch and store plates.\n",
     "\n",
     "## When to use\n",
     "\n",
@@ -18,7 +18,7 @@
   {
    "cell_type": "code",
    "metadata": {},
-   "source": "from pylabrobot.capabilities.automated_retrieval import AutomatedRetrieval\nfrom pylabrobot.capabilities.automated_retrieval.chatterbox import AutomatedRetrievalChatterboxBackend\nfrom pylabrobot.resources import Coordinate, Cor_96_wellplate_360ul_Fb, PlateCarrier, PlateHolder\n\n# The capability owns the storage racks and the loading tray, and implements the\n# site bookkeeping (free-site counting, lookup, selection). A device that composes\n# AutomatedRetrieval passes these in; here we build them by hand.\nsite = PlateHolder(name=\"site_0\", size_x=127.76, size_y=85.48, size_z=30, pedestal_size_z=0)\nsite.location = Coordinate(0, 0, 0)\nrack = PlateCarrier(name=\"rack_0\", size_x=135, size_y=95, size_z=400, sites={0: site})\nloading_tray = PlateHolder(\n    name=\"loading_tray\", size_x=127.76, size_y=85.48, size_z=0, pedestal_size_z=0\n)\n\nretrieval = AutomatedRetrieval(\n    backend=AutomatedRetrievalChatterboxBackend(),\n    racks=[rack],\n    loading_tray=loading_tray,\n)\nawait retrieval._on_setup()\n\n# place a plate into storage so we have something to retrieve\nsite.assign_child_resource(Cor_96_wellplate_360ul_Fb(name=\"my_plate\"))\nprint(\"free sites:\", retrieval.get_num_free_sites())",
+   "source": "from pylabrobot.capabilities.automated_retrieval import RandomAccessRetrieval\nfrom pylabrobot.capabilities.automated_retrieval.chatterbox import AutomatedRetrievalChatterboxBackend\nfrom pylabrobot.resources import Coordinate, Cor_96_wellplate_360ul_Fb, PlateCarrier, PlateHolder\n\n# The capability owns the storage racks and the loading tray, and implements the\n# site bookkeeping (free-site counting, lookup, selection).  A device that composes\n# RandomAccessRetrieval passes these in; here we build them by hand.\nsite = PlateHolder(name=\"site_0\", size_x=127.76, size_y=85.48, size_z=30, pedestal_size_z=0)\nsite.location = Coordinate(0, 0, 0)\nrack = PlateCarrier(name=\"rack_0\", size_x=135, size_y=95, size_z=400, sites={0: site})\nloading_tray = PlateHolder(\n    name=\"loading_tray\", size_x=127.76, size_y=85.48, size_z=0, pedestal_size_z=0\n)\n\nretrieval = RandomAccessRetrieval(\n    backend=AutomatedRetrievalChatterboxBackend(),\n    racks=[rack],\n    loading_tray=loading_tray,\n)\nawait retrieval._on_setup()\n\n# place a plate into storage so we have something to retrieve\nsite.assign_child_resource(Cor_96_wellplate_360ul_Fb(name=\"my_plate\"))\nprint(\"free sites:\", retrieval.get_num_free_sites())",
    "execution_count": null,
    "outputs": []
   },
@@ -40,7 +40,7 @@
     "\n",
     "## API reference\n",
     "\n",
-    "See {class}`~pylabrobot.capabilities.automated_retrieval.automated_retrieval.AutomatedRetrieval` and {class}`~pylabrobot.capabilities.automated_retrieval.automated_retrieval.AutomatedRetrievalBackend`."
+    "See {class}`~pylabrobot.capabilities.automated_retrieval.RandomAccessRetrieval` and {class}`~pylabrobot.capabilities.automated_retrieval.RandomAccessRetrievalBackend`."
    ]
   }
  ],

pylabrobot/capabilities/automated_retrieval/__init__.py

@@ -1,2 +1,5 @@
-from .automated_retrieval import AutomatedRetrieval, NoFreeSiteError
-from .backend import AutomatedRetrievalBackend
+from .automated_retrieval import AutomatedRetrieval
+from .backend import AutomatedRetrievalBackend, StackerBackend
+from .chatterbox import AutomatedRetrievalChatterboxBackend, StackerChatterboxBackend
+from .random_access_retrieval import NoFreeSiteError, RandomAccessRetrieval
+from .stacker_retrieval import EmptyStackError, LoadingTrayOccupiedError, StackerRetrieval

pylabrobot/capabilities/automated_retrieval/automated_retrieval.py

@@ -1,159 +1,59 @@
-import random
-from typing import List, Literal, Optional, Union, cast
+from typing import Optional
 
-from pylabrobot.capabilities.capability import Capability, need_capability_ready
-from pylabrobot.resources import (
-  Plate,
-  PlateCarrier,
-  PlateHolder,
-  ResourceNotFoundError,
-)
-
-from .backend import AutomatedRetrievalBackend
-
-
-class NoFreeSiteError(Exception):
-  pass
+from pylabrobot.capabilities.capability import Capability, CapabilityBackend
+from pylabrobot.resources import Plate, PlateHolder, ResourceNotFoundError
 
 
 class AutomatedRetrieval(Capability):
-  """Automated plate retrieval/storage capability.
+  """Shared base for storage-retrieval capabilities that move plates to and from a single
+  transfer position -- the "loading tray".
+
+  Concrete capabilities differ only in how storage locations are addressed:
 
-  Owns the storage racks and the loading tray, and implements the site
-  bookkeeping (free-site counting, lookup and selection) shared by all
-  automated storage systems so devices composing this capability do not have to
-  reimplement it.
+  * :class:`~pylabrobot.capabilities.automated_retrieval.RandomAccessRetrieval` is *random access*
+    -- individually addressable rack sites.
+  * :class:`~pylabrobot.capabilities.automated_retrieval.StackerRetrieval` is *sequential* --
+    single-ended LIFO stacks.
 
-  See :doc:`/user_guide/capabilities/automated-retrieval` for a walkthrough.
+  This base owns the loading tray and the small amount of plate-movement plumbing the two share
+  (loading-tray access and the summary table), so the concrete capabilities only implement their
+  location-addressing logic.
   """
 
-  def __init__(
-    self,
-    backend: AutomatedRetrievalBackend,
-    racks: Optional[List[PlateCarrier]] = None,
-    loading_tray: Optional[PlateHolder] = None,
-  ):
+  def __init__(self, backend: CapabilityBackend, loading_tray: Optional[PlateHolder] = None):
     super().__init__(backend=backend)
-    self.backend: AutomatedRetrievalBackend = backend
-    self._racks: List[PlateCarrier] = racks if racks is not None else []
     self.loading_tray = loading_tray
 
-  @property
-  def racks(self) -> List[PlateCarrier]:
-    return self._racks
-
-  # -- site bookkeeping --
-
-  def get_num_free_sites(self) -> int:
-    return sum(len(rack.get_free_sites()) for rack in self._racks)
-
-  def get_site_by_plate_name(self, plate_name: str) -> PlateHolder:
-    for rack in self._racks:
-      for site in rack.sites.values():
-        if site.resource is not None and site.resource.name == plate_name:
-          return site
-    raise ResourceNotFoundError(f"Plate {plate_name} not found in automated storage")
-
-  def _find_available_sites_sorted(self, plate: Plate) -> List[PlateHolder]:
-    """Find all sites that are free and fit the plate, sorted by size."""
-
-    def _plate_height(p: Plate):
-      if p.has_lid():
-        # TODO: we can use plr nesting height
-        # lid.location.z + lid.get_anchor(z="t").z
-        return p.get_size_z() + 3
-      return p.get_size_z()
-
-    available = [
-      site
-      for rack in self._racks
-      for site in rack.get_free_sites()
-      if site.get_size_z() >= _plate_height(plate)
-    ]
-    if len(available) == 0:
-      raise NoFreeSiteError(f"No free site found for plate '{plate.name}'")
-    return sorted(available, key=lambda site: site.get_size_z())
-
-  def find_smallest_site_for_plate(self, plate: Plate) -> PlateHolder:
-    return self._find_available_sites_sorted(plate)[0]
-
-  def find_random_site(self, plate: Plate) -> PlateHolder:
-    return random.choice(self._find_available_sites_sorted(plate))
-
-  # -- storage operations --
-
-  @need_capability_ready
-  async def fetch_plate_to_loading_tray(self, plate_name: str) -> Plate:
-    """Retrieve the plate with the given name from storage onto the loading tray."""
+  def _require_loading_tray(self) -> PlateHolder:
     if self.loading_tray is None:
-      raise RuntimeError("No loading tray configured for this automated retrieval.")
-    site = self.get_site_by_plate_name(plate_name)
-    plate = cast(Plate, site.resource)
-    await self.backend.fetch_plate_to_loading_tray(plate)
-    plate.unassign()
-    self.loading_tray.assign_child_resource(plate)
-    return plate
+      raise RuntimeError("No loading tray configured for this capability.")
+    return self.loading_tray
 
-  @need_capability_ready
-  async def take_in_plate(
-    self, site: Union[PlateHolder, Literal["random", "smallest"]] = "smallest"
-  ):
-    """Take the plate from the loading tray and store it into storage.
-
-    `site` may be an explicit free `PlateHolder`, or `"smallest"`/`"random"` to
-    let the capability pick a fitting free site.
-    """
-    if self.loading_tray is None:
-      raise RuntimeError("No loading tray configured for this automated retrieval.")
-    plate = cast(Optional[Plate], self.loading_tray.resource)
-    if plate is None:
+  def _plate_on_loading_tray(self) -> Plate:
+    tray = self._require_loading_tray()
+    plate = tray.resource
+    if not isinstance(plate, Plate):
       raise ResourceNotFoundError("No plate on the loading tray.")
+    return plate
 
-    if site == "random":
-      site = self.find_random_site(plate)
-    elif site == "smallest":
-      site = self.find_smallest_site_for_plate(plate)
-    elif isinstance(site, PlateHolder):
-      if site not in self._find_available_sites_sorted(plate):
-        raise ValueError(f"Site {site.name} is not available for plate {plate.name}")
-    else:
-      raise ValueError(f"Invalid site: {site}")
-
-    await self.backend.store_plate(plate, site)
-    plate.unassign()
-    site.assign_child_resource(plate)
-
-  def summary(self) -> str:
-    def create_pretty_table(header, *columns) -> str:
-      col_widths = [
-        max(len(str(item)) for item in [header[i]] + list(columns[i])) for i in range(len(header))
-      ]
-
-      def format_row(row, border="|") -> str:
-        return (
-          f"{border} "
-          + " | ".join(f"{str(row[i]).ljust(col_widths[i])}" for i in range(len(row)))
-          + f" {border}"
-        )
-
-      def separator_line(cross: str = "+", line: str = "-") -> str:
-        return cross + cross.join(line * (width + 2) for width in col_widths) + cross
-
-      table = []
-      table.append(separator_line())  # Top border
-      table.append(format_row(header))
-      table.append(separator_line())  # Header separator
-      for row in zip(*columns):
-        table.append(format_row(row))
-      table.append(separator_line())  # Bottom border
-      return "\n".join(table)
-
-    header = [f"Rack {i}" for i in range(len(self._racks))]
-    sites = [
-      [site.resource.name if site.resource else "<empty>" for site in reversed(rack.sites.values())]
-      for rack in self._racks
+  @staticmethod
+  def _pretty_table(header, *columns) -> str:
+    col_widths = [
+      max(len(str(item)) for item in [header[i]] + list(columns[i])) for i in range(len(header))
     ]
-    return create_pretty_table(header, *sites)
 
-  async def _on_stop(self):
-    await super()._on_stop()
+    def format_row(row, border="|") -> str:
+      return (
+        f"{border} "
+        + " | ".join(f"{str(row[i]).ljust(col_widths[i])}" for i in range(len(row)))
+        + f" {border}"
+      )
+
+    def separator_line(cross: str = "+", line: str = "-") -> str:
+      return cross + cross.join(line * (width + 2) for width in col_widths) + cross
+
+    table = [separator_line(), format_row(header), separator_line()]
+    for row in zip(*columns):
+      table.append(format_row(row))
+    table.append(separator_line())
+    return "\n".join(table)

pylabrobot/capabilities/automated_retrieval/backend.py

@@ -2,10 +2,11 @@
 
 from pylabrobot.capabilities.capability import CapabilityBackend
 from pylabrobot.resources import Plate, PlateHolder
+from pylabrobot.resources.resource_stack import ResourceStack
 
 
 class AutomatedRetrievalBackend(CapabilityBackend, metaclass=ABCMeta):
-  """Abstract backend for automated plate retrieval/storage devices."""
+  """Abstract backend for random-access automated plate retrieval/storage devices."""
 
   @abstractmethod
   async def fetch_plate_to_loading_tray(self, plate: Plate):
@@ -14,3 +15,20 @@ async def fetch_plate_to_loading_tray(self, plate: Plate):
   @abstractmethod
   async def store_plate(self, plate: Plate, site: PlateHolder):
     """Store a plate from the loading tray into the given site."""
+
+
+class StackerBackend(CapabilityBackend, metaclass=ABCMeta):
+  """Abstract backend for a sequential ("stacking access") plate stacker.
+
+  A stacker stores plates in one or more single-ended LIFO stacks; only the accessible (top) plate
+  of a stack can be moved without first moving the plates above it. The device exposes two
+  transfers between a stack and the loading tray.
+  """
+
+  @abstractmethod
+  async def downstack(self, stack: ResourceStack):
+    """Move the accessible (top) plate of ``stack`` onto the loading tray."""
+
+  @abstractmethod
+  async def upstack(self, stack: ResourceStack, plate: Plate):
+    """Move a plate from the loading tray onto ``stack``."""

pylabrobot/capabilities/automated_retrieval/chatterbox.py

@@ -2,8 +2,9 @@
 
 from pylabrobot.resources.carrier import PlateHolder
 from pylabrobot.resources.plate import Plate
+from pylabrobot.resources.resource_stack import ResourceStack
 
-from .backend import AutomatedRetrievalBackend
+from .backend import AutomatedRetrievalBackend, StackerBackend
 
 logger = logging.getLogger(__name__)
 
@@ -16,3 +17,13 @@ async def fetch_plate_to_loading_tray(self, plate: Plate):
 
   async def store_plate(self, plate: Plate, site: PlateHolder):
     logger.info("Storing plate %s at site %s.", plate.name, site.name)
+
+
+class StackerChatterboxBackend(StackerBackend):
+  """Chatterbox backend for device-free testing."""
+
+  async def downstack(self, stack: ResourceStack):
+    logger.info("Downstacking accessible plate from stack %s.", stack.name)
+
+  async def upstack(self, stack: ResourceStack, plate: Plate):
+    logger.info("Upstacking plate %s onto stack %s.", plate.name, stack.name)