Skip to content

session

Backend-agnostic fluid-control session for manual operation.

This module is intentionally free of prompt_toolkit and rich dependencies so it can be imported and unit-tested without those packages installed. The interactive REPL is in fluid_control.cli.cli.

FluidControlSession

Backend-agnostic session wrapper for manual fluid-control operations.

Wraps a PressureOverLiquidControl instance (Dispenser or Pipettor) with optional [Gantry][applied_motion.applied_motion.Gantry] support for mount-arm motion. Has no dependency on prompt_toolkit or rich; it can be used programmatically without an interactive terminal.

Parameters:

Name Type Description Default
component PressureOverLiquidControl

The PressureOverLiquidControl instance to control.

 required
gantry Gantry | None

Optional connected [Gantry][applied_motion.applied_motion.Gantry] for mount-arm motion commands. Required for raise/lower/move and home commands.

 None
mount_axis_name str

Name of the axis in gantry that acts as the mount arm (raise/lower axis). Defaults to "Z".

 'Z'

Attributes:

Name Type Description
component

The bound fluid-control component.
gantry

The bound gantry, or None if not provided.
mount_axis_name

Name of the mount arm axis.
Source code in src/fluid_control/cli/session.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
class FluidControlSession:
    """
    Backend-agnostic session wrapper for manual fluid-control operations.

    Wraps a [`PressureOverLiquidControl`][fluid_control.fluid_control.PressureOverLiquidControl]
    instance (Dispenser or Pipettor) with optional
    [`Gantry`][applied_motion.applied_motion.Gantry] support for mount-arm motion.
    Has no dependency on ``prompt_toolkit`` or ``rich``; it can be used
    programmatically without an interactive terminal.

    Args:
        component: The [`PressureOverLiquidControl`][fluid_control.fluid_control.PressureOverLiquidControl]
            instance to control.
        gantry: Optional connected [`Gantry`][applied_motion.applied_motion.Gantry] for
            mount-arm motion commands.  Required for ``raise``/``lower``/``move``
            and ``home`` commands.
        mount_axis_name: Name of the axis in *gantry* that acts as the mount arm
            (raise/lower axis).  Defaults to ``"Z"``.

    Attributes:
        component: The bound fluid-control component.
        gantry: The bound gantry, or ``None`` if not provided.
        mount_axis_name: Name of the mount arm axis.

    """

    def __init__(
        self,
        component: PressureOverLiquidControl,
        gantry: Gantry | None = None,
        mount_axis_name: str = "Z",
    ) -> None:
        """
        Initialise a session for the given fluid-control component.

        Args:
            component: The
                [`PressureOverLiquidControl`][fluid_control.fluid_control.PressureOverLiquidControl]
                instance to control.
            gantry: Optional connected [`Gantry`][applied_motion.applied_motion.Gantry]
                for mount-arm motion commands.  Defaults to ``None``.
            mount_axis_name: Name of the mount arm axis in *gantry*.
                Defaults to ``"Z"``.

        """
        self.component = component
        self.gantry = gantry
        self.mount_axis_name = mount_axis_name
        logger.debug(
            "FluidControlSession created: component=%r, gantry=%r, mount_axis=%s",
            component,
            gantry,
            mount_axis_name,
        )

    # ------------------------------------------------------------------
    # Valve / pressure operations
    # ------------------------------------------------------------------

    def valve_timed(self, channel: int, time_ms: int, pressure: int = 0) -> OperationResult:
        """
        Open a single valve for a fixed duration.

        Wraps [`PressureOverLiquidControl.direct_command`][fluid_control.fluid_control.PressureOverLiquidControl.direct_command]
        for the common single-channel case.

        Args:
            channel: Valve channel ID (must be in active channels).
            time_ms: Valve opening duration in milliseconds.  Must be > 0.
            pressure: Output pressure in mbar to hold during the operation.
                Defaults to ``0`` (neutral).

        Returns:
            ``[status_code, message]`` from the underlying direct command.

        """
        logger.info("valve_timed: channel=%d, time_ms=%d, pressure=%d", channel, time_ms, pressure)
        return self.component.direct_command(channel_times={channel: time_ms}, pressure=pressure)

    def direct(self, channel_times: dict[int, int], pressure: int) -> OperationResult:
        """
        Send a multi-channel raw valve command, bypassing volume calibration.

        Args:
            channel_times: Mapping of channel ID → opening time in ms.
            pressure: Output pressure in mbar for the operation.

        Returns:
            ``[status_code, message]`` from the underlying direct command.

        """
        logger.info("direct: channel_times=%s, pressure=%d", channel_times, pressure)
        return self.component.direct_command(channel_times=channel_times, pressure=pressure)

    def dispense(self, channel: int, volume_ul: float, liquid_class: str) -> OperationResult:
        """
        Dispense a volume on a single channel.

        Args:
            channel: Valve channel ID.
            volume_ul: Volume to dispense in microlitres.
            liquid_class: Liquid class key matching a calibration entry.

        Returns:
            ``[status_code, message]`` describing the outcome.

        """
        logger.info("dispense: channel=%d, volume_ul=%.2f, liquid_class=%r", channel, volume_ul, liquid_class)
        dispense_dict: dict[int, ChannelCommand] = {channel: {"volume": volume_ul, "liquid_class": liquid_class}}
        self.component.dispense(dispense_dict)
        return OperationResult(self.component.fluid_control_status.get_status(), "Dispense complete")

    def aspirate(self, channel: int, volume_ul: float, liquid_class: str) -> OperationResult:
        """
        Aspirate a volume on a single channel (Pipettor only).

        Args:
            channel: Valve channel ID.
            volume_ul: Volume to aspirate in microlitres.
            liquid_class: Liquid class key matching a calibration entry.

        Returns:
            ``[status_code, message]`` describing the outcome.

        Raises:
            NotImplementedError: If the component does not support aspiration
                (e.g. a [`Dispenser`][fluid_control.Dispenser]).

        """
        logger.info("aspirate: channel=%d, volume_ul=%.2f, liquid_class=%r", channel, volume_ul, liquid_class)
        aspirate_dict: dict[int, ChannelCommand] = {channel: {"volume": volume_ul, "liquid_class": liquid_class}}
        self.component.aspirate(aspirate_dict)
        return OperationResult(self.component.fluid_control_status.get_status(), "Aspirate complete")

    def mix(self, channel: int, volume_ul: float, liquid_class: str, cycles: int) -> OperationResult:
        """
        Mix by repeatedly aspirating and dispensing on a channel.

        Args:
            channel: Valve channel ID.
            volume_ul: Volume per cycle in microlitres.
            liquid_class: Liquid class key matching a calibration entry.
            cycles: Number of aspirate/dispense cycles.

        Returns:
            ``[status_code, message]`` describing the outcome.

        Raises:
            NotImplementedError: If the component does not support mixing
                (e.g. a static [`Dispenser`][fluid_control.Dispenser]).

        """
        logger.info(
            "mix: channel=%d, volume_ul=%.2f, liquid_class=%r, cycles=%d",
            channel,
            volume_ul,
            liquid_class,
            cycles,
        )
        mix_dict: dict[int, ChannelCommand] = {channel: {"volume": volume_ul, "liquid_class": liquid_class}}
        self.component.mix(mix_dict, cycles)
        return OperationResult(self.component.fluid_control_status.get_status(), f"Mix complete ({cycles} cycle(s))")

    def set_pressure(self, pressure_mbar: int) -> None:
        """
        Set output pressure and block until it stabilises.

        Args:
            pressure_mbar: Target pressure in mbar.

        """
        logger.info("set_pressure: %d mbar", pressure_mbar)
        self.component._wait_output_pressure(pressure_mbar)

    # ------------------------------------------------------------------
    # Axis / gantry operations
    # ------------------------------------------------------------------

    @require_attr("gantry")
    def move_axis(self, axis_name: str, position_mm: float, velocity: float = _DEFAULT_VELOCITY) -> dict[str, float]:
        """
        Move a gantry axis to an absolute position.

        Args:
            axis_name: Name of the axis as registered in the gantry.
            position_mm: Target position in mm (absolute).
            velocity: Move speed in mm/s.  Defaults to ``10.0``.

        Returns:
            Gantry location dict after the move completes.

        Raises:
            GantryNotConfiguredError: If no gantry is configured.
            KeyError: If *axis_name* is not registered with the gantry.

        """
        logger.info("move_axis: axis=%s, position=%.3f mm, velocity=%.1f mm/s", axis_name, position_mm, velocity)
        axis = self.gantry.axes[axis_name]  # type: ignore[ty:unresolved-attribute]
        clamped = max(axis.min_position, min(axis.max_position, position_mm))
        if clamped != position_mm:
            logger.warning(
                "move_axis: position %.3f mm clamped to %.3f mm (axis limits [%.3f, %.3f])",
                position_mm,
                clamped,
                axis.min_position,
                axis.max_position,
            )
        if abs(clamped - self.gantry.get_location()[axis_name]) < 1e-3:  # type: ignore[ty:unresolved-attribute]
            return self.gantry.get_location()  # type: ignore[ty:unresolved-attribute]
        self.gantry.move_to(  # type: ignore[ty:unresolved-attribute]
            deque([{axis_name: {"position": clamped, "velocity": velocity}}]),
        )
        return self.gantry.get_location()  # # type: ignore[ty:unresolved-attribute]

    @require_attr("gantry")
    def raise_arm(self, delta_mm: float, velocity: float = _DEFAULT_VELOCITY) -> dict[str, float]:
        """
        Move the mount arm by a relative offset from its current position.

        TODO: add flag for checking if axis is mounted inverted and support for positioning
        Reads the current axis position and applies *delta_mm* as a signed
        offset.  Positive values move the arm in the positive axis direction;
        negative values move it in the opposite direction.

        The ``raise`` REPL command passes a positive delta; the ``lower``
        command passes a negative delta.  Absolute moves can be achieved with
        [`move_axis`][move_axis] directly.

        Args:
            delta_mm: Relative move distance in mm.  Positive = higher encoder
                count; negative = lower encoder count.
            velocity: Move speed in mm/s.  Defaults to ``10.0``.

        Returns:
            Gantry location dict after the move completes.

        Raises:
            GantryNotConfiguredError: If no gantry is configured.

        """
        current = self.gantry.get_location()  # type: ignore[ty:unresolved-attribute]
        new_position = current[self.mount_axis_name] + delta_mm
        return self.move_axis(self.mount_axis_name, new_position, velocity)

    @require_attr("gantry")
    def where(self) -> dict[str, float]:
        """
        Return the current position of all gantry axes.

        Returns:
            Mapping of axis name → position in mm.

        Raises:
            GantryNotConfiguredError: If no gantry is configured.

        """
        return self.gantry.get_location()  # type: ignore[ty:unresolved-attribute]

    @require_attr("gantry")
    def home(self) -> None:
        """
        Home all axes on the gantry.

        Raises:
            GantryNotConfiguredError: If no gantry is configured.

        """
        logger.info("home: homing all gantry axes")
        self.gantry.home()  # type: ignore[ty:unresolved-attribute]

    def enable_axes(self) -> None:
        """
        Enable the powerstage on the component's ``disable_axes`` list.

        Raises:
            NotImplementedError: If the component is static (no axes configured).

        """
        logger.info("enable_axes: enabling powerstage on disable_axes")
        self.component._enable_lateral_axes()

    def disable_axes(self) -> None:
        """
        Disable the powerstage on the component's ``disable_axes`` list.

        Raises:
            NotImplementedError: If the component is static (no axes configured).

        """
        logger.info("disable_axes: disabling powerstage on disable_axes")
        self.component._disable_lateral_axes()

    # ------------------------------------------------------------------
    # Tip operations (Pipettor)
    # ------------------------------------------------------------------

    def pickup_tips(self, duration: float) -> OperationResult:
        """
        Pick up tips by driving the mount arm downward until stall detection.

        Args:
            duration: Jog duration in seconds per stall-detection cycle.

        Returns:
            ``[status_code, message]`` from the pickup operation.

        Raises:
            NotImplementedError: If the component is static or does not support
                tip pickup (e.g. a [`Dispenser`][fluid_control.Dispenser]).

        """
        logger.info("pickup_tips: duration=%.2f s", duration)
        return self.component.pickup_tips(duration)

    def eject_tips(self) -> OperationResult:
        """
        Eject tips using pneumatic actuation.

        Returns:
            ``[status_code, message]`` from the eject operation.

        Raises:
            NotImplementedError: If the component does not support tip ejection
                (e.g. a [`Dispenser`][fluid_control.Dispenser]).

        """
        logger.info("eject_tips: ejecting tips")
        return self.component.eject_tips()

    # ------------------------------------------------------------------
    # Introspection
    # ------------------------------------------------------------------

    def get_status(self) -> dict:
        """
        Return combined fluid-control, pressure, and valve status.

        Returns:
            Status dict from
            [`PressureOverLiquidControl.get_status`][fluid_control.fluid_control.PressureOverLiquidControl.get_status].

        """
        return self.component.get_status()

    def get_liquid_classes(self) -> list[str]:
        """
        Return all available liquid class names.

        Returns:
            List of liquid class name strings from the current calibration config.

        """
        return list(self.component.get_liquid_classes())

    def get_channels(self) -> list[int]:
        """
        Return the list of active valve channel IDs.

        Returns:
            List of active channel integers.

        """
        return list(self.component.active_channels)

