Skip to content

capabilities

Capability mixins for pressure-over-liquid fluid-handling devices.

Each mixin contributes a single public operation (or a cohesive group of operations) and delegates into the engine primitives provided by PressureOverLiquidControl (e.g. _handle_liquid, _require_arm, _disable_lateral_axes). Mixins are pure object subclasses at runtime and are never instantiated directly; concrete devices compose them onto the engine.

At type-check time the mixins are treated as subclasses of PressureOverLiquidControl so that references to engine members resolve, while at runtime they remain lightweight traits with no base of their own.

AspirateMixin

Bases: _EngineBase

Adds pressure-over-liquid aspiration to a device.

Source code in src/fluid_control/capabilities.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class AspirateMixin(_EngineBase):
    """Adds pressure-over-liquid aspiration to a device."""

    def aspirate(self, aspirate_dict: dict[int, ChannelCommand]) -> None:
        """
        Aspirate liquid across one or more valve controller channels.

        Args:
            aspirate_dict (dict): Mapping of channel IDs to channel-operation parameters.
                Each value is a [`ChannelCommand`][fluid_control.fluid_control.ChannelCommand],
                i.e. ``{"volume": <uL>, "liquid_class": <str>}``.

        Examples:
            Aspirate 50 uL of water on channel 1:

            >>> pipettor.aspirate({1: {"volume": 50.0, "liquid_class": "water"}})

        """
        logger.info(f"ASPIRATE START: {aspirate_dict}")
        # TODO: Enable ability to set timing PER CLASS
        self._handle_liquid(aspirate_dict, process="aspirate")

aspirate(aspirate_dict)

Aspirate liquid across one or more valve controller channels.

Parameters:

Name Type Description Default
aspirate_dict dict

Mapping of channel IDs to channel-operation parameters. Each value is a ChannelCommand, i.e. {"volume": <uL>, "liquid_class": <str>}.

 required

Examples:

Aspirate 50 uL of water on channel 1:

1
>>> pipettor.aspirate({1: {"volume": 50.0, "liquid_class": "water"}})
Source code in src/fluid_control/capabilities.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def aspirate(self, aspirate_dict: dict[int, ChannelCommand]) -> None:
    """
    Aspirate liquid across one or more valve controller channels.

    Args:
        aspirate_dict (dict): Mapping of channel IDs to channel-operation parameters.
            Each value is a [`ChannelCommand`][fluid_control.fluid_control.ChannelCommand],
            i.e. ``{"volume": <uL>, "liquid_class": <str>}``.

    Examples:
        Aspirate 50 uL of water on channel 1:

        >>> pipettor.aspirate({1: {"volume": 50.0, "liquid_class": "water"}})

    """
    logger.info(f"ASPIRATE START: {aspirate_dict}")
    # TODO: Enable ability to set timing PER CLASS
    self._handle_liquid(aspirate_dict, process="aspirate")

DispenseMixin

Bases: _EngineBase

Adds pressure-over-liquid dispensing to a device.

Source code in src/fluid_control/capabilities.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class DispenseMixin(_EngineBase):
    """Adds pressure-over-liquid dispensing to a device."""

    def dispense(self, dispense_dict: dict[int, ChannelCommand]) -> None:
        """
        Dispense liquid across one or more valve controller channels.

        Args:
            dispense_dict (dict): Mapping of channel IDs to channel-operation parameters.
                Each value is a [`ChannelCommand`][fluid_control.fluid_control.ChannelCommand],
                i.e. ``{"volume": <uL>, "liquid_class": <str>}``.

        Examples:
            Dispense 25 uL of water on channel 1:

            >>> dispenser.dispense({1: {"volume": 25.0, "liquid_class": "water"}})

            Dispense on two channels in parallel (armed together, fired once):

            >>> dispenser.dispense(
            ...     {
            ...         1: {"volume": 10.0, "liquid_class": "water"},
            ...         2: {"volume": 15.0, "liquid_class": "water"},
            ...     }
            ... )

        """
        logger.info(f"DISPENSE START: {dispense_dict}")
        # TODO: Enable ability to set timing PER CLASS
        self._handle_liquid(dispense_dict, process="dispense")

dispense(dispense_dict)

Dispense liquid across one or more valve controller channels.

Parameters:

Name Type Description Default
dispense_dict dict

Mapping of channel IDs to channel-operation parameters. Each value is a ChannelCommand, i.e. {"volume": <uL>, "liquid_class": <str>}.

 required

Examples:

Dispense 25 uL of water on channel 1:

1
>>> dispenser.dispense({1: {"volume": 25.0, "liquid_class": "water"}})

