AFL.automation.prepare.OT2HTTPDriver#

Functions

`ceil`(x, /)	Return the ceiling of x as an Integral.
`listify`(obj)

Classes

`Driver`(name[, defaults, overrides, ...])
`OT2DeckWebAppMixin`()
`OT2HTTPDriver`([overrides])	HTTP-backed Opentrons OT-2 driver.
`Path`(args, *kwargs)	PurePath subclass that can make system calls.

class AFL.automation.prepare.OT2HTTPDriver.OT2HTTPDriver(overrides=None)[source]#

HTTP-backed Opentrons OT-2 driver.

This driver wraps the Opentrons HTTP API and persists deck state in the AFL driver configuration so labware, modules, instruments, tip usage, and preparation targets can survive run recreation.

Parameters:: overrides (dict, optional) – Configuration overrides merged into defaults during driver initialization.

Notes

The driver recreates robot runs on demand and reloads previously configured deck state when a run expires or is recreated.

Examples

>>> driver = OT2HTTPDriver({"robot_ip": "192.168.1.50"})
>>> driver.load_labware("opentrons_96_tiprack_300ul", "1")
>>> driver.load_instrument("p300_single", "left", ["1"])
>>> driver.transfer("2A1", "3A1", 100)

PIPETTE_NAME_ALIASES = {'p10': 'p10_single', 'p1000': 'p1000_single', 'p1000_single': 'p1000_single', 'p10_single': 'p10_single', 'p10_single_gen1': 'p10_single', 'p300': 'p300_single', 'p300_single': 'p300_single'}#

EXPECTED_TIPRACK_TOKEN = {'p1000_single': '1000ul', 'p10_single': '10ul', 'p300_single': '300ul'}#

defaults = {'available_tips': {}, 'loaded_instruments': {}, 'loaded_labware': {}, 'loaded_modules': {}, 'occupied_sample_locations': [], 'prep_targets': [], 'reserved_stock_tips': [], 'robot_ip': '127.0.0.1', 'robot_port': '31950', 'stock_tip_locations': {}, 'stock_tip_reservations': {}, 'tip_rack_offset': {'x': 0, 'y': 0, 'z': 0}}#

__init__(overrides=None)[source]#

Initialize the OT-2 HTTP driver.

Parameters:: overrides (dict, optional) – Configuration values that override the class defaults.

Examples

>>> driver = OT2HTTPDriver({"robot_ip": "127.0.0.1", "robot_port": "31950"})
>>> driver.base_url
'http://127.0.0.1:31950'

log_info(message)[source]#

Log an informational message.

Parameters:: message (str) – Message to emit.

log_error(message)[source]#

Log an error message.

Parameters:: message (str) – Message to emit.

log_debug(message)[source]#

Log a debug message.

Parameters:: message (str) – Message to emit.

log_warning(message)[source]#

Log a warning message.

Parameters:: message (str) – Message to emit.

reset_prep_targets()[source]#

Clear queued preparation targets.

Examples

>>> driver.reset_prep_targets()

add_prep_targets(targets, reset=False)[source]#

Append preparation target locations.

Parameters:

targets (str or sequence of str) – Target well locations to queue.
reset (bool, default=False) – If True, clear existing targets before appending.

Examples

>>> driver.add_prep_targets(["4A1", "4A2"], reset=True)

get_prep_target()[source]#

Pop and return the next queued preparation target.

Returns:: Next queued target location.
Return type:: str

status()[source]#

Return human-readable OT-2 status lines.

Returns:: Status lines describing prep targets, tip state, session state, active pipettes, and loaded labware.
Return type:: list of str

reset_tipracks(mount='both')[source]#

Reset available-tip tracking for one or more mounts.

Parameters:: mount ({"left", "right", "both"}, default="both") – Mount selection to reset.

reset()[source]#: Reset the active OT-2 session, protocol, and persisted deck state.

reset_deck()[source]#: Clear persisted deck configuration and related in-memory state.

home(**kwargs)[source]#

Home the robot’s axes using the dedicated /robot/home endpoint.

This endpoint is a direct control endpoint and doesn’t require creating a run. It can be used to home all axes at once or specific axes as needed.

parse_well(loc)[source]#

Split a deck location into slot and well components.

Parameters:: loc (str) – Deck location such as "1A1".
Returns:: Two-item tuple (slot, well).
Return type:: tuple

get_wells(locs)[source]#

Convert deck locations into validated HTTP API well descriptors.

Parameters:

locs (str or sequence of str) – Deck locations in slot+well form, for example "1A1".

Returns:

Well descriptors containing labwareId and wellName.

Return type:

list of dict

Raises:

ValueError – If the slot has no loaded labware or the stored labware metadata is malformed.
AssertionError – If the requested well is not valid for the loaded labware.

send_labware(labware_def, check_run_status=True, reload_loaded_labware=True)[source]#