__init__(component, gantry=None, mount_axis_name='Z')

Initialise a session for the given fluid-control component.

Parameters:

Name Type Description Default
component PressureOverLiquidControl

The PressureOverLiquidControl instance to control.

 required
gantry Gantry | None

Optional connected [Gantry][applied_motion.applied_motion.Gantry] for mount-arm motion commands. Defaults to None.

 None
mount_axis_name str

Name of the mount arm axis in gantry. Defaults to "Z".

 'Z'
Source code in src/fluid_control/cli/session.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def __init__(
    self,
    component: PressureOverLiquidControl,
    gantry: Gantry | None = None,
    mount_axis_name: str = "Z",
) -> None:
    """
    Initialise a session for the given fluid-control component.

    Args:
        component: The
            [`PressureOverLiquidControl`][fluid_control.fluid_control.PressureOverLiquidControl]
            instance to control.
        gantry: Optional connected [`Gantry`][applied_motion.applied_motion.Gantry]
            for mount-arm motion commands.  Defaults to ``None``.
        mount_axis_name: Name of the mount arm axis in *gantry*.
            Defaults to ``"Z"``.

    """
    self.component = component
    self.gantry = gantry
    self.mount_axis_name = mount_axis_name
    logger.debug(
        "FluidControlSession created: component=%r, gantry=%r, mount_axis=%s",
        component,
        gantry,
        mount_axis_name,
    )