Dispense on two channels in parallel (armed together, fired once):

1
2
3
4
5
6
>>> dispenser.dispense(
...     {
...         1: {"volume": 10.0, "liquid_class": "water"},
...         2: {"volume": 15.0, "liquid_class": "water"},
...     }
... )
Source code in src/fluid_control/capabilities.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def dispense(self, dispense_dict: dict[int, ChannelCommand]) -> None:
    """
    Dispense liquid across one or more valve controller channels.

    Args:
        dispense_dict (dict): Mapping of channel IDs to channel-operation parameters.
            Each value is a [`ChannelCommand`][fluid_control.fluid_control.ChannelCommand],
            i.e. ``{"volume": <uL>, "liquid_class": <str>}``.

    Examples:
        Dispense 25 uL of water on channel 1:

        >>> dispenser.dispense({1: {"volume": 25.0, "liquid_class": "water"}})

        Dispense on two channels in parallel (armed together, fired once):

        >>> dispenser.dispense(
        ...     {
        ...         1: {"volume": 10.0, "liquid_class": "water"},
        ...         2: {"volume": 15.0, "liquid_class": "water"},
        ...     }
        ... )

    """
    logger.info(f"DISPENSE START: {dispense_dict}")
    # TODO: Enable ability to set timing PER CLASS
    self._handle_liquid(dispense_dict, process="dispense")

MixMixin

Bases: _EngineBase

Adds mixing (repeated aspirate/dispense) to a device.

Source code in src/fluid_control/capabilities.py 
 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
class MixMixin(_EngineBase):
    """Adds mixing (repeated aspirate/dispense) to a device."""

    def mix(self, mix_dict: dict[int, ChannelCommand], cycles: int) -> None:
        """
        Aspirate and dispense repeatedly to mix liquid in the channels.

        Args:
            mix_dict (dict): Mapping of channel IDs to channel-operation parameters.
            cycles (int): Number of aspirate/dispense cycles to execute.

        Examples:
            Mix 20 uL up and down for 3 cycles on channel 1:

            >>> pipettor.mix({1: {"volume": 20.0, "liquid_class": "water"}}, cycles=3)

        """
        logger.info(f"MIX START: {mix_dict}")

        for _ in range(cycles):
            self._handle_liquid(mix_dict, "aspirate")
            # TODO: Raise fluid_control arms so no bubbles
            self._handle_liquid(
                mix_dict, "dispense"
            )  # TODO: This needs to be an instance attribute dictionary that ensures all liquid is clear.

mix(mix_dict, cycles)

Aspirate and dispense repeatedly to mix liquid in the channels.

Parameters:

Name Type Description Default
mix_dict dict

Mapping of channel IDs to channel-operation parameters.

 required
cycles int

Number of aspirate/dispense cycles to execute.

 required

Examples:

Mix 20 uL up and down for 3 cycles on channel 1:

1
>>> pipettor.mix({1: {"volume": 20.0, "liquid_class": "water"}}, cycles=3)
Source code in src/fluid_control/capabilities.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def mix(self, mix_dict: dict[int, ChannelCommand], cycles: int) -> None:
    """
    Aspirate and dispense repeatedly to mix liquid in the channels.

    Args:
        mix_dict (dict): Mapping of channel IDs to channel-operation parameters.
        cycles (int): Number of aspirate/dispense cycles to execute.

    Examples:
        Mix 20 uL up and down for 3 cycles on channel 1:

        >>> pipettor.mix({1: {"volume": 20.0, "liquid_class": "water"}}, cycles=3)

    """
    logger.info(f"MIX START: {mix_dict}")

    for _ in range(cycles):
        self._handle_liquid(mix_dict, "aspirate")
        # TODO: Raise fluid_control arms so no bubbles
        self._handle_liquid(
            mix_dict, "dispense"
        )  # TODO: This needs to be an instance attribute dictionary that ensures all liquid is clear.

TipHandlingMixin

Bases: _EngineBase

Adds tip pickup and ejection to a device that has a motion axis.