Persist and upload a custom labware definition.

Parameters:

labware_def (dict) – Opentrons labware definition.
check_run_status (bool, default=True) – If False, skip the run-status GET check when ensuring a run.
reload_loaded_labware (bool, default=True) – If True, reload matching active labware after upload.

Returns:

Upload metadata including definition URI, version, and content hash.

Return type:

dict

load_labware(name, slot, module=None, check_run_status=True, **kwargs)[source]#

Load labware into a deck slot or module.

Parameters:

name (str) – Labware load name or namespace/load_name key.
slot (str) – Deck slot identifier.
module (str, optional) – Module identifier when loading onto a module.
check_run_status (bool, default=True) – If False, skip the run-status GET check when ensuring a run.
**kwargs – Additional options, including labware_json for custom labware.

Returns:

Loaded labware identifier returned by the robot.

Return type:

str

load_module(name, slot, check_run_status=True, **kwargs)[source]#

Load a module into a deck slot.

Parameters:

name (str) – Module model name.
slot (str) – Deck slot identifier.
check_run_status (bool, default=True) – If False, skip the run-status GET check when ensuring a run.

Returns:

Loaded module identifier returned by the robot.

Return type:

str

load_instrument(name, mount, tip_rack_slots, reload=False, check_run_status=True, update_pipettes=True, **kwargs)[source]#

Load a pipette and initialize tip tracking.

Parameters:

name (str) – Pipette name or alias.
mount ({"left", "right"}) – Mount on which to load the pipette.
tip_rack_slots (sequence of str) – Slots containing compatible tipracks.
reload (bool, default=False) – If True, preserve existing tip availability during run reload.
check_run_status (bool, default=True) – If False, skip the run-status GET check when ensuring a run.
update_pipettes (bool, default=True) – If False, skip refreshing attached pipette metadata.

Returns:

Loaded pipette identifier returned by the robot.

Return type:

str

mix(volume, location, repetitions=1, **kwargs)[source]#

Mix liquid in place by repeated aspirate/dispense cycles.

Parameters:

volume (float) – Mix volume in microliters.
location (str) – Deck location to mix.
repetitions (int, default=1) – Number of aspirate/dispense cycles.
**kwargs – Reserved for future compatibility.

transfer(source, dest, volume, mix_before=None, mix_after=None, air_gap=0, aspirate_rate=None, dispense_rate=None, mix_aspirate_rate=None, mix_dispense_rate=None, blow_out=False, post_aspirate_delay=0.0, aspirate_equilibration_delay=0.0, post_dispense_delay=0.0, drop_tip=True, return_tip=False, force_new_tip=False, to_top=True, to_center=False, to_top_z_offset=0, source_z_offset=0, tip_rack_offset=None, return_tip_z_offset=None, fast_mixing=False, touch_tip=False, tip_location=None, **kwargs)[source]#

Transfer liquid between two deck locations.

Parameters:

source (str) – Source deck location such as "2A1".
dest (str) – Destination deck location such as "3B1".
volume (float) – Requested transfer volume in microliters.
mix_before (sequence of int and float, optional) – Two-item sequence (repetitions, volume_ul) applied before the aspirate step.
mix_after (sequence of int and float, optional) – Two-item sequence (repetitions, volume_ul) applied after the dispense step.
air_gap (float, default=0) – Air gap volume in microliters.
aspirate_rate (float, optional) – Aspirate flow rate in microliters per second.
dispense_rate (float, optional) – Dispense flow rate in microliters per second.
mix_aspirate_rate (float, optional) – Aspirate flow rate used during mix cycles.
mix_dispense_rate (float, optional) – Dispense flow rate used during mix cycles.
blow_out (bool, default=False) – If True, perform a blow-out after dispensing.
post_aspirate_delay (float, default=0.0) – Delay in seconds after moving above the source well.
aspirate_equilibration_delay (float, default=0.0) – Delay in seconds while the tip remains in the source liquid after aspirating.
post_dispense_delay (float, default=0.0) – Delay in seconds after dispensing.
drop_tip (bool, default=True) – If True, discard the tip after the transfer.
return_tip (bool, default=False) – If True, return the tip to its origin instead of discarding it.
force_new_tip (bool, default=False) – If True, force a fresh tip between split sub-transfers.
to_top (bool, default=True) – Dispense at the top of the destination well.
to_center (bool, default=False) – Dispense at the center of the destination well.
to_top_z_offset (float, default=0) – Additional z-offset applied when dispensing to the top.
source_z_offset (float, default=0) – Additional z-offset applied when aspirating from the source.
tip_rack_offset (dict, optional) – Offset mapping with x, y, and z keys used for tip pickup and tip return.
return_tip_z_offset (float, optional) – Return-only z-offset applied when returning a tip to its origin. If omitted, the existing tip_rack_offset z value is used.
fast_mixing (bool, default=False) – Reserved flag for higher-level callers.
touch_tip (bool, default=False) – If True, touch the tip to the destination well after dispense.
tip_location (str, optional) – Explicit tip location to use, for example "1A1".
**kwargs – Additional compatibility aliases such as blowout and touchTip.