aspirate(channel, volume_ul, liquid_class)

Aspirate a volume on a single channel (Pipettor only).

Parameters:

Name Type Description Default
channel int

Valve channel ID.

 required
volume_ul float

Volume to aspirate in microlitres.

 required
liquid_class str

Liquid class key matching a calibration entry.

 required

Returns:

Type Description
OperationResult

[status_code, message] describing the outcome.

Raises:

Type Description
NotImplementedError

If the component does not support aspiration (e.g. a Dispenser).
Source code in src/fluid_control/cli/session.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def aspirate(self, channel: int, volume_ul: float, liquid_class: str) -> OperationResult:
    """
    Aspirate a volume on a single channel (Pipettor only).

    Args:
        channel: Valve channel ID.
        volume_ul: Volume to aspirate in microlitres.
        liquid_class: Liquid class key matching a calibration entry.

    Returns:
        ``[status_code, message]`` describing the outcome.

    Raises:
        NotImplementedError: If the component does not support aspiration
            (e.g. a [`Dispenser`][fluid_control.Dispenser]).

    """
    logger.info("aspirate: channel=%d, volume_ul=%.2f, liquid_class=%r", channel, volume_ul, liquid_class)
    aspirate_dict: dict[int, ChannelCommand] = {channel: {"volume": volume_ul, "liquid_class": liquid_class}}
    self.component.aspirate(aspirate_dict)
    return OperationResult(self.component.fluid_control_status.get_status(), "Aspirate complete")

