diff --git a/pylabrobot/hamilton/liquid_handlers/star/driver.py b/pylabrobot/hamilton/liquid_handlers/star/driver.py index 860ffafe11e..a684c79a1e5 100644 --- a/pylabrobot/hamilton/liquid_handlers/star/driver.py +++ b/pylabrobot/hamilton/liquid_handlers/star/driver.py @@ -148,8 +148,8 @@ class _FirmwareLock: — but you never have two X0, two H0, etc. at once. Modules without a dedicated mutex are gated but not per-module serialized. - Read-only request commands (``R*``) are not coordinated here at all — they take no lock - and run fully in parallel (see ``STARDriver.send_command``). + Read-only request (``R*``) and query (``Q*``) commands are not coordinated here at all — + they take no lock and run fully in parallel (see ``STARDriver.send_command``). The first slave-module command acquires the exclusive lock and the last one releases it, so a C0 command simply takes the same lock and automatically waits the slaves out. @@ -268,13 +268,18 @@ def __init__( async def send_command(self, module: str, command: str, *args, **kwargs): """Send a firmware command under the firmware lock. - Request commands (command starting with "R") are read-only on every module, so they - take no lock and run fully in parallel. A C0 master command runs exclusively — nothing - else is in flight. Every other (slave-module) command blocks the C0 master, overlaps - commands on *other* modules (so an X0 X-arm move, an H0 head move, and Px channel - commands can run together), but serializes against other commands on its *own* module. + Request ("R*") and query ("Q*") commands are read-only on every module, so they take no + lock and run fully in parallel — this lets a query (e.g. a Px:QN TADM-buffer read) run + concurrently with an in-flight C0 master command such as an aspiration. A C0 master + command runs exclusively — nothing else is in flight. Every other (slave-module) command + blocks the C0 master, overlaps commands on *other* modules (so an X0 X-arm move, an H0 + head move, and Px channel commands can run together), but serializes against other + commands on its *own* module. + + Caveat: C0:QS with an ``on`` parameter (cover.py reset_output) is a write, not a query; + it is rare and never issued during pipetting, so it is treated lock-free with the rest. """ - if command.startswith("R"): + if command.startswith(("R", "Q")): return await super().send_command(module, command, *args, **kwargs) if module == "C0": diff --git a/pylabrobot/hamilton/liquid_handlers/star/pip_backend.py b/pylabrobot/hamilton/liquid_handlers/star/pip_backend.py index 4482d05f4b5..de73bb08ab3 100644 --- a/pylabrobot/hamilton/liquid_handlers/star/pip_backend.py +++ b/pylabrobot/hamilton/liquid_handlers/star/pip_backend.py @@ -6,7 +6,17 @@ import logging from contextlib import contextmanager from dataclasses import dataclass -from typing import TYPE_CHECKING, Dict, List, Literal, Optional, Sequence, Tuple, Union +from typing import ( + TYPE_CHECKING, + Dict, + List, + Literal, + Optional, + Protocol, + Sequence, + Tuple, + Union, +) from pylabrobot.capabilities.capability import BackendParams from pylabrobot.capabilities.liquid_handling.pip_backend import PIPBackend @@ -27,7 +37,7 @@ STARFirmwareError, convert_star_firmware_error_to_plr_error, ) -from .pip_channel import PIPChannel +from .pip_channel import _TADM_RECORDING_FW, PIPChannel, TADMRecordingMode if TYPE_CHECKING: from .driver import STARDriver @@ -196,6 +206,12 @@ def _assert_range(values, lo, hi, name): raise ValueError(f"{name} values must be between {lo} and {hi}, got {values}") +class _RecordsTADM(Protocol): + """Structural type for aspirate/dispense params that can request TADM recording.""" + + tadm_recording_mode: TADMRecordingMode + + class STARPIPBackend(PIPBackend): """Translates PIP operations into STAR firmware commands via the driver.""" @@ -579,9 +595,7 @@ class AspirateParams(BackendParams): z_drive_speed_during_2nd_section_search: Z drive speed during 2nd section search in mm/s. Must be between 0.3 and 160.0. cup_upper_edge: Cup upper edge in mm. Must be between 0 and 360.0. - tadm_algorithm: Whether to use the TADM algorithm. Default False. - recording_mode: Recording mode (0 = no recording, 1 = TADM errors only, - 2 = all TADM measurements). Must be between 0 and 2. + tadm_recording_mode: How much of the trace to store: "off", "errors_only", or "all". probe_liquid_height: If True, use gamma LLD to probe the liquid height before aspirating. Cannot be used when liquid heights are already set on operations. auto_surface_following_distance: If True, automatically compute the surface @@ -624,8 +638,7 @@ class AspirateParams(BackendParams): dosing_drive_speed_during_2nd_section_search: Optional[List[float]] = None z_drive_speed_during_2nd_section_search: Optional[List[float]] = None cup_upper_edge: Optional[List[float]] = None - tadm_algorithm: bool = False - recording_mode: int = 0 + tadm_recording_mode: TADMRecordingMode = "off" probe_liquid_height: bool = False auto_surface_following_distance: bool = False @@ -874,8 +887,6 @@ async def aspirate( "mix_surface_following_distance", ) _assert_range(limit_curve_index, 0, 999, "limit_curve_index") - if not 0 <= backend_params.recording_mode <= 2: - raise ValueError("recording_mode must be between 0 and 2") # 2nd section aspiration range checks _assert_range( [ @@ -920,6 +931,8 @@ async def aspirate( "cup_upper_edge", ) + await self._begin_tadm_if_recording(use_channels, backend_params) + try: await self.driver.send_command( module="C0", @@ -960,8 +973,8 @@ async def aspirate( ms=[f"{round(s * 10):04}" for s in mix_speed], mh=[f"{round(d * 10):04}" for d in mix_surface_following_distance], gi=[f"{i:03}" for i in limit_curve_index], - gj=backend_params.tadm_algorithm, - gk=backend_params.recording_mode, + gj=0, + gk=_TADM_RECORDING_FW[backend_params.tadm_recording_mode], lk=[1 if x else 0 for x in _fill(backend_params.use_2nd_section_aspiration, [False] * n)], ik=[ f"{round(x * 10):04}" @@ -1059,9 +1072,7 @@ class DispenseParams(BackendParams): 0 and 360.0. min_z_endpos: Minimum Z position in mm at end of command. If None, uses backend's ``traversal_height``. Must be between 0 and 360.0. - tadm_algorithm: Whether to use the TADM algorithm. Default False. - recording_mode: Recording mode (0 = no recording, 1 = TADM errors only, - 2 = all TADM measurements). Must be between 0 and 2. + tadm_recording_mode: How much of the trace to store: "off", "errors_only", or "all". probe_liquid_height: If True, use gamma LLD to probe the liquid height before dispensing. Cannot be used when liquid heights are already set on operations. auto_surface_following_distance: If True, automatically compute the surface @@ -1097,8 +1108,7 @@ class DispenseParams(BackendParams): limit_curve_index: Optional[List[int]] = None minimum_traverse_height_at_beginning_of_a_command: Optional[float] = None min_z_endpos: Optional[float] = None - tadm_algorithm: bool = False - recording_mode: int = 0 + tadm_recording_mode: TADMRecordingMode = "off" probe_liquid_height: bool = False auto_surface_following_distance: bool = False @@ -1335,8 +1345,7 @@ async def dispense( "mix_surface_following_distance", ) _assert_range(limit_curve_index, 0, 999, "limit_curve_index") - if not 0 <= backend_params.recording_mode <= 2: - raise ValueError("recording_mode must be between 0 and 2") + await self._begin_tadm_if_recording(use_channels, backend_params) try: await self.driver.send_command( @@ -1378,8 +1387,8 @@ async def dispense( ms=[f"{round(s * 10):04}" for s in mix_speed], mh=[f"{round(d * 10):04}" for d in mix_surface_following_distance], gi=[f"{i:03}" for i in limit_curve_index], - gj=backend_params.tadm_algorithm, - gk=backend_params.recording_mode, + gj=0, + gk=_TADM_RECORDING_FW[backend_params.tadm_recording_mode], ) except STARFirmwareError as e: if plr_e := convert_star_firmware_error_to_plr_error(e): @@ -1696,6 +1705,21 @@ async def request_tip_presence(self) -> List[Optional[bool]]: resp = await self.driver.send_command(module="C0", command="RT", fmt="rt# (n)") return [bool(v) for v in resp.get("rt")] + async def _begin_tadm_if_recording( + self, use_channels: List[int], backend_params: _RecordsTADM + ) -> None: + """Auto-arm TADM monitoring (Px:BG) on each involved channel when the op records TADM. + + So callers get the "begin monitoring" step for free on any aspirate/dispense that enables + TADM recording or limit-curve evaluation, rather than having to issue BG themselves. + """ + if backend_params.tadm_recording_mode == "off": + return + for channel in use_channels: + await self.channels[channel].begin_tadm_monitoring( + recording_mode=backend_params.tadm_recording_mode + ) + async def request_tadm_status(self) -> List[int]: """Request TADM enable/disable status across all PIP channels. diff --git a/pylabrobot/hamilton/liquid_handlers/star/pip_channel.py b/pylabrobot/hamilton/liquid_handlers/star/pip_channel.py index 1dd3ca45f84..aa1b091e7c4 100644 --- a/pylabrobot/hamilton/liquid_handlers/star/pip_channel.py +++ b/pylabrobot/hamilton/liquid_handlers/star/pip_channel.py @@ -2,9 +2,10 @@ from __future__ import annotations +import asyncio import datetime import enum -from typing import TYPE_CHECKING, Dict, Literal, Optional, Tuple +from typing import TYPE_CHECKING, Dict, List, Literal, Optional, Tuple from .errors import STARFirmwareError from .fw_parsing import parse_star_firmware_version_date @@ -14,6 +15,11 @@ from .pip_backend import STARPIPBackend +# TADM recording mode: how much of the pressure trace the firmware stores (``gk``). +TADMRecordingMode = Literal["off", "errors_only", "all"] +_TADM_RECORDING_FW = {"off": 0, "errors_only": 1, "all": 2} + + # --------------------------------------------------------------------------- # Drive-unit conversion helpers (mirrored from legacy STARBackend) # --------------------------------------------------------------------------- @@ -493,6 +499,170 @@ async def request_tadm_enabled(self) -> bool: ) return bool(resp["qs"][self.index]) + # -- Px:QL/QN read recorded TADM pressure trace ---------------------------- + + async def request_tadm_recording_length(self, slot: int = 0) -> int: + """Number of TADM samples currently recorded in a measurement slot (Px:QL). + + The firmware keeps one counter per measurement slot (the ``gi`` slot a recording was + started in). Use this to find how many samples ``request_tadm_recording`` will read. + + Args: + slot: TADM measurement slot (0-4), matching the ``gi`` slot the recording was + started in. Defaults to 0. + + Returns: + The number of samples recorded in that slot. + + Raises: + ValueError: If ``slot`` is out of range. + """ + if not 0 <= slot <= 4: + raise ValueError("slot must be between 0 and 4") + resp = await self.driver.send_command( + module=self.module_id, + command="QL", + fmt="ql#### (n)", + ) + return int(resp["ql"][slot]) + + async def request_tadm_recording( + self, + num_samples: Optional[int] = None, + start: int = 0, + batch_size: int = 50, + slot: int = 0, + ) -> List[float]: + """Read a recorded TADM pressure trace from this channel into a list of floats. + + After an aspiration or dispense run with ``tadm_recording_mode="all"`` (record all TADM + measurements), the channel buffers one pressure sample per firmware tick. This reads + them via the ``Px:QN`` firmware command and returns them in acquisition order. + + When ``num_samples`` is ``None`` (the default), the recording length is first queried + with ``Px:QL`` and the whole trace is read. + + The values are the raw signed TADM pressure samples the firmware reports, cast to + ``float``. They are in the instrument's internal pressure units and are not scaled to + a physical unit. + + Args: + num_samples: Number of samples to read. ``None`` (default) reads every recorded + sample from ``start`` to the end of the buffer. + start: Index of the first sample to read (0-based). Defaults to 0. + batch_size: Samples requested per firmware command. The firmware caps a single QN + read at 50 samples. Defaults to 50. + slot: TADM measurement slot (0-4) whose length is queried when ``num_samples`` is + ``None``. Defaults to 0. + + Returns: + The recorded pressure trace as a list of floats. May be shorter than + ``num_samples`` if the recording buffer holds fewer samples. + + Raises: + ValueError: If ``num_samples``, ``start``, ``batch_size``, or ``slot`` are out of + range. + """ + if start < 0: + raise ValueError("start must be >= 0") + if not 1 <= batch_size <= 50: + raise ValueError("batch_size must be between 1 and 50") + + if num_samples is None: + num_samples = max(0, await self.request_tadm_recording_length(slot) - start) + elif num_samples < 0: + raise ValueError("num_samples must be >= 0") + + trace: List[float] = [] + index = start + remaining = num_samples + while remaining > 0: + n = min(batch_size, remaining) + resp = await self.driver.send_command( + module=self.module_id, + command="QN", + li=f"{index:04}", # index of first sample to read + ln=f"{n:02}", # number of samples to read + fmt="qn#### (n)", + ) + values = resp.get("qn", []) + trace.extend(float(v) for v in values) + if len(values) < n: # buffer exhausted before num_samples + break + index += n + remaining -= n + return trace + + # -- Px:BG begin TADM monitoring ------------------------------------------- + + async def begin_tadm_monitoring( + self, + measurement_id: str = "0001", + slot: int = 0, + recording_mode: TADMRecordingMode = "all", + ): + """Arm TADM recording for the next pipetting step (Px:BG). + + A recording aspirate/dispense issues this automatically; call it directly only for custom + setups. + + Args: + measurement_id: Measurement label stored with the recording (firmware ``nr``). + slot: Measurement slot 0-4 to record into (firmware ``gi``); read back with the same slot. + recording_mode: How much of the trace to store (firmware ``gk``): ``"off"``, + ``"errors_only"``, or ``"all"``. + """ + if not 0 <= slot <= 4: + raise ValueError("slot must be between 0 and 4") + await self.driver.send_command( + self.module_id, + "BG", + nr=measurement_id, + gi=f"{slot:03}", + gj=0, + gk=_TADM_RECORDING_FW[recording_mode], + ) + + # -- Px:QL/QN TADM streaming ----------------------------------------------- + + async def _tadm_recording_finalized(self, slot: int = 0) -> Tuple[bool, int]: + """Whether the current TADM recording has finalized, and its sample count (Px:QL). + + The firmware's ``qm`` flag is 0 while a recording is in progress (count withheld, 0) and 1 + once the recording phase ends and the count is published. + """ + resp = await self.driver.send_command(self.module_id, "QL", fmt="qm#ql#### (n)") + return bool(resp["qm"]), int(resp["ql"][slot]) + + async def stream_tadm(self, slot: int = 0, poll_interval: float = 0.03): + """Yield the recorded TADM trace as soon as it finalizes during an in-flight op. + + Run the aspiration/dispense as a background task (do not await it) and iterate this + concurrently. ``Q*`` queries are lock-free, so the poll runs alongside the in-flight C0 + command; the trace is yielded when the recording phase finalizes, which happens partway + through the command — before it returns. + + The firmware does not expose a live, growing sample count mid-recording (``QL`` reports + ``qm=0`` and count 0 until the recording finalizes) and the buffer cannot be cleared, so + samples cannot be surfaced strictly one-at-a-time without diffing the previous trace. This + waits for the recording to start (``qm`` -> 0) and then finalize (``qm`` -> 1), then yields + the whole trace in order. + + Yields: + ``(index, value)`` pairs in acquisition order; value is in raw firmware pressure units. + """ + started = False + while True: + finalized, count = await self._tadm_recording_finalized(slot) + if not finalized: + started = True # qm=0: this recording is in progress + elif started: # qm back to 1 after starting: this recording has finalized + trace = await self.request_tadm_recording(count, slot=slot) + for index, value in enumerate(trace): + yield index, value + return + await asyncio.sleep(poll_interval) + # -- Px:ZL cLLD Z search (low-level, head-space) -------------------------- async def search_z_using_clld(