import requests
import time
import logging
import copy
import hashlib
import json
import shutil
from pathlib import Path
from math import ceil
from AFL.automation.APIServer.Driver import Driver
from AFL.automation.prepare.OT2DeckWebAppMixin import OT2DeckWebAppMixin
from AFL.automation.shared.utilities import listify
# Add this constant at the top of the file, after the imports
TIPRACK_WELLS = [f"{row}{col}" for col in range(1, 13) for row in "ABCDEFGH"]
FIXED_TRASH_ADDRESSABLE_AREA = "fixedTrash"
[docs]
class OT2HTTPDriver(OT2DeckWebAppMixin, Driver):
"""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 :attr:`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",
"p10_single": "p10_single",
"p10_single_gen1": "p10_single",
"p300": "p300_single",
"p300_single": "p300_single",
"p1000": "p1000_single",
"p1000_single": "p1000_single",
}
EXPECTED_TIPRACK_TOKEN = {
"p10_single": "10ul",
"p300_single": "300ul",
"p1000_single": "1000ul",
}
defaults = {}
defaults["robot_ip"] = "127.0.0.1" # Default to localhost, should be overridden
defaults["robot_port"] = "31950" # Default Opentrons HTTP API port
defaults["loaded_labware"] = {} # Persistent storage for loaded labware
defaults["loaded_instruments"] = {} # Persistent storage for loaded instruments
defaults["loaded_modules"] = {} # Persistent storage for loaded modules
defaults["available_tips"] = {} # Persistent storage for available tips, Format: {mount: [(tiprack_id, well_name), ...]}
defaults["stock_tip_locations"] = {} # Configured stock tip candidates, Format: {stock_name: ["6A4", "9A4"]}
defaults["stock_tip_reservations"] = {} # Activated stock tip reservations, Format: {stock_name: ["6A4"]}
defaults["reserved_stock_tips"] = [] # Tip locations reserved for stock pipetting, e.g. ["6A4"]
defaults["occupied_sample_locations"] = [] # Sample destinations already populated on deck
defaults["prep_targets"] = [] # Persistent storage for prep target well locations
defaults["tip_rack_offset"] = {"x": 0, "y": 0, "z": 0} # Default offset for tip pickup/return at tiprack wells
[docs]
def __init__(self, overrides=None):
"""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'
"""
self.app = None
Driver.__init__(
self,
name="OT2_HTTP_Driver",
defaults=self.gather_defaults(),
overrides=overrides,
)
self.name = "OT2_HTTP_Driver"
# Initialize state variables
self.session_id = None
self.protocol_id = None
self.max_transfer = None
self.min_transfer = None
self.has_tip = False
self.last_pipette = None
self.current_tip = None
self.modules = {}
self.pipette_info = {}
# Custom labware handling
self.custom_labware_files = {}
self.sent_custom_labware = {}
self.custom_labware_dir = self._get_custom_labware_dir()
self._load_custom_labware_defs()
# Base URL for HTTP requests
self.base_url = f"http://{self.config['robot_ip']}:{self.config['robot_port']}"
self.headers = {"Opentrons-Version": "2"}
# Initialize the robot connection
self._initialize_robot()
self.useful_links['View Deck'] = '/visualize_deck'
def _log(self, level, message):
"""Log a message safely with or without a Flask app.
Parameters
----------
level : str
Logger method name such as ``"info"`` or ``"error"``.
message : str
Message to emit.
"""
if self.app is not None and hasattr(self.app, "logger"):
log_method = getattr(self.app.logger, level, None)
if log_method:
log_method(message)
else:
print(f"[{level.upper()}] {message}")
[docs]
def log_info(self, message):
"""Log an informational message.
Parameters
----------
message : str
Message to emit.
"""
self._log("info", message)
[docs]
def log_error(self, message):
"""Log an error message.
Parameters
----------
message : str
Message to emit.
"""
self._log("error", message)
[docs]
def log_debug(self, message):
"""Log a debug message.
Parameters
----------
message : str
Message to emit.
"""
self._log("debug", message)
[docs]
def log_warning(self, message):
"""Log a warning message.
Parameters
----------
message : str
Message to emit.
"""
self._log("warning", message)
def _get_custom_labware_dir(self) -> Path:
"""Return the user-scoped custom labware directory.
Returns
-------
pathlib.Path
Directory used to persist custom labware JSON definitions.
"""
custom_labware_dir = Path.home() / ".afl" / "opentrons_labware"
self._bootstrap_custom_labware_dir(custom_labware_dir)
return custom_labware_dir
def _get_seed_custom_labware_dir(self):
"""Locate packaged labware definitions used for first-run seeding.
Returns
-------
pathlib.Path or None
Seed directory when found, otherwise ``None``.
"""
current = Path(__file__).resolve()
for parent in current.parents:
candidate = parent / "support" / "labware"
if candidate.is_dir():
return candidate
candidate = Path.cwd() / "support" / "labware"
if candidate.is_dir():
return candidate
return None
def _bootstrap_custom_labware_dir(self, custom_labware_dir: Path):
"""Create and seed the user custom-labware directory.
Parameters
----------
custom_labware_dir : pathlib.Path
Destination directory to create and seed.
"""
if custom_labware_dir.exists():
return
custom_labware_dir.mkdir(parents=True, exist_ok=True)
seed_dir = self._get_seed_custom_labware_dir()
if seed_dir is None:
self.log_warning(
f"Custom labware seed directory not found; leaving {custom_labware_dir} empty"
)
return
for json_file in sorted(seed_dir.glob("*.json")):
shutil.copy2(json_file, custom_labware_dir / json_file.name)
def _load_custom_labware_defs(self):
"""Index locally available custom labware definitions.
Notes
-----
Definitions are keyed by ``namespace/loadName`` and duplicate keys are
rejected to avoid ambiguous uploads.
"""
self.custom_labware_files = {}
duplicates = []
for json_file in self.custom_labware_dir.glob("*.json"):
try:
with open(json_file, "r") as f:
definition = json.load(f)
_, _, key = self._custom_labware_key(definition)
existing_path = self.custom_labware_files.get(key)
if existing_path is not None and existing_path != json_file:
duplicates.append((key, existing_path, json_file))
continue
self.custom_labware_files[key] = json_file
except Exception:
continue
if duplicates:
details = "; ".join(
f"{key}: {first} vs {second}"
for key, first, second in duplicates
)
raise ValueError(
f"Duplicate custom labware definitions found in {self.custom_labware_dir}: {details}"
)
def _custom_labware_key(self, labware_def):
"""Extract the canonical key for a labware definition.
Parameters
----------
labware_def : dict
Opentrons labware definition.
Returns
-------
tuple
``(namespace, load_name, key)`` where ``key`` is
``"namespace/load_name"``.
"""
namespace = labware_def.get("namespace", "custom_beta")
load_name = labware_def.get("parameters", {}).get("loadName")
if not load_name:
raise ValueError("labware_def missing parameters.loadName")
return namespace, load_name, f"{namespace}/{load_name}"
def _canonical_labware_def(self, labware_def):
"""Return a normalized copy of a labware definition for hashing.
Parameters
----------
labware_def : dict
Labware definition to normalize.
Returns
-------
dict
Deep-copied normalized definition.
"""
canonical = copy.deepcopy(labware_def)
canonical.pop("version", None)
return json.dumps(canonical, sort_keys=True, separators=(",", ":"))
def _hash_labware_def(self, labware_def):
"""Compute a stable content hash for a labware definition.
Parameters
----------
labware_def : dict
Labware definition to hash.
Returns
-------
str
SHA-256 hex digest.
"""
canonical = self._canonical_labware_def(labware_def)
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
def _labware_upload_info(self, key):
"""Return cached upload metadata for a custom labware key.
Parameters
----------
key : str
``namespace/load_name`` key.
Returns
-------
dict or None
Cached upload metadata when available.
"""
info = self.sent_custom_labware.get(key)
if isinstance(info, dict):
return info
return None
def _next_custom_labware_version(self, key, labware_def):
"""Choose the next upload version for a custom labware definition.
Parameters
----------
key : str
``namespace/load_name`` key.
labware_def : dict
Labware definition being uploaded.
Returns
-------
int
Version number to upload for the current run.
"""
requested_version = int(labware_def.get("version", 1) or 1)
if self._labware_upload_info(key) is not None:
return max(requested_version, int(self._labware_upload_info(key)["version"]) + 1)
return requested_version
def _custom_labware_file_path(self, labware_def):
"""Return the on-disk JSON path for a custom labware definition.
Parameters
----------
labware_def : dict
Labware definition.
Returns
-------
pathlib.Path
Destination JSON path in the user custom-labware directory.
"""
_, load_name, _ = self._custom_labware_key(labware_def)
return self.custom_labware_dir / f"{load_name}.json"
def _loaded_labware_key(self, labware_info):
"""Derive a ``namespace/load_name`` key from persisted labware metadata.
Parameters
----------
labware_info : tuple
Persisted labware tuple stored in ``config["loaded_labware"]``.
Returns
-------
str or None
Derived key when enough metadata is available.
"""
if not isinstance(labware_info, tuple) or len(labware_info) < 2:
return None
load_name = labware_info[1]
result = labware_info[2] if len(labware_info) >= 3 else {}
definition = result.get("definition", {}) if isinstance(result, dict) else {}
namespace = definition.get("namespace", "custom_beta")
return namespace, load_name
def _remap_tip_availability(self, old_uuid_to_slot, slot_to_new_tiprack_uuid):
"""Remap available-tip tracking after tiprack reload.
Parameters
----------
old_uuid_to_slot : dict
Mapping from old tiprack UUIDs to deck slots.
slot_to_new_tiprack_uuid : dict
Mapping from deck slots to newly loaded tiprack UUIDs.
"""
old_available_tips = self.config.get("available_tips", {})
new_available_tips = {}
for mount in self.config["loaded_instruments"].keys():
new_available_tips[mount] = []
for tiprack_uuid, well in old_available_tips.get(mount, []):
slot = old_uuid_to_slot.get(tiprack_uuid)
new_uuid = slot_to_new_tiprack_uuid.get(slot)
if new_uuid is not None:
new_available_tips[mount].append((new_uuid, well))
self.log_info(f"Remapped {len(new_available_tips[mount])} available tips for {mount} mount after reload.")
self.config["available_tips"] = new_available_tips
def _reload_matching_labware_definition(self, labware_def, run_id=None, check_run_status=True):
"""Reload active labware instances that match an updated definition.
Parameters
----------
labware_def : dict
Updated labware definition.
run_id : str, optional
Existing run identifier to reuse.
check_run_status : bool, default=True
If ``False``, skip the run-status GET check.
Returns
-------
bool
``True`` when reload succeeds or no reload is needed.
"""
namespace, load_name, key = self._custom_labware_key(labware_def)
matching_slots = []
original_labware = copy.deepcopy(self.config["loaded_labware"])
for slot, labware_info in original_labware.items():
loaded_key = self._loaded_labware_key(labware_info)
if loaded_key is None:
continue
loaded_namespace, loaded_name = loaded_key
if loaded_namespace == namespace and loaded_name == load_name:
matching_slots.append(str(slot))
if not matching_slots:
return False
if run_id is None:
run_id = self._ensure_run_exists(check_run_status=check_run_status)
original_instruments = copy.deepcopy(self.config["loaded_instruments"])
old_uuid_to_slot = {}
slot_to_new_tiprack_uuid = {}
affected_mounts = {}
for mount, instrument in original_instruments.items():
tiprack_slots = []
for tiprack_uuid in instrument.get("tip_racks", []):
slot = self._slot_by_labware_uuid(tiprack_uuid)
if slot is not None:
slot = str(slot)
old_uuid_to_slot[tiprack_uuid] = slot
tiprack_slots.append(slot)
if any(slot in matching_slots for slot in tiprack_slots):
affected_mounts[mount] = {
"name": instrument["name"],
"tiprack_slots": tiprack_slots,
}
for slot in matching_slots:
module_id = None
if slot in self.config["loaded_modules"]:
module_id = self.config["loaded_modules"][slot][0]
self.log_info(
f"Reloading active labware '{namespace}/{load_name}' in slot {slot}"
)
new_labware_id = self.load_labware(
f"{namespace}/{load_name}",
slot,
module=module_id,
labware_json=labware_def,
check_run_status=False,
)
slot_to_new_tiprack_uuid[str(slot)] = new_labware_id
for mount, instrument in affected_mounts.items():
self.log_info(
f"Reloading pipette '{instrument['name']}' on {mount} mount after tiprack update"
)
self.load_instrument(
instrument["name"],
mount,
instrument["tiprack_slots"],
reload=True,
check_run_status=False,
)
if slot_to_new_tiprack_uuid:
self._remap_tip_availability(old_uuid_to_slot, slot_to_new_tiprack_uuid)
if affected_mounts:
self.has_tip = False
self.last_pipette = None
self.current_tip = None
return True
def _initialize_robot(self):
"""Probe robot connectivity and refresh attached pipette metadata.
Raises
------
RuntimeError
If the robot cannot be reached or pipette metadata cannot be read.
"""
self.log_info("Initializing OT2 HTTP Driver")
try:
# Check if the robot is reachable
response = requests.get(url=f"{self.base_url}/health", headers=self.headers)
if response.status_code != 200:
raise ConnectionError(f"Failed to connect to robot at {self.base_url}")
# Get attached pipettes
self._update_pipettes()
except requests.exceptions.RequestException as e:
self.log_error(f"Error connecting to robot: {str(e)}")
raise ConnectionError(
f"Error connecting to robot at {self.base_url}: {str(e)}"
)
def _update_pipettes(self):
"""Get information about attached pipettes and their settings"""
try:
if self.app is not None:
self.log_info("Fetching pipette information from robot")
# Get basic pipette information
response = requests.get(
url=f"{self.base_url}/instruments", headers=self.headers
)
if response.status_code != 200:
raise RuntimeError(f"Failed to get pipettes: {response.text}")
pipettes_data = response.json()['data']
self.pipette_info = {}
# Update min/max transfer values based on attached pipettes
self.min_transfer = None
self.max_transfer = None
for pipette in pipettes_data:
mount = pipette['mount']
try:
pipette_id = self.config["loaded_instruments"][mount]["pipette_id"] # the id from this run
except KeyError:
pipette_id = None
# Store basic pipette info
self.pipette_info[mount] = {
"id": pipette_id,
"name": pipette["instrumentName"],
"model": pipette["instrumentModel"],
"serial": pipette["serialNumber"],
"mount": mount,
"min_volume": pipette.get("data",{}).get("min_volume", None),
"max_volume": pipette.get("data",{}).get("max_volume", None),
"aspirate_flow_rate": pipette.get("data",{}).get(
"aspirateFlowRate", {}
).get("value",150),
"dispense_flow_rate": pipette.get("data",{}).get(
"dispenseFlowRate", {}
).get("value",150),
"channels": pipette.get("data",{}).get("channels", 1),
}
if pipette_id is None:
continue
# Update global min/max transfer values
min_volume = self.pipette_info[mount]['min_volume']
max_volume = self.pipette_info[mount]['max_volume']
if (self.min_transfer is None) or (self.min_transfer > min_volume):
self.min_transfer = min_volume
if self.app is not None:
self.log_info(
f"Setting minimum transfer to {self.min_transfer}"
)
if (self.max_transfer is None) or (self.max_transfer < max_volume):
self.max_transfer = max_volume
if self.app is not None:
self.log_info(
f"Setting maximum transfer to {self.max_transfer}"
)
if self.app is not None:
self.log_info(f"Pipette information updated: {self.pipette_info}")
except Exception as e:
raise RuntimeError(f"Error getting pipettes: {str(e)}")
def _get_active_pipettes(self):
"""Return pipettes that are attached and loaded into the active run.
Returns
-------
dict
Mapping from mount name to pipette metadata.
"""
active_pipettes = {}
loaded_instruments = self.config.get("loaded_instruments", {})
for mount, info in self.pipette_info.items():
if not info:
continue
if mount not in loaded_instruments:
continue
if info.get("id") is None:
continue
active_pipettes[mount] = info
return active_pipettes
def _get_active_pipette_info(self, mount):
"""Return active pipette metadata for a mount.
Parameters
----------
mount : str
Pipette mount, typically ``"left"`` or ``"right"``.
Returns
-------
dict
Active pipette metadata.
Raises
------
ValueError
If no loaded pipette is available on the requested mount.
"""
mount = str(mount).strip().lower()
info = self._get_active_pipettes().get(mount)
if info is None:
raise ValueError(f"No loaded pipette available on {mount} mount")
return info
[docs]
def reset_prep_targets(self):
"""Clear queued preparation targets.
Examples
--------
>>> driver.reset_prep_targets()
"""
self.config["prep_targets"] = []
[docs]
def add_prep_targets(self, targets, reset=False):
"""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)
"""
if reset:
self.reset_prep_targets()
self.config.setdefault("prep_targets", [])
self.config["prep_targets"].extend(listify(targets))
self.config._update_history()
[docs]
def get_prep_target(self):
"""Pop and return the next queued preparation target.
Returns
-------
str
Next queued target location.
"""
return self.config["prep_targets"].pop(0)
[docs]
def status(self):
"""Return human-readable OT-2 status lines.
Returns
-------
list of str
Status lines describing prep targets, tip state, session state,
active pipettes, and loaded labware.
"""
status = []
prep_targets = self.config.get("prep_targets", [])
if len(prep_targets) > 0:
status.append(f"Next prep target: {prep_targets[0]}")
status.append(f"Remaining prep targets: {len(prep_targets)}")
else:
status.append("No prep targets loaded")
status.append(self.get_tip_status())
# Get current session status if available
if self.session_id:
try:
response = requests.get(
url=f"{self.base_url}/sessions/{self.session_id}",
headers=self.headers,
)
if response.status_code == 200:
session_data = response.json().get("data", {})
current_state = session_data.get("details", {}).get(
"currentState", "unknown"
)
status.append(f"Session state: {current_state}")
except requests.exceptions.RequestException:
status.append("Unable to get session status")
# Get pipette information for mounts that are loaded into the active run
for mount, pipette in self._get_active_pipettes().items():
if pipette:
status.append(
f"Pipette on {mount} mount: {pipette.get('model', 'unknown')}"
)
# Get loaded labware information
try:
for slot, (labware_id, name, _) in self.config["loaded_labware"].items():
status.append(f"Labware in slot {slot}: {name}")
except Exception:
print(self.config["loaded_labware"])
return status
[docs]
@Driver.quickbar(
qb={
"button_text": "Refill Tipracks",
"params": {
"mount": {
"label": "Which Pipet left/right/both",
"type": "text",
"default": "both",
},
},
}
)
def reset_tipracks(self, mount="both"):
"""Reset available-tip tracking for one or more mounts.
Parameters
----------
mount : {"left", "right", "both"}, default="both"
Mount selection to reset.
"""
self.log_info(f"Resetting tipracks for {mount} mount")
mounts_to_reset = []
if mount == "both":
mounts_to_reset = list(self.config["loaded_instruments"].keys())
else:
mounts_to_reset = [mount]
for m in mounts_to_reset:
if m in self.config["loaded_instruments"]:
# Reinitialize available tips for this mount
self.config["available_tips"][m] = []
for tiprack in self.config["loaded_instruments"][m]["tip_racks"]:
for well in TIPRACK_WELLS:
self.config["available_tips"][m].append((tiprack, well))
self.log_info(f"Reset {len(self.config['available_tips'][m])} tips for {m} mount")
# Reset tip status
self.has_tip = False
self.current_tip = None
[docs]
def reset(self):
"""Reset the active OT-2 session, protocol, and persisted deck state."""
self.log_info("Resetting the protocol context")
# Delete any active session
if self.session_id:
try:
requests.delete(
url=f"{self.base_url}/sessions/{self.session_id}",
headers=self.headers,
)
except requests.exceptions.RequestException as e:
self.log_error(f"Error deleting session: {str(e)}")
# Delete any uploaded protocol
if self.protocol_id:
try:
requests.delete(
url=f"{self.base_url}/protocols/{self.protocol_id}",
headers=self.headers,
)
except requests.exceptions.RequestException as e:
self.log_error(f"Error deleting protocol: {str(e)}")
# Reset state variables
self.session_id = None
self.protocol_id = None
self.has_tip = False
self.last_pipette = None
self.current_tip = None
# Reset deck configuration too
self.reset_deck()
# Re-initialize robot connection
self._initialize_robot()
[docs]
def reset_deck(self):
"""Clear persisted deck configuration and related in-memory state."""
self.log_info("Resetting the deck configuration")
# Clear the deck configuration
self.config["loaded_labware"] = {}
self.config["loaded_instruments"] = {}
self.config["loaded_modules"] = {}
self.config["available_tips"] = {}
self.config["prep_targets"] = []
# Clear internal state variables
self.modules = {}
self.sent_custom_labware = {}
self.run_id = None
self.current_tip = None
[docs]
@Driver.quickbar(qb={"button_text": "Home"})
def home(self, **kwargs):
"""
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.
"""
self.log_info("Homing the robot's axes")
try:
# Call the dedicated home endpoint
response = requests.post(
url=f"{self.base_url}/robot/home",
headers=self.headers,
json={
"target": "robot", # Home the entire robot
},
)
if response.status_code != 200:
self.log_error(f"Failed to home robot: {response.status_code}")
self.log_error(f"Response: {response.text}")
raise RuntimeError(f"Failed to home robot: {response.text}")
self.log_info("Robot homing completed successfully")
return True
except requests.exceptions.RequestException as e:
self.log_error(f"Error during homing: {str(e)}")
raise RuntimeError(f"Error during homing: {str(e)}")
[docs]
def parse_well(self, loc):
"""Split a deck location into slot and well components.
Parameters
----------
loc : str
Deck location such as ``"1A1"``.
Returns
-------
tuple
Two-item tuple ``(slot, well)``.
"""
# Default value in case no alphabetic character is found
i = 0
for i, loc_part in enumerate(list(loc)):
if loc_part.isalpha():
break
slot = loc[:i]
well = loc[i:]
return slot, well
[docs]
def get_wells(self, locs):
"""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
-------
list of dict
Well descriptors containing ``labwareId`` and ``wellName``.
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.
"""
self.log_debug(f"Converting locations to well objects: {locs}")
wells = []
for loc in listify(locs):
slot, well = self.parse_well(loc)
# Get labware info from the slot
labware_info = self.config['loaded_labware'].get(slot)
if not labware_info:
raise ValueError(f"No labware found in slot {slot}")
if not isinstance(labware_info, tuple) or len(labware_info) < 1:
raise ValueError(f"Invalid labware info format in slot {slot}")
labware_id = labware_info[0]
wells.append({"labwareId": labware_id, "wellName": well})
self.log_debug(f"Created well objects: {wells}")
# Check well validity here
assert slot in self.config["loaded_labware"].keys(), f"Slot {slot} does not have any loaded labware"
assert well in self.config["loaded_labware"][slot][2]['definition']['wells'].keys(), f"Well {well} is not a valid well for slot {slot}, {self.config['loaded_labware'][slot][2]['definition']['metadata']['displayName']}"
return wells
def _check_cmd_success(self, response):
"""Raise when an HTTP command response indicates failure.
Parameters
----------
response : requests.Response
Response returned by the robot server.
"""
if response.status_code != 201:
self.log_error(
f"Failed to execute command : {response.status_code}"
)
self.log_error(f"Response: {response.text}")
raise RuntimeError(
f"Failed to execute command: {response.text}"
)
if 'status' in response.json()['data'].keys():
if response.json()['data']['status'] == 'failed':
self.log_error(
f"Command returned error : {response.status_code}"
)
self.log_error(f"Response: {response.text}")
raise RuntimeError(
f"Command returned error: {response.text}"
)
[docs]
def send_labware(
self, labware_def, check_run_status=True, reload_loaded_labware=True
):
"""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
-------
dict
Upload metadata including definition URI, version, and content hash.
"""
self.log_debug(f"Sending custom labware definition: {labware_def}")
ns, load_name, key = self._custom_labware_key(labware_def)
content_hash = self._hash_labware_def(labware_def)
# Persist the definition for future use
self.custom_labware_dir.mkdir(parents=True, exist_ok=True)
file_path = self._custom_labware_file_path(labware_def)
file_path.parent.mkdir(parents=True, exist_ok=True)
with open(file_path, "w") as f:
json.dump(labware_def, f, indent=2)
self._load_custom_labware_defs()
existing = self._labware_upload_info(key)
if existing is not None and existing.get("content_hash") == content_hash:
self.log_debug(
f"Labware {key} already sent to robot as version {existing['version']}"
)
return copy.deepcopy(existing)
# Ensure we have a valid run
run_id = self._ensure_run_exists(check_run_status=check_run_status)
try:
upload_def = copy.deepcopy(labware_def)
upload_version = self._next_custom_labware_version(key, upload_def)
upload_def["version"] = upload_version
command_dict = {"data": upload_def}
response = requests.post(
url=f"{self.base_url}/runs/{run_id}/labware_definitions",
headers=self.headers,
params={"waitUntilComplete": True},
json=command_dict,
)
self._check_cmd_success(response)
response_data = response.json()
labware_name = response_data["data"]["definitionUri"]
upload_info = {
"definition_uri": labware_name,
"version": upload_version,
"content_hash": content_hash,
}
self.sent_custom_labware[key] = upload_info
if reload_loaded_labware:
self._reload_matching_labware_definition(
upload_def, run_id=run_id, check_run_status=False
)
self.log_info(
f"Successfully sent custom labware with name/URI {labware_name}"
)
return copy.deepcopy(upload_info)
except (requests.exceptions.RequestException, KeyError) as e:
self.log_error(f"Error sending custom labware: {str(e)}")
raise RuntimeError(f"Error sending custom labware: {str(e)}")
[docs]
def load_labware(self, name, slot, module=None, check_run_status=True, **kwargs):
"""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
-------
str
Loaded labware identifier returned by the robot.
"""
self.log_debug(f"Loading labware '{name}' into slot '{slot}'")
# Ensure we have a valid run
run_id = self._ensure_run_exists(check_run_status=check_run_status)
labware_json = kwargs.pop("labware_json", None)
version = 1
if labware_json is not None:
namespace = labware_json.get("namespace", "custom_beta")
load_name = labware_json.get("parameters", {}).get("loadName")
if not load_name:
raise ValueError("labware_json missing parameters.loadName")
name = load_name
version = int(labware_json.get("version", 1) or 1)
if namespace != "opentrons":
upload_info = self.send_labware(
labware_json,
check_run_status=check_run_status,
reload_loaded_labware=False,
)
version = int(upload_info["version"])
else:
if "/" in name:
namespace, load_name = name.split("/", 1)
name = load_name
else:
load_name = name
if f"custom_beta/{load_name}" in self.custom_labware_files:
namespace = "custom_beta"
elif f"opentrons/{load_name}" in self.custom_labware_files:
namespace = "opentrons"
else:
namespace = "opentrons"
key = f"{namespace}/{load_name}"
if namespace != "opentrons":
path = self.custom_labware_files.get(key)
if path and Path(path).exists():
with open(path, "r") as f:
definition = json.load(f)
upload_info = self.send_labware(
definition,
check_run_status=check_run_status,
reload_loaded_labware=False,
)
version = int(upload_info["version"])
else:
self.log_warning(f"Custom labware definition not found for {key}")
try:
# Check if there's existing labware in the slot
if slot in self.config["loaded_labware"]:
self.log_info(
f"Found existing labware in slot {slot}, moving it off-deck first"
)
existing_labware_id = self.config["loaded_labware"][slot][
0
] # Get the ID of existing labware
# Create command to move existing labware off-deck
move_command = {
"data": {
"commandType": "moveLabware",
"params": {
"labwareId": existing_labware_id,
"newLocation": "offDeck",
"strategy": "manualMoveWithoutPause", # Allow user to manually move the labware
},
"intent": "setup",
}
}
# Execute the move command
move_response = requests.post(
url=f"{self.base_url}/runs/{run_id}/commands",
headers=self.headers,
params={"waitUntilComplete": True},
json=move_command,
)
self._check_cmd_success(move_response)
# Remove from our tracking
del self.config["loaded_labware"][slot]
if str(slot) in self.config["loaded_modules"].keys():
# we need to load into a module, not a slot
location = {"moduleId": self.config["loaded_modules"][str(slot)][0]}
else:
location = {"slotName": str(slot)}
# Prepare the loadLabware command
command_dict = {
"data": {
"commandType": "loadLabware",
"params": {
"location": location,
"loadName": name,
"namespace": namespace,
"version": version,
},
"intent": "setup",
}
}
# If this is a module, we need to specify the moduleId
if module:
command_dict["data"]["params"]["moduleId"] = module
# Execute the command
response = requests.post(
url=f"{self.base_url}/runs/{run_id}/commands",
headers=self.headers,
params={"waitUntilComplete": True},
json=command_dict,
)
self._check_cmd_success(response)
# Get the labware ID from the response
response_data = response.json()
# Debug log the response structure
self.log_debug(f"Load labware response: {response_data}")
# Handle different response structures that might occur
try:
if "data" in response_data and "result" in response_data["data"]:
labware_id = response_data["data"]["result"]["labwareId"]
elif "data" in response_data and "labwareId" in response_data["data"]:
labware_id = response_data["data"]["labwareId"]
elif "data" in response_data and "id" in response_data["data"]:
labware_id = response_data["data"]["id"]
else:
# Try to find labware ID in any structure
self.log_warning(f"Unexpected response structure: {response_data}")
for key, value in response_data.items():
if isinstance(value, dict) and "labwareId" in value:
labware_id = value["labwareId"]
break
else:
raise KeyError("Could not find labwareId in response")
except KeyError as e:
self.log_error(f"Error extracting labware ID from response: {str(e)}")
self.log_error(f"Response data: {response_data}")
raise RuntimeError(
f"Failed to extract labware ID from response: {str(e)}"
)
result = response_data["data"]["result"]
# Store the labware information directly in config
self.config["loaded_labware"][slot] = (labware_id, name, result)
# If this is a module, store it
if module:
self.modules[slot] = module
self.log_info(
f"Successfully loaded labware '{name}' in slot {slot} with ID {labware_id}"
)
self.config._update_history()
return labware_id
except (requests.exceptions.RequestException, KeyError) as e:
self.log_error(f"Error loading labware: {str(e)}")
raise RuntimeError(f"Error loading labware: {str(e)}")
[docs]
def load_module(self, name, slot, check_run_status=True, **kwargs):
"""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
-------
str
Loaded module identifier returned by the robot.
"""
slot = str(slot)
self.log_debug(f"Loading module '{name}' into slot '{slot}'")
existing_module = self.config["loaded_modules"].get(slot)
if existing_module is not None:
try:
existing_module_id, existing_name = existing_module
except (TypeError, ValueError) as exc:
raise RuntimeError(
f"Cannot load module {name!r} in deck slot {slot!r}: the stored "
f"module record is invalid: {existing_module!r}"
) from exc
if existing_name == name:
self.log_info(
f"Module {name!r} is already loaded in deck slot {slot!r} "
f"with ID {existing_module_id!r}; reusing it."
)
return existing_module_id
raise RuntimeError(
f"Cannot load module {name!r} in deck slot {slot!r}: slot already "
f"contains module {existing_name!r} with ID {existing_module_id!r}. "
"Unload or reset the existing module before replacing it."
)
# Ensure we have a valid run
run_id = self._ensure_run_exists(check_run_status=check_run_status)
try:
# Prepare the loadLabware command
command_dict = {
"data": {
"commandType": "loadModule",
"params": {
"location": {"slotName": str(slot)},
"model": name,
},
"intent": "setup",
}
}
# Execute the command
response = requests.post(
url=f"{self.base_url}/runs/{run_id}/commands",
headers=self.headers,
params={"waitUntilComplete": True},
json=command_dict,
)
try:
self._check_cmd_success(response)
except RuntimeError:
message = self._module_load_failure_message(name, slot, response)
self.log_error(message)
raise RuntimeError(message) from None
# Get the labware ID from the response
response_data = response.json()
# Debug log the response structure
self.log_debug(f"Load labware response: {response_data}")
# Handle different response structures that might occur
try:
if "data" in response_data and "result" in response_data["data"]:
module_id = response_data["data"]["result"]["moduleId"]
elif "data" in response_data and "moduleId" in response_data["data"]:
module_id = response_data["data"]["moduleId"]
elif "data" in response_data and "id" in response_data["data"]:
module_id = response_data["data"]["id"]
else:
# Try to find labware ID in any structure
self.log_warning(f"Unexpected response structure: {response_data}")
for key, value in response_data.items():
if isinstance(value, dict) and "moduleId" in value:
module_id = value["moduleId"]
break
else:
raise KeyError("Could not find moduleId in response")
except KeyError as e:
self.log_error(f"Error extracting module ID from response: {str(e)}")
self.log_error(f"Response data: {response_data}")
raise RuntimeError(
f"Failed to extract module ID from response: {str(e)}"
)
# Store the module information directly in config
self.config["loaded_modules"][slot] = (module_id, name)
self.log_info(
f"Successfully loaded module '{name}' in slot {slot} with ID {module_id}"
)
self.config._update_history()
return module_id
except (requests.exceptions.RequestException, KeyError) as e:
self.log_error(f"Error loading module: {str(e)}")
raise RuntimeError(f"Error loading module: {str(e)}")
@staticmethod
def _module_load_failure_message(name, slot, response):
"""Create an actionable error message from an OT-2 load-module response."""
error_type = None
error_code = None
detail = None
try:
error = response.json().get("data", {}).get("error", {})
error_type = error.get("errorType")
error_code = error.get("errorCode")
detail = error.get("detail")
except (AttributeError, TypeError, ValueError):
pass
reported_error = ""
if error_type:
reported_error = f" Robot reported {error_type}"
if error_code:
reported_error += f" (code {error_code})"
if detail:
reported_error += f": {detail}"
reported_error += "."
elif detail:
reported_error = f" Robot reported: {detail}."
return (
f"Unable to load OT-2 module {name!r} in deck slot {str(slot)!r}."
f"{reported_error} Verify that the module is connected to the OT-2, "
"powered if required, matches the requested model, and is detected by "
"the Opentrons hardware server before retrying."
)
[docs]
def load_instrument(self, name, mount, tip_rack_slots, reload=False, check_run_status=True, update_pipettes=True, **kwargs):
"""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
-------
str
Loaded pipette identifier returned by the robot.
"""
pipette_name = self._normalize_pipette_name(name)
mount = str(mount).strip().lower()
if mount not in {"left", "right"}:
raise ValueError(f"Mount must be 'left' or 'right'. Received: {mount!r}")
tip_rack_slots = [str(slot) for slot in listify(tip_rack_slots)]
if len(tip_rack_slots) == 0:
raise ValueError("At least one tip rack slot must be provided.")
for slot in tip_rack_slots:
if slot not in self.config["loaded_labware"]:
raise ValueError(
f"Tip rack slot {slot!r} is not loaded. Load a tiprack first."
)
labware_name = str(self.config["loaded_labware"][slot][1]).lower()
if "tiprack" not in labware_name:
self.log_warning(
f"Slot {slot} contains labware '{labware_name}', which may not be a tiprack."
)
self._warn_on_tiprack_mismatch(pipette_name, tip_rack_slots)
self.log_debug(
f"Loading pipette '{pipette_name}' on '{mount}' mount with tip_racks in slots {tip_rack_slots}"
)
# Ensure we have a valid run
run_id = self._ensure_run_exists(check_run_status=check_run_status)
try:
# First, load the pipette using the HTTP API
command_dict = {
"data": {
"commandType": "loadPipette",
"params": {
"pipetteName": pipette_name,
"mount": mount,
"tip_racks": [self.config["loaded_labware"][str(slot)][0] for slot in tip_rack_slots],
},
"intent": "setup",
}
}
# Execute the loadPipette command
response = requests.post(
url=f"{self.base_url}/runs/{run_id}/commands",
headers=self.headers,
params={"waitUntilComplete": True},
json=command_dict,
)
self._check_cmd_success(response)
# Get the pipette ID from the response
response_data = response.json()
logging.debug(f'loadPipette response: {response_data}')
pipette_id = response_data["data"]["result"]["pipetteId"]
# Make sure we have the latest pipette information (unless disabled for optimization)
if update_pipettes:
self._update_pipettes()
# Ensure pipette_info entry exists before patching
if mount not in self.pipette_info:
self.pipette_info[mount] = {}
self.pipette_info[mount][
"id"
] = pipette_id # patch the correct pipette id to the pipette_info dict
# Get the tip rack IDs - note that loaded_labware now stores tuples of (id, name)
tip_racks = []
for slot in listify(tip_rack_slots):
labware_info = self.config["loaded_labware"].get(slot)
if (
labware_info
and isinstance(labware_info, tuple)
and len(labware_info) >= 1
):
tip_racks.append(labware_info[0])
if not tip_racks:
self.log_warning(f"No valid tip racks found in slots {tip_rack_slots}")
# Store the instrument information
self.config["loaded_instruments"][mount] = {
"name": pipette_name,
"pipette_id": pipette_id,
"tip_racks": tip_racks,
}
# If not reloading, initialize available tips for this mount
if not reload:
self.config["available_tips"][mount] = []
for tiprack in tip_racks:
for well in TIPRACK_WELLS:
self.config["available_tips"][mount].append((tiprack, well))
# Verify that there's actually a pipette in this mount
if mount not in self.pipette_info or self.pipette_info[mount] is None:
self.log_warning(
f"No physical pipette detected in {mount} mount, but pipette information stored"
)
# Update min/max values for largest and smallest pipettes
self._update_pipette_ranges()
self.log_info(
f"Successfully loaded pipette '{pipette_name}' on {mount} mount with ID {pipette_id}"
)
self.config._update_history()
return pipette_id
except (requests.exceptions.RequestException, KeyError) as e:
self.log_error(f"Error loading pipette: {str(e)}")
raise RuntimeError(f"Error loading pipette: {str(e)}")
def _normalize_pipette_name(self, name):
"""Normalize a pipette alias to the canonical Opentrons name."""
key = str(name).strip().lower()
normalized = self.PIPETTE_NAME_ALIASES.get(key, key)
return normalized
def _warn_on_tiprack_mismatch(self, pipette_name, tip_rack_slots):
"""Warn when tiprack names appear incompatible with a pipette.
Parameters
----------
pipette_name : str
Canonical pipette name.
tip_rack_slots : sequence of str
Slots containing candidate tipracks.
"""
token = self.EXPECTED_TIPRACK_TOKEN.get(pipette_name)
if token is None:
return
mismatched_slots = []
for slot in tip_rack_slots:
labware_info = self.config["loaded_labware"].get(str(slot))
if labware_info is None:
continue
labware_name = str(labware_info[1]).lower()
if "tiprack" in labware_name and token not in labware_name:
mismatched_slots.append(slot)
if mismatched_slots:
self.log_warning(
f"Loaded pipette '{pipette_name}' with potentially mismatched tipracks in slots {mismatched_slots}. "
f"Expected tiprack names containing '{token}'."
)
def _update_pipette_ranges(self):
"""Update helper ranges used for transfer splitting across pipettes."""
self.min_largest_pipette = None
self.max_smallest_pipette = None
# Get all available pipettes with their volumes
available_pipettes = self._get_active_pipettes()
if available_pipettes:
# Get min and max volumes for each pipette
min_vols = {
mount: info.get("min_volume", float("inf"))
for mount, info in available_pipettes.items()
}
max_vols = {
mount: info.get("max_volume", 0)
for mount, info in available_pipettes.items()
}
# Find the smallest and largest pipettes
if max_vols:
# Use list and regular max/min functions with a key function
mounts = list(max_vols.keys())
if mounts:
largest_pipette_mount = max(
mounts, key=lambda m: max_vols.get(m, 0)
)
smallest_pipette_mount = min(
mounts, key=lambda m: max_vols.get(m, float("inf"))
)
# Set global min/max values
if min_vols and largest_pipette_mount in min_vols:
self.min_largest_pipette = min_vols[largest_pipette_mount]
self.log_info(
f"Setting min_largest_pipette to {self.min_largest_pipette}"
)
if max_vols and smallest_pipette_mount in max_vols:
self.max_smallest_pipette = max_vols[smallest_pipette_mount]
self.log_info(
f"Setting max_smallest_pipette to {self.max_smallest_pipette}"
)
[docs]
def mix(self, volume, location, repetitions=1, **kwargs):
"""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.
"""
self.log_info(f"Mixing {volume}uL {repetitions} times at {location}")
# Verify run exists once at the start, then skip checks for all atomic commands
self._ensure_run_exists()
# Get pipette based on volume
pipette = self.get_pipette(volume)
pipette_mount = pipette["mount"] # Get the mount from the pipette object
# Get the pipette ID
pipette_id = None
for mount, data in self.pipette_info.items():
if mount == pipette_mount and data:
pipette_id = data.get("id")
break
if not pipette_id:
raise ValueError(f"Could not find ID for pipette on {pipette_mount} mount")
# Get well location
wells = self.get_wells(location)
if not wells:
raise ValueError("Invalid location")
well = wells[0]
# Pick up tip if needed
if not self.has_tip:
self._execute_atomic_command(
"pickUpTip",
{
"pipetteId": pipette_id,
"pipetteMount": pipette_mount,
"wellLocation": None,
},
check_run_status=False,
)
self.has_tip = True
# Execute mix by performing repetitions of aspirate/dispense
for _ in range(repetitions):
self._execute_atomic_command(
"aspirate",
{
"pipetteId": pipette_id,
"volume": volume,
"labwareId": well["labwareId"],
"wellName": well["wellName"],
"wellLocation": {
"origin": "bottom",
"offset": {"x": 0, "y": 0, "z": 0},
},
},
check_run_status=False,
)
self._execute_atomic_command(
"dispense",
{
"pipetteId": pipette_id,
"volume": volume,
"labwareId": well["labwareId"],
"wellName": well["wellName"],
"wellLocation": {
"origin": "bottom",
"offset": {"x": 0, "y": 0, "z": 0},
},
},
check_run_status=False,
)
[docs]
def transfer(
self,
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,
):
"""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
-------
dict
Structured transfer metadata including selected pipette, subtransfer
volumes, source and destination well metadata, and applied options.
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},
... )
"""
self.log_info(f"Transferring {volume}uL from {source} to {dest}")
if drop_tip and return_tip:
raise ValueError("Only one of drop_tip and return_tip can be True")
if "blowout" in kwargs and not blow_out:
blow_out = bool(kwargs["blowout"])
if "touchTip" in kwargs and not touch_tip:
touch_tip = bool(kwargs["touchTip"])
volume_ul = float(volume)
if volume_ul <= 0:
self.log_info(f"Skipping transfer with nonpositive volume {volume_ul}uL from {source} to {dest}")
return {
"source": source,
"dest": dest,
"requested_volume_ul": volume_ul,
"subtransfers_ul": [],
"status": "skipped_nonpositive_volume",
}
self._ensure_run_exists()
if aspirate_rate is not None:
self.set_aspirate_rate(aspirate_rate)
if dispense_rate is not None:
self.set_dispense_rate(dispense_rate)
pipette = self.get_pipette(volume_ul)
pipette_mount = pipette["mount"]
resolve_tip_rack_offset = getattr(self, "_resolve_tip_rack_offset", None)
if resolve_tip_rack_offset is not None:
resolved_tip_rack_offset = resolve_tip_rack_offset(
tip_rack_offset,
mount=pipette_mount,
)
elif tip_rack_offset is None:
resolved_tip_rack_offset = dict(self.config.get("tip_rack_offset", {"x": 0, "y": 0, "z": 0}))
else:
resolved_tip_rack_offset = dict(tip_rack_offset)
requested_tip = None
if tip_location is not None:
tip_location_str = str(tip_location).strip().upper()
resolve_tip_location = getattr(self, "_resolve_tip_location", None)
if resolve_tip_location is not None:
requested_tip = resolve_tip_location(pipette_mount, tip_location_str)
else:
requested_tip = {
"labware_id": self.config["loaded_labware"][str(tip_location_str[0])][0],
"well_name": tip_location_str[1:],
}
requested_tip["location"] = tip_location_str
pipette_id = None
for mount, data in self.pipette_info.items():
if mount == pipette_mount and data:
pipette_id = data.get("id")
break
if not pipette_id:
raise ValueError(f"Could not find ID for pipette on {pipette_mount} mount")
source_wells = self.get_wells(source)
if len(source_wells) > 1:
raise ValueError("Transfer only accepts one source well at a time!")
source_well = source_wells[0]
dest_wells = self.get_wells(dest)
if len(dest_wells) > 1:
raise ValueError("Transfer only accepts one dest well at a time!")
dest_well = dest_wells[0]
source_position = "bottom"
dest_position = "bottom"
if to_top and to_center:
raise ValueError("Cannot dispense to_top and to_center simultaneously")
elif to_top:
dest_position = "top"
elif to_center:
dest_position = "center"
split_up_transfers = getattr(self, "_split_up_transfers", None)
if split_up_transfers is not None:
transfers = split_up_transfers(volume_ul)
else:
transfers = [volume_ul]
transfer_record = {
"source": source,
"dest": dest,
"requested_volume_ul": volume_ul,
"subtransfers_ul": [],
"subtransfer_count": 0,
"pipette_mount": pipette_mount,
"pipette_name": pipette.get("name"),
"pipette_id": pipette_id,
"source_well": {
"labware_id": source_well["labwareId"],
"well_name": source_well["wellName"],
"position": source_position,
},
"dest_well": {
"labware_id": dest_well["labwareId"],
"well_name": dest_well["wellName"],
"position": dest_position,
"offset": {"x": 0, "y": 0, "z": to_top_z_offset if dest_position == "top" else 0},
},
"options": {
"mix_before": list(mix_before) if mix_before is not None else None,
"mix_after": list(mix_after) if mix_after is not None else None,
"air_gap": air_gap,
"aspirate_rate": aspirate_rate,
"dispense_rate": dispense_rate,
"mix_aspirate_rate": mix_aspirate_rate,
"mix_dispense_rate": mix_dispense_rate,
"blow_out": blow_out,
"post_aspirate_delay": post_aspirate_delay,
"aspirate_equilibration_delay": aspirate_equilibration_delay,
"post_dispense_delay": post_dispense_delay,
"drop_tip": drop_tip,
"return_tip": return_tip,
"force_new_tip": force_new_tip,
"to_top": to_top,
"to_center": to_center,
"to_top_z_offset": to_top_z_offset,
"tip_rack_offset": dict(resolved_tip_rack_offset),
"fast_mixing": fast_mixing,
"touch_tip": touch_tip,
"tip_location": tip_location,
},
"status": "executed",
}
if requested_tip is not None:
transfer_record["requested_tip"] = requested_tip.copy()
for i, sub_volume in enumerate(transfers):
if sub_volume <= 0:
self.log_warning(
f"Skipping nonpositive sub-transfer volume {sub_volume}uL from {source} to {dest}"
)
continue
transfer_record["subtransfers_ul"].append(float(sub_volume))
# Final tip disposal is controlled by drop_tip / return_tip alone.
# force_new_tip only controls whether a used tip is cleared before
# the next subtransfer picks up a fresh one.
is_last_subtransfer = i == (len(transfers) - 1)
effective_drop_tip = bool(drop_tip and is_last_subtransfer)
effective_return_tip = bool(return_tip and is_last_subtransfer)
# Keep tip handling consistent with the non-HTTP driver:
# reuse the current tip across split transfers unless force_new_tip is set.
if force_new_tip and self.has_tip:
tip_mount = self.last_pipette if self.last_pipette is not None else pipette_mount
tip_pipette_id = self.pipette_info.get(tip_mount, {}).get("id", pipette_id)
if return_tip:
self._return_tip_to_origin(
tip_pipette_id,
mount=tip_mount,
tip_rack_offset=resolved_tip_rack_offset,
return_tip_z_offset=return_tip_z_offset,
)
else:
self._drop_tip_to_trash(tip_pipette_id)
# If a tip is on a different mount, drop it before switching mounts.
if self.has_tip and self.last_pipette not in (None, pipette_mount):
tip_pipette_id = self.pipette_info.get(self.last_pipette, {}).get("id", pipette_id)
if return_tip:
self._return_tip_to_origin(
tip_pipette_id,
mount=self.last_pipette,
tip_rack_offset=resolved_tip_rack_offset,
return_tip_z_offset=return_tip_z_offset,
)
else:
self._drop_tip_to_trash(tip_pipette_id)
if (
requested_tip is None
and self.has_tip
and self.last_pipette == pipette_mount
and self._current_tip_is_reserved_stock_tip()
):
if return_tip:
self._return_tip_to_origin(
pipette_id,
mount=pipette_mount,
tip_rack_offset=resolved_tip_rack_offset,
return_tip_z_offset=return_tip_z_offset,
)
else:
self._drop_tip_to_trash(pipette_id)
needs_requested_tip = (
requested_tip is not None
and (
not self.has_tip
or self.last_pipette != pipette_mount
or self.current_tip is None
or self.current_tip.get("labware_id") != requested_tip["labware_id"]
or self.current_tip.get("well_name") != requested_tip["well_name"]
)
)
if needs_requested_tip and self.has_tip:
if return_tip:
self.return_tip(
tip_location=self.current_tip.get("location") if self.current_tip is not None else None,
tip_rack_offset=resolved_tip_rack_offset,
return_tip_z_offset=return_tip_z_offset,
)
else:
self._drop_tip_to_trash(pipette_id)
if not self.has_tip:
if requested_tip is not None:
requested_tip_location = (
f"{self._slot_by_labware_uuid(requested_tip['labware_id'])}{requested_tip['well_name']}"
)
self.pickup_tip(
requested_tip_location,
tip_rack_offset=resolved_tip_rack_offset,
)
else:
self._execute_atomic_command(
"pickUpTip",
{
"pipetteId": pipette_id,
"pipetteMount": pipette_mount,
"tipRackOffset": dict(resolved_tip_rack_offset),
},
check_run_status=False,
)
self.has_tip = True
self.last_pipette = pipette_mount
# 1a. If destination is on a heater-shaker, stop the shaking and latch the latch pre-flight
was_shaking = False
dest_well_slot = self._slot_by_labware_uuid(dest_well['labwareId'])
source_well_slot = self._slot_by_labware_uuid(source_well['labwareId'])
heater_shaker_slots = [slot for (slot,(uuid,name)) in self.config["loaded_modules"].items() if "heaterShaker" in name]
if dest_well_slot in heater_shaker_slots or source_well_slot in heater_shaker_slots:
# latch heater-shaker
# this is contextual, maybe - seems to not cause trouble to run without conditional
#if 'closed' not in self.get_shake_latch_status():
self.latch_shaker()
# store current shake rpm and stop shake
if self.get_shake_rpm()[0] != 'idle':
shake_rpm = self.get_shake_rpm()[2]
was_shaking = True
self.stop_shake()
# 2. Mix before if specified
if mix_before is not None:
n_mixes, mix_volume = mix_before
# Set mix aspirate rate if specified
if mix_aspirate_rate is not None:
self.set_aspirate_rate(mix_aspirate_rate, pipette_mount)
# Set mix dispense rate if specified
if mix_dispense_rate is not None:
self.set_dispense_rate(mix_dispense_rate, pipette_mount)
# Mix before transfer - implement by executing multiple aspirate/dispense
for _ in range(n_mixes):
self._execute_atomic_command(
"aspirate",
{
"pipetteId": pipette_id,
"volume": mix_volume,
"labwareId": source_well["labwareId"],
"wellName": source_well["wellName"],
"wellLocation": {
"origin": source_position,
"offset": {"x": 0, "y": 0, "z": 0},
},
"flowRate": self.pipette_info[pipette_mount]['aspirate_flow_rate'],
},
check_run_status=False,
)
self._execute_atomic_command(
"dispense",
{
"pipetteId": pipette_id,
"volume": mix_volume,
"labwareId": source_well["labwareId"],
"wellName": source_well["wellName"],
"wellLocation": {
"origin": source_position,
"offset": {"x": 0, "y": 0, "z": 0},
},
"flowRate": self.pipette_info[pipette_mount]['dispense_flow_rate'],
},
check_run_status=False,
)
# Restore original rates
if mix_aspirate_rate is not None or mix_dispense_rate is not None:
# Reset rates to default or specified rates
if aspirate_rate is not None:
self.set_aspirate_rate(aspirate_rate, pipette_mount)
if dispense_rate is not None:
self.set_dispense_rate(dispense_rate, pipette_mount)
# 3. Aspirate
self._execute_atomic_command(
"aspirate",
{
"pipetteId": pipette_id,
"volume": sub_volume,
"labwareId": source_well["labwareId"],
"wellName": source_well["wellName"],
"wellLocation": {
"origin": source_position,
"offset": {"x": 0, "y": 0, "z": source_z_offset},
},
"flowRate": self.pipette_info[pipette_mount]['aspirate_flow_rate'],
},
check_run_status=False,
)
# 4. Aspirate equilibration delay (while tip is in liquid)
if aspirate_equilibration_delay > 0:
time.sleep(aspirate_equilibration_delay)
# self._execute_atomic_command("delay", {"seconds": aspirate_equilibration_delay})
# 5. Move tip above liquid and post-aspirate delay (tip above liquid)
self._execute_atomic_command(
"moveToWell",
{
"pipetteId": pipette_id,
"labwareId": source_well["labwareId"],
"wellName": source_well["wellName"],
"wellLocation": {
"origin": "top",
"offset": {"x": 0, "y": 0, "z": 0},
},
},
check_run_status=False,
)
if post_aspirate_delay > 0:
time.sleep(post_aspirate_delay)
# self._execute_atomic_command("delay", {"seconds": post_aspirate_delay})
# 6. Air gap if specified
if air_gap > 0:
# Air gap is implemented as aspirate at the top of the source well
self._execute_atomic_command(
"aspirate",
{
"pipetteId": pipette_id,
"volume": air_gap,
"labwareId": source_well["labwareId"],
"wellName": source_well["wellName"],
"wellLocation": {
"origin": "top",
"offset": {"x": 0, "y": 0, "z": 0},
},
"flowRate": self.pipette_info[pipette_mount]['aspirate_flow_rate'],
},
check_run_status=False,
)
# 7. Dispense
offset = {
"x": 0,
"y": 0,
"z": (
to_top_z_offset
if dest_position == "top" and to_top_z_offset != 0
else 0
),
}
self._execute_atomic_command(
"dispense",
{
"pipetteId": pipette_id,
"volume": sub_volume
+ air_gap, # Include air gap in dispense volume
"labwareId": dest_well["labwareId"],
"wellName": dest_well["wellName"],
"wellLocation": {"origin": dest_position, "offset": offset},
"flowRate": self.pipette_info[pipette_mount]['dispense_flow_rate'],
},
check_run_status=False,
)
# 8. Post-dispense delay
if post_dispense_delay > 0:
time.sleep(post_dispense_delay)
# self._execute_atomic_command("delay", {"seconds": post_dispense_delay})
# 9. Mix after if specified
if mix_after is not None:
n_mixes, mix_volume = mix_after
# Set mix aspirate rate if specified
if mix_aspirate_rate is not None:
self.set_aspirate_rate(mix_aspirate_rate, pipette_mount)
# Set mix dispense rate if specified
if mix_dispense_rate is not None:
self.set_dispense_rate(mix_dispense_rate, pipette_mount)
# Mix after transfer should be performed from the bottom of the destination well
mix_well_location = {
"origin": "bottom",
"offset": {"x": 0, "y": 0, "z": 0},
}
# Mix after transfer - implement by executing multiple aspirate/dispense
for _ in range(n_mixes):
self._execute_atomic_command(
"aspirate",
{
"pipetteId": pipette_id,
"volume": mix_volume,
"labwareId": dest_well["labwareId"],
"wellName": dest_well["wellName"],
"wellLocation": mix_well_location,
"flowRate": self.pipette_info[pipette_mount]['aspirate_flow_rate'],
},
check_run_status=False,
)
self._execute_atomic_command(
"dispense",
{
"pipetteId": pipette_id,
"volume": mix_volume,
"labwareId": dest_well["labwareId"],
"wellName": dest_well["wellName"],
"wellLocation": mix_well_location,
"flowRate": self.pipette_info[pipette_mount]['dispense_flow_rate'],
},
check_run_status=False,
)
# Restore original rates
if mix_aspirate_rate is not None or mix_dispense_rate is not None:
# Reset rates to default or specified rates
if aspirate_rate is not None:
self.set_aspirate_rate(aspirate_rate, pipette_mount)
if dispense_rate is not None:
self.set_dispense_rate(dispense_rate, pipette_mount)
# 10. Blow out if specified
if blow_out:
flow_rate = self.pipette_info.get(pipette_mount, {}).get(
"dispense_flow_rate", 300
)
self._execute_atomic_command(
"blowout",
{
"pipetteId": pipette_id,
"labwareId": dest_well["labwareId"],
"wellName": dest_well["wellName"],
"wellLocation": {"origin": dest_position, "offset": offset},
"flowRate": flow_rate,
},
check_run_status=False,
)
# 10b. Optionally touch the tip to destination well edge.
if touch_tip:
self._touch_tip_well(pipette_id=pipette_id, well=dest_well)
if was_shaking:
self.set_shake(shake_rpm)
# back to running :)
# 11. Drop tip if specified
if effective_drop_tip:
self._drop_tip_to_trash(pipette_id)
elif effective_return_tip:
self._return_tip_to_origin(
pipette_id,
mount=pipette_mount,
tip_rack_offset=resolved_tip_rack_offset,
return_tip_z_offset=return_tip_z_offset,
)
# Update last pipette
self.last_pipette = pipette_mount
transfer_record["subtransfer_count"] = len(transfer_record["subtransfers_ul"])
return transfer_record
def _split_up_transfers(self, volume):
"""Split a requested transfer volume into pipette-safe subtransfers.
Parameters
----------
volume : float
Requested transfer volume in microliters.
Returns
-------
list of float
One or more subtransfer volumes.
"""
volume_ul = float(volume)
if volume_ul <= 0:
return []
if self.max_transfer is None or volume_ul <= self.max_transfer:
return [volume_ul]
transfers = []
remaining = volume_ul
while remaining > 0:
sub_volume = min(self.max_transfer, remaining)
transfers.append(float(sub_volume))
remaining -= sub_volume
return transfers
def _resolve_tip_rack_offset(self, tip_rack_offset=None, mount=None):
"""Resolve the configured tip-rack offset mapping.
Parameters
----------
tip_rack_offset : dict, optional
Explicit offset override.
mount : str, optional
Pipette mount used when the configured offsets are stored per mount.
Returns
-------
dict
Offset mapping with ``x``, ``y``, and ``z`` keys.
"""
offset = self.config.get("tip_rack_offset", {"x": 0, "y": 0, "z": 0})
if tip_rack_offset is not None:
offset = tip_rack_offset
if (
mount is not None
and isinstance(offset, dict)
and "x" not in offset
and "y" not in offset
and "z" not in offset
and mount in offset
):
offset = offset[mount]
resolved = copy.deepcopy(offset)
resolved.setdefault("x", 0)
resolved.setdefault("y", 0)
resolved.setdefault("z", 0)
return resolved
def _resolve_tip_location(self, mount, tip_location):
"""Resolve a deck tip location into a tracked tiprack/well pair.
Parameters
----------
mount : str
Pipette mount.
tip_location : str
Deck location such as ``"1A2"``.
Returns
-------
dict
Mapping with ``labware_id`` and ``well_name``.
Raises
------
ValueError
If the requested tip location is not available for the mount.
"""
normalized = str(tip_location).strip().upper()
if len(normalized) < 3:
raise ValueError(f"Requested tip location {tip_location} is invalid")
slot = normalized[0]
well_name = normalized[1:]
labware_info = self.config.get("loaded_labware", {}).get(str(slot))
if labware_info is None:
raise ValueError(f"Requested tip location {normalized} is not available")
labware_id = labware_info[0]
available = self.config.get("available_tips", {}).get(mount, [])
current_tip_matches = (
self.has_tip
and self.last_pipette == mount
and self.current_tip is not None
and self.current_tip.get("labware_id") == labware_id
and self.current_tip.get("well_name") == well_name
)
if (labware_id, well_name) not in available and not current_tip_matches:
raise ValueError(f"Requested tip location {normalized} is not available")
return {"labware_id": labware_id, "well_name": well_name}
def _resolve_tip_mount(self, tip_location):
"""Resolve a tip location to the unique loaded mount and pipette.
Parameters
----------
tip_location : str
Deck tip location such as ``"1A2"``.
Returns
-------
dict
Mapping with ``mount``, ``pipette_id``, ``tip_location``,
``labware_id``, and ``well_name``.
Raises
------
ValueError
If the tip location is invalid, unavailable, or maps to zero or
multiple loaded mounts.
"""
normalized = str(tip_location).strip().upper()
if len(normalized) < 3:
raise ValueError(f"Requested tip location {tip_location} is invalid")
slot = normalized[0]
labware_info = self.config.get("loaded_labware", {}).get(str(slot))
if labware_info is None:
raise ValueError(f"Requested tip location {normalized} is not available")
labware_id = labware_info[0]
matches = []
for mount, instrument in self.config.get("loaded_instruments", {}).items():
if labware_id not in instrument.get("tip_racks", []):
continue
pipette_id = self.pipette_info.get(mount, {}).get("id")
if pipette_id is None:
pipette_id = instrument.get("pipette_id")
requested_tip = self._resolve_tip_location(mount, normalized)
matches.append(
{
"mount": mount,
"pipette_id": pipette_id,
"tip_location": normalized,
"labware_id": requested_tip["labware_id"],
"well_name": requested_tip["well_name"],
}
)
if not matches:
raise ValueError(
f"No loaded instrument is configured for tip location {normalized}"
)
if len(matches) > 1:
mounts = ", ".join(sorted(match["mount"] for match in matches))
raise ValueError(
f"Tip location {normalized} is ambiguous across loaded mounts: {mounts}"
)
if not matches[0]["pipette_id"]:
raise ValueError(
f"Could not find pipette ID for mount {matches[0]['mount']}"
)
return matches[0]
[docs]
def pickup_tip(self, tip_location, tip_rack_offset=None):
"""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
-------
dict
Pickup metadata including mount, pipette, and requested tip
location.
Raises
------
ValueError
If the requested tip location is invalid or unavailable.
RuntimeError
If another tip is already attached.
"""
tip_target = self._resolve_tip_mount(tip_location)
current_tip_matches = (
self.has_tip
and self.last_pipette == tip_target["mount"]
and self.current_tip is not None
and self.current_tip.get("labware_id") == tip_target["labware_id"]
and self.current_tip.get("well_name") == tip_target["well_name"]
)
if current_tip_matches:
return {
"mount": tip_target["mount"],
"pipette_id": tip_target["pipette_id"],
"tip_location": tip_target["tip_location"],
"labware_id": tip_target["labware_id"],
"well_name": tip_target["well_name"],
"status": "already_attached",
}
if self.has_tip:
raise RuntimeError(
f"Cannot pick up tip {tip_target['tip_location']} while a tip is already attached on "
f"{self.last_pipette} mount"
)
self._execute_atomic_command(
"pickUpTip",
{
"pipetteId": tip_target["pipette_id"],
"pipetteMount": tip_target["mount"],
"labwareId": tip_target["labware_id"],
"wellName": tip_target["well_name"],
"tipRackOffset": self._resolve_tip_rack_offset(
tip_rack_offset,
mount=tip_target["mount"],
),
},
check_run_status=False,
)
self.has_tip = True
self.last_pipette = tip_target["mount"]
return {
"mount": tip_target["mount"],
"pipette_id": tip_target["pipette_id"],
"tip_location": tip_target["tip_location"],
"labware_id": tip_target["labware_id"],
"well_name": tip_target["well_name"],
"status": "picked_up",
}
[docs]
def return_tip(
self,
tip_location=None,
tip_rack_offset=None,
return_tip_z_offset=None,
):
"""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
-------
dict
Return metadata including the tip origin and status.
Raises
------
ValueError
If the requested tip location does not match the attached tip.
"""
if not self.has_tip or self.current_tip is None:
status = {"status": "no_tip_attached"}
if tip_location is not None:
status["tip_location"] = str(tip_location).strip().upper()
return status
attached_location = self.current_tip.get("location")
if attached_location is None:
slot = self.current_tip.get("slot")
well_name = self.current_tip.get("well_name")
if slot is not None and well_name is not None:
attached_location = f"{slot}{well_name}"
expected_tip_target = None
if tip_location is not None:
expected_tip_target = self._resolve_tip_mount(tip_location)
if (
self.current_tip.get("mount") != expected_tip_target["mount"]
or self.current_tip.get("labware_id") != expected_tip_target["labware_id"]
or self.current_tip.get("well_name") != expected_tip_target["well_name"]
):
raise ValueError(
f"Attached tip does not match requested return location {expected_tip_target['tip_location']}"
)
else:
mount = self.current_tip.get("mount")
expected_tip_target = {
"mount": mount,
"pipette_id": self.pipette_info.get(mount, {}).get("id"),
"tip_location": attached_location,
"labware_id": self.current_tip.get("labware_id"),
"well_name": self.current_tip.get("well_name"),
}
if expected_tip_target["pipette_id"] is None and mount in self.config.get(
"loaded_instruments", {}
):
expected_tip_target["pipette_id"] = self.config["loaded_instruments"][mount].get(
"pipette_id"
)
if not expected_tip_target.get("pipette_id"):
raise ValueError(
f"Could not find pipette ID for mount {expected_tip_target['mount']}"
)
self._return_tip_to_origin(
expected_tip_target["pipette_id"],
mount=expected_tip_target["mount"],
tip_rack_offset=tip_rack_offset,
return_tip_z_offset=return_tip_z_offset,
)
return {
"mount": expected_tip_target["mount"],
"pipette_id": expected_tip_target["pipette_id"],
"tip_location": expected_tip_target["tip_location"],
"labware_id": expected_tip_target["labware_id"],
"well_name": expected_tip_target["well_name"],
"status": "returned",
"offset": tip_rack_offset,
"z_offset" : return_tip_z_offset
}
def _reserve_tip(self, mount, labware_id, well_name):
"""Reserve a specific available tip for pickup.
Parameters
----------
mount : str
Pipette mount.
labware_id : str
Tiprack identifier.
well_name : str
Tip well name.
Returns
-------
tuple
Reserved ``(labware_id, well_name)`` pair.
Raises
------
ValueError
If the requested tip is not available.
"""
available = list(self.config.get("available_tips", {}).get(mount, []))
requested = (labware_id, well_name)
if requested not in available:
raise ValueError(
f"Requested tip location {well_name} in {labware_id} is not available for {mount} mount"
)
available.remove(requested)
self.config.setdefault("available_tips", {})[mount] = available
return requested
def _touch_tip_well(self, pipette_id, well):
"""Touch the tip to a well edge or fall back to a top-edge move.
Parameters
----------
pipette_id : str
Pipette identifier.
well : dict
Well descriptor containing ``labwareId`` and ``wellName``.
"""
params = {
"pipetteId": pipette_id,
"labwareId": well["labwareId"],
"wellName": well["wellName"],
"wellLocation": {"origin": "top", "offset": {"x": 0, "y": 0, "z": 0}},
"mmFromEdge": 1,
}
try:
self._execute_atomic_command("touchTip", params, check_run_status=False)
except RuntimeError as exc:
self.log_warning(
f"touchTip command unavailable; using moveToWell fallback. Error: {exc}"
)
self._execute_atomic_command(
"moveToWell",
{
"pipetteId": pipette_id,
"labwareId": well["labwareId"],
"wellName": well["wellName"],
"wellLocation": {"origin": "top", "offset": {"x": 0, "y": 0, "z": -2}},
},
check_run_status=False,
)
def _move_above_tiprack(
self,
pipette_id,
labware_id,
well_name,
tip_rack_offset,
approach_z_offset = 40.0,
check_run_status=True,
):
"""Move above a target tip before issuing the pickup command."""
approach_offset = dict(tip_rack_offset)
approach_offset["z"] = approach_z_offset
self._execute_atomic_command(
"moveToWell",
{
"pipetteId": pipette_id,
"labwareId": labware_id,
"wellName": well_name,
"wellLocation": {
"origin": "top",
"offset": approach_offset,
},
},
check_run_status=check_run_status,
)
def _execute_atomic_command(
self, command_type, params=None, wait_until_complete=True, timeout=None, check_run_status=True
):
"""Execute one atomic HTTP API command.
Parameters
----------
command_type : str
Opentrons command type.
params : dict, optional
Command parameters.
wait_until_complete : bool, default=True
If ``True``, wait for command completion before returning.
timeout : float, optional
Command timeout forwarded to the robot server.
check_run_status : bool, default=True
If ``False``, skip the run-status GET check when ensuring a run.
Returns
-------
bool or str
``True`` when a waited command succeeds, otherwise the command ID for
asynchronous tracking.
"""
if params is None:
params = {}
# Track tip usage for pick up and drop commands
if command_type == "pickUpTip":
mount = params.get("pipetteMount")
if not mount:
raise RuntimeError("pickUpTip requires pipetteMount for tip tracking")
if "labwareId" in params and "wellName" in params:
tiprack_id, well = self._reserve_tip(
mount, params["labwareId"], params["wellName"]
)
elif mount in self.config["available_tips"] and self.config["available_tips"][mount]:
tiprack_id, well = self.get_tip(mount)
else:
raise RuntimeError(f"No tips available for {mount} mount")
self.log_debug(
f"Using tip from {tiprack_id} well {well} for {mount} mount"
)
tip_rack_offset = self._resolve_tip_rack_offset(
params.get("tipRackOffset"),
mount=mount,
)
self._move_above_tiprack(
pipette_id=params["pipetteId"],
labware_id=tiprack_id,
well_name=well,
tip_rack_offset=tip_rack_offset,
approach_z_offset= 40.0,
check_run_status=check_run_status,
)
params["labwareId"] = tiprack_id
params["wellName"] = well
params["wellLocation"] = {
"origin": "top",
"offset": tip_rack_offset,
}
params.pop("tipRackOffset", None)
slot = self._slot_by_labware_uuid(tiprack_id)
self.current_tip = {
"mount": mount,
"labware_id": tiprack_id,
"well_name": well,
"slot": slot,
"location": f"{slot}{well}" if slot is not None else None,
}
del params["pipetteMount"]
self.log_debug(
f"Executing atomic command: {command_type} with params: {params}"
)
# Ensure we have a valid run
run_id = self._ensure_run_exists(check_run_status=check_run_status)
# Build the query parameters
query_params = {"waitUntilComplete": wait_until_complete}
if timeout is not None:
query_params["timeout"] = timeout
try:
# Send the command
command_response = requests.post(
url=f"{self.base_url}/runs/{run_id}/commands",
params=query_params,
headers=self.headers,
json={
"data": {
"commandType": command_type,
"params": params,
"intent": "setup",
}
},
)
self._check_cmd_success(command_response)
command_data = command_response.json()["data"]
command_id = command_data["id"]
self.log_debug(
f"Command {command_id} executed with status: {command_data['status']}"
)
# If wait_until_complete is True, the command has already completed
if wait_until_complete:
if command_data["status"] == "succeeded":
return True
elif command_data["status"] in ["failed", "error"]:
error_info = command_data.get("error", "Unknown error")
self.log_error(f"Command failed: {error_info}")
raise RuntimeError(f"Command failed: {error_info}")
# If we're not waiting or the command is still running, return the command ID for tracking
return command_id
except requests.exceptions.RequestException as e:
self.log_error(f"Error executing command: {str(e)}")
raise RuntimeError(f"Error executing command: {str(e)}")
[docs]
def set_aspirate_rate(self, rate=150, pipette=None):
"""Set stored aspirate flow rate for one or more active pipettes."""
self.log_info(f"Setting aspirate rate to {rate} uL/s")
if pipette is None:
active_pipettes = self._get_active_pipettes()
if not active_pipettes:
self.log_warning("No loaded pipettes available to update aspirate rate")
return
for info in active_pipettes.values():
info["aspirate_flow_rate"] = rate
return
self._get_active_pipette_info(pipette)["aspirate_flow_rate"] = rate
[docs]
def set_dispense_rate(self, rate=300, pipette=None):
"""Set stored dispense flow rate for one or more active pipettes."""
self.log_info(f"Setting dispense rate to {rate} uL/s")
if pipette is None:
active_pipettes = self._get_active_pipettes()
if not active_pipettes:
self.log_warning("No loaded pipettes available to update dispense rate")
return
for info in active_pipettes.values():
info["dispense_flow_rate"] = rate
return
self._get_active_pipette_info(pipette)["dispense_flow_rate"] = rate
[docs]
def set_gantry_speed(self, speed=400):
"""Record a requested gantry speed change.
Notes
-----
The HTTP driver currently logs the request but does not apply it through
the robot server.
"""
self.log_info(f"Setting gantry speed to {speed} mm/s")
[docs]
def get_pipette(self, volume, method="min_transfers"):
"""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
-------
dict
Selected pipette metadata including mount, volume range, and number
of required transfers.
"""
self.log_debug(f"Looking for a pipette for volume {volume}")
# Make sure we have the latest pipette information
self._update_pipettes()
pipettes = []
for mount, pipette_data in self._get_active_pipettes().items():
if not pipette_data:
continue
min_volume = pipette_data.get("min_volume", 1)
max_volume = pipette_data.get("max_volume", 300)
if volume >= min_volume:
pipettes.append(
{
"mount": mount, # Use mount as the identifier
"min_volume": min_volume,
"max_volume": max_volume,
"name": pipette_data.get("name"),
"model": pipette_data.get("model"),
"channels": pipette_data.get("channels", 1),
"pipette_id": pipette_data.get("id"),
}
)
if not pipettes:
raise ValueError("No suitable loaded pipettes found!\n")
# Calculate transfers and uncertainties
for pipette in pipettes:
max_volume = pipette["max_volume"]
ntransfers = ceil(volume / max_volume)
vol_per_transfer = volume / ntransfers
pipette["ntransfers"] = ntransfers
# Calculate uncertainty (simplified from original)
pipette["uncertainty"] = (
ntransfers * 0.1
) # Simplified uncertainty calculation
if self.data is not None:
self.data["transfer_method"] = method
self.data["pipette_options"] = str(pipettes)
# Choose pipette based on method
if method == "uncertainty":
pipette = min(pipettes, key=lambda x: x["uncertainty"])
elif method == "min_transfers":
min_xfers = min(pipettes, key=lambda x: x["ntransfers"])["ntransfers"]
acceptable_pipettes = [p for p in pipettes if p["ntransfers"] == min_xfers]
pipette = min(acceptable_pipettes, key=lambda x: x["max_volume"])
else:
raise ValueError(f"Pipette selection method {method} was not recognized.")
self.log_debug(f"Chosen pipette: {pipette}")
if self.data is not None:
self.data["chosen_pipette"] = str(pipette)
return pipette
[docs]
def get_aspirate_rate(self, pipette=None):
"""Return the stored aspirate flow rate for a pipette."""
active_pipettes = self._get_active_pipettes()
if pipette is None:
# Return the rate of the first pipette found
for mount, pipette_data in active_pipettes.items():
if pipette_data:
pipette = mount
break
if pipette is None:
return None
try:
for mount, pipette_data in active_pipettes.items():
if mount == pipette and pipette_data:
return pipette_data.get("aspirate_flow_rate", 150)
except requests.exceptions.RequestException:
pass
return 150 # Default value
[docs]
def get_dispense_rate(self, pipette=None):
"""Return the stored dispense flow rate for a pipette."""
active_pipettes = self._get_active_pipettes()
if pipette is None:
# Return the rate of the first pipette found
for mount, pipette_data in active_pipettes.items():
if pipette_data:
pipette = mount
break
if pipette is None:
return None
try:
for mount, pipette_data in active_pipettes.items():
if mount == pipette and pipette_data:
return pipette_data.get("dispense_flow_rate", 300)
except requests.exceptions.RequestException:
pass
return 300 # Default value
# HTTP API communication with heater-shaker module
[docs]
def set_shake(self, rpm, module_id = None):
"""Set heater-shaker speed and wait for the target RPM."""
self.log_info(f"Setting heater-shaker speed to {rpm} RPM")
if module_id is None:
module_id = self._find_module_by_type("heaterShaker")
self._execute_atomic_command("heaterShaker/setAndWaitForShakeSpeed",
params= {
"moduleId": module_id,
"rpm": rpm,
},
)
[docs]
def stop_shake(self, module_id = None):
"""Stop heater-shaker motion."""
self.log_info("Stopping heater-shaker")
if module_id is None:
module_id = self._find_module_by_type("heaterShaker")
self._execute_atomic_command("heaterShaker/deactivateShaker",
params= {
"moduleId": module_id,
},
)
[docs]
def set_shaker_temp(self, temp, module_id = None):
"""Set heater-shaker target temperature."""
self.log_info(f"Setting heater-shaker temperature to {temp}°C")
if module_id is None:
module_id = self._find_module_by_type("heaterShaker")
self._execute_atomic_command("heaterShaker/setTargetTemperature",
params= {
"moduleId": module_id,
"celsius": temp,
},
)
[docs]
def stop_shaker_heat(self, module_id = None):
"""Deactivate heater-shaker heating."""
self.log_info(f"Deactivating heater-shaker heating")
if module_id is None:
module_id = self._find_module_by_type("heaterShaker")
self._execute_atomic_command("heaterShaker/deactivateHeater",
params= {
"moduleId": module_id,
},
)
[docs]
def unlatch_shaker(self, module_id = None):
"""Open the heater-shaker labware latch."""
self.log_info("Unlatching heater-shaker")
if module_id is None:
module_id = self._find_module_by_type("heaterShaker")
self._execute_atomic_command("heaterShaker/openLabwareLatch",
params= {
"moduleId": module_id,
},
)
[docs]
def latch_shaker(self, module_id = None):
"""Close the heater-shaker labware latch."""
self.log_info("Latching heater-shaker")
if module_id is None:
module_id = self._find_module_by_type("heaterShaker")
self._execute_atomic_command("heaterShaker/closeLabwareLatch",
params= {
"moduleId": module_id,
},
)
def _find_module_by_type(self,partial_name):
"""Return the first loaded module ID whose name contains a token."""
module_id = None
for module in self.config["loaded_modules"].values():
if partial_name in module[1]:
module_id = module[0]
return module_id
[docs]
def get_shaker_temp(self):
"""Return current and target heater-shaker temperatures."""
self.log_info("Getting heater-shaker temperature")
# For get operations, we still need to use the modules API directly
try:
# Get modules to find the heater-shaker module
modules_response = requests.get(
url=f"{self.base_url}/modules", headers=self.headers
)
if modules_response.status_code != 200:
self.log_error(f"Failed to get modules: {modules_response.status_code}")
return f"Error getting modules: {modules_response.status_code}"
modules = modules_response.json().get("modules", [])
heater_shaker_module = next(
(m for m in modules if "heaterShaker" in m.get("moduleModel")),
None,
)
if not heater_shaker_module:
self.log_error("No heater-shaker module found")
return "No heater-shaker module found"
logging.debug(heater_shaker_module)
current_temp = heater_shaker_module.get("data", {}).get("currentTemp")
target_temp = heater_shaker_module.get("data", {}).get("targetTemp")
self.log_info(
f"Heater-shaker temperature - Current: {current_temp}°C, Target: {target_temp}°C"
)
return (current_temp,target_temp)
except Exception as e:
self.log_error(f"Error getting temperature: {str(e)}")
return f"Error: {str(e)}"
[docs]
def get_shake_rpm(self):
"""Return heater-shaker speed status, current RPM, and target RPM."""
# For get operations, we just use the modules API
try:
# Get modules to find the heater-shaker module
modules_response = requests.get(
url=f"{self.base_url}/modules", headers=self.headers
)
if modules_response.status_code != 200:
self.log_error(f"Failed to get modules: {modules_response.status_code}")
return f"Error getting modules: {modules_response.status_code}"
modules = modules_response.json().get("modules", [])
heater_shaker_module = next(
(m for m in modules if "heaterShaker" in m.get("moduleModel")),
None,
)
if not heater_shaker_module:
self.log_error("No heater-shaker module found")
return "No heater-shaker module found"
current_rpm = heater_shaker_module.get("data", {}).get("currentSpeed")
target_rpm = heater_shaker_module.get("data", {}).get("targetSpeed")
status = heater_shaker_module.get("data", {}).get("speedStatus")
return (status,current_rpm,target_rpm)
except Exception as e:
self.log_error(f"Error getting RPM: {str(e)}")
return f"Error: {str(e)}"
[docs]
def get_shake_latch_status(self):
"""Return the heater-shaker latch status string."""
# For get operations, we just use the modules API
try:
# Get modules to find the heater-shaker module
modules_response = requests.get(
url=f"{self.base_url}/modules", headers=self.headers
)
if modules_response.status_code != 200:
self.log_error(f"Failed to get modules: {modules_response.status_code}")
return f"Error getting modules: {modules_response.status_code}"
modules = modules_response.json().get("modules", [])
heater_shaker_module = next(
(m for m in modules if "heaterShaker" in m.get("moduleModel")),
None,
)
if not heater_shaker_module:
self.log_error("No heater-shaker module found")
return "No heater-shaker module found"
status = heater_shaker_module.get("data", {}).get("labwareLatchStatus")
return status
except Exception as e:
self.log_error(f"Error getting RPM: {str(e)}")
return f"Error: {str(e)}"
[docs]
def set_tempmodule_temperature(
self,
module_id,
temperature_c,
hold_time = 0.0,
wait = True,
):
"""Set a temperature module target and optionally wait to stabilize.
Returns
-------
tuple
Current and target temperatures after stabilization.
"""
self.log_info(f"Setting temperature module to {temperature_c}°C")
if module_id is None:
module_id = self._find_module_by_type("tempdeck")
self._execute_atomic_command(
"temperatureModule/setTargetTemperature",
params={"moduleId": module_id, "celsius": float(temperature_c)},
wait_until_complete=wait,
)
time.sleep(60) # wait for a minute before querying status
data = self.get_tempmodule_status(log=False)
# Some OT-2 API versions report no target temperature immediately
# after setting it. Return in that case so callers can perform their
# own equilibration hold instead of crashing on float - None.
current_temp = data.get("currentTemp")
target_temp = data.get("targetTemp")
if current_temp is None or target_temp is None:
self.log_warning(
"Temperature module did not report both currentTemp and targetTemp; "
"skipping driver-side stabilization wait."
)
else:
while abs(current_temp - target_temp) > 1.0:
time.sleep(5)
data = self.get_tempmodule_status(log=False)
current_temp = data.get("currentTemp")
target_temp = data.get("targetTemp")
if current_temp is None or target_temp is None:
self.log_warning(
"Temperature module stopped reporting currentTemp or targetTemp; "
"skipping driver-side stabilization wait."
)
return current_temp, target_temp
self.log_debug(f"Waiting for temperature to stabilize... "
f"(Current: {current_temp}°C, Target: {target_temp}°C)")
if hold_time > 0:
self.log_info(f"Holding the command exceution for {hold_time}")
time.sleep(hold_time)
return current_temp, target_temp
[docs]
def deactivate_tempmodule(self, module_id, timeout_s=120, wait=True):
"""Deactivate a temperature module.
Returns
-------
bool or str
Result returned by :meth:`_execute_atomic_command`.
"""
if module_id is None:
module_id = self._find_module_by_type("tempdeck")
return self._execute_atomic_command(
"temperatureModule/deactivate",
params={"moduleId": module_id},
wait_until_complete=wait,
timeout=timeout_s,
)
[docs]
def get_tempmodule_status(self, log=True):
"""Return raw tempdeck status data from the modules endpoint."""
response = requests.get(
url=f"{self.base_url}/modules",
headers=self.headers,
timeout=30,
)
response.raise_for_status()
modules = response.json().get("modules", [])
for m in modules:
if m.get("name") == "tempdeck":
temp_module = m
data = temp_module.get("data", {})
status = temp_module.get("status", "unknown")
current_temp = data.get("currentTemp")
target_temp = data.get("targetTemp")
current_temp_str = f"{current_temp:g}°C" if current_temp is not None else "None"
target_temp_str = f"{target_temp:g}°C" if target_temp is not None else "None"
if log:
self.log_info(
f"Current status: {status} "
f"(Current temperature : {current_temp_str},"
f" Target temperature: {target_temp_str})"
)
else:
data = {}
return data
def _create_run(self):
"""Create a new robot run and reload persisted deck state.
Returns
-------
str
Newly created run identifier.
"""
self.log_info("Creating a new run for commands")
try:
# Clear custom labware tracking so definitions are re-uploaded for the new run
self.sent_custom_labware = {}
# Create a run
import datetime
run_response = requests.post(
url=f"{self.base_url}/runs",
headers=self.headers,
)
if run_response.status_code != 201:
self.log_error(f"Failed to create run: {run_response.status_code}")
self.log_error(f"Response: {run_response.text}")
raise RuntimeError(f"Failed to create run: {run_response.text}")
self.run_id = run_response.json()["data"]["id"]
self.log_debug(f"Created run: {self.run_id}")
# Reload previously configured labware, instruments, and modules
self._reload_deck_configuration()
return self.run_id
except requests.exceptions.RequestException as e:
self.log_error(f"Error creating run: {str(e)}")
raise RuntimeError(f"Error creating run: {str(e)}")
def _reload_deck_configuration(self):
"""Reload persisted modules, labware, instruments, and tip state.
Returns
-------
bool
``True`` on success, otherwise ``False`` after restoring the prior
persisted configuration.
"""
self.log_info("Reloading previously configured deck setup")
# Store original configuration for recovery if needed
original_modules = self.config["loaded_modules"].copy()
original_labware = self.config["loaded_labware"].copy()
original_instruments = self.config["loaded_instruments"].copy()
old_uuid_to_slot = {}
tiprack_slots = {}
for (mount,instrument) in original_instruments.items():
tiprack_slots[mount] = [self._slot_by_labware_uuid(uuid) for uuid in instrument['tip_racks']]
old_uuid_to_slot.update({uuid:self._slot_by_labware_uuid(uuid) for uuid in instrument['tip_racks']})
# Clear current state for reloading
self.config["loaded_modules"] = {}
self.config["loaded_labware"] = {}
self.config["loaded_instruments"] = {}
try:
# Step 1: Load modules first
# We know the run exists because _create_run just created it, so skip status checks
self.log_info("Reloading modules")
for slot, (_, module_name) in original_modules.items():
try:
self.log_info(f"Reloading module {module_name} in slot {slot}")
self.load_module(module_name, slot, check_run_status=False)
# New module ID will be stored in config["loaded_modules"]
except Exception as e:
self.log_error(f"Error reloading module {module_name} in slot {slot}: {str(e)}")
raise
# Step 2: Load labware
# We know the run exists because _create_run just created it, so skip status checks
self.log_info("Reloading labware")
for slot, (_, labware_name, labware_data) in original_labware.items():
# Check if this labware is on a module
module_id = None
if str(slot) in self.config["loaded_modules"]:
module_id = self.config["loaded_modules"][str(slot)][0] # Get new module ID
try:
self.log_info(f"Reloading labware {labware_name} in slot {slot}")
self.load_labware(labware_name, slot, module=module_id, check_run_status=False)
# New labware ID will be stored in config["loaded_labware"]
except Exception as e:
self.log_error(f"Error reloading labware {labware_name} in slot {slot}: {str(e)}")
raise
# Step 3: Load instruments
# We know the run exists because _create_run just created it, so skip status checks
# Also skip pipette updates since _initialize_robot already fetched pipette info
self.log_info("Reloading instruments")
for mount, instrument_data in original_instruments.items():
instrument_name = instrument_data['name']
try:
self.log_info(f"Reloading instrument {instrument_name} on {mount} mount")
self.load_instrument(instrument_name, mount, tiprack_slots[mount], reload=True, check_run_status=False, update_pipettes=False)
# New instrument ID will be stored in config["loaded_instruments"]
except Exception as e:
self.log_error(f"Error reloading instrument {instrument_name} on {mount} mount: {str(e)}")
raise
self.log_info("Deck configuration successfully reloaded")
# Update tiprack lists
# Build slot->new_uuid mapping from new loaded_instruments
slot_to_new_tiprack_uuid = {}
for instrument in self.config["loaded_instruments"].values():
for new_uuid in instrument.get('tip_racks', []):
slot = self._slot_by_labware_uuid(new_uuid)
slot_to_new_tiprack_uuid[slot] = new_uuid
# Remap available tips
old_available_tips = self.config.get("available_tips", {})
new_available_tips = {}
for mount in self.config["loaded_instruments"].keys():
new_available_tips[mount] = []
for tiprack_uuid, well in old_available_tips.get(mount, []):
slot = old_uuid_to_slot.get(tiprack_uuid)
new_uuid = slot_to_new_tiprack_uuid.get(slot)
if new_uuid is not None:
new_available_tips[mount].append((new_uuid, well))
self.log_info(f"Remapped {len(new_available_tips[mount])} available tips for {mount} mount after reload.")
self.config["available_tips"] = new_available_tips
return True
except Exception as e:
self.log_error(f"Failed to reload deck configuration: {str(e)}")
# Restore original configuration in config
self.config["loaded_modules"] = original_modules
self.config["loaded_labware"] = original_labware
self.config["loaded_instruments"] = original_instruments
return False
def _ensure_run_exists(self, check_run_status=True):
"""Return a valid run identifier, creating a run when needed.
Parameters
----------
check_run_status : bool, default=True
If ``False``, trust the cached run ID without a GET request.
Returns
-------
str
Valid run identifier.
"""
if not hasattr(self, "run_id") or not self.run_id:
return self._create_run()
# Skip status check if requested (optimization for bulk operations)
if not check_run_status:
return self.run_id
# Check if the run is still valid
try:
response = requests.get(
url=f"{self.base_url}/runs/{self.run_id}", headers=self.headers
)
if response.status_code != 200:
# Run doesn't exist, create a new one
return self._create_run()
# Check run state
run_data = response.json()["data"]
current_state = run_data.get("status")
if current_state in ["failed", "error", "succeeded", "stopped"]:
# Run is in a terminal state, create a new one
return self._create_run()
return self.run_id
except requests.exceptions.RequestException:
# Error checking run, create a new one
return self._create_run()
def _slot_by_labware_uuid(self, labware_id):
"""Return the deck slot for a loaded labware identifier."""
for slot, labware_info in self.config.get("loaded_labware", {}).items():
if labware_info and labware_info[0] == labware_id:
return str(slot)
return None
def _current_tip_is_reserved_stock_tip(self):
"""Return whether the currently attached tip is reserved for stock use."""
if not self.current_tip:
return False
location = self.current_tip.get("location")
if location is None:
slot = self.current_tip.get("slot")
well_name = self.current_tip.get("well_name")
if slot is not None and well_name is not None:
location = f"{slot}{well_name}"
return location in set(self.config.get("reserved_stock_tips", []))
def _drop_tip_to_trash(self, pipette_id):
"""Drop the current tip into trash and clear local tracking."""
try:
self._execute_atomic_command(
"moveToAddressableAreaForDropTip",
{
"pipetteId": pipette_id,
"addressableAreaName": FIXED_TRASH_ADDRESSABLE_AREA,
"alternateDropLocation": False,
},
check_run_status=False,
)
except Exception as exc:
self.log_warning(
"Explicit fixed-trash move unavailable; falling back to "
f"dropTipInPlace only. Error: {exc}"
)
self._execute_atomic_command(
"dropTipInPlace",
{"pipetteId": pipette_id},
check_run_status=False,
)
self.has_tip = False
self.current_tip = None
def _return_tip_to_origin(
self,
pipette_id,
mount=None,
tip_rack_offset=None,
return_tip_z_offset=None,
):
"""Return the current tip to its original tiprack location."""
if not self.current_tip:
return
tip_mount = mount or self.current_tip.get("mount")
labware_id = self.current_tip.get("labware_id")
well_name = self.current_tip.get("well_name")
if labware_id is None or well_name is None:
self._drop_tip_to_trash(pipette_id)
return
offset = self._resolve_tip_rack_offset(tip_rack_offset, mount=tip_mount)
return_offset = copy.deepcopy(offset)
# approch the tiprack safely
self._move_above_tiprack(
pipette_id=pipette_id,
labware_id=labware_id,
well_name=well_name,
tip_rack_offset=offset,
approach_z_offset=50.0,
check_run_status=False,
)
# Apply any return-only z adjustment without mutating the configured or
# caller-provided offset mapping used for future pickups.
if return_tip_z_offset is not None:
return_offset["z"] += return_tip_z_offset
# Move to the base well location before issuing the return/drop command.
self._execute_atomic_command(
"moveToWell",
{
"pipetteId": pipette_id,
"labwareId": labware_id,
"wellName": well_name,
"wellLocation": {
"origin": "center",
"offset": dict(offset),
},
},
check_run_status=False,
)
# drop the tip
self._execute_atomic_command(
"dropTipInPlace",
{
"pipetteId": pipette_id,
"labwareId": labware_id,
"wellName": well_name,
"wellLocation": {
"origin": "center",
"offset": dict(return_offset),
},
},
check_run_status=False,
)
available = list(self.config.get("available_tips", {}).get(tip_mount, []))
tip_entry = (labware_id, well_name)
if tip_entry not in available:
available.insert(0, tip_entry)
self.config.setdefault("available_tips", {})[tip_mount] = available
self.has_tip = False
self.current_tip = None
[docs]
def get_tip(self, mount):
"""Reserve and return the next available tip for a mount."""
available = list(self.config.get("available_tips", {}).get(mount, []))
if not available:
raise ValueError(f"No tips available for mount {mount}")
reserved_locations = {
str(location).strip().upper()
for location in self.config.get("reserved_stock_tips", [])
}
selected_index = None
for index, (tiprack_id, well_name) in enumerate(available):
slot = self._slot_by_labware_uuid(tiprack_id)
normalized_location = None if slot is None else f"{slot}{well_name}".upper()
if normalized_location in reserved_locations:
continue
selected_index = index
break
if selected_index is None:
raise RuntimeError(f"No unreserved tips available for {mount} mount")
tiprack_id, well_name = available.pop(selected_index)
self.config.setdefault("available_tips", {})[mount] = available
return tiprack_id, well_name
def _tip_status_counts(self, mount):
"""Summarize general and reserved tip availability for a mount.
Parameters
----------
mount : str
Pipette mount name such as ``"left"`` or ``"right"``.
Returns
-------
dict
Counts keyed by ``general_available`` and ``reserved_available``.
"""
available = list(self.config.get("available_tips", {}).get(mount, []))
reserved_locations = {
str(location).strip().upper()
for location in self.config.get("reserved_stock_tips", [])
}
reserved_available = 0
general_available = 0
for tiprack_id, well_name in available:
slot = self._slot_by_labware_uuid(tiprack_id)
normalized_location = None if slot is None else f"{slot}{well_name}".upper()
if normalized_location in reserved_locations:
reserved_available += 1
else:
general_available += 1
return {
"general_available": general_available,
"reserved_available": reserved_available,
}
[docs]
def get_tip_status(self, mount=None):
"""Return human-readable tip availability status.
Parameters
----------
mount : str, optional
Specific mount to report. If omitted, report all mounts.
Returns
-------
str
Tip availability summary.
"""
if mount:
if mount not in self.config["available_tips"]:
return f"No tipracks loaded for {mount} mount"
if mount not in self.config["loaded_instruments"]:
return f"No instrument defined for {mount} mount"
total_tips = len(TIPRACK_WELLS) * len(
self.config["loaded_instruments"][mount]["tip_racks"]
)
counts = self._tip_status_counts(mount)
return (
f"{counts['general_available']}/{total_tips} general tips available on {mount} mount "
f"({counts['reserved_available']} reserved for stock pipetting)"
)
# Return status for all mounts
status = []
for m in self.config["available_tips"]:
status.append(self.get_tip_status(m))
return "\n".join(status)
[docs]
def make_align_script(self, filename: str):
"""
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")
"""
script = []
# Header
script.append("from opentrons import protocol_api")
script.append("")
script.append("metadata = {")
script.append(" 'protocolName': 'Alignment Check',")
script.append(" 'author': 'AFL Auto-Generated',")
script.append(" 'description': 'Script for aligning and testing deck configuration',")
script.append(" 'apiLevel': '2.13'")
script.append("}")
script.append("")
script.append("def run(protocol: protocol_api.ProtocolContext):")
indent = " "
# Track labware variable names by ID
labware_var_by_id = {}
# 1. Modules
loaded_modules = self.config.get("loaded_modules", {})
if loaded_modules:
script.append(f"{indent}# Modules")
# Sort by slot
for slot in sorted(loaded_modules.keys(), key=lambda x: int(x) if x.isdigit() else 99):
module_id, module_name = loaded_modules[slot]
script.append(f"{indent}module_{slot} = protocol.load_module('{module_name}', '{slot}')")
script.append("")
# 2. Labware
loaded_labware = self.config.get("loaded_labware", {})
regular_labware_vars = []
if loaded_labware:
script.append(f"{indent}# Labware")
# Sort by slot
for slot in sorted(loaded_labware.keys(), key=lambda x: int(x) if x.isdigit() else 99):
labware_id, _, labware_data = loaded_labware[slot]
# Get precise load info from definition if available
definition = labware_data.get('definition', {})
params = definition.get('parameters', {})
load_name = params.get('loadName', 'unknown_labware')
namespace = definition.get('namespace', 'opentrons')
version = definition.get('version', 1)
# Determine if tiprack
labware_type = definition.get('metadata', {}).get('displayCategory', 'default')
is_tiprack = (
params.get('isTiprack') or
labware_type == 'tipRack' or
'tiprack' in load_name.lower()
)
var_name = f"labware_{slot}"
labware_var_by_id[labware_id] = var_name
if not is_tiprack:
regular_labware_vars.append(var_name)
# Check if on module
if str(slot) in loaded_modules:
parent = f"module_{slot}"
script.append(f"{indent}{var_name} = {parent}.load_labware('{load_name}', namespace='{namespace}', version={version})")
else:
script.append(f"{indent}{var_name} = protocol.load_labware('{load_name}', '{slot}', namespace='{namespace}', version={version})")
script.append("")
# 3. Pipettes
loaded_instruments = self.config.get("loaded_instruments", {})
pipette_vars = []
if loaded_instruments:
script.append(f"{indent}# Pipettes")
for mount, instrument_data in loaded_instruments.items():
name = instrument_data['name']
tip_rack_ids = instrument_data.get('tip_racks', [])
# Resolve tip rack variables
tip_rack_vars = [labware_var_by_id[tid] for tid in tip_rack_ids if tid in labware_var_by_id]
tip_racks_arg = f"[{', '.join(tip_rack_vars)}]"
var_name = f"pipette_{mount}"
pipette_vars.append(var_name)
script.append(f"{indent}{var_name} = protocol.load_instrument('{name}', '{mount}', tip_racks={tip_racks_arg})")
script.append("")
# 4. Alignment Moves
if pipette_vars and regular_labware_vars:
script.append(f"{indent}# Alignment Verification")
script.append(f"{indent}# Move to top of A1 for each labware")
for pip in pipette_vars:
for lab in regular_labware_vars:
script.append(f"{indent}protocol.comment(f'Checking {lab} with {pip}')")
script.append(f"{indent}{pip}.move_to({lab}['A1'].top())")
script.append(f"{indent}protocol.delay(seconds=0.5)")
# Write to file
with open(filename, 'w') as f:
f.write('\n'.join(script))
self.log_info(f"Generated alignment script at {filename}")
if __name__ == "__main__":
from AFL.automation.shared.launcher import *