direct(channel_times, pressure)

Send a multi-channel raw valve command, bypassing volume calibration.

Parameters:

Name Type Description Default
channel_times dict[int, int]

Mapping of channel ID → opening time in ms.

 required
pressure int

Output pressure in mbar for the operation.

 required

Returns:

Type Description
OperationResult

[status_code, message] from the underlying direct command.
Source code in src/fluid_control/cli/session.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def direct(self, channel_times: dict[int, int], pressure: int) -> OperationResult:
    """
    Send a multi-channel raw valve command, bypassing volume calibration.

    Args:
        channel_times: Mapping of channel ID → opening time in ms.
        pressure: Output pressure in mbar for the operation.

    Returns:
        ``[status_code, message]`` from the underlying direct command.

    """
    logger.info("direct: channel_times=%s, pressure=%d", channel_times, pressure)
    return self.component.direct_command(channel_times=channel_times, pressure=pressure)

disable_axes()

Disable the powerstage on the component's disable_axes list.

Raises:

Type Description
NotImplementedError

If the component is static (no axes configured).
Source code in src/fluid_control/cli/session.py 
357
358
359
360
361
362
363
364
365
366
def disable_axes(self) -> None:
    """
    Disable the powerstage on the component's ``disable_axes`` list.

    Raises:
        NotImplementedError: If the component is static (no axes configured).

    """
    logger.info("disable_axes: disabling powerstage on disable_axes")
    self.component._disable_lateral_axes()

