diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py index 8ad0392546e..e2f2d3953c2 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py @@ -20,6 +20,7 @@ Sequence, Tuple, Type, + TypedDict, TypeVar, Union, cast, @@ -1604,6 +1605,58 @@ def ensure_can_reach_position( "Try the operation with different channels or a different target position (i.e. different labware placement)." ) + class ChannelCycleCounts(TypedDict): + tip_pick_up_cycles: int + tip_discard_cycles: int + aspiration_cycles: int + dispensing_cycles: int + + async def channel_request_cycle_counts(self, channel_idx: int) -> ChannelCycleCounts: + """Request cycle counters for a single channel. + + Returns the number of tip pick-up, tip discard, aspiration, and dispensing cycles + performed by the channel. + + Args: + channel_idx: The channel index to query (0-indexed). + + Returns: + A dict with keys ``tip_pick_up_cycles``, ``tip_discard_cycles``, + ``aspiration_cycles``, and ``dispensing_cycles``. + """ + + if not (0 <= channel_idx < self.num_channels): + raise ValueError( + f"channel_idx must be between 0 and {self.num_channels - 1}, got {channel_idx}." + ) + + resp = await self.send_command( + module=self.channel_id(channel_idx), + command="RV", + fmt="na##########nb##########nc##########nd##########", + ) + return { + "tip_pick_up_cycles": resp["na"], + "tip_discard_cycles": resp["nb"], + "aspiration_cycles": resp["nc"], + "dispensing_cycles": resp["nd"], + } + + async def channels_request_cycle_counts(self) -> List[ChannelCycleCounts]: + """Request cycle counters for all channels. + + Returns: + A list of dicts (one per channel, ordered by channel index), each with keys + ``tip_pick_up_cycles``, ``tip_discard_cycles``, ``aspiration_cycles``, + and ``dispensing_cycles``. + """ + + return list( + await asyncio.gather( + *(self.channel_request_cycle_counts(channel_idx=idx) for idx in range(self.num_channels)) + ) + ) + # # # ACTION Commands # # # async def pick_up_tips(