Source code in src/fluid_control/capabilities.py 
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
class TipHandlingMixin(_EngineBase):
    """Adds tip pickup and ejection to a device that has a motion axis."""

    def eject_tips(self) -> OperationResult:
        """
        Eject tips from the fluid control module.

        Requires a mounted (non-static) device with a configured motion axis.

        Returns:
            OperationResult: ``(code, message)`` describing the outcome, where
            ``code`` is ``0`` (clear), ``1`` (error), or ``2`` (busy).

        Examples:
            >>> result = pipettor.eject_tips()
            >>> result.code
            0
            >>> result.message
            'Tips ejected successfully'

        """
        # TODO: How to consider static, mechanical, deck-based, fixed-point ejection mode
        # Make optional?
        # class Ejector
        # if ejector is not None:
        #   ....
        # TODO: Minimum: Add warning that this class assumes a co-mounted, dynamic tip ejection mechanism. Others are not yet supported.
        logger.info("EJECT TIPS START")
        self._require_arm()
        self.fluid_control_status.set_busy()
        self._disable_lateral_axes()
        try:
            for cycle in range(_EJECT_ACTUATION_CYCLES):
                logger.debug(
                    f"Eject tips: actuation cycle {cycle + 1}/{_EJECT_ACTUATION_CYCLES}"
                )  # TODO: Parameterize the total number of eject cycles for testing
                self._wait_output_pressure(
                    _EJECT_MAX_PRESSURE_MBAR
                )  # TODO: Change to Pressure Control Library max pressure. Will need slight modificiation of PGVA library
                self.pressure_control.trigger_actuation_valve(10)
                self.pressure_control.trigger_actuation_valve(1000)
                self._wait_output_pressure(
                    _EJECT_MIN_PRESSURE_MBAR
                )  # TODO: Change to Pressure Control Library min pressure. Will need slight modificiation of PGVA library
                self.pressure_control.trigger_actuation_valve(10)
                self.pressure_control.trigger_actuation_valve(2000)
            self.pressure_control.set_output_pressure(0)
            self.fluid_control_status.set_clear()
            self._enable_lateral_axes()
            logger.info("EJECT TIPS COMPLETE")
            return OperationResult(self.fluid_control_status.get_status(), "Tips ejected successfully")
        except Exception as e:
            logger.error(f"EJECT TIPS FAILED: {e}")
            self.fluid_control_status.set_error()
            self._enable_lateral_axes()
            return OperationResult(self.fluid_control_status.get_status(), str(e))

    def _pickup_action(self, duration: float) -> None:
        """Jog the mount arm downward until tip engagement stalls its motion."""
        arm = self._require_arm()
        delta = _PICKUP_STALL_DELTA_MM  # mm — FestoAxis.current_position() returns mm
        arm.acknowledge_faults()  # TODO: We need a way to NOT dig this deep into internals of the edcon library
        # self.mount_arm.disable_powerstage()
        current_position = arm.current_position()
        logger.debug(f"_pickup_action: start position={current_position}, duration={duration}, delta={delta}")
        repeat = True
        count = 0
        self._disable_lateral_axes()
        while repeat:
            arm.acknowledge_faults()
            arm.enable_powerstage()
            arm.jog_task(
                True, False, duration=duration
            )  # TODO: Passing all the way to jog_task defeats the purpose of the axis class and make the parameters passed / API confusing. Think of a way to fix this.
            # self.mount_arm.position_task(position=5000, velocity=duration, absolute=False, nonblocking=False)
            # self.mount_arm.jog_task(True, False, duration=0.5)
            new_position = arm.current_position()
            movement = abs(new_position - current_position)
            logger.debug(f"_pickup_action: new_position={new_position}, movement={movement}, stall_count={count}")
            if movement <= delta:
                count += 1
                current_position = new_position
            else:
                current_position = new_position
            if count >= _PICKUP_STALL_CONSECUTIVE:  # TODO: Parameterize this for testing
                logger.debug("_pickup_action: stall detected — tip engagement complete")
                repeat = False
        self._enable_lateral_axes()
        arm.acknowledge_faults()

    def pickup_tips(self, duration: float) -> OperationResult:
        """
        Pick up tips with the fluid_control.

        Requires a mounted (non-static) device with a configured motion axis.

        Args:
            duration (float): Duration in seconds of each downward jog toward the tips.

        Returns:
            OperationResult: ``(code, message)`` describing the outcome, where
            ``code`` is ``0`` (clear), ``1`` (error), or ``2`` (busy).

        Examples:
            >>> result = pipettor.pickup_tips(duration=0.5)
            >>> result.code
            0
            >>> result.message
            'Tips picked up successfully'

        """
        logger.info(f"PICKUP TIPS START: duration={duration}")
        self._require_arm()
        self.fluid_control_status.set_busy()
        try:
            self._pickup_action(duration=duration)
            self.fluid_control_status.set_clear()
            logger.info("PICKUP TIPS COMPLETE")
            return OperationResult(self.fluid_control_status.get_status(), "Tips picked up successfully")
        except Exception as e:
            logger.error(f"PICKUP TIPS FAILED: {e}")
            self.fluid_control_status.set_error()
            return OperationResult(self.fluid_control_status.get_status(), str(e))