dispense(channel, volume_ul, liquid_class)

Dispense a volume on a single channel.

Parameters:

Name Type Description Default
channel int

Valve channel ID.

 required
volume_ul float

Volume to dispense in microlitres.

 required
liquid_class str

Liquid class key matching a calibration entry.

 required

Returns:

Type Description
OperationResult

[status_code, message] describing the outcome.
Source code in src/fluid_control/cli/session.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def dispense(self, channel: int, volume_ul: float, liquid_class: str) -> OperationResult:
    """
    Dispense a volume on a single channel.

    Args:
        channel: Valve channel ID.
        volume_ul: Volume to dispense in microlitres.
        liquid_class: Liquid class key matching a calibration entry.

    Returns:
        ``[status_code, message]`` describing the outcome.

    """
    logger.info("dispense: channel=%d, volume_ul=%.2f, liquid_class=%r", channel, volume_ul, liquid_class)
    dispense_dict: dict[int, ChannelCommand] = {channel: {"volume": volume_ul, "liquid_class": liquid_class}}
    self.component.dispense(dispense_dict)
    return OperationResult(self.component.fluid_control_status.get_status(), "Dispense complete")

eject_tips()

Eject tips using pneumatic actuation.

Returns:

Type Description
OperationResult

[status_code, message] from the eject operation.

Raises:

Type Description
NotImplementedError

If the component does not support tip ejection (e.g. a Dispenser).
Source code in src/fluid_control/cli/session.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
def eject_tips(self) -> OperationResult:
    """
    Eject tips using pneumatic actuation.

    Returns:
        ``[status_code, message]`` from the eject operation.

    Raises:
        NotImplementedError: If the component does not support tip ejection
            (e.g. a [`Dispenser`][fluid_control.Dispenser]).

    """
    logger.info("eject_tips: ejecting tips")
    return self.component.eject_tips()

enable_axes()

Enable the powerstage on the component's disable_axes list.

Raises:

Type Description
NotImplementedError

If the component is static (no axes configured).
Source code in src/fluid_control/cli/session.py 
346
347
348
349
350
351
352
353
354
355
def enable_axes(self) -> None:
    """
    Enable the powerstage on the component's ``disable_axes`` list.

    Raises:
        NotImplementedError: If the component is static (no axes configured).

    """
    logger.info("enable_axes: enabling powerstage on disable_axes")
    self.component._enable_lateral_axes()

get_channels()

Return the list of active valve channel IDs.

Returns:

Type Description
list[int]

List of active channel integers.
Source code in src/fluid_control/cli/session.py 
430
431
432
433
434
435
436
437
438
def get_channels(self) -> list[int]:
    """
    Return the list of active valve channel IDs.

    Returns:
        List of active channel integers.

    """
    return list(self.component.active_channels)

