feat(STAR): read recorded TADM pressure traces from a PIP channel

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":

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.
 

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(