diff --git a/docs/user_guide/capabilities/automated-retrieval.ipynb b/docs/user_guide/capabilities/automated-retrieval.ipynb index 7d321c9b9e4..d6b739c4a24 100644 --- a/docs/user_guide/capabilities/automated-retrieval.ipynb +++ b/docs/user_guide/capabilities/automated-retrieval.ipynb @@ -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 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(s), 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. Pass one loading\n# tray per transfer nest; most devices have just one.\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_trays=[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": [] }, diff --git a/docs/user_guide/liconic/stx/hello-world.ipynb b/docs/user_guide/liconic/stx/hello-world.ipynb index f5a5e0648cd..dbaf1351faa 100644 --- a/docs/user_guide/liconic/stx/hello-world.ipynb +++ b/docs/user_guide/liconic/stx/hello-world.ipynb @@ -35,7 +35,7 @@ { "cell_type": "code", "id": "562vs3qn86", - "source": "from pylabrobot.resources import Azenta4titudeFrameStar_96_wellplate_200ul_Vb\n\nplate = Azenta4titudeFrameStar_96_wellplate_200ul_Vb(name=\"my_plate\")\nincubator.loading_tray.assign_child_resource(plate)\n\nawait incubator.take_in_plate(\"smallest\") # store in the smallest free site that fits", + "source": "from pylabrobot.resources import Azenta4titudeFrameStar_96_wellplate_200ul_Vb\n\nplate = Azenta4titudeFrameStar_96_wellplate_200ul_Vb(name=\"my_plate\")\nincubator.loading_tray.assign_child_resource(plate)\n\nawait incubator.retrieval.take_in_plate(\"smallest\") # store in the smallest free site that fits", "metadata": {}, "execution_count": null, "outputs": [] @@ -49,7 +49,7 @@ { "cell_type": "code", "id": "2w9vsj6i8bp", - "source": "await incubator.fetch_plate_to_loading_tray(plate_name=\"my_plate\")\nretrieved = incubator.loading_tray.resource", + "source": "await incubator.retrieval.fetch_plate_to_loading_tray(plate_name=\"my_plate\")\nretrieved = incubator.loading_tray.resource", "metadata": {}, "execution_count": null, "outputs": [] @@ -63,7 +63,7 @@ { "cell_type": "code", "id": "rl1sabpoa6i", - "source": "await incubator.take_in_plate(\"random\") # random free site\n# await incubator.take_in_plate(racks[3].sites[0]) # specific rack and position", + "source": "await incubator.retrieval.take_in_plate(\"random\") # random free site\n# await incubator.retrieval.take_in_plate(racks[3].sites[0]) # specific rack and position", "metadata": {}, "execution_count": null, "outputs": [] @@ -77,7 +77,7 @@ { "cell_type": "code", "id": "rvqkhqaii3r", - "source": "print(incubator.summary())", + "source": "print(incubator.retrieval.summary())", "metadata": {}, "execution_count": null, "outputs": [] diff --git a/docs/user_guide/thermo_fisher/cytomat/hello-world.ipynb b/docs/user_guide/thermo_fisher/cytomat/hello-world.ipynb index 8292ff494b0..551915480dc 100644 --- a/docs/user_guide/thermo_fisher/cytomat/hello-world.ipynb +++ b/docs/user_guide/thermo_fisher/cytomat/hello-world.ipynb @@ -29,13 +29,13 @@ { "cell_type": "markdown", "id": "snc2ihvinf", - "source": "## Storing a plate\n\nPlace a plate on the loading tray and call {meth}`~pylabrobot.thermo_fisher.cytomat.cytomat.Cytomat.take_in_plate` to move it into storage. You can specify `\"smallest\"` (smallest free site that fits), `\"random\"`, or an explicit site. See [Automated Retrieval](../../capabilities/automated-retrieval) for the full capability API.", + "source": "## Storing a plate\n\nPlace a plate on the loading tray and call {meth}`~pylabrobot.capabilities.automated_retrieval.automated_retrieval.AutomatedRetrieval.take_in_plate` (via `cytomat.retrieval`) to move it into storage. You can specify `\"smallest\"` (smallest free site that fits), `\"random\"`, or an explicit site. See [Automated Retrieval](../../capabilities/automated-retrieval) for the full capability API.", "metadata": {} }, { "cell_type": "code", "id": "7mnnfp1pup6", - "source": "from pylabrobot.resources.corning.plates import Cor_96_wellplate_360ul_Fb\n\nplate = Cor_96_wellplate_360ul_Fb(\"my_plate\")\ncytomat.loading_tray.assign_child_resource(plate)\n\nawait cytomat.take_in_plate(\"smallest\") # choose the smallest free site\n\n# other options:\n# await cytomat.take_in_plate(\"random\") # random free site\n# await cytomat.take_in_plate(rack[3]) # store at rack position 3", + "source": "from pylabrobot.resources.corning.plates import Cor_96_wellplate_360ul_Fb\n\nplate = Cor_96_wellplate_360ul_Fb(\"my_plate\")\ncytomat.loading_tray.assign_child_resource(plate)\n\nawait cytomat.retrieval.take_in_plate(\"smallest\") # choose the smallest free site\n\n# other options:\n# await cytomat.retrieval.take_in_plate(\"random\") # random free site\n# await cytomat.retrieval.take_in_plate(rack[3]) # store at rack position 3", "metadata": {}, "execution_count": null, "outputs": [] @@ -43,13 +43,13 @@ { "cell_type": "markdown", "id": "jtiua5orv2m", - "source": "## Retrieving a plate\n\nUse {meth}`~pylabrobot.thermo_fisher.cytomat.cytomat.Cytomat.fetch_plate_to_loading_tray` to move a plate from storage back to the loading tray.", + "source": "## Retrieving a plate\n\nUse {meth}`~pylabrobot.capabilities.automated_retrieval.automated_retrieval.AutomatedRetrieval.fetch_plate_to_loading_tray` (via `cytomat.retrieval`) to move a plate from storage back to the loading tray.", "metadata": {} }, { "cell_type": "code", "id": "7o3qh6ahz29", - "source": "await cytomat.fetch_plate_to_loading_tray(\"my_plate\")\nretrieved = cytomat.loading_tray.resource", + "source": "await cytomat.retrieval.fetch_plate_to_loading_tray(\"my_plate\")\nretrieved = cytomat.loading_tray.resource", "metadata": {}, "execution_count": null, "outputs": [] @@ -105,7 +105,7 @@ { "cell_type": "code", "id": "6ur6evj2pt", - "source": "print(cytomat.summary())", + "source": "print(cytomat.retrieval.summary())", "metadata": {}, "execution_count": null, "outputs": [] diff --git a/pylabrobot/capabilities/automated_retrieval/automated_retrieval.py b/pylabrobot/capabilities/automated_retrieval/automated_retrieval.py index e505900664c..5432f17112c 100644 --- a/pylabrobot/capabilities/automated_retrieval/automated_retrieval.py +++ b/pylabrobot/capabilities/automated_retrieval/automated_retrieval.py @@ -19,11 +19,16 @@ class NoFreeSiteError(Exception): class AutomatedRetrieval(Capability): """Automated plate retrieval/storage capability. - Owns the storage racks and the loading tray, and implements the site + Owns the storage racks and the loading tray(s), 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. + Most devices have a single loading tray and pass it as a one-element list. + Devices with several transfer nests (e.g. the TundraStore) pass one + :class:`PlateHolder` per nest and address them by ``tray_index`` (0-based, + defaulting to the first tray). + See :doc:`/user_guide/capabilities/automated-retrieval` for a walkthrough. """ @@ -31,17 +36,27 @@ def __init__( self, backend: AutomatedRetrievalBackend, racks: Optional[List[PlateCarrier]] = None, - loading_tray: Optional[PlateHolder] = None, + loading_trays: Optional[List[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 + self.loading_trays: List[PlateHolder] = loading_trays if loading_trays is not None else [] @property def racks(self) -> List[PlateCarrier]: return self._racks + def _loading_tray(self, tray_index: int) -> PlateHolder: + if not self.loading_trays: + raise RuntimeError("No loading tray configured for this automated retrieval.") + if not 0 <= tray_index < len(self.loading_trays): + raise ValueError( + f"tray_index {tray_index} out of range; this device has " + f"{len(self.loading_trays)} loading tray(s)." + ) + return self.loading_trays[tray_index] + # -- site bookkeeping -- def get_num_free_sites(self) -> int: @@ -83,29 +98,29 @@ def find_random_site(self, plate: Plate) -> PlateHolder: # -- 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.""" - if self.loading_tray is None: - raise RuntimeError("No loading tray configured for this automated retrieval.") + async def fetch_plate_to_loading_tray(self, plate_name: str, tray_index: int = 0) -> Plate: + """Retrieve the named plate from storage onto loading tray ``tray_index``.""" + tray = self._loading_tray(tray_index) site = self.get_site_by_plate_name(plate_name) plate = cast(Plate, site.resource) - await self.backend.fetch_plate_to_loading_tray(plate) + await self.backend.fetch_plate_to_loading_tray(plate, tray_index=tray_index) plate.unassign() - self.loading_tray.assign_child_resource(plate) + tray.assign_child_resource(plate) return plate @need_capability_ready async def take_in_plate( - self, site: Union[PlateHolder, Literal["random", "smallest"]] = "smallest" + self, + site: Union[PlateHolder, Literal["random", "smallest"]] = "smallest", + tray_index: int = 0, ): - """Take the plate from the loading tray and store it into storage. + """Take the plate from loading tray ``tray_index`` 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) + tray = self._loading_tray(tray_index) + plate = cast(Optional[Plate], tray.resource) if plate is None: raise ResourceNotFoundError("No plate on the loading tray.") @@ -119,7 +134,7 @@ async def take_in_plate( else: raise ValueError(f"Invalid site: {site}") - await self.backend.store_plate(plate, site) + await self.backend.store_plate(plate, site, tray_index=tray_index) plate.unassign() site.assign_child_resource(plate) diff --git a/pylabrobot/capabilities/automated_retrieval/backend.py b/pylabrobot/capabilities/automated_retrieval/backend.py index 5ad8179d946..bd477ecf8f0 100644 --- a/pylabrobot/capabilities/automated_retrieval/backend.py +++ b/pylabrobot/capabilities/automated_retrieval/backend.py @@ -1,4 +1,5 @@ from abc import ABCMeta, abstractmethod +from typing import Optional from pylabrobot.capabilities.capability import CapabilityBackend from pylabrobot.resources import Plate, PlateHolder @@ -8,9 +9,23 @@ class AutomatedRetrievalBackend(CapabilityBackend, metaclass=ABCMeta): """Abstract backend for automated plate retrieval/storage devices.""" @abstractmethod - async def fetch_plate_to_loading_tray(self, plate: Plate): - """Retrieve a plate from storage and place it on the loading tray.""" + async def fetch_plate_to_loading_tray(self, plate: Plate, tray_index: Optional[int] = None): + """Retrieve a plate from storage and place it on a loading tray. + + Args: + plate: The plate to retrieve. + tray_index: 0-based index of the loading tray to deliver the plate to. ``None`` + selects the device's default tray. Devices with a single loading tray + accept ``None``/``0`` and reject any other value. + """ @abstractmethod - async def store_plate(self, plate: Plate, site: PlateHolder): - """Store a plate from the loading tray into the given site.""" + async def store_plate(self, plate: Plate, site: PlateHolder, tray_index: Optional[int] = None): + """Store a plate from a loading tray into the given site. + + Args: + plate: The plate to store. + site: The destination storage site. + tray_index: 0-based index of the loading tray the plate is currently on. ``None`` + selects the device's default tray. + """ diff --git a/pylabrobot/capabilities/automated_retrieval/chatterbox.py b/pylabrobot/capabilities/automated_retrieval/chatterbox.py index d1ac1c8d463..bb1959f6585 100644 --- a/pylabrobot/capabilities/automated_retrieval/chatterbox.py +++ b/pylabrobot/capabilities/automated_retrieval/chatterbox.py @@ -1,4 +1,5 @@ import logging +from typing import Optional from pylabrobot.resources.carrier import PlateHolder from pylabrobot.resources.plate import Plate @@ -11,8 +12,8 @@ class AutomatedRetrievalChatterboxBackend(AutomatedRetrievalBackend): """Chatterbox backend for device-free testing.""" - async def fetch_plate_to_loading_tray(self, plate: Plate): - logger.info("Fetching plate %s to loading tray.", plate.name) + async def fetch_plate_to_loading_tray(self, plate: Plate, tray_index: Optional[int] = None): + logger.info("Fetching plate %s to loading tray %s.", plate.name, tray_index) - async def store_plate(self, plate: Plate, site: PlateHolder): - logger.info("Storing plate %s at site %s.", plate.name, site.name) + async def store_plate(self, plate: Plate, site: PlateHolder, tray_index: Optional[int] = None): + logger.info("Storing plate %s at site %s (tray %s).", plate.name, site.name, tray_index) diff --git a/pylabrobot/highres/__init__.py b/pylabrobot/highres/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/pylabrobot/highres/sample_storage/__init__.py b/pylabrobot/highres/sample_storage/__init__.py new file mode 100644 index 00000000000..17997bf53da --- /dev/null +++ b/pylabrobot/highres/sample_storage/__init__.py @@ -0,0 +1,23 @@ +from .driver import ( + HighResSampleStorageAutomatedRetrievalBackend, + HighResSampleStorageDriver, + HighResSampleStorageHumidityControllerBackend, + HighResSampleStorageTemperatureControllerBackend, +) +from .errors import ( + PlateNotFoundError, + HighResSampleStorageAbortedError, + HighResSampleStorageError, + HighResSampleStorageFault, +) +from .settings import MachineType, HighResSampleStorageSettings +from .types import ( + DoorState, + EnvironmentParameter, + NestState, + StackerDimensions, + VersionInfo, +) +from pylabrobot.capabilities.automated_retrieval import NoFreeSiteError + +from .sample_storage import AmbiStore, SteriStore, TundraStore diff --git a/pylabrobot/highres/sample_storage/driver/__init__.py b/pylabrobot/highres/sample_storage/driver/__init__.py new file mode 100644 index 00000000000..a262b404768 --- /dev/null +++ b/pylabrobot/highres/sample_storage/driver/__init__.py @@ -0,0 +1,4 @@ +from .automated_retrieval import HighResSampleStorageAutomatedRetrievalBackend +from .driver import HighResSampleStorageDriver +from .humidity import HighResSampleStorageHumidityControllerBackend +from .temperature import HighResSampleStorageTemperatureControllerBackend diff --git a/pylabrobot/highres/sample_storage/driver/automated_retrieval.py b/pylabrobot/highres/sample_storage/driver/automated_retrieval.py new file mode 100644 index 00000000000..3882a5d06c0 --- /dev/null +++ b/pylabrobot/highres/sample_storage/driver/automated_retrieval.py @@ -0,0 +1,298 @@ +from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, cast + +from pylabrobot.capabilities.automated_retrieval.backend import AutomatedRetrievalBackend +from pylabrobot.resources import Plate, PlateCarrier, PlateHolder + +from ..errors import ( + HighResSampleStorageError, + HighResSampleStorageFault, + PlateNotFoundError, + left_unsafe, +) +from ..settings import HighResSampleStorageSettings +from ..types import DOOR_STATES, NEST_STATES, DoorState, NestState, StackerDimensions +from .protocol import parse_kv + +if TYPE_CHECKING: + from .driver import HighResSampleStorageDriver + + +class HighResSampleStorageAutomatedRetrievalBackend(AutomatedRetrievalBackend): + """Plate storage/motion (automated retrieval) for a HighRes sample store. + + Plates are stored in a refrigerated carousel of *stackers*, each holding a + number of *slots*. An external robot hands plates to/from one of the device's + *nests* (transfer stations); the internal spatula moves plates between a nest + and a (stacker, slot). The low-level :meth:`pick` / :meth:`place` take those + three indices directly. + + The store has two nests, exposed through the multi-tray + :class:`AutomatedRetrieval` capability: its ``tray_index`` argument is a + 0-based nest index (tray ``i`` -> device nest ``i + 1``), and ``tray_index=None`` + selects :attr:`loading_tray_nest`. + + All commands are issued through the owning :class:`HighResSampleStorageDriver`. + """ + + def __init__( + self, + driver: "HighResSampleStorageDriver", + loading_tray_nest: int = 1, + num_nests: int = 2, + ): + super().__init__() + self._driver = driver + self.loading_tray_nest = loading_tray_nest + self.num_nests = num_nests + # Slide (Y) below this is "retracted"; a spatula stuck in a stacker sits at + # the ~256mm slide-in depth, home is 0. Used by request_is_parked()/recover(). + self._retracted_y_max = 50.0 + # stacker/slot lookup, built from racks by set_racks(). + self._site_locations: Dict[str, Tuple[int, int]] = {} + + # --- queries (verified against firmware 3.0.0.119) ------------------------ + + async def request_axis_positions(self) -> Dict[str, float]: + """Return the ``status`` report: carousel/theta/Y/Z positions.""" + out: Dict[str, float] = {} + for key, value in parse_kv(await self._driver.send_command("status")).items(): + try: + out[key] = float(value) + except ValueError: + continue + return out + + async def request_is_homed(self) -> bool: + lines = await self._driver.send_command("homedstatus") + return any(line.strip().lower() == "homed" for line in lines) + + async def request_door_status(self) -> Dict[str, DoorState]: + """Parsed ``doorstatus`` output, keyed by door name.""" + doors: Dict[str, DoorState] = {} + for name, value in parse_kv(await self._driver.send_command("doorstatus")).items(): + state = value.lower() + doors[name] = cast(DoorState, state) if state in DOOR_STATES else "unknown" + return doors + + async def request_nest_status(self) -> Dict[int, NestState]: + """Parsed ``neststatus`` output, keyed by nest number.""" + nests: Dict[int, NestState] = {} + for key, value in parse_kv(await self._driver.send_command("neststatus")).items(): + try: + nest = int(key) + except ValueError: + continue + state = value.lower() + nests[nest] = cast(NestState, state) if state in NEST_STATES else "unknown" + return nests + + async def request_spatula_is_holding(self) -> bool: + """Whether a plate is currently held on the spatula (``platestatus``).""" + lines = await self._driver.send_command("platestatus") + return not any("NO_PLATE" in line for line in lines) + + async def request_nest_is_holding(self, nest: int) -> bool: + """Whether a plate is present on ``nest`` (per its plate sensor). + + Note: this unit reports an occupied nest as ``UNKNOWN`` rather than + ``OCCUPIED``; anything other than ``CLEAR`` counts as holding. + """ + state = (await self.request_nest_status()).get(nest) + return state is not None and state != "clear" + + async def probe_presence(self, stacker: int, slot: int, to_nest: int = 1) -> bool: + """Probe whether a plate is present in ``(stacker, slot)`` by attempting a + pick. Returns ``True`` if a plate was there, ``False`` if the slot is empty. + + SIDE EFFECT: a plate that is found is moved to ``to_nest`` (the only way to + sense a stacker slot is to pick it). Only safe for non-top slots, where an + empty pick is graceful; the top slot (24) faults when empty — see + :meth:`pick`. For nests, use :meth:`request_nest_is_holding` instead (a + non-destructive sensor read). + """ + try: + await self.pick(stacker, slot, to_nest) + return True + except PlateNotFoundError: + return False + + async def request_stacker_dimensions(self) -> List[StackerDimensions]: + """Parse ``getstackerdimensions`` (``: + ``).""" + dims: List[StackerDimensions] = [] + for line in await self._driver.send_command("getstackerdimensions"): + key, _, rest = line.partition(":") + try: + stacker = int(key) + zero_offset, slot_height, slot_count = rest.split() + dims.append( + StackerDimensions( + stacker=stacker, + zero_offset=float(zero_offset), + slot_height=float(slot_height), + slot_count=int(slot_count), + ) + ) + except ValueError: + continue + return dims + + async def request_settings(self) -> HighResSampleStorageSettings: + """Read the device's full settings file (``NAME = value`` pairs) into a + frozen :class:`HighResSampleStorageSettings`.""" + lines = await self._driver.send_command("settings", timeout=self._driver.read_timeout) + return HighResSampleStorageSettings.from_lines(lines) + + async def request_stacker_barcodes(self, stacker, slot: Optional[int] = None) -> List[str]: + """Scan a stacker (or a single slot) for barcodes. + + Args: + stacker: Stacker number, or the string ``"all"`` to scan the whole + inventory. + slot: Optional single slot to scan. + """ + command = f"barcode {stacker}" + if slot is not None: + command += f" {slot}" + return await self._driver.send_command(command, timeout=self._driver.motion_timeout) + + # --- motion --------------------------------------------------------------- + + async def home(self): + """Home the system. The first step closes all doors, which requires the + pneumatic supply (clean dry air >80 psi); without it this raises + :class:`HighResSampleStorageError` ("Unable to close all doors").""" + await self._driver.send_command("home", timeout=self._driver.motion_timeout) + + async def request_is_parked(self) -> bool: + """Whether the machine is genuinely safe to move: homed AND the spatula + retracted out of the carousel. + + Prefer this over :meth:`request_is_homed`. ``homedstatus`` reports homed even while + the spatula is stuck extended in a stacker after a faulted top-slot pick, so + it alone is not a safe-state check; this also verifies the slide (Y) axis is + near its home position (a stuck spatula sits at the ~256mm slide-in depth). + """ + if not await self.request_is_homed(): + return False + y = (await self.request_axis_positions()).get("Y axis") + return y is not None and abs(y) < self._retracted_y_max + + async def recover(self) -> bool: + """Retract the spatula and re-home after a motion fault. + + A faulted command (e.g. an empty-slot ``pick`` in the top few slots) can + leave the spatula extended. This ALWAYS issues the retract (``spatulaout``) + + ``home`` — it does not trust ``homedstatus`` to decide whether recovery is + needed, because that reports homed even while the spatula is stuck extended. + Retries a few times. Returns ``True`` once :meth:`request_is_parked`. + """ + for _ in range(3): + for command in ("enable", "spatulaout"): + try: + await self._driver.send_command(command, timeout=self._driver.motion_timeout) + except HighResSampleStorageError: + pass + try: + await self._driver.send_command("home", timeout=self._driver.motion_timeout) + except HighResSampleStorageError: + pass + if await self.request_is_parked(): + return True + return False + + async def pick(self, stacker: int, slot: int, nest: int, close_door: bool = True): + """Retrieve a plate from ``(stacker, slot)`` to ``nest``. + + ``close_door=False`` re-opens the doors after the transfer (see :meth:`place`). + + On failure the error is classified; no automatic motion is performed: + + - :class:`PlateNotFoundError` — the slot was empty ("No plate detected") + and the store retracted cleanly; the machine is safe to keep using. + - :class:`HighResSampleStorageFault` — the machine was left unsafe (spatula extended + / unhomed), e.g. an empty *top* slot where the firmware can't complete its + safe-travel retract. Call :meth:`recover` before any further motion. + + Note: ``homedstatus`` reports homed even when the spatula is stuck extended + at a top slot, so the firmware's own "unsafe for rotation" signal is used + (not just :meth:`request_is_homed`) to detect that case. + """ + command = f"pick {stacker} {slot} {nest}" + try: + await self._driver.send_command(command, timeout=self._driver.motion_timeout) + except HighResSampleStorageError as exc: + if left_unsafe(exc.error_lines) or not await self.request_is_homed(): + raise HighResSampleStorageFault(command, exc.error_lines) from exc + if any("no plate detected" in line.lower() for line in exc.error_lines): + raise PlateNotFoundError(command, exc.error_lines) from exc + raise + if not close_door: + await self.open_all_doors() + + async def place(self, stacker: int, slot: int, nest: int, close_door: bool = True): + """Place the plate at ``nest`` into ``(stacker, slot)``. + + The store re-seals its doors as part of every transfer, so ``close_door`` + controls only the *end* state: with ``close_door=False`` the doors are + re-opened after the place, leaving the carousel accessible for a following + operation (handy when the cold environment doesn't matter). The default + leaves it sealed. + """ + await self._driver.send_command( + f"place {stacker} {slot} {nest}", timeout=self._driver.motion_timeout + ) + if not close_door: + await self.open_all_doors() + + async def open_all_doors(self): + await self._driver.send_command("openalldoors", timeout=self._driver.motion_timeout) + + async def close_all_doors(self): + await self._driver.send_command("closealldoors", timeout=self._driver.motion_timeout) + + async def abort(self): + """Stop current machine operations. ``clear_abort`` is required afterward.""" + await self._driver.send_command("abort") + + async def clear_abort(self): + await self._driver.send_command("clearabort") + + # --- AutomatedRetrieval capability ---------------------------------------- + + async def set_racks(self, racks: List[PlateCarrier]): + """Register the storage racks so the capability can resolve a plate/site to + a ``(stacker, slot)``. Rack *i* (0-based) maps to stacker ``i + 1``; site + *j* within a rack maps to slot ``j + 1``.""" + self._site_locations = {} + for rack_index, rack in enumerate(racks): + for slot_index, site in enumerate(rack.sites.values()): + self._site_locations[site.name] = (rack_index + 1, slot_index + 1) + + def _locate(self, site: PlateHolder) -> Tuple[int, int]: + if site.name not in self._site_locations: + raise ValueError(f"Site '{site.name}' is not a known stacker slot; call set_racks() first.") + return self._site_locations[site.name] + + def _nest_for_tray(self, tray_index: Optional[int]) -> int: + """Map a 0-based capability tray index to a 1-based device nest number. + + ``None`` selects :attr:`loading_tray_nest` (the configured default nest).""" + if tray_index is None: + return self.loading_tray_nest + if not 0 <= tray_index < self.num_nests: + raise ValueError( + f"sample store has trays 0..{self.num_nests - 1}; got tray_index={tray_index}." + ) + return tray_index + 1 + + async def fetch_plate_to_loading_tray(self, plate: Plate, tray_index: Optional[int] = None): + site = plate.parent + if not isinstance(site, PlateHolder): + raise ValueError(f"Plate '{plate.name}' is not in a stacker slot.") + stacker, slot = self._locate(site) + await self.pick(stacker, slot, self._nest_for_tray(tray_index)) + + async def store_plate(self, plate: Plate, site: PlateHolder, tray_index: Optional[int] = None): + stacker, slot = self._locate(site) + await self.place(stacker, slot, self._nest_for_tray(tray_index)) diff --git a/pylabrobot/highres/sample_storage/driver/driver.py b/pylabrobot/highres/sample_storage/driver/driver.py new file mode 100644 index 00000000000..c6edd7137ea --- /dev/null +++ b/pylabrobot/highres/sample_storage/driver/driver.py @@ -0,0 +1,207 @@ +import asyncio +import logging +from dataclasses import dataclass +from typing import Dict, List, Optional + +from pylabrobot.capabilities.capability import BackendParams +from pylabrobot.device import Driver +from pylabrobot.io.socket import Socket + +from ..errors import HighResSampleStorageAbortedError, HighResSampleStorageError +from ..types import EnvironmentParameter, VersionInfo +from .automated_retrieval import HighResSampleStorageAutomatedRetrievalBackend +from .humidity import HighResSampleStorageHumidityControllerBackend +from .protocol import ( + ACK_TOKEN, + COMPLETION_ABORTED, + COMPLETION_ERROR, + COMPLETION_OK, + COMPLETION_TOKENS, + parse_kv, +) +from .temperature import HighResSampleStorageTemperatureControllerBackend + +logger = logging.getLogger(__name__) + + +class HighResSampleStorageDriver(Driver): + """Transport for HighRes Biosolutions sample stores (TundraStore / SteriStore + / AmbiStore). + + The store exposes a text-based remote-control server over TCP, port 1000. + Commands are case-sensitive, space-separated, terminated with ``\\r\\n``. Each + command is answered with an ``ACK!`` echo, optional data lines, then exactly + one completion line (``OK!`` / ``ABORTED!`` / ``ERROR!``). See the User + Manual, section "Message Formatting". + + :meth:`send_command` is the shared primitive. The driver owns the per-capability + backends (:attr:`automated_retrieval`, :attr:`temperature`, :attr:`humidity`), + which build their commands on top of it. + """ + + @dataclass + class SetupParams(BackendParams): + """Optional parameters for :meth:`setup`.""" + + home_on_setup: bool = False + + def __init__( + self, + host: str, + port: int = 1000, + read_timeout: float = 30.0, + motion_timeout: float = 240.0, + loading_tray_nest: int = 1, + num_nests: int = 2, + ): + """ + Args: + host: IP address of the store. The factory default is ``192.168.127.60``; + all HighRes devices also answer on the backdoor ``10.253.253.253``. + port: Remote-control server port (always 1000). + read_timeout: Timeout (s) for query/status commands. + motion_timeout: Timeout (s) for long-running motion commands + (``home``, ``pick``, ``place``, door moves). + loading_tray_nest: Which nest the :class:`AutomatedRetrieval` capability + uses as its default loading tray (1 or 2). + """ + super().__init__() + self.io = Socket( + human_readable_device_name="HighRes sample store", + host=host, + port=port, + read_timeout=read_timeout, + write_timeout=read_timeout, + ) + self._read_timeout = read_timeout + self._motion_timeout = motion_timeout + self._command_lock = asyncio.Lock() + + self.automated_retrieval = HighResSampleStorageAutomatedRetrievalBackend( + self, loading_tray_nest=loading_tray_nest, num_nests=num_nests + ) + self.temperature = HighResSampleStorageTemperatureControllerBackend(self) + self.humidity = HighResSampleStorageHumidityControllerBackend(self) + + @property + def read_timeout(self) -> float: + return self._read_timeout + + @property + def motion_timeout(self) -> float: + return self._motion_timeout + + def serialize(self) -> dict: + return { + **super().serialize(), + "host": self.io._host, + "port": self.io._port, + "read_timeout": self._read_timeout, + "motion_timeout": self._motion_timeout, + "loading_tray_nest": self.automated_retrieval.loading_tray_nest, + } + + # --- lifecycle ------------------------------------------------------------ + + async def setup(self, backend_params: Optional[BackendParams] = None): + if backend_params is None: + backend_params = HighResSampleStorageDriver.SetupParams() + if not isinstance(backend_params, HighResSampleStorageDriver.SetupParams): + raise TypeError(f"backend_params must be {HighResSampleStorageDriver.SetupParams}") + + await self.io.setup() + version = await self.request_version() + logger.info( + "Connected to %s (serial %s, firmware %s)", + version.product_name, + version.serial_number, + version.firmware_version, + ) + if backend_params.home_on_setup: + await self.automated_retrieval.home() + + async def stop(self): + await self.io.stop() + + # --- transport ------------------------------------------------------------ + + async def _readline(self, timeout: Optional[float]) -> str: + raw = await self.io.readuntil(b"\n", timeout=timeout) + return raw.decode("ascii", errors="replace").rstrip("\r\n") + + async def send_command(self, command: str, timeout: Optional[float] = None) -> List[str]: + """Send a command and return its data lines (those between the ``ACK!`` echo + and the completion line). + + Raises: + HighResSampleStorageError: if the device replies ``ERROR!``. + HighResSampleStorageAbortedError: if the device replies ``ABORTED!``. + """ + if timeout is None: + timeout = self._read_timeout + async with self._command_lock: + await self.io.write(command.encode("ascii") + b"\r\n") + + data_lines: List[str] = [] + completion: Optional[str] = None + seen_ack = False + while completion is None: + line = await self._readline(timeout) + if line.startswith(ACK_TOKEN) and not seen_ack: + seen_ack = True + continue + if line.startswith(COMPLETION_TOKENS): + completion = line + break + data_lines.append(line) + + if completion.startswith(COMPLETION_ERROR): + # Firmware 3.0.x emits the ``Error : ...`` stack as data lines *before* + # the ERROR! completion, so they are already collected in data_lines. + error_lines = [ln for ln in data_lines if ln.startswith("Error")] or data_lines + raise HighResSampleStorageError(command, error_lines) + if completion.startswith(COMPLETION_ABORTED): + raise HighResSampleStorageAbortedError(command) + assert completion.startswith(COMPLETION_OK) + return data_lines + + # --- shared device queries ------------------------------------------------ + + async def request_version(self) -> VersionInfo: + raw = parse_kv(await self.send_command("version")) + return VersionInfo( + product_name=raw.get("Product Name"), + serial_number=raw.get("Serial Number"), + firmware_version=raw.get("Firmware Version"), + firmware_build=raw.get("Firmware Build"), + raw=raw, + ) + + async def request_environment(self) -> Dict[str, EnvironmentParameter]: + """Parse ``environmentstatus`` into ``{name: EnvironmentParameter}``. + + Each channel reports ``NAME:current/setpoint/limit``; sensor-only channels + (e.g. the gas tank pressures) report only a current value. Shared by the + temperature and humidity capability backends. + """ + out: Dict[str, EnvironmentParameter] = {} + for line in await self.send_command("environmentstatus"): + if ":" not in line: + continue + name, _, rest = line.partition(":") + parts = rest.strip().rstrip(":").split("/") + try: + current = float(parts[0]) + except (ValueError, IndexError): + continue + + def _opt(i: int, parts=parts) -> Optional[float]: + try: + return float(parts[i]) + except (ValueError, IndexError): + return None + + out[name.strip()] = EnvironmentParameter( + name=name.strip(), current=current, setpoint=_opt(1), limit=_opt(2) + ) + return out diff --git a/pylabrobot/highres/sample_storage/driver/humidity.py b/pylabrobot/highres/sample_storage/driver/humidity.py new file mode 100644 index 00000000000..45d176a1f2e --- /dev/null +++ b/pylabrobot/highres/sample_storage/driver/humidity.py @@ -0,0 +1,29 @@ +from typing import TYPE_CHECKING + +from pylabrobot.capabilities.humidity_controlling.backend import HumidityControllerBackend + +from ..errors import HighResSampleStorageError + +if TYPE_CHECKING: + from .driver import HighResSampleStorageDriver + + +class HighResSampleStorageHumidityControllerBackend(HumidityControllerBackend): + """Humidity monitoring for a HighRes sample store (read-only; no active control).""" + + def __init__(self, driver: "HighResSampleStorageDriver"): + super().__init__() + self._driver = driver + + @property + def supports_humidity_control(self) -> bool: + return False + + async def request_current_humidity(self) -> float: + env = await self._driver.request_environment() + if "RH" not in env: + raise HighResSampleStorageError("environmentstatus", ["no RH channel reported"]) + return env["RH"].current / 100.0 + + async def set_humidity(self, humidity: float): + raise NotImplementedError("HighRes sample stores do not support active humidity control.") diff --git a/pylabrobot/highres/sample_storage/driver/protocol.py b/pylabrobot/highres/sample_storage/driver/protocol.py new file mode 100644 index 00000000000..a55bb07e270 --- /dev/null +++ b/pylabrobot/highres/sample_storage/driver/protocol.py @@ -0,0 +1,24 @@ +"""Low-level wire-protocol constants and helpers shared by the driver and the +per-capability backends.""" + +from typing import Dict, List + +# Completion-status tokens that terminate a command's reply (see the manual, +# "Message Formatting"). Every command ends with exactly one of these. +COMPLETION_OK = "OK!" +COMPLETION_ABORTED = "ABORTED!" +COMPLETION_ERROR = "ERROR!" +COMPLETION_TOKENS = (COMPLETION_OK, COMPLETION_ABORTED, COMPLETION_ERROR) + +# Immediate command-receipt echo prefix. +ACK_TOKEN = "ACK!" + + +def parse_kv(lines: List[str]) -> Dict[str, str]: + """Parse ``Key: value`` lines into a dict (first colon splits).""" + out: Dict[str, str] = {} + for line in lines: + if ":" in line: + key, _, value = line.partition(":") + out[key.strip()] = value.strip() + return out diff --git a/pylabrobot/highres/sample_storage/driver/temperature.py b/pylabrobot/highres/sample_storage/driver/temperature.py new file mode 100644 index 00000000000..c82b9a6b75f --- /dev/null +++ b/pylabrobot/highres/sample_storage/driver/temperature.py @@ -0,0 +1,32 @@ +from typing import TYPE_CHECKING + +from pylabrobot.capabilities.temperature_controlling.backend import TemperatureControllerBackend + +from ..errors import HighResSampleStorageError + +if TYPE_CHECKING: + from .driver import HighResSampleStorageDriver + + +class HighResSampleStorageTemperatureControllerBackend(TemperatureControllerBackend): + """Temperature control for a HighRes sample store (refrigerated, -20 to 4 C).""" + + def __init__(self, driver: "HighResSampleStorageDriver"): + super().__init__() + self._driver = driver + + @property + def supports_active_cooling(self) -> bool: + return True + + async def request_current_temperature(self) -> float: + env = await self._driver.request_environment() + if "TEMP" not in env: + raise HighResSampleStorageError("environmentstatus", ["no TEMP channel reported"]) + return env["TEMP"].current + + async def set_temperature(self, temperature: float): + await self._driver.send_command(f"environmentset TEMP {temperature}") + + async def deactivate(self): + await self._driver.send_command("environment TEMP off") diff --git a/pylabrobot/highres/sample_storage/errors.py b/pylabrobot/highres/sample_storage/errors.py new file mode 100644 index 00000000000..1966cbccd87 --- /dev/null +++ b/pylabrobot/highres/sample_storage/errors.py @@ -0,0 +1,70 @@ +from typing import List + + +class HighResSampleStorageError(Exception): + """A command returned an ``ERROR!`` completion status. + + The TundraStore reports failures as an error stack: the completion line is + preceded (firmware 3.0.x) by one or more ``Error : ...`` lines, the last of + which is generally the most pertinent. Those lines are preserved in + :attr:`error_lines`. + """ + + def __init__(self, command: str, error_lines: List[str]): + self.command = command + self.error_lines = error_lines + detail = error_lines[-1] if error_lines else "no error detail returned" + super().__init__(f"'{command}' failed: {detail}") + + +class HighResSampleStorageAbortedError(Exception): + """A command returned an ``ABORTED!`` completion status (e.g. after ``abort``).""" + + def __init__(self, command: str): + self.command = command + super().__init__(f"'{command}' was aborted") + + +class PlateNotFoundError(HighResSampleStorageError): + """A pick found no plate in the target slot ("No plate detected"). + + This is the normal *empty slot* outcome: the store's height detector reports + the absence and the machine stays homed and operational — not a fault. + Contrast with :class:`HighResSampleStorageFault` (the machine de-homed). Note that an + empty *top* slot raises a fault instead, because the firmware can't complete + its safe-travel retract from the topmost position. + """ + + +class HighResSampleStorageFault(HighResSampleStorageError): + """A motion command faulted and left the machine UNHOMED/extended. + + The canonical trigger is picking an empty *top* slot. The machine is not + usable until recovered — call + :meth:`HighResSampleStorageAutomatedRetrievalBackend.recover` (retract the + spatula and re-home) before issuing further motion. + """ + + def __init__(self, command: str, error_lines: List[str]): + super().__init__(command, error_lines) + self.args = (f"{self.args[0]}; machine is unsafe — call recover()",) + + +# Error-stack substrings meaning the spatula was left extended / unsafe to move, +# even when ``homedstatus`` still reports homed (it does exactly that when the +# spatula is stuck extended at a top slot). Recover before any further motion. +_UNSAFE_SIGNATURES = ( + "unsafe for rotation", + "safe travel position", + "crash occurred", + "sensor was tripped", + "machine must be homed", +) + + +def left_unsafe(error_lines: List[str]) -> bool: + """Whether an error stack indicates the machine was left unsafe (spatula + extended / unhomed), requiring + :meth:`HighResSampleStorageAutomatedRetrievalBackend.recover`.""" + blob = " ".join(error_lines).lower() + return any(sig in blob for sig in _UNSAFE_SIGNATURES) diff --git a/pylabrobot/highres/sample_storage/sample_storage.py b/pylabrobot/highres/sample_storage/sample_storage.py new file mode 100644 index 00000000000..c6e00e2621c --- /dev/null +++ b/pylabrobot/highres/sample_storage/sample_storage.py @@ -0,0 +1,141 @@ +import warnings +from typing import List, Optional + +from pylabrobot.capabilities.automated_retrieval import AutomatedRetrieval +from pylabrobot.capabilities.capability import BackendParams +from pylabrobot.capabilities.humidity_controlling import HumidityController +from pylabrobot.capabilities.temperature_controlling import TemperatureController +from pylabrobot.device import Device +from pylabrobot.resources import ( + Coordinate, + PlateCarrier, + PlateHolder, + Rotation, +) +from pylabrobot.resources.resource import Resource + +from .driver import HighResSampleStorageDriver + + +class _HighResSampleStorage(Resource, Device): + """Base device for HighRes Biosolutions sample stores. + + The TundraStore, SteriStore and AmbiStore are the same machine family behind a + shared port-1000 API, so all of the implementation lives here and the concrete + devices are thin subclasses. Each rack is a *stacker* (a vertical column of + plate slots); plates enter and leave through one of the device's *nests* + (transfer stations), exposed as the loading trays of the + :class:`AutomatedRetrieval` capability (:attr:`retrieval`). Storage bookkeeping + and the fetch/store operations live on the capability; address a particular + nest with its ``tray_index`` (0-based, defaulting to the first nest). + + Subclasses set :attr:`_model_name` and :attr:`_has_environment_control` (the + latter controls whether the temperature/humidity capabilities are wired). + """ + + _model_name: str = "HighResSampleStorage" + _has_environment_control: bool = True + + def __init__( + self, + name: str, + driver: HighResSampleStorageDriver, + racks: List[PlateCarrier], + nest_locations: List[Coordinate], + size_x: float = 0, + size_y: float = 0, + size_z: float = 0, + rotation: Optional[Rotation] = None, + category: Optional[str] = "plate_store", + model: Optional[str] = None, + ): + """ + Args: + racks: Storage racks; rack *i* maps to device stacker ``i + 1``. + nest_locations: One :class:`Coordinate` per transfer nest (the device has + two). ``nest_locations[i]`` is the location of nest/tray ``i``. + """ + Resource.__init__( + self, + name=name, + size_x=size_x, + size_y=size_y, + size_z=size_z, + rotation=rotation, + category=category, + model=model or self._model_name, + ) + Device.__init__(self, driver=driver) + self.driver: HighResSampleStorageDriver = driver + + self.nests: List[PlateHolder] = [] + for i, location in enumerate(nest_locations): + nest = PlateHolder( + name=f"{name}_nest_{i + 1}", size_x=127.76, size_y=85.48, size_z=0, pedestal_size_z=0 + ) + self.assign_child_resource(nest, location=location) + self.nests.append(nest) + + self._racks = racks + for rack in self._racks: + self.assign_child_resource(rack, location=None) + + self.retrieval = AutomatedRetrieval( + backend=driver.automated_retrieval, racks=self._racks, loading_trays=self.nests + ) + self._capabilities = [self.retrieval] + + if self._has_environment_control: + self.tc = TemperatureController(backend=driver.temperature) + self.humidity = HumidityController(backend=driver.humidity) + self._capabilities = [self.tc, self.humidity, self.retrieval] + + @property + def racks(self) -> List[PlateCarrier]: + return self._racks + + async def setup(self, backend_params: Optional[BackendParams] = None): + await super().setup(backend_params=backend_params) + await self.driver.automated_retrieval.set_racks(self._racks) + + def serialize(self) -> dict: + from pylabrobot.serializer import serialize + + return { + **Device.serialize(self), + **Resource.serialize(self), + "racks": [rack.serialize() for rack in self._racks], + "nest_locations": [serialize(nest.location) for nest in self.nests], + } + + +class TundraStore(_HighResSampleStorage): + """HighRes Biosolutions TundraStore refrigerated plate store.""" + + _model_name = "TundraStore" + + +class SteriStore(_HighResSampleStorage): + """HighRes Biosolutions SteriStore plate store (same API as the TundraStore).""" + + _model_name = "SteriStore" + + +class AmbiStore(_HighResSampleStorage): + """HighRes Biosolutions AmbiStore plate store. + + WORK IN PROGRESS: the AmbiStore is ambient (no refrigeration), so it exposes + only the retrieval capability — no temperature/humidity control. Whether it + has any environment control at all is not yet confirmed against hardware. + """ + + _model_name = "AmbiStore" + _has_environment_control = False + + def __init__(self, *args, **kwargs): + warnings.warn( + "AmbiStore support is a work in progress and unverified against hardware; " + "it currently exposes only the retrieval capability (no environment control).", + stacklevel=2, + ) + super().__init__(*args, **kwargs) diff --git a/pylabrobot/highres/sample_storage/settings.py b/pylabrobot/highres/sample_storage/settings.py new file mode 100644 index 00000000000..6d3ba32d5ef --- /dev/null +++ b/pylabrobot/highres/sample_storage/settings.py @@ -0,0 +1,697 @@ +"""Typed, immutable view of a TundraStore's on-device settings. + +The device exposes its full calibration/configuration via the ``settings`` +command as ``NAME = value`` text. Each key is surfaced here as one explicitly +typed attribute (the device ``NAME`` lower-cased); types are inferred from the +device's own values. A :class:`HighResSampleStorageSettings` is loaded whole from the +device (or a capture) and is frozen once built. +""" + +import warnings +from dataclasses import dataclass, fields +from typing import Dict, Iterable, Tuple + +try: + from typing import Literal +except ImportError: # pragma: no cover + from typing_extensions import Literal # type: ignore + + +# Known TundraStore/SteriStore models, as reported by the device's MACHINE_TYPE +# setting. Extend this (and MachineType) when a new model is encountered. +MachineType = Literal["SteriStore2"] +KNOWN_MACHINE_TYPES: Tuple[str, ...] = ("SteriStore2",) + + +@dataclass(frozen=True) +class HighResSampleStorageSettings: + """All on-device settings, one typed attribute per device key.""" + + product_name: str + product_description: str + + serial_number: str + + machine_type: MachineType + + rest_server_port: int + + syslog_server: str + syslog_level: int + + internal_log_level: int + + carousel_home_speed_fast: float + carousel_home_speed_slow: float + carousel_home_acceleration: float + carousel_velocity: float + carousel_idle_velocity: float + carousel_acceleration: float + carousel_abort_deceleration: float + carousel_jerk: float + carousel_final_drive_jerk: float + carousel_stacker_0_pos: float + carousel_stacker_1_pos: float + carousel_stacker_2_pos: float + carousel_stacker_count: int + carousel_count: int + carousel_calibration_offset: float + + spatula_home_speed_fast: float + spatula_home_speed_slow: float + spatula_home_acceleration: float + spatula_velocity: float + spatula_velocity_with_plate: float + spatula_acceleration: float + spatula_abort_deceleration: float + spatula_jerk: float + spatula_rot_home_speed_fast: float + spatula_rot_home_speed_slow: float + spatula_rot_home_acceleration: float + spatula_rot_velocity: float + spatula_rot_acceleration: float + spatula_rot_abort_deceleration: float + spatula_rot_jerk: float + spatula_rot_zero_pos: float + spatula_rot_stack_pos_0: float + spatula_rot_stack_pos_1: float + spatula_rot_stack_pos_2: float + spatula_rot_nest_1_pos: float + spatula_rot_nest_2_pos: float + spatula_rot_nest_3_pos: float + spatula_rot_nest_4_pos: float + spatula_rot_nest_5_pos: float + spatula_rot_nest_6_pos: float + spatula_rot_nest_7_pos: float + spatula_rot_nest_8_pos: float + spatula_rot_nest_9_pos: float + spatula_rot_nest_10_pos: float + spatula_rot_nest_21_pos: float + spatula_rot_nest_22_pos: float + spatula_rot_nest_23_pos: float + spatula_rot_nest_24_pos: float + spatula_rot_nest_51_pos: float + spatula_rot_nest_52_pos: float + spatula_rot_nest_61_pos: float + spatula_rot_nest_62_pos: float + spatula_rot_nest_63_pos: float + spatula_rot_nest_64_pos: float + spatula_rot_nest_65_pos: float + spatula_rot_nest_66_pos: float + spatula_rot_nest_67_pos: float + spatula_rot_nest_68_pos: float + spatula_rot_nest_69_pos: float + spatula_slide_home_speed_fast: float + spatula_slide_home_speed_slow: float + spatula_slide_home_acceleration: float + spatula_slide_home_offset: float + spatula_slide_velocity: float + spatula_slide_acceleration: float + spatula_slide_abort_deceleration: float + spatula_slide_jerk: float + spatula_slide_in_pos_0: float + spatula_slide_in_pos_1: float + spatula_slide_in_pos_2: float + spatula_slide_nest_1_pos: float + spatula_slide_nest_2_pos: float + spatula_slide_nest_3_pos: float + spatula_slide_nest_4_pos: float + spatula_slide_nest_5_pos: float + spatula_slide_nest_6_pos: float + spatula_slide_nest_7_pos: float + spatula_slide_nest_8_pos: float + spatula_slide_nest_9_pos: float + spatula_slide_nest_10_pos: float + spatula_slide_nest_21_pos: float + spatula_slide_nest_22_pos: float + spatula_slide_nest_23_pos: float + spatula_slide_nest_24_pos: float + spatula_slide_nest_51_pos: float + spatula_slide_nest_52_pos: float + spatula_slide_nest_61_pos: float + spatula_slide_nest_62_pos: float + spatula_slide_nest_63_pos: float + spatula_slide_nest_64_pos: float + spatula_slide_nest_65_pos: float + spatula_slide_nest_66_pos: float + spatula_slide_nest_67_pos: float + spatula_slide_nest_68_pos: float + spatula_slide_nest_69_pos: float + spatula_valve_hold: int + spatula_plate_sensor: int + spatula_plate_release_sensor: int + + inner_user_door_sensor: int + + door_open_sensor_output: int + + nest_count: int + nest_1_height: float + nest_2_height: float + nest_3_height: float + nest_4_height: float + nest_5_height: float + nest_6_height: float + nest_7_height: float + nest_8_height: float + nest_9_height: float + nest_10_height: float + nest_21_height: float + nest_22_height: float + nest_23_height: float + nest_24_height: float + nest_51_height: float + nest_52_height: float + nest_61_height: float + nest_62_height: float + nest_63_height: float + nest_64_height: float + nest_65_height: float + nest_66_height: float + nest_67_height: float + nest_68_height: float + nest_69_height: float + nest_1_style: str + nest_2_style: str + nest_3_style: str + nest_4_style: str + nest_5_style: str + nest_6_style: str + nest_7_style: str + nest_8_style: str + nest_9_style: str + nest_10_style: str + nest_clearance_above: float + nest_clearance_below: float + + handover_nest_clearance_above: float + handover_nest_clearance_below: float + handover_y_rotation_position: float + + conveyor_clearance_above: float + conveyor_clearance_below: float + + static_nest_clearance_above: float + static_nest_clearance_below: float + + io_nest_clearance_above: float + io_nest_clearance_below: float + + nest_1_sense_input: int + nest_2_sense_input: int + nest_3_sense_input: int + nest_4_sense_input: int + nest_5_sense_input: int + nest_6_sense_input: int + nest_7_sense_input: int + nest_8_sense_input: int + nest_9_sense_input: int + nest_10_sense_input: int + + stacker_base_0: float + stacker_base_1: float + stacker_base_2: float + + barcode_base_0: float + barcode_base_1: float + + stacker_clearance_above: float + stacker_clearance_below: float + + barcode_scanner: str + barcode_laser_start: int + barcode_laser_stop: int + barcode_velocity: float + barcode_acceleration: float + barcode_config_itf_enable: str + barcode_config_itf_status: str + barcode_config_itf_length_1: int + barcode_config_itf_length_2: int + barcode_config_itf_range: str + + plate_hold_settle_time: int + + door_0_position: float + door_height: float + door_overlap_negative: float + door_overlap_positive: float + door_gasket_valve: int + + big_door_valve: int + + door_1_valve: int + door_2_valve: int + door_3_valve: int + door_4_valve: int + door_5_valve: int + door_6_valve: int + door_7_valve: int + door_8_valve: int + door_1_open_sensor: int + door_2_open_sensor: int + door_3_open_sensor: int + door_4_open_sensor: int + door_5_open_sensor: int + door_6_open_sensor: int + door_7_open_sensor: int + door_8_open_sensor: int + door_1_close_sensor: int + door_2_close_sensor: int + door_3_close_sensor: int + door_4_close_sensor: int + door_5_close_sensor: int + door_6_close_sensor: int + door_7_close_sensor: int + door_8_close_sensor: int + door_ri_open_sensor: int + door_ri_close_sensor: int + + gasket_deflate_delay_ms: int + gasket_inflate_delay_ms: int + + big_door_open_delay_ms: int + big_door_close_delay_ms: int + + door_open_delay_ms: int + door_close_delay_ms: int + door_open_signal_active_level: int + + vac_1_enable: int + vac_2_enable: int + vac_1_purge: int + vac_2_purge: int + + lift_1_enable: int + lift_2_enable: int + + nest_1_sense: int + nest_2_sense: int + + vac_3_enable: int + vac_4_enable: int + vac_3_purge: int + vac_4_purge: int + + lift_3_enable: int + lift_4_enable: int + + nest_3_sense: int + nest_4_sense: int + + vac_5_enable: int + vac_6_enable: int + vac_5_purge: int + vac_6_purge: int + + lift_5_enable: int + lift_6_enable: int + + nest_5_sense: int + nest_6_sense: int + + vac_7_enable: int + vac_8_enable: int + vac_7_purge: int + vac_8_purge: int + + lift_7_enable: int + lift_8_enable: int + + nest_7_sense: int + nest_8_sense: int + + active_hotels: int + + nest_rot_home_speed_fast: int + nest_rot_home_speed_slow: int + nest_rot_home_acceleration: int + nest_rot_velocity: int + nest_rot_acceleration: int + nest_rot_abort_deceleration: int + nest_rot_jerk: int + nest_rot_zero_pos: float + + microspin_door_closed: float + microspin_door_open: float + microspin_spindle_home_offset: float + microspin_bucket_radius_m: float + microspin_spindle_counts_per_rev: int + microspin_spindle_position_window: int + microspin_door_velocity: int + microspin_door_home_velocity: int + microspin_door_accel: int + microspin_door_abort_decel: int + microspin_door_jerk: int + microspin_spindle_home_velocity: int + microspin_spindle_velocity: int + microspin_spindle_accel: float + microspin_spindle_decel: float + microspin_spindle_slow_accel: float + microspin_spindle_slow_decel: float + microspin_spindle_abort_decel: float + microspin_spindle_jerk: int + microspin_spindle_max_accel: float + microspin_spindle_max_decel: float + microspin_idle_spindle_threshold: float + microspin_bucket_rise_rpm: int + + pico_rot_home_speed_fast: int + pico_rot_home_speed_slow: int + pico_rot_home_acceleration: int + pico_rot_velocity: int + pico_rot_acceleration: int + pico_rot_abort_deceleration: int + pico_rot_jerk: int + pico_rot_zero_pos: int + pico_stacker_count: int + + def_plate_height: float + def_stack_height: float + def_plate_thickness: float + + jiggle_count: int + jiggle_size: int + + has_lock_sensor: str + + lock_sensor_input: int + + carousel_max_position: int + carousel_max_velocity: int + carousel_max_acceleration: int + carousel_max_deceleration: int + carousel_max_jerk: int + carousel_home_pos_offset: int + carousel_home_neg_offset: float + carousel_homing_speed: int + carousel_home_fast: int + carousel_home_slow: int + carousel_home_accel: int + carousel_stacker_width_default: float + carousel_small_flag_width: float + carousel_large_flag_check_distance: float + carousel_large_flag_width: float + carousel_ring_numbering: str + + effectuator_extended_position: float + effectuator_max_position: float + effectuator_lock_position: float + effectuator_unlock_position: int + effectuator_max_velocity: int + effectuator_max_acceleration: int + effectuator_abort_deceleration: int + effectuator_max_jerk: int + effectuator_home_offset: int + effectuator_home_fast: int + effectuator_home_slow: int + effectuator_home_accel: int + + spatula_max_position: float + spatula_max_velocity: int + spatula_measure_velocity: int + spatula_max_acceleration: int + spatula_max_jerk: int + spatula_home_offset: int + spatula_home_fast: int + spatula_home_slow: int + spatula_home_accel: int + spatula_nest_offset: float + spatula_measurement_tolerance: float + spatula_beam_break_height: float + + max_transparency_width_um: int + + spatula_lock_position: float + spatula_base_position: float + + barcode_max_position: float + barcode_max_velocity: int + barcode_max_move_velocity: int + barcode_abort_deceleration: int + barcode_max_move_jerk: int + barcode_max_read_velocity: int + barcode_home_offset: int + barcode_home_fast: int + barcode_home_slow: int + barcode_home_accel: int + + spatula_slide_safe_position: float + spatula_slide_barcode_position: float + + nest_safe_rotation_clearance: float + + limit_x_min: float + limit_x_max: float + limit_y_min: float + limit_y_max: float + limit_z_min: float + limit_z_max: float + limit_theta_min: float + limit_theta_max: float + limit_g_min: float + limit_g_max: float + limit_barcode_min: float + limit_barcode_max: float + + tundra_door_cycle_active: str + tundra_door_cycle_time_sec: int + tundra_door_cycle_open_time_sec: int + + barcode_height_adjust: float + + tundra_outer_door_cycle_active: str + tundra_outer_door_cycle_time_sec: int + tundra_outer_door_cycle_open_time_sec: int + + spatula_door_clearance_below: float + spatula_door_clearance_above: float + + weigh_cell_door_output: int + weigh_cell_door_input_status_0: int + weigh_cell_door_input_status_1: int + weigh_cell_door_input_status_2: int + weigh_cell_led_red_output: int + weigh_cell_led_green_output: int + weigh_cell_led_blue_output: int + + axis_x_home_speed_fast: float + axis_x_home_speed_slow: float + axis_x_home_acceleration: float + axis_x_home_offset: float + axis_x2_home_offset: float + + tray_to_gantry_0_distance: float + + axis_x2_calibration_adjustment: float + axis_x_calibration_pos: float + axis_x_velocity: float + axis_x_acceleration: float + axis_x_abort_deceleration: float + axis_x_jerk: float + axis_z_home_speed_fast: float + axis_z_home_speed_slow: float + axis_z_home_acceleration: float + axis_z_home_offset: float + axis_z_home_offset_hardstop: float + axis_z_home_hardstop_current_ma: int + axis_z_home_hardstop_current_time_ms: int + axis_z_velocity: float + axis_z_acceleration: float + axis_z_abort_deceleration: float + axis_z_jerk: float + axis_barcode_home_speed_fast: float + axis_barcode_home_speed_slow: float + axis_barcode_home_acceleration: float + axis_barcode_home_offset: float + axis_barcode_velocity: float + axis_barcode_acceleration: float + axis_barcode_abort_deceleration: float + axis_barcode_jerk: float + axis_gripper_home_speed_fast: float + axis_gripper_home_speed_slow: float + axis_gripper_home_acceleration: float + axis_gripper_home_offset: float + axis_gripper_home_offset_hardstop: float + axis_gripper_home_hardstop_current_ma: int + axis_gripper_home_hardstop_current_time_ms: int + axis_gripper_close_position: float + axis_gripper_velocity: float + axis_gripper_acceleration: float + axis_gripper_abort_deceleration: float + axis_gripper_jerk: float + + height_detect_base: float + height_detect_positive_adjustment: float + height_detect_negative_adjustment: float + height_detect_enable_address: int + + barcode_input_number: int + + height_detect_input_number: int + + stacker_code_clearance_above: float + stacker_code_clearance_below: float + stacker_code_height: float + + barcode_fixture_height: float + barcode_fixture_groove_height: float + barcode_ideal_stacker_tier_1: float + barcode_ideal_stacker_tier_2: float + + muting_bank_input_1: int + muting_bank_input_2: int + muting_bank_input_3: int + muting_input_1: int + muting_input_2: int + + tool_head_sel_0: int + tool_head_sel_1: int + tool_head_addr_0: int + tool_head_addr_1: int + tool_head_addr_2: int + tool_head_addr_3: int + + ion_bar_air_output: int + ion_bar_power_output: int + + busybox_mode: str + + microserve_bus_voltage_threshold: float + microserve_recover_after_estop: str + + randomserve_bus_voltage_threshold: float + randomserve_recover_after_estop: str + + psp_packet_delay: int + + microspin_spindle_voltage_delay: int + microspin_bus_voltage_threshold: float + + home_trays_to_hardstop: str + + dc_out_1_default_on: str + dc_out_2_default_on: str + dc_out_3_default_on: str + + lid_discard_drop_wait_time_ms: int + + lidvalet_plate_dropped_threshold: int + lidvalet_hold_time_after_unlid: int + lidvalet_purge_time_ms: int + lidvalet_drop_down_time_ms: int + lidvalet_drop_up_time_ms: int + lidvalet_pickup_wait_ms: int + + disable_blink_function: str + + oled_blink_time_ms: int + + suppress_copley_debug_statements: str + + prime_waste_chute_installed: str + prime_waste_chute_position: str + + plate_sensor_high_is_plate_present: str + + stacker_1_speed_multiplier: float + stacker_2_speed_multiplier: float + stacker_3_speed_multiplier: float + stacker_4_speed_multiplier: float + stacker_5_speed_multiplier: float + stacker_6_speed_multiplier: float + stacker_7_speed_multiplier: float + stacker_8_speed_multiplier: float + stacker_9_speed_multiplier: float + stacker_10_speed_multiplier: float + stacker_11_speed_multiplier: float + stacker_12_speed_multiplier: float + stacker_13_speed_multiplier: float + stacker_14_speed_multiplier: float + stacker_15_speed_multiplier: float + stacker_16_speed_multiplier: float + stacker_17_speed_multiplier: float + stacker_18_speed_multiplier: float + stacker_19_speed_multiplier: float + stacker_20_speed_multiplier: float + stacker_21_speed_multiplier: float + stacker_22_speed_multiplier: float + stacker_23_speed_multiplier: float + stacker_24_speed_multiplier: float + stacker_25_speed_multiplier: float + stacker_26_speed_multiplier: float + stacker_27_speed_multiplier: float + stacker_28_speed_multiplier: float + stacker_1_clearance_above_offset: float + stacker_2_clearance_above_offset: float + stacker_3_clearance_above_offset: float + stacker_4_clearance_above_offset: float + stacker_5_clearance_above_offset: float + stacker_6_clearance_above_offset: float + stacker_7_clearance_above_offset: float + stacker_8_clearance_above_offset: float + stacker_9_clearance_above_offset: float + stacker_10_clearance_above_offset: float + stacker_11_clearance_above_offset: float + stacker_12_clearance_above_offset: float + stacker_13_clearance_above_offset: float + stacker_14_clearance_above_offset: float + stacker_15_clearance_above_offset: float + stacker_16_clearance_above_offset: float + stacker_17_clearance_above_offset: float + stacker_18_clearance_above_offset: float + stacker_19_clearance_above_offset: float + stacker_20_clearance_above_offset: float + stacker_21_clearance_above_offset: float + stacker_22_clearance_above_offset: float + stacker_23_clearance_above_offset: float + stacker_24_clearance_above_offset: float + stacker_25_clearance_above_offset: float + stacker_26_clearance_above_offset: float + stacker_27_clearance_above_offset: float + stacker_28_clearance_above_offset: float + + mfg_door_time_low_limit_ms: int + mfg_door_time_high_limit_ms: int + + automation_door_time_ms: int + + carousel_home_adj_low_limit: int + carousel_home_adj_high_limit: int + + store_calibration_fixture_y_distance: float + store_y_teach_minimum: float + store_y_teach_maximum: float + + lidvalet_wait_for_lift_rise_ms: int + + @classmethod + def from_lines(cls, lines: Iterable[str]) -> "HighResSampleStorageSettings": + """Build from the device's ``settings`` output (``NAME = value`` lines).""" + data: Dict[str, str] = {} + for line in lines: + if "=" in line: + key, _, value = line.partition("=") + data[key.strip()] = value.strip() + + values = {} + missing = [] + for f in fields(cls): + key = f.name.upper() + if key not in data: + missing.append(key) + continue + values[f.name] = f.type(data[key]) if f.type in (int, float) else data[key] + if missing: + raise ValueError( + f"settings is missing {len(missing)} expected key(s): " + + ", ".join(missing[:8]) + + ("..." if len(missing) > 8 else "") + ) + machine_type = values["machine_type"] + if machine_type not in KNOWN_MACHINE_TYPES: + warnings.warn( + f"unknown TundraStore model {machine_type!r}; please contribute it to " + "MachineType / KNOWN_MACHINE_TYPES", + stacklevel=2, + ) + return cls(**values) diff --git a/pylabrobot/highres/sample_storage/tests/__init__.py b/pylabrobot/highres/sample_storage/tests/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/pylabrobot/highres/sample_storage/tests/driver_tests.py b/pylabrobot/highres/sample_storage/tests/driver_tests.py new file mode 100644 index 00000000000..f4fc583e816 --- /dev/null +++ b/pylabrobot/highres/sample_storage/tests/driver_tests.py @@ -0,0 +1,175 @@ +import unittest +from typing import Dict, List + +from pylabrobot.highres.sample_storage.driver import HighResSampleStorageDriver +from pylabrobot.highres.sample_storage.errors import HighResSampleStorageError + +# Real responses captured from a TundraStore (firmware 3.0.0.119, serial +# HRB-2209-35148) over the port-1000 remote-control server. +CAPTURES: Dict[str, List[str]] = { + "version": [ + "ACK! version 1", + "Product Name: SteriStore", + "Serial Number: HRB-2209-35148", + "libcommon Version: 1.1.0.119", + "libts7600 Version: 1.0.0.119", + "Firmware Version: 3.0.0.119", + "Firmware Build: D9BE232A", + "OK! version 1", + ], + "homedstatus": ["ACK! homedstatus 5", "not homed", "OK! homedstatus 5"], + "doorstatus": [ + "ACK! doorstatus 17", + "User Door: CLOSED", + "RI: CLOSED", + "SEAL: CLOSING", + "RO1: CLOSING", + "RO2: CLOSED", + "RO3: CLOSED", + "RO4: CLOSED", + "OK! doorstatus 17", + ], + "platestatus": ["ACK! platestatus 9", "NO_PLATE", "OK! platestatus 9"], + "neststatus": ["ACK! neststatus 11", "1: CLEAR", "2: CLEAR", "OK! neststatus 11"], + "environmentstatus": [ + "ACK! environmentstatus 31", + "TEMP:21.9/22.0/100.0", + "RH:54.7/0.0/-100.0", + "CO2:0.0/5.0/100.0", + "O2:20.5/5.0/100.0", + "TANK1:135.0:", + "TANK2:135.0:", + "OK! environmentstatus 31", + ], + "getstackerdimensions": [ + "ACK! getstackerdimensions 35", + "1: 0.000 28.940 0", + "2: 0.000 22.867 24", + "13: 0.000 22.867 0", + "OK! getstackerdimensions 35", + ], + # The home command failed because the pneumatic doors could not close (no air). + "home": [ + "ACK! home 13", + "Error 1: (00:32:44) 13: Unable to close all doors", + "ERROR! home 13", + ], +} + + +class FakeSocket: + """Replays scripted line responses keyed by the command written to it.""" + + def __init__(self, captures: Dict[str, List[str]]): + self.captures = captures + self.written: List[str] = [] + self._queue: List[str] = [] + + async def setup(self): + pass + + async def stop(self): + pass + + async def write(self, data: bytes, timeout=None): + command = data.decode("ascii").rstrip("\r\n") + self.written.append(command) + self._queue = list(self.captures[command]) + + async def readuntil(self, separator: bytes = b"\n", timeout=None) -> bytes: + return self._queue.pop(0).encode("ascii") + b"\r\n" + + +class HighResSampleStorageBackendTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.driver = HighResSampleStorageDriver(host="10.253.253.253") + self.socket = FakeSocket(CAPTURES) + self.driver.io = self.socket # type: ignore[assignment] + self.retrieval = self.driver.automated_retrieval + + async def test_send_command_strips_ack_and_completion(self): + data = await self.driver.send_command("neststatus") + self.assertEqual(data, ["1: CLEAR", "2: CLEAR"]) + self.assertEqual(self.socket.written, ["neststatus"]) + + async def test_version(self): + v = await self.driver.request_version() + self.assertEqual(v.product_name, "SteriStore") + self.assertEqual(v.serial_number, "HRB-2209-35148") + self.assertEqual(v.firmware_version, "3.0.0.119") + self.assertEqual(v.firmware_build, "D9BE232A") + + async def test_request_is_homed(self): + self.assertFalse(await self.retrieval.request_is_homed()) + + async def test_door_status(self): + doors = await self.retrieval.request_door_status() + self.assertEqual(doors["User Door"], "closed") + self.assertEqual(doors["SEAL"], "closing") + self.assertEqual(doors["RO1"], "closing") + self.assertFalse(all(state == "closed" for state in doors.values())) + + async def test_nest_status(self): + nests = await self.retrieval.request_nest_status() + self.assertEqual(nests, {1: "clear", 2: "clear"}) + + async def test_plate_on_spatula(self): + self.assertFalse(await self.retrieval.request_spatula_is_holding()) + + async def test_environment_parsing(self): + env = await self.driver.request_environment() + self.assertAlmostEqual(env["TEMP"].current, 21.9) + self.assertIsNotNone(env["TEMP"].setpoint) + assert env["TEMP"].setpoint is not None # narrow for type checker + self.assertAlmostEqual(env["TEMP"].setpoint, 22.0) + self.assertAlmostEqual(env["O2"].current, 20.5) + # Sensor-only channel: current value, no setpoint. + self.assertAlmostEqual(env["TANK1"].current, 135.0) + self.assertIsNone(env["TANK1"].setpoint) + + async def test_temperature_capability_reads_temp_channel(self): + self.assertAlmostEqual(await self.driver.temperature.request_current_temperature(), 21.9) + self.assertTrue(self.driver.temperature.supports_active_cooling) + + async def test_humidity_capability_reads_rh_as_fraction(self): + self.assertAlmostEqual(await self.driver.humidity.request_current_humidity(), 0.547) + self.assertFalse(self.driver.humidity.supports_humidity_control) + + async def test_stacker_dimensions(self): + dims = await self.retrieval.request_stacker_dimensions() + self.assertEqual(dims[0].stacker, 1) + self.assertEqual(dims[0].slot_count, 0) + self.assertEqual(dims[1].stacker, 2) + self.assertAlmostEqual(dims[1].slot_height, 22.867) + self.assertEqual(dims[1].slot_count, 24) + + async def test_home_error_raises_with_stack_detail(self): + with self.assertRaises(HighResSampleStorageError) as ctx: + await self.retrieval.home() + self.assertIn("Unable to close all doors", str(ctx.exception)) + self.assertEqual(ctx.exception.command, "home") + + async def test_pick_formats_command(self): + self.socket.captures["pick 3 12 1"] = ["ACK! pick 3 12 1 99", "OK! pick 3 12 1 99"] + await self.retrieval.pick(3, 12, 1) + self.assertEqual(self.socket.written, ["pick 3 12 1"]) + + def test_tray_maps_to_nest(self): + # 0-based capability tray -> 1-based device nest; None uses the default. + self.assertEqual(self.retrieval._nest_for_tray(None), 1) + self.assertEqual(self.retrieval._nest_for_tray(0), 1) + self.assertEqual(self.retrieval._nest_for_tray(1), 2) + with self.assertRaises(ValueError): + self.retrieval._nest_for_tray(2) + + def test_default_tray_follows_loading_tray_nest(self): + driver = HighResSampleStorageDriver(host="10.253.253.253", loading_tray_nest=2) + self.assertEqual(driver.automated_retrieval._nest_for_tray(None), 2) + + async def test_set_humidity_unsupported(self): + with self.assertRaises(NotImplementedError): + await self.driver.humidity.set_humidity(0.5) + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/highres/sample_storage/tests/recovery_tests.py b/pylabrobot/highres/sample_storage/tests/recovery_tests.py new file mode 100644 index 00000000000..b1b6698fde8 --- /dev/null +++ b/pylabrobot/highres/sample_storage/tests/recovery_tests.py @@ -0,0 +1,182 @@ +import unittest +from typing import List + +from pylabrobot.highres.sample_storage.driver import HighResSampleStorageDriver +from pylabrobot.highres.sample_storage.errors import HighResSampleStorageFault, PlateNotFoundError + + +def _ok(command: str, cid: int) -> List[str]: + return [f"ACK! {command} {cid}", f"OK! {command} {cid}"] + + +class ScriptedSocket: + """Replays an ordered (expected_command, response_lines) script, asserting the + exact command sequence — used to verify multi-step recovery flows.""" + + def __init__(self, script): + self.script = list(script) + self.i = 0 + self.commands: List[str] = [] + self._queue: List[str] = [] + + async def setup(self): + pass + + async def stop(self): + pass + + async def write(self, data: bytes, timeout=None): + command = data.decode("ascii").rstrip("\r\n") + self.commands.append(command) + expected, lines = self.script[self.i] + self.i += 1 + assert command == expected, f"expected {expected!r}, got {command!r}" + self._queue = list(lines) + + async def readuntil(self, separator: bytes = b"\n", timeout=None) -> bytes: + return self._queue.pop(0).encode("ascii") + b"\r\n" + + +class HighResSampleStorageRecoveryTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.driver = HighResSampleStorageDriver(host="10.253.253.253") + self.retrieval = self.driver.automated_retrieval + + async def test_empty_slot_pick_raises_plate_not_found_and_stays_homed(self): + # The store reports "No plate detected" and stays homed (graceful empty). + empty = [ + "ACK! pick 5 12 1 50", + "Error 1: (00:00:01) 50: No plate detected", + "ERROR! pick 5 12 1 50", + ] + sock = ScriptedSocket( + [ + ("pick 5 12 1", empty), + ("homedstatus", ["ACK! homedstatus 51", "homed", "OK! homedstatus 51"]), + ] + ) + self.driver.io = sock # type: ignore[assignment] + with self.assertRaises(PlateNotFoundError): + await self.retrieval.pick(5, 12, 1) + # classified by state, no recovery motion issued + self.assertEqual(sock.commands, ["pick 5 12 1", "homedstatus"]) + + async def test_top_slot_stuck_raises_fault_despite_homed_lie(self): + # Empty TOP slot: "No plate detected" but the spatula is left extended and + # the firmware reports "unsafe for rotation" while homedstatus still says + # homed. The "unsafe" signal must win -> HighResSampleStorageFault (no homedstatus + # query needed, the signature short-circuits). + stuck = [ + "ACK! pick 5 24 1 60", + "Error 1: 60: No plate detected", + "Error 2: 60: Z height is unsafe for rotation, check machine", + "ERROR! pick 5 24 1 60", + ] + sock = ScriptedSocket([("pick 5 24 1", stuck)]) + self.driver.io = sock # type: ignore[assignment] + with self.assertRaises(HighResSampleStorageFault): + await self.retrieval.pick(5, 24, 1) + self.assertEqual(sock.commands, ["pick 5 24 1"]) + + async def test_dehomed_pick_raises_fault(self): + # An error with no "unsafe" signature but the machine reports unhomed. + fault = [ + "ACK! pick 5 24 1 70", + "Error 1: 70: motor fault", + "ERROR! pick 5 24 1 70", + ] + sock = ScriptedSocket( + [ + ("pick 5 24 1", fault), + ("homedstatus", ["ACK! homedstatus 71", "not homed", "OK! homedstatus 71"]), + ] + ) + self.driver.io = sock # type: ignore[assignment] + with self.assertRaises(HighResSampleStorageFault): + await self.retrieval.pick(5, 24, 1) + self.assertEqual(sock.commands, ["pick 5 24 1", "homedstatus"]) + + def _homed(self, cid: int) -> List[str]: + return [f"ACK! homedstatus {cid}", "homed", f"OK! homedstatus {cid}"] + + def _status(self, cid: int, y: float) -> List[str]: + return [ + f"ACK! status {cid}", + "Carousel: 0.0", + f"Y axis: {y}", + "Z axis: 0.0", + f"OK! status {cid}", + ] + + async def test_request_is_parked_catches_the_homed_lie(self): + # homedstatus says homed, but the spatula is stuck extended (Y=256) -> NOT parked. + sock = ScriptedSocket([("homedstatus", self._homed(1)), ("status", self._status(2, 255.9999))]) + self.driver.io = sock # type: ignore[assignment] + self.assertFalse(await self.retrieval.request_is_parked()) + + async def test_recover_always_retracts_and_rehomes(self): + # recover() must issue retract+home even when homedstatus already says homed + # (the lie), then confirm parked via the slide position. + sock = ScriptedSocket( + [ + ("enable", _ok("enable", 1)), + ("spatulaout", _ok("spatulaout", 2)), + ("home", _ok("home", 3)), + ("homedstatus", self._homed(4)), + ("status", self._status(5, 0.0)), + ] + ) + self.driver.io = sock # type: ignore[assignment] + self.assertTrue(await self.retrieval.recover()) + self.assertEqual(sock.commands, ["enable", "spatulaout", "home", "homedstatus", "status"]) + + async def test_recover_retries_until_parked(self): + # First round still reads extended (homed-lie); recover retries and succeeds. + sock = ScriptedSocket( + [ + ("enable", _ok("enable", 1)), + ("spatulaout", _ok("spatulaout", 2)), + ("home", _ok("home", 3)), + ("homedstatus", self._homed(4)), + ("status", self._status(5, 255.9999)), # still extended -> retry + ("enable", _ok("enable", 6)), + ("spatulaout", _ok("spatulaout", 7)), + ("home", _ok("home", 8)), + ("homedstatus", self._homed(9)), + ("status", self._status(10, 0.0)), # now retracted + ] + ) + self.driver.io = sock # type: ignore[assignment] + self.assertTrue(await self.retrieval.recover()) + + async def test_place_default_leaves_doors_sealed(self): + sock = ScriptedSocket([("place 2 5 1", _ok("place 2 5 1", 1))]) + self.driver.io = sock # type: ignore[assignment] + await self.retrieval.place(2, 5, 1) # close_door=True default + self.assertEqual(sock.commands, ["place 2 5 1"]) + + async def test_place_close_door_false_reopens(self): + sock = ScriptedSocket( + [ + ("place 2 5 1", _ok("place 2 5 1", 1)), + ("openalldoors", _ok("openalldoors", 2)), + ] + ) + self.driver.io = sock # type: ignore[assignment] + await self.retrieval.place(2, 5, 1, close_door=False) + self.assertEqual(sock.commands, ["place 2 5 1", "openalldoors"]) + + async def test_pick_close_door_false_reopens(self): + sock = ScriptedSocket( + [ + ("pick 2 5 1", _ok("pick 2 5 1", 1)), + ("openalldoors", _ok("openalldoors", 2)), + ] + ) + self.driver.io = sock # type: ignore[assignment] + await self.retrieval.pick(2, 5, 1, close_door=False) + self.assertEqual(sock.commands, ["pick 2 5 1", "openalldoors"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/highres/sample_storage/types.py b/pylabrobot/highres/sample_storage/types.py new file mode 100644 index 00000000000..c5a7a7e64ff --- /dev/null +++ b/pylabrobot/highres/sample_storage/types.py @@ -0,0 +1,51 @@ +from dataclasses import dataclass +from typing import Dict, Optional, Tuple + +try: + from typing import Literal +except ImportError: # pragma: no cover + from typing_extensions import Literal # type: ignore + + +# State of a single TundraStore door, as reported by ``doorstatus``. +DoorState = Literal["open", "closed", "opening", "closing", "unknown"] +DOOR_STATES: Tuple[DoorState, ...] = ("open", "closed", "opening", "closing", "unknown") + +# State of a transfer nest, as reported by ``neststatus``. +NestState = Literal["clear", "occupied", "unknown"] +NEST_STATES: Tuple[NestState, ...] = ("clear", "occupied", "unknown") + + +@dataclass +class VersionInfo: + """Parsed output of the ``version`` command.""" + + product_name: Optional[str] + serial_number: Optional[str] + firmware_version: Optional[str] + firmware_build: Optional[str] + raw: Dict[str, str] + + +@dataclass +class EnvironmentParameter: + """One row of ``environmentstatus`` (e.g. TEMP, RH, CO2, O2). + + The device reports ``NAME:current/setpoint/limit``. ``setpoint``/``limit`` are + ``None`` for sensor-only channels (e.g. the gas tank pressures). + """ + + name: str + current: float + setpoint: Optional[float] = None + limit: Optional[float] = None + + +@dataclass +class StackerDimensions: + """One stacker's geometry, from ``getstackerdimensions``.""" + + stacker: int + zero_offset: float + slot_height: float + slot_count: int diff --git a/pylabrobot/liconic/backend.py b/pylabrobot/liconic/backend.py index ddd0cbcd581..66569be0880 100644 --- a/pylabrobot/liconic/backend.py +++ b/pylabrobot/liconic/backend.py @@ -149,7 +149,11 @@ async def set_racks(self, racks: List[PlateCarrier]): # -- AutomatedRetrievalBackend -- - async def fetch_plate_to_loading_tray(self, plate: Plate): + async def fetch_plate_to_loading_tray(self, plate: Plate, tray_index: Optional[int] = None): + if tray_index not in (None, 0): + raise ValueError( + f"This device has a single loading tray; got tray_index={tray_index}. Use None or 0." + ) site = plate.parent assert isinstance(site, PlateHolder), "Plate not in storage" @@ -172,7 +176,11 @@ async def fetch_plate_to_loading_tray(self, plate: Plate): n, ) - async def store_plate(self, plate: Plate, site: PlateHolder): + async def store_plate(self, plate: Plate, site: PlateHolder, tray_index: Optional[int] = None): + if tray_index not in (None, 0): + raise ValueError( + f"This device has a single loading tray; got tray_index={tray_index}. Use None or 0." + ) m, n = self._site_to_m_n(site) step_size, pos_num = self._carrier_to_steps_pos(site) diff --git a/pylabrobot/liconic/liconic.py b/pylabrobot/liconic/liconic.py index 3e7955ea602..1eb60e43eec 100644 --- a/pylabrobot/liconic/liconic.py +++ b/pylabrobot/liconic/liconic.py @@ -64,7 +64,7 @@ def __init__( self.assign_child_resource(rack, location=None) self.retrieval = AutomatedRetrieval( - backend=backend, racks=self._racks, loading_tray=self.loading_tray + backend=backend, racks=self._racks, loading_trays=[self.loading_tray] ) self.tc = ( TemperatureController(backend=backend) if liconic_model.has_temperature_control else None diff --git a/pylabrobot/thermo_fisher/cytomat/backend.py b/pylabrobot/thermo_fisher/cytomat/backend.py index f6af01f4695..5284ab19a98 100644 --- a/pylabrobot/thermo_fisher/cytomat/backend.py +++ b/pylabrobot/thermo_fisher/cytomat/backend.py @@ -122,7 +122,11 @@ async def set_racks(self, racks: List[PlateCarrier]): # -- AutomatedRetrievalBackend -- - async def fetch_plate_to_loading_tray(self, plate: Plate): + async def fetch_plate_to_loading_tray(self, plate: Plate, tray_index: Optional[int] = None): + if tray_index not in (None, 0): + raise ValueError( + f"This device has a single loading tray; got tray_index={tray_index}. Use None or 0." + ) logger.info( "[Cytomat %s %s] fetch plate to loading tray: plate='%s'", self.model.value, @@ -133,7 +137,11 @@ async def fetch_plate_to_loading_tray(self, plate: Plate): assert isinstance(site, PlateHolder) await self.action_storage_to_transfer(site) - async def store_plate(self, plate: Plate, site: PlateHolder): + async def store_plate(self, plate: Plate, site: PlateHolder, tray_index: Optional[int] = None): + if tray_index not in (None, 0): + raise ValueError( + f"This device has a single loading tray; got tray_index={tray_index}. Use None or 0." + ) logger.info( "[Cytomat %s %s] store plate: plate='%s', site='%s'", self.model.value, diff --git a/pylabrobot/thermo_fisher/cytomat/cytomat.py b/pylabrobot/thermo_fisher/cytomat/cytomat.py index 9c7c0071c1f..a6d5dcc724b 100644 --- a/pylabrobot/thermo_fisher/cytomat/cytomat.py +++ b/pylabrobot/thermo_fisher/cytomat/cytomat.py @@ -66,7 +66,7 @@ def __init__( self.assign_child_resource(rack, location=None) self.retrieval = AutomatedRetrieval( - backend=driver, racks=self._racks, loading_tray=self.loading_tray + backend=driver, racks=self._racks, loading_trays=[self.loading_tray] ) self.tc = TemperatureController(backend=driver) self.humidity = HumidityController(backend=driver) diff --git a/pylabrobot/thermo_fisher/cytomat/heraeus_backend.py b/pylabrobot/thermo_fisher/cytomat/heraeus_backend.py index f9a07687c68..4fd00785563 100644 --- a/pylabrobot/thermo_fisher/cytomat/heraeus_backend.py +++ b/pylabrobot/thermo_fisher/cytomat/heraeus_backend.py @@ -121,7 +121,11 @@ async def set_racks(self, racks: List[PlateCarrier]): # -- AutomatedRetrievalBackend -- - async def fetch_plate_to_loading_tray(self, plate: Plate): + async def fetch_plate_to_loading_tray(self, plate: Plate, tray_index: Optional[int] = None): + if tray_index not in (None, 0): + raise ValueError( + f"This device has a single loading tray; got tray_index={tray_index}. Use None or 0." + ) logger.info("[Heraeus %s] fetch plate to loading tray: plate='%s'", self.io.port, plate.name) site = plate.parent assert isinstance(site, PlateHolder), "Plate not in storage" @@ -132,7 +136,11 @@ async def fetch_plate_to_loading_tray(self, plate: Plate): await self._wait_ready() await self._send_command("ST 1903") - async def store_plate(self, plate: Plate, site: PlateHolder): + async def store_plate(self, plate: Plate, site: PlateHolder, tray_index: Optional[int] = None): + if tray_index not in (None, 0): + raise ValueError( + f"This device has a single loading tray; got tray_index={tray_index}. Use None or 0." + ) logger.info( "[Heraeus %s] store plate: plate='%s', site='%s'", self.io.port, plate.name, site.name )