get_liquid_classes()

Return all available liquid class names.

Returns:

Type Description
list[str]

List of liquid class name strings from the current calibration config.
Source code in src/fluid_control/cli/session.py 
420
421
422
423
424
425
426
427
428
def get_liquid_classes(self) -> list[str]:
    """
    Return all available liquid class names.

    Returns:
        List of liquid class name strings from the current calibration config.

    """
    return list(self.component.get_liquid_classes())

get_status()

Return combined fluid-control, pressure, and valve status.

Returns:

Type Description
dict

Status dict from
dict

PressureOverLiquidControl.get_status.
Source code in src/fluid_control/cli/session.py 
409
410
411
412
413
414
415
416
417
418
def get_status(self) -> dict:
    """
    Return combined fluid-control, pressure, and valve status.

    Returns:
        Status dict from
        [`PressureOverLiquidControl.get_status`][fluid_control.fluid_control.PressureOverLiquidControl.get_status].

    """
    return self.component.get_status()

home()

Home all axes on the gantry.

Raises:

Type Description
GantryNotConfiguredError

If no gantry is configured.
Source code in src/fluid_control/cli/session.py 
334
335
336
337
338
339
340
341
342
343
344
@require_attr("gantry")
def home(self) -> None:
    """
    Home all axes on the gantry.

    Raises:
        GantryNotConfiguredError: If no gantry is configured.

    """
    logger.info("home: homing all gantry axes")
    self.gantry.home()  # type: ignore[ty:unresolved-attribute]

mix(channel, volume_ul, liquid_class, cycles)

Mix by repeatedly aspirating and dispensing on a channel.

Parameters:

Name Type Description Default
channel int

Valve channel ID.

 required
volume_ul float

Volume per cycle in microlitres.

 required
liquid_class str

Liquid class key matching a calibration entry.

 required
cycles int

Number of aspirate/dispense cycles.

 required

Returns:

Type Description
OperationResult

[status_code, message] describing the outcome.

Raises:

Type Description
NotImplementedError

If the component does not support mixing (e.g. a static Dispenser).
Source code in src/fluid_control/cli/session.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def mix(self, channel: int, volume_ul: float, liquid_class: str, cycles: int) -> OperationResult:
    """
    Mix by repeatedly aspirating and dispensing on a channel.

    Args:
        channel: Valve channel ID.
        volume_ul: Volume per cycle in microlitres.
        liquid_class: Liquid class key matching a calibration entry.
        cycles: Number of aspirate/dispense cycles.

    Returns:
        ``[status_code, message]`` describing the outcome.

    Raises:
        NotImplementedError: If the component does not support mixing
            (e.g. a static [`Dispenser`][fluid_control.Dispenser]).

    """
    logger.info(
        "mix: channel=%d, volume_ul=%.2f, liquid_class=%r, cycles=%d",
        channel,
        volume_ul,
        liquid_class,
        cycles,
    )
    mix_dict: dict[int, ChannelCommand] = {channel: {"volume": volume_ul, "liquid_class": liquid_class}}
    self.component.mix(mix_dict, cycles)
    return OperationResult(self.component.fluid_control_status.get_status(), f"Mix complete ({cycles} cycle(s))")

move_axis(axis_name, position_mm, velocity=_DEFAULT_VELOCITY)

Move a gantry axis to an absolute position.

Parameters:

Name Type Description Default
axis_name str

Name of the axis as registered in the gantry.

 required
position_mm float

Target position in mm (absolute).

 required
velocity float

Move speed in mm/s. Defaults to 10.0.

 _DEFAULT_VELOCITY

Returns:

Type Description
dict[str, float]

Gantry location dict after the move completes.

Raises:

Type Description
GantryNotConfiguredError

If no gantry is configured.
KeyError