Returns:

Structured transfer metadata including selected pipette, subtransfer volumes, source and destination well metadata, and applied options.

Return type:

dict

Raises:

ValueError – If the transfer request is invalid or no suitable pipette is loaded.
RuntimeError – If the underlying robot command fails.

Examples

>>> driver.transfer("2A1", "3A1", 150)
>>> driver.transfer(
...     "2A1",
...     "3A1",
...     50,
...     mix_before=(3, 40),
...     return_tip=True,
...     tip_rack_offset={"x": 0, "y": 0, "z": -1},
... )

pickup_tip(tip_location, tip_rack_offset=None)[source]#

Pick up a specific tip from a deck tip location.

Parameters:

tip_location (str) – Deck tip location such as "1A2".
tip_rack_offset (dict, optional) – Offset mapping with x, y, and z keys applied during pickup.

Returns:

Pickup metadata including mount, pipette, and requested tip location.

Return type:

dict

Raises:

ValueError – If the requested tip location is invalid or unavailable.
RuntimeError – If another tip is already attached.

return_tip(tip_location=None, tip_rack_offset=None, return_tip_z_offset=None)[source]#

Return the currently attached tip to its original tiprack well.

Parameters:

tip_location (str, optional) – Expected current tip location such as "1A2". When provided, this is validated against the attached tip before returning it.
tip_rack_offset (dict, optional) – Offset mapping with x, y, and z keys applied while moving to the return location.
return_tip_z_offset (float, optional) – Return-only z-offset applied during the return operation.

Returns:

Return metadata including the tip origin and status.

Return type:

dict

Raises:

ValueError – If the requested tip location does not match the attached tip.

set_aspirate_rate(rate=150, pipette=None)[source]#: Set stored aspirate flow rate for one or more active pipettes.

set_dispense_rate(rate=300, pipette=None)[source]#: Set stored dispense flow rate for one or more active pipettes.

set_gantry_speed(speed=400)[source]#

Record a requested gantry speed change.

Notes

The HTTP driver currently logs the request but does not apply it through the robot server.

get_pipette(volume, method='min_transfers')[source]#

Select the best loaded pipette for a requested volume.

Parameters:

volume (float) – Requested transfer volume in microliters.
method ({"min_transfers", "uncertainty"}, default="min_transfers") – Selection strategy.

Returns:

Selected pipette metadata including mount, volume range, and number of required transfers.

Return type:

dict

get_aspirate_rate(pipette=None)[source]#: Return the stored aspirate flow rate for a pipette.

get_dispense_rate(pipette=None)[source]#: Return the stored dispense flow rate for a pipette.

set_shake(rpm, module_id=None)[source]#: Set heater-shaker speed and wait for the target RPM.

stop_shake(module_id=None)[source]#: Stop heater-shaker motion.

set_shaker_temp(temp, module_id=None)[source]#: Set heater-shaker target temperature.

stop_shaker_heat(module_id=None)[source]#: Deactivate heater-shaker heating.

unlatch_shaker(module_id=None)[source]#: Open the heater-shaker labware latch.

latch_shaker(module_id=None)[source]#: Close the heater-shaker labware latch.

get_shaker_temp()[source]#: Return current and target heater-shaker temperatures.

get_shake_rpm()[source]#: Return heater-shaker speed status, current RPM, and target RPM.

get_shake_latch_status()[source]#: Return the heater-shaker latch status string.

set_tempmodule_temperature(module_id, temperature_c, hold_time=0.0, wait=True)[source]#

Set a temperature module target and optionally wait to stabilize.

Returns:: Current and target temperatures after stabilization.
Return type:: tuple

deactivate_tempmodule(module_id, timeout_s=120, wait=True)[source]#

Deactivate a temperature module.

Returns:: Result returned by _execute_atomic_command().
Return type:: bool or str

get_tempmodule_status(log=True)[source]#: Return raw tempdeck status data from the modules endpoint.

get_tip(mount)[source]#: Reserve and return the next available tip for a mount.

get_tip_status(mount=None)[source]#

Return human-readable tip availability status.

Parameters:: mount (str, optional) – Specific mount to report. If omitted, report all mounts.
Returns:: Tip availability summary.
Return type:: str

make_align_script(filename: str)[source]#

Generate an Opentrons Python Protocol API script to verify alignment.

Parameters:: filename (str) – Output path for the generated protocol script.

Notes

The generated script recreates the current deck state and moves each loaded pipette to the top of well A1 for each non-tiprack labware.

Examples

>>> driver.make_align_script("align_check.py")