eject_tips()

Eject tips from the fluid control module.

Requires a mounted (non-static) device with a configured motion axis.

Returns:

Name Type Description
OperationResult OperationResult

(code, message) describing the outcome, where
OperationResult

code is 0 (clear), 1 (error), or 2 (busy).

Examples:

1
2
3
4
5
>>> result = pipettor.eject_tips()
>>> result.code
0
>>> result.message
'Tips ejected successfully'
Source code in src/fluid_control/capabilities.py 
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
def eject_tips(self) -> OperationResult:
    """
    Eject tips from the fluid control module.

    Requires a mounted (non-static) device with a configured motion axis.

    Returns:
        OperationResult: ``(code, message)`` describing the outcome, where
        ``code`` is ``0`` (clear), ``1`` (error), or ``2`` (busy).

    Examples:
        >>> result = pipettor.eject_tips()
        >>> result.code
        0
        >>> result.message
        'Tips ejected successfully'

    """
    # TODO: How to consider static, mechanical, deck-based, fixed-point ejection mode
    # Make optional?
    # class Ejector
    # if ejector is not None:
    #   ....
    # TODO: Minimum: Add warning that this class assumes a co-mounted, dynamic tip ejection mechanism. Others are not yet supported.
    logger.info("EJECT TIPS START")
    self._require_arm()
    self.fluid_control_status.set_busy()
    self._disable_lateral_axes()
    try:
        for cycle in range(_EJECT_ACTUATION_CYCLES):
            logger.debug(
                f"Eject tips: actuation cycle {cycle + 1}/{_EJECT_ACTUATION_CYCLES}"
            )  # TODO: Parameterize the total number of eject cycles for testing
            self._wait_output_pressure(
                _EJECT_MAX_PRESSURE_MBAR
            )  # TODO: Change to Pressure Control Library max pressure. Will need slight modificiation of PGVA library
            self.pressure_control.trigger_actuation_valve(10)
            self.pressure_control.trigger_actuation_valve(1000)
            self._wait_output_pressure(
                _EJECT_MIN_PRESSURE_MBAR
            )  # TODO: Change to Pressure Control Library min pressure. Will need slight modificiation of PGVA library
            self.pressure_control.trigger_actuation_valve(10)
            self.pressure_control.trigger_actuation_valve(2000)
        self.pressure_control.set_output_pressure(0)
        self.fluid_control_status.set_clear()
        self._enable_lateral_axes()
        logger.info("EJECT TIPS COMPLETE")
        return OperationResult(self.fluid_control_status.get_status(), "Tips ejected successfully")
    except Exception as e:
        logger.error(f"EJECT TIPS FAILED: {e}")
        self.fluid_control_status.set_error()
        self._enable_lateral_axes()
        return OperationResult(self.fluid_control_status.get_status(), str(e))

pickup_tips(duration)

Pick up tips with the fluid_control.

Requires a mounted (non-static) device with a configured motion axis.

Parameters:

Name Type Description Default
duration float

Duration in seconds of each downward jog toward the tips.

 required

Returns:

Name Type Description
OperationResult OperationResult

(code, message) describing the outcome, where
OperationResult

code is 0 (clear), 1 (error), or 2 (busy).

Examples:

1
2
3
4
5
>>> result = pipettor.pickup_tips(duration=0.5)
>>> result.code
0
>>> result.message
'Tips picked up successfully'
Source code in src/fluid_control/capabilities.py 
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
def pickup_tips(self, duration: float) -> OperationResult:
    """
    Pick up tips with the fluid_control.

    Requires a mounted (non-static) device with a configured motion axis.

    Args:
        duration (float): Duration in seconds of each downward jog toward the tips.

    Returns:
        OperationResult: ``(code, message)`` describing the outcome, where
        ``code`` is ``0`` (clear), ``1`` (error), or ``2`` (busy).

    Examples:
        >>> result = pipettor.pickup_tips(duration=0.5)
        >>> result.code
        0
        >>> result.message
        'Tips picked up successfully'

    """
    logger.info(f"PICKUP TIPS START: duration={duration}")
    self._require_arm()
    self.fluid_control_status.set_busy()
    try:
        self._pickup_action(duration=duration)
        self.fluid_control_status.set_clear()
        logger.info("PICKUP TIPS COMPLETE")
        return OperationResult(self.fluid_control_status.get_status(), "Tips picked up successfully")
    except Exception as e:
        logger.error(f"PICKUP TIPS FAILED: {e}")
        self.fluid_control_status.set_error()
        return OperationResult(self.fluid_control_status.get_status(), str(e))