If axis_name is not registered with the gantry.
Source code in src/fluid_control/cli/session.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
@require_attr("gantry")
def move_axis(self, axis_name: str, position_mm: float, velocity: float = _DEFAULT_VELOCITY) -> dict[str, float]:
    """
    Move a gantry axis to an absolute position.

    Args:
        axis_name: Name of the axis as registered in the gantry.
        position_mm: Target position in mm (absolute).
        velocity: Move speed in mm/s.  Defaults to ``10.0``.

    Returns:
        Gantry location dict after the move completes.

    Raises:
        GantryNotConfiguredError: If no gantry is configured.
        KeyError: If *axis_name* is not registered with the gantry.

    """
    logger.info("move_axis: axis=%s, position=%.3f mm, velocity=%.1f mm/s", axis_name, position_mm, velocity)
    axis = self.gantry.axes[axis_name]  # type: ignore[ty:unresolved-attribute]
    clamped = max(axis.min_position, min(axis.max_position, position_mm))
    if clamped != position_mm:
        logger.warning(
            "move_axis: position %.3f mm clamped to %.3f mm (axis limits [%.3f, %.3f])",
            position_mm,
            clamped,
            axis.min_position,
            axis.max_position,
        )
    if abs(clamped - self.gantry.get_location()[axis_name]) < 1e-3:  # type: ignore[ty:unresolved-attribute]
        return self.gantry.get_location()  # type: ignore[ty:unresolved-attribute]
    self.gantry.move_to(  # type: ignore[ty:unresolved-attribute]
        deque([{axis_name: {"position": clamped, "velocity": velocity}}]),
    )
    return self.gantry.get_location()  # # type: ignore[ty:unresolved-attribute]

pickup_tips(duration)

Pick up tips by driving the mount arm downward until stall detection.

Parameters:

Name Type Description Default
duration float

Jog duration in seconds per stall-detection cycle.

 required

Returns:

Type Description
OperationResult

[status_code, message] from the pickup operation.

Raises:

Type Description
NotImplementedError

If the component is static or does not support tip pickup (e.g. a Dispenser).
Source code in src/fluid_control/cli/session.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
def pickup_tips(self, duration: float) -> OperationResult:
    """
    Pick up tips by driving the mount arm downward until stall detection.

    Args:
        duration: Jog duration in seconds per stall-detection cycle.

    Returns:
        ``[status_code, message]`` from the pickup operation.

    Raises:
        NotImplementedError: If the component is static or does not support
            tip pickup (e.g. a [`Dispenser`][fluid_control.Dispenser]).

    """
    logger.info("pickup_tips: duration=%.2f s", duration)
    return self.component.pickup_tips(duration)

raise_arm(delta_mm, velocity=_DEFAULT_VELOCITY)

Move the mount arm by a relative offset from its current position.

TODO: add flag for checking if axis is mounted inverted and support for positioning Reads the current axis position and applies delta_mm as a signed offset. Positive values move the arm in the positive axis direction; negative values move it in the opposite direction.

The raise REPL command passes a positive delta; the lower command passes a negative delta. Absolute moves can be achieved with [move_axis][] directly.

Parameters:

Name Type Description Default
delta_mm float

Relative move distance in mm. Positive = higher encoder count; negative = lower encoder count.

 required
velocity float

Move speed in mm/s. Defaults to 10.0.

 _DEFAULT_VELOCITY

Returns:

Type Description
dict[str, float]

Gantry location dict after the move completes.

Raises:

Type Description
GantryNotConfiguredError

If no gantry is configured.
Source code in src/fluid_control/cli/session.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
@require_attr("gantry")
def raise_arm(self, delta_mm: float, velocity: float = _DEFAULT_VELOCITY) -> dict[str, float]:
    """
    Move the mount arm by a relative offset from its current position.

    TODO: add flag for checking if axis is mounted inverted and support for positioning
    Reads the current axis position and applies *delta_mm* as a signed
    offset.  Positive values move the arm in the positive axis direction;
    negative values move it in the opposite direction.

    The ``raise`` REPL command passes a positive delta; the ``lower``
    command passes a negative delta.  Absolute moves can be achieved with
    [`move_axis`][move_axis] directly.

    Args:
        delta_mm: Relative move distance in mm.  Positive = higher encoder
            count; negative = lower encoder count.
        velocity: Move speed in mm/s.  Defaults to ``10.0``.

    Returns:
        Gantry location dict after the move completes.

    Raises:
        GantryNotConfiguredError: If no gantry is configured.

    """
    current = self.gantry.get_location()  # type: ignore[ty:unresolved-attribute]
    new_position = current[self.mount_axis_name] + delta_mm
    return self.move_axis(self.mount_axis_name, new_position, velocity)

