diff --git a/CHANGELOG.md b/CHANGELOG.md index 220209251b6..91509eb3fa9 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/api/pylabrobot.capabilities.rst b/docs/api/pylabrobot.capabilities.rst index 666ce292e20..6ce20ae27df 100644 --- a/docs/api/pylabrobot.capabilities.rst +++ b/docs/api/pylabrobot.capabilities.rst @@ -209,7 +209,7 @@ Microscopy Automated Retrieval ------------------- -.. currentmodule:: pylabrobot.capabilities.automated_retrieval.automated_retrieval +.. currentmodule:: pylabrobot.capabilities.automated_retrieval .. autosummary:: :toctree: _autosummary @@ -217,7 +217,10 @@ Automated Retrieval :recursive: AutomatedRetrieval + RandomAccessRetrieval + StackerRetrieval AutomatedRetrievalBackend + StackerBackend Plate Reading - Absorbance diff --git a/docs/user_guide/capabilities/automated-retrieval.ipynb b/docs/user_guide/capabilities/automated-retrieval.ipynb index 7d321c9b9e4..3c7539d84a9 100644 --- a/docs/user_guide/capabilities/automated-retrieval.ipynb +++ b/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`." ] } ], diff --git a/pylabrobot/capabilities/automated_retrieval/__init__.py b/pylabrobot/capabilities/automated_retrieval/__init__.py index 1b32adda8cd..0d6d56b0c62 100644 --- a/pylabrobot/capabilities/automated_retrieval/__init__.py +++ b/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 diff --git a/pylabrobot/capabilities/automated_retrieval/automated_retrieval.py b/pylabrobot/capabilities/automated_retrieval/automated_retrieval.py index e505900664c..db01584a626 100644 --- a/pylabrobot/capabilities/automated_retrieval/automated_retrieval.py +++ b/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 "" 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) diff --git a/pylabrobot/capabilities/automated_retrieval/backend.py b/pylabrobot/capabilities/automated_retrieval/backend.py index 5ad8179d946..77c7234d0a8 100644 --- a/pylabrobot/capabilities/automated_retrieval/backend.py +++ b/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``.""" diff --git a/pylabrobot/capabilities/automated_retrieval/chatterbox.py b/pylabrobot/capabilities/automated_retrieval/chatterbox.py index d1ac1c8d463..6ca176c5dde 100644 --- a/pylabrobot/capabilities/automated_retrieval/chatterbox.py +++ b/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) diff --git a/pylabrobot/capabilities/automated_retrieval/random_access_retrieval.py b/pylabrobot/capabilities/automated_retrieval/random_access_retrieval.py new file mode 100644 index 00000000000..3e249911390 --- /dev/null +++ b/pylabrobot/capabilities/automated_retrieval/random_access_retrieval.py @@ -0,0 +1,133 @@ +import random +from typing import List, Literal, Optional, Union, cast + +from pylabrobot.capabilities.automated_retrieval.automated_retrieval import AutomatedRetrieval +from pylabrobot.capabilities.capability import need_capability_ready +from pylabrobot.resources import ( + Plate, + PlateCarrier, + PlateHolder, + ResourceNotFoundError, +) + +from .backend import AutomatedRetrievalBackend + + +class NoFreeSiteError(Exception): + pass + + +class RandomAccessRetrieval(AutomatedRetrieval): + """Automated plate retrieval/storage capability (random access). + + Owns the storage racks and the loading tray, and implements the site + bookkeeping (free-site counting, lookup and selection) shared by all + random-access automated storage systems so devices composing this capability + do not have to reimplement it. The loading tray and the shared plate-movement + plumbing live on :class:`~pylabrobot.capabilities.automated_retrieval.AutomatedRetrieval`, + which the sequential :class:`~pylabrobot.capabilities.automated_retrieval.StackerRetrieval` + also uses. + + See :doc:`/user_guide/capabilities/automated-retrieval` for a walkthrough. + """ + + def __init__( + self, + backend: AutomatedRetrievalBackend, + racks: Optional[List[PlateCarrier]] = None, + loading_tray: Optional[PlateHolder] = None, + ): + super().__init__(backend=backend, loading_tray=loading_tray) + self.backend: AutomatedRetrievalBackend = backend + self._racks: List[PlateCarrier] = racks if racks is not None else [] + + @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.""" + loading_tray = self._require_loading_tray() + 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() + loading_tray.assign_child_resource(plate) + return plate + + @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. + """ + plate = self._plate_on_loading_tray() + + 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: + header = [f"Rack {i}" for i in range(len(self._racks))] + sites = [ + [site.resource.name if site.resource else "" for site in reversed(rack.sites.values())] + for rack in self._racks + ] + return self._pretty_table(header, *sites) + + async def _on_stop(self): + await super()._on_stop() diff --git a/pylabrobot/capabilities/automated_retrieval/stacker_retrieval.py b/pylabrobot/capabilities/automated_retrieval/stacker_retrieval.py new file mode 100644 index 00000000000..f5aa2a1b105 --- /dev/null +++ b/pylabrobot/capabilities/automated_retrieval/stacker_retrieval.py @@ -0,0 +1,100 @@ +from typing import List, Optional, Union + +from pylabrobot.capabilities.automated_retrieval.automated_retrieval import AutomatedRetrieval +from pylabrobot.capabilities.capability import need_capability_ready +from pylabrobot.resources import Plate, PlateHolder, ResourceNotFoundError +from pylabrobot.resources.resource_stack import ResourceStack + +from .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 StackerRetrieval(AutomatedRetrieval): + """Sequential ("stacking access") plate-storage capability. + + Owns one or more single-ended LIFO stacks -- each a + :class:`~pylabrobot.resources.resource_stack.ResourceStack` (``direction="z"``) -- plus the + loading tray shared with + :class:`~pylabrobot.capabilities.automated_retrieval.RandomAccessRetrieval` via + :class:`~pylabrobot.capabilities.automated_retrieval.AutomatedRetrieval`. Only the accessible + (top) plate of a stack can be moved without first moving the plates above it. + + Devices that are stackers (e.g. the Agilent BenchCel or HighRes MicroServe) compose this + capability and provide a + :class:`~pylabrobot.capabilities.automated_retrieval.backend.StackerBackend`. + """ + + def __init__( + self, + backend: StackerBackend, + stacks: Optional[List[ResourceStack]] = None, + loading_tray: Optional[PlateHolder] = None, + ): + super().__init__(backend=backend, loading_tray=loading_tray) + self.backend: StackerBackend = backend + self._stacks: List[ResourceStack] = stacks if stacks is not None else [] + + @property + def stacks(self) -> List[ResourceStack]: + return 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 this stacker") + 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") + + @need_capability_ready + async def downstack(self, stack: Union[ResourceStack, int]) -> Plate: + """Move the accessible (top) plate of ``stack`` onto the loading tray and return it.""" + loading_tray = self._require_loading_tray() + stack = self._resolve_stack(stack) + plate = self.get_accessible_plate(stack) + if plate is None: + raise EmptyStackError(f"Stack {stack.name!r} is empty") + if loading_tray.resource is not None: + raise LoadingTrayOccupiedError(f"Loading tray already holds '{loading_tray.resource.name}'") + await self.backend.downstack(stack) + plate.unassign() + loading_tray.assign_child_resource(plate) + return plate + + @need_capability_ready + async def upstack(self, stack: Union[ResourceStack, int], plate: Optional[Plate] = None) -> None: + """Move a plate from the loading tray onto ``stack`` (defaults to the plate on the tray).""" + stack = self._resolve_stack(stack) + if plate is None: + plate = self._plate_on_loading_tray() + await self.backend.upstack(stack, plate) + plate.unassign() + stack.assign_child_resource(plate) + + def summary(self) -> str: + columns = [[child.name for child in reversed(stack.children)] for stack in self._stacks] + height = max((len(c) for c in columns), default=0) + columns = [c + [""] * (height - len(c)) for c in columns] + header = [f"Stack {i}" for i in range(len(self._stacks))] + return self._pretty_table(header, *columns) diff --git a/pylabrobot/capabilities/automated_retrieval/stacker_retrieval_tests.py b/pylabrobot/capabilities/automated_retrieval/stacker_retrieval_tests.py new file mode 100644 index 00000000000..40b2c5f5b2c --- /dev/null +++ b/pylabrobot/capabilities/automated_retrieval/stacker_retrieval_tests.py @@ -0,0 +1,116 @@ +"""Tests for the sequential StackerRetrieval capability.""" + +import unittest + +from pylabrobot.resources import Plate, PlateHolder, ResourceNotFoundError +from pylabrobot.resources.resource_stack import ResourceStack +from pylabrobot.resources.utils import create_ordered_items_2d +from pylabrobot.resources.well import Well + +from .chatterbox import StackerChatterboxBackend +from .stacker_retrieval import EmptyStackError, LoadingTrayOccupiedError, StackerRetrieval + + +def _plate(name: str) -> Plate: + return Plate( + name, + size_x=127.76, + size_y=85.48, + size_z=14.0, + ordered_items=create_ordered_items_2d( + Well, + num_items_x=1, + num_items_y=1, + dx=0, + dy=0, + dz=0, + item_dx=9, + item_dy=9, + size_x=9, + size_y=9, + size_z=10, + ), + ) + + +def _make_stacker(num_stacks: int = 2) -> StackerRetrieval: + stacks = [ResourceStack(f"stack_{i}", "z") for i in range(num_stacks)] + loading_tray = PlateHolder(name="tray", size_x=127.76, size_y=85.48, size_z=0, pedestal_size_z=0) + return StackerRetrieval( + backend=StackerChatterboxBackend(), + stacks=stacks, + loading_tray=loading_tray, + ) + + +class StackerTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.stacker = _make_stacker() + await self.stacker._on_setup() + assert self.stacker.loading_tray is not None + self.tray: PlateHolder = self.stacker.loading_tray + + async def test_requires_setup(self): + stacker = _make_stacker() # not set up + with self.assertRaises(RuntimeError): + await stacker.downstack(0) + + async def test_upstack_moves_plate_from_tray_to_stack(self): + plate = _plate("p1") + self.tray.assign_child_resource(plate) + await self.stacker.upstack(0) + self.assertIsNone(self.tray.resource) + self.assertIs(self.stacker.stacks[0].get_top_item(), plate) + self.assertIs(self.stacker.get_accessible_plate(0), plate) + + async def test_downstack_moves_accessible_plate_to_tray(self): + plate = _plate("p1") + self.tray.assign_child_resource(plate) + await self.stacker.upstack(0) + returned = await self.stacker.downstack(0) + self.assertIs(returned, plate) + self.assertIs(self.tray.resource, plate) + self.assertEqual(len(self.stacker.stacks[0].children), 0) + + async def test_lifo_order(self): + for name in ("A", "B"): + self.tray.assign_child_resource(_plate(name)) + await self.stacker.upstack(0) + accessible = self.stacker.get_accessible_plate(0) + assert accessible is not None + self.assertEqual(accessible.name, "B") + first_out = await self.stacker.downstack(0) + self.assertEqual(first_out.name, "B") + + async def test_downstack_empty_raises(self): + with self.assertRaises(EmptyStackError): + await self.stacker.downstack(0) + + async def test_downstack_with_occupied_tray_raises(self): + self.tray.assign_child_resource(_plate("in_stack")) + await self.stacker.upstack(0) + self.tray.assign_child_resource(_plate("on_tray")) + with self.assertRaises(LoadingTrayOccupiedError): + await self.stacker.downstack(0) + + async def test_upstack_without_plate_raises(self): + with self.assertRaises(ResourceNotFoundError): + await self.stacker.upstack(0) + + async def test_get_stack_by_plate_name(self): + self.tray.assign_child_resource(_plate("findme")) + await self.stacker.upstack(1) + self.assertIs(self.stacker.get_stack_by_plate_name("findme"), self.stacker.stacks[1]) + with self.assertRaises(ResourceNotFoundError): + self.stacker.get_stack_by_plate_name("nope") + + async def test_resolve_stack_rejects_foreign_stack(self): + foreign = ResourceStack("foreign", "z") + with self.assertRaises(ValueError): + await self.stacker.upstack(foreign) + + async def test_no_loading_tray_raises(self): + stacker = StackerRetrieval(backend=StackerChatterboxBackend(), stacks=[ResourceStack("s", "z")]) + await stacker._on_setup() + with self.assertRaises(RuntimeError): + await stacker.downstack(0) diff --git a/pylabrobot/liconic/liconic.py b/pylabrobot/liconic/liconic.py index 3e7955ea602..456832354a3 100644 --- a/pylabrobot/liconic/liconic.py +++ b/pylabrobot/liconic/liconic.py @@ -1,6 +1,6 @@ from typing import List, Optional, Union -from pylabrobot.capabilities.automated_retrieval import AutomatedRetrieval, NoFreeSiteError +from pylabrobot.capabilities.automated_retrieval import NoFreeSiteError, RandomAccessRetrieval from pylabrobot.capabilities.barcode_scanning import BarcodeScanner from pylabrobot.capabilities.capability import BackendParams from pylabrobot.capabilities.humidity_controlling import HumidityController @@ -63,7 +63,7 @@ def __init__( for rack in self._racks: self.assign_child_resource(rack, location=None) - self.retrieval = AutomatedRetrieval( + self.retrieval = RandomAccessRetrieval( backend=backend, racks=self._racks, loading_tray=self.loading_tray ) self.tc = ( diff --git a/pylabrobot/thermo_fisher/cytomat/cytomat.py b/pylabrobot/thermo_fisher/cytomat/cytomat.py index 9c7c0071c1f..855636cac56 100644 --- a/pylabrobot/thermo_fisher/cytomat/cytomat.py +++ b/pylabrobot/thermo_fisher/cytomat/cytomat.py @@ -1,6 +1,6 @@ from typing import List, Optional -from pylabrobot.capabilities.automated_retrieval import AutomatedRetrieval, NoFreeSiteError +from pylabrobot.capabilities.automated_retrieval import NoFreeSiteError, RandomAccessRetrieval from pylabrobot.capabilities.capability import BackendParams from pylabrobot.capabilities.humidity_controlling import HumidityController from pylabrobot.capabilities.shaking import Shaker @@ -24,7 +24,7 @@ class Cytomat(Resource, Device): _racks: List[PlateCarrier] driver: CytomatBackend loading_tray: PlateHolder - retrieval: AutomatedRetrieval + retrieval: RandomAccessRetrieval tc: TemperatureController humidity: HumidityController shaker: Shaker @@ -65,7 +65,7 @@ def __init__( for rack in self._racks: self.assign_child_resource(rack, location=None) - self.retrieval = AutomatedRetrieval( + self.retrieval = RandomAccessRetrieval( backend=driver, racks=self._racks, loading_tray=self.loading_tray ) self.tc = TemperatureController(backend=driver)