set_pressure(pressure_mbar)

Set output pressure and block until it stabilises.

Parameters:

Name Type Description Default
pressure_mbar int

Target pressure in mbar.

 required
Source code in src/fluid_control/cli/session.py 
239
240
241
242
243
244
245
246
247
248
def set_pressure(self, pressure_mbar: int) -> None:
    """
    Set output pressure and block until it stabilises.

    Args:
        pressure_mbar: Target pressure in mbar.

    """
    logger.info("set_pressure: %d mbar", pressure_mbar)
    self.component._wait_output_pressure(pressure_mbar)

valve_timed(channel, time_ms, pressure=0)

Open a single valve for a fixed duration.

Wraps PressureOverLiquidControl.direct_command for the common single-channel case.

Parameters:

Name Type Description Default
channel int

Valve channel ID (must be in active channels).

 required
time_ms int

Valve opening duration in milliseconds. Must be > 0.

 required
pressure int

Output pressure in mbar to hold during the operation. Defaults to 0 (neutral).

 0

Returns:

Type Description
OperationResult

[status_code, message] from the underlying direct command.
Source code in src/fluid_control/cli/session.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def valve_timed(self, channel: int, time_ms: int, pressure: int = 0) -> OperationResult:
    """
    Open a single valve for a fixed duration.

    Wraps [`PressureOverLiquidControl.direct_command`][fluid_control.fluid_control.PressureOverLiquidControl.direct_command]
    for the common single-channel case.

    Args:
        channel: Valve channel ID (must be in active channels).
        time_ms: Valve opening duration in milliseconds.  Must be > 0.
        pressure: Output pressure in mbar to hold during the operation.
            Defaults to ``0`` (neutral).

    Returns:
        ``[status_code, message]`` from the underlying direct command.

    """
    logger.info("valve_timed: channel=%d, time_ms=%d, pressure=%d", channel, time_ms, pressure)
    return self.component.direct_command(channel_times={channel: time_ms}, pressure=pressure)

where()

Return the current position of all gantry axes.

Returns:

Type Description
dict[str, float]

Mapping of axis name → position in mm.

Raises:

Type Description
GantryNotConfiguredError

If no gantry is configured.
Source code in src/fluid_control/cli/session.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
@require_attr("gantry")
def where(self) -> dict[str, float]:
    """
    Return the current position of all gantry axes.

    Returns:
        Mapping of axis name → position in mm.

    Raises:
        GantryNotConfiguredError: If no gantry is configured.

    """
    return self.gantry.get_location()  # type: ignore[ty:unresolved-attribute]

GantryNotConfiguredError

Bases: AttributeError

Raise when a gantry-dependent method is called without a gantry.

Subclasses [AttributeError][builtins.AttributeError] because the underlying failure is an attempted attribute access on None (self.gantry is unset), while remaining precisely catchable by callers that only care about this case.

Source code in src/fluid_control/cli/session.py 
21
22
23
24
25
26
27
28
class GantryNotConfiguredError(AttributeError):
    """
    Raise when a gantry-dependent method is called without a gantry.

    Subclasses [`AttributeError`][builtins.AttributeError] because the underlying failure is an
    attempted attribute access on ``None`` (``self.gantry`` is unset), while
    remaining precisely catchable by callers that only care about this case.
    """

require_attr(attr)

Raise GantryNotConfiguredError via this decorator if self.<attr> is None.

Source code in src/fluid_control/cli/session.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def require_attr(attr):
    """Raise [`GantryNotConfiguredError`][fluid_control.cli.session.GantryNotConfiguredError] via this decorator if ``self.<attr>`` is ``None``."""

    def decorator(func):
        @functools.wraps(func)
        def wrapper(self, *args, **kwargs):
            if getattr(self, attr) is None:
                raise GantryNotConfiguredError(
                    f"No gantry configured: method '{func.__name__}' is disabled because '{attr}' was not provided."
                )
            return func(self, *args, **kwargs)

        return wrapper

    return decorator