Chapter 2: Fundamentals of Robotic Experiments
Study time: 25-30 minutes
Introduction
The heart of experimental automation lies in robotic arms, liquid handling systems, and sensor networks. In this chapter, we learn these foundational technologies hands-on through Python programming.
Centered on the OpenTrons OT-2 liquid handling robot, and through code examples that actually run, we master the basic operations of automated experiments: reagent dispensing, plate movement, sensor data acquisition, and more.
Learning Objectives
By studying this chapter, you will be able to master the following:
- Robotic arm control: Fundamentals of inverse kinematics and path planning, with Python implementation
- Liquid handling: Precise pipetting with the OpenTrons OT-2
- Solid handling: Automation methods for powder weighing and tablet forming
- Sensor integration: Interfacing with cameras, spectrometers, and XRD
- Safety design: Error handling, emergency stops, and anomaly detection
- Labware standardization: Unified standards for microplates, vials, and cuvettes
2.1 Fundamentals of Robotic Arm Control
2.1.1 Forward and Inverse Kinematics
Controlling a robotic arm involves two kinematics problems.
Forward Kinematics: Compute the end-effector position and orientation $(x, y, z, roll, pitch, yaw)$ from the joint angles $\theta_1, \theta_2, ..., \theta_n$
Inverse Kinematics (IK): Compute the joint angles that achieve a target position and orientation (typically used in experiments)
import numpy as np
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D
class SimpleRobotArm:
"""
Simulation of a 2-link 2D planar robotic arm
Intended for tasks such as accessing reagent bottles and moving samples in materials science experiments
"""
def __init__(self, link1_length=0.3, link2_length=0.25):
"""
Args:
link1_length: Length of the first link (meters)
link2_length: Length of the second link (meters)
"""
self.L1 = link1_length
self.L2 = link2_length
def forward_kinematics(self, theta1, theta2):
"""
Forward kinematics: Compute the end-effector position from joint angles
Args:
theta1: Angle of the first joint (degrees)
theta2: Angle of the second joint (degrees)
Returns:
(x, y): End-effector position
"""
# Convert degrees to radians
th1 = np.radians(theta1)
th2 = np.radians(theta2)
# Position of the tip of the first link
x1 = self.L1 * np.cos(th1)
y1 = self.L1 * np.sin(th1)
# End-effector position
x = x1 + self.L2 * np.cos(th1 + th2)
y = y1 + self.L2 * np.sin(th1 + th2)
return x, y
def inverse_kinematics(self, target_x, target_y):
"""
Inverse kinematics: Compute joint angles from a target position
Args:
target_x: Target X coordinate
target_y: Target Y coordinate
Returns:
(theta1, theta2): Joint angles (degrees) or None (unreachable)
"""
# Reachability check
distance = np.sqrt(target_x**2 + target_y**2)
if distance > (self.L1 + self.L2) or distance < abs(self.L1 - self.L2):
print(f"Warning: Target position ({target_x:.2f}, {target_y:.2f}) is unreachable")
return None
# Compute the second joint angle using the law of cosines
cos_theta2 = (target_x**2 + target_y**2 - self.L1**2 - self.L2**2) / (2 * self.L1 * self.L2)
# Numerical error safeguard
cos_theta2 = np.clip(cos_theta2, -1.0, 1.0)
# Select the elbow-up solution (common in experiments)
theta2_rad = np.arccos(cos_theta2)
# Compute the first joint angle
k1 = self.L1 + self.L2 * np.cos(theta2_rad)
k2 = self.L2 * np.sin(theta2_rad)
theta1_rad = np.arctan2(target_y, target_x) - np.arctan2(k2, k1)
# Convert radians to degrees
theta1 = np.degrees(theta1_rad)
theta2 = np.degrees(theta2_rad)
return theta1, theta2
def plot_arm(self, theta1, theta2, target_point=None):
"""Visualize the current position of the arm"""
th1 = np.radians(theta1)
th2 = np.radians(theta2)
# Position of each joint
x0, y0 = 0, 0 # Base
x1 = self.L1 * np.cos(th1)
y1 = self.L1 * np.sin(th1)
x2 = x1 + self.L2 * np.cos(th1 + th2)
y2 = y1 + self.L2 * np.sin(th1 + th2)
plt.figure(figsize=(8, 8))
plt.plot([x0, x1, x2], [y0, y1, y2], 'o-', linewidth=3, markersize=10, label='Robotic arm')
plt.plot(x0, y0, 'ro', markersize=15, label='Base')
plt.plot(x2, y2, 'go', markersize=12, label='End-effector')
if target_point:
plt.plot(target_point[0], target_point[1], 'r*', markersize=20, label='Target position')
# Reachable-range circles
theta = np.linspace(0, 2*np.pi, 100)
r_max = self.L1 + self.L2
r_min = abs(self.L1 - self.L2)
plt.plot(r_max * np.cos(theta), r_max * np.sin(theta), 'k--', alpha=0.3, label='Maximum reach')
plt.plot(r_min * np.cos(theta), r_min * np.sin(theta), 'k--', alpha=0.3)
plt.axis('equal')
plt.grid(True, alpha=0.3)
plt.xlabel('X position (m)', fontsize=12)
plt.ylabel('Y position (m)', fontsize=12)
plt.title('2-link robotic arm', fontsize=14, fontweight='bold')
plt.legend()
plt.tight_layout()
# Usage example
robot = SimpleRobotArm(link1_length=0.3, link2_length=0.25)
# Inverse kinematics usage example: access reagent bottle at (0.4, 0.2)
target_x, target_y = 0.4, 0.2
angles = robot.inverse_kinematics(target_x, target_y)
if angles:
theta1, theta2 = angles
print(f"Target position: ({target_x}, {target_y})")
print(f"Required joint angles: θ1 = {theta1:.2f}°, θ2 = {theta2:.2f}°")
# Verification: check the position with forward kinematics
x_check, y_check = robot.forward_kinematics(theta1, theta2)
error = np.sqrt((x_check - target_x)**2 + (y_check - target_y)**2)
print(f"Verification: actual position ({x_check:.4f}, {y_check:.4f}), error {error:.6f}m")
# Visualization
robot.plot_arm(theta1, theta2, target_point=(target_x, target_y))
plt.savefig('robot_arm_ik.png', dpi=300, bbox_inches='tight')
plt.show()
Code explanation: 1. Forward kinematics: Compute the position of each link with trigonometric functions 2. Inverse kinematics: Back-calculate the joint angles using the law of cosines and arctangent 3. Reachability: Check whether the target distance lies within the range $|L_1 - L_2| \leq d \leq L_1 + L_2$ 4. Elbow-up/down: When two solutions exist for the same target position, experiments usually select elbow-up
2.1.2 Path Planning
Plan smooth and safe trajectories, such as moving from a reagent bottle to a reaction vessel.
def linear_trajectory(start_pos, end_pos, num_points=50):
"""
Generate a straight-line trajectory between two points
Args:
start_pos: Start position (x, y)
end_pos: End position (x, y)
num_points: Number of points along the trajectory
Returns:
List of points along the trajectory [(x1, y1), (x2, y2), ...]
"""
x_traj = np.linspace(start_pos[0], end_pos[0], num_points)
y_traj = np.linspace(start_pos[1], end_pos[1], num_points)
trajectory = list(zip(x_traj, y_traj))
return trajectory
def execute_trajectory(robot, trajectory, plot=True):
"""
Execute the trajectory (simulation)
Args:
robot: RobotArm instance
trajectory: List of target positions
plot: Whether to visualize the trajectory
"""
joint_angles = []
successful_points = []
for i, (x, y) in enumerate(trajectory):
angles = robot.inverse_kinematics(x, y)
if angles:
joint_angles.append(angles)
successful_points.append((x, y))
else:
print(f"Warning: point {i} ({x:.3f}, {y:.3f}) is unreachable")
if plot and successful_points:
plt.figure(figsize=(10, 8))
# Reachable range
theta = np.linspace(0, 2*np.pi, 100)
r_max = robot.L1 + robot.L2
plt.plot(r_max * np.cos(theta), r_max * np.sin(theta), 'k--', alpha=0.2, label='Maximum reach')
# Trajectory
traj_x, traj_y = zip(*successful_points)
plt.plot(traj_x, traj_y, 'b-', linewidth=2, alpha=0.6, label='Planned trajectory')
plt.plot(traj_x[0], traj_y[0], 'go', markersize=15, label='Start position')
plt.plot(traj_x[-1], traj_y[-1], 'ro', markersize=15, label='End position')
# Display several intermediate poses
for i in range(0, len(joint_angles), len(joint_angles)//5):
theta1, theta2 = joint_angles[i]
th1 = np.radians(theta1)
th2 = np.radians(theta2)
x1 = robot.L1 * np.cos(th1)
y1 = robot.L1 * np.sin(th1)
x2 = x1 + robot.L2 * np.cos(th1 + th2)
y2 = y1 + robot.L2 * np.sin(th1 + th2)
plt.plot([0, x1, x2], [0, y1, y2], 'gray', alpha=0.3, linewidth=1)
plt.axis('equal')
plt.grid(True, alpha=0.3)
plt.xlabel('X position (m)', fontsize=12)
plt.ylabel('Y position (m)', fontsize=12)
plt.title('Robotic arm path planning', fontsize=14, fontweight='bold')
plt.legend()
plt.tight_layout()
plt.savefig('trajectory_planning.png', dpi=300, bbox_inches='tight')
plt.show()
return joint_angles
# Usage example: move from reagent bottle (0.35, 0.15) to reaction vessel (0.25, 0.35)
robot = SimpleRobotArm()
start = (0.35, 0.15)
end = (0.25, 0.35)
trajectory = linear_trajectory(start, end, num_points=30)
print(f"Trajectory generated: {len(trajectory)} points")
joint_angles = execute_trajectory(robot, trajectory, plot=True)
print(f"Execution successful: {len(joint_angles)}/{len(trajectory)} points")
2.2 Liquid Handling: OpenTrons OT-2
2.2.1 Overview of the OpenTrons OT-2
The OpenTrons OT-2 is an open-source liquid handling robot widely used in research laboratories.
Key specifications: - Accuracy: ±1 µL (1-20 µL), ±2% (20-300 µL) - Volume: 1-1000 µL (covered by swapping pipettes) - Deck size: 11 slots (microplates, tube racks, reagent bottles) - Price: About $10,000 (academic discount available) - Programming: Python API (intuitive and easy to learn)
2.2.2 Basic Pipetting
from opentrons import protocol_api
# OT-2 protocol: dispensing reagent into a 96-well plate
metadata = {
'protocolName': 'Basic pipetting',
'author': 'Materials Lab',
'description': 'Dispense reagent into a 96-well plate',
'apiLevel': '2.13'
}
def run(protocol: protocol_api.ProtocolContext):
"""
Basic pipetting protocol
Args:
protocol: OpenTrons ProtocolContext
"""
# Configure the deck layout
# Slot 1: 96-well plate (for reactions)
plate = protocol.load_labware('corning_96_wellplate_360ul_flat', location='1')
# Slot 2: tube rack (reagent bottles)
tuberack = protocol.load_labware('opentrons_24_tuberack_eppendorf_1.5ml_safelock_snapcap', location='2')
# Slot 3: pipette tip rack
tiprack = protocol.load_labware('opentrons_96_tiprack_300ul', location='3')
# Mount the pipette (P300 Single-Channel)
pipette = protocol.load_instrument('p300_single_gen2', mount='left', tip_racks=[tiprack])
# Pipetting operation
# Dispense 50 µL of reagent A (tube A1) into each well
reagent_a = tuberack.wells_by_name()['A1']
for well in plate.wells():
pipette.pick_up_tip() # Pick up a new tip
pipette.aspirate(50, reagent_a) # Aspirate reagent (50 µL)
pipette.dispense(50, well) # Dispense into the well
pipette.blow_out(well.top()) # Expel any residual liquid
pipette.drop_tip() # Discard the tip
protocol.comment("Protocol complete: dispensed reagent A into all 96 wells")
# Protocol simulation (verify operation without hardware)
# Run in a terminal: opentrons_simulate basic_pipetting.py
Code explanation: 1. Loading labware: Place the plate, tube rack, and tip rack on the deck 2. Mounting the pipette: Single-channel 300 µL pipette 3. Pipetting loop: Dispense reagent into each well 4. Cross-contamination prevention: Use a new tip every time
2.2.3 Multi-Channel Pipetting
For high-speed dispensing across an entire 96-well plate, use an 8-channel pipette.
def run(protocol: protocol_api.ProtocolContext):
"""
Multi-channel pipetting (simultaneous dispensing of 8 wells at once)
"""
# Deck layout
source_plate = protocol.load_labware('nest_12_reservoir_15ml', location='1') # Reagent reservoir
dest_plate = protocol.load_labware('corning_96_wellplate_360ul_flat', location='2')
tiprack = protocol.load_labware('opentrons_96_tiprack_300ul', location='3')
# 8-channel pipette
p300_multi = protocol.load_instrument('p300_multi_gen2', mount='left', tip_racks=[tiprack])
# From the reagent reservoir to the 96-well plate
# The 8-channel pipette processes one column (8 wells) at a time
p300_multi.pick_up_tip()
# Process the 12 columns sequentially (96 wells = 12 columns × 8 rows)
for col in dest_plate.columns():
p300_multi.aspirate(100, source_plate['A1']) # Aspirate from the reservoir
p300_multi.dispense(100, col[0]) # First well of the column (A1, A2, ..., A12)
p300_multi.blow_out()
p300_multi.drop_tip()
protocol.comment("Finished dispensing 100 µL into all 96 wells")
# Efficiency comparison
print("Dispensing time into a 96-well plate:")
print(" Single-channel: 96 wells × 20 s = 32 min")
print(" Multi-channel: 12 columns × 20 s = 4 min")
print(" Speedup: 8x faster")
2.2.4 Serial Dilution
Automate serial dilution, which is important for tasks such as catalyst screening.
def run(protocol: protocol_api.ProtocolContext):
"""
Serial dilution protocol (10-fold dilution series)
Concentrations: 10^0, 10^-1, 10^-2, ..., 10^-7 M
"""
plate = protocol.load_labware('corning_96_wellplate_360ul_flat', location='1')
tiprack = protocol.load_labware('opentrons_96_tiprack_300ul', location='2')
reservoir = protocol.load_labware('nest_12_reservoir_15ml', location='3')
pipette = protocol.load_instrument('p300_single_gen2', mount='left', tip_racks=[tiprack])
# Create a dilution series (column A: 10^0 → 10^-7)
# Well A1: stock solution, A2-A8: diluted solutions
# 1. Dispense solvent into each well (A2-A8)
solvent = reservoir['A1']
for well in plate.columns()[0][1:8]: # A2 to A8
pipette.pick_up_tip()
pipette.transfer(180, solvent, well, new_tip='never')
pipette.drop_tip()
# 2. Dispense the stock solution into A1
stock_solution = reservoir['A2']
pipette.transfer(200, stock_solution, plate['A1'], new_tip='once')
# 3. Perform the serial dilution
# A1 → A2 → A3 → ... → A8
pipette.pick_up_tip()
for i in range(7):
source_well = plate.columns()[0][i] # A1, A2, ..., A7
dest_well = plate.columns()[0][i+1] # A2, A3, ..., A8
# Transfer 20 µL to the next well
pipette.aspirate(20, source_well)
pipette.dispense(20, dest_well)
pipette.mix(3, 100, dest_well) # Mix 3 times (100 µL)
pipette.drop_tip()
protocol.comment("Serial dilution complete: 10^0 → 10^-7 M")
# Concentration calculation for the serial dilution
import pandas as pd
dilution_factor = 10 # 10-fold dilution
num_dilutions = 8
initial_concentration = 1.0 # M
concentrations = [initial_concentration / (dilution_factor ** i) for i in range(num_dilutions)]
wells = [f'A{i+1}' for i in range(num_dilutions)]
df_dilution = pd.DataFrame({
'Well': wells,
'Concentration (M)': concentrations,
'Log concentration': [f'10^{int(np.log10(c))}' if c >= 1e-10 else '0' for c in concentrations]
})
print("Serial dilution series:")
print(df_dilution.to_string(index=False))
2.3 Solid Handling
2.3.1 Automating Powder Weighing
Automated weighing of solid reagents requires coordination between a precision balance and a robotic arm.
class PowderDispenserSimulator:
"""
Simulator for a powder dispensing system
Actual equipment: Mettler Toledo Balance + Robot Arm
"""
def __init__(self, accuracy=0.001):
"""
Args:
accuracy: Weighing accuracy (grams)
"""
self.accuracy = accuracy
self.dispensed_amounts = []
def dispense_powder(self, target_mass, powder_name='Reagent A'):
"""
Dispense powder up to a target mass
Args:
target_mass: Target mass (grams)
powder_name: Name of the powder
Returns:
actual_mass: Actually dispensed mass
"""
# Simulation: emulate variation with a normal distribution
actual_mass = np.random.normal(target_mass, self.accuracy)
self.dispensed_amounts.append(actual_mass)
error = actual_mass - target_mass
print(f"{powder_name} dispensed: target {target_mass:.3f}g, measured {actual_mass:.3f}g (error: {error:+.4f}g)")
return actual_mass
def multi_component_dispensing(self, composition_dict):
"""
Automated blending of multi-component powders
Args:
composition_dict: {component name: mass (g)}
Returns:
actual_composition: Actual composition
"""
print("Starting multi-component blending:")
actual_composition = {}
for component, target_mass in composition_dict.items():
actual_mass = self.dispense_powder(target_mass, component)
actual_composition[component] = actual_mass
total_mass = sum(actual_composition.values())
print(f"\nTotal mass: {total_mass:.3f}g")
# Composition ratio (weight %)
print("\nActual composition ratio (wt%):")
for component, mass in actual_composition.items():
percentage = (mass / total_mass) * 100
print(f" {component}: {percentage:.2f}%")
return actual_composition
# Usage example: blending a ternary catalyst (NiO, CoO, MnO2)
dispenser = PowderDispenserSimulator(accuracy=0.002)
# Target composition: Ni:Co:Mn = 60:20:20 (wt%)
total_mass = 1.0 # 1.0 g total
composition = {
'NiO': 0.6,
'CoO': 0.2,
'MnO2': 0.2
}
actual_comp = dispenser.multi_component_dispensing(composition)
# Accuracy evaluation
print("\nAccuracy evaluation:")
for component, target_mass in composition.items():
actual_mass = actual_comp[component]
error_percent = abs((actual_mass - target_mass) / target_mass) * 100
print(f" {component}: error {error_percent:.2f}%")
2.3.2 Transferring Solid Samples
def solid_sample_transfer_protocol():
"""
Automated solid sample transfer protocol (pseudocode)
The actual implementation depends on the robotic arm's API
"""
protocol_steps = [
"1. The robotic arm grips the sample holder",
"2. Move to the XRD measurement position",
"3. Place the sample on the XRD stage",
"4. Start the XRD measurement (external trigger)",
"5. Wait for measurement completion",
"6. Retrieve the sample",
"7. Move to the next sample position",
"8. Repeat steps 1-7"
]
for step in protocol_steps:
print(step)
# Pseudocode: actual robot control
"""
# Example with the Universal Robots UR5e
import urx
robot = urx.Robot("192.168.1.100") # Robot IP address
# Sample position (XYZ coordinates, millimeters)
sample_position = [300, 200, 100, 0, 0, 0] # X, Y, Z, RX, RY, RZ
xrd_position = [500, 200, 150, 0, 0, 0]
# Move
robot.movel(sample_position, acc=0.1, vel=0.1) # Linear move
# Grip the sample with the gripper
robot.set_digital_out(0, True) # Control the gripper with a digital output
robot.movel(xrd_position, acc=0.1, vel=0.1)
# Place the sample
robot.set_digital_out(0, False)
robot.close()
"""
solid_sample_transfer_protocol()
2.4 Sensor Integration
2.4.1 Linking with a Spectrometer
Monitor reactions in real time with a UV-Vis spectrometer.
import time
class SpectrometerSimulator:
"""
UV-Vis spectrometer simulator
Actual equipment: Ocean Optics USB4000, Agilent Cary 60
"""
def __init__(self, wavelength_range=(200, 800), resolution=1):
"""
Args:
wavelength_range: Wavelength range (nm)
resolution: Wavelength resolution (nm)
"""
self.wavelengths = np.arange(wavelength_range[0], wavelength_range[1], resolution)
def measure_absorbance(self, sample_id, concentration=0.1):
"""
Absorbance measurement (simulation)
Args:
sample_id: Sample ID
concentration: Concentration (M)
Returns:
wavelengths, absorbance: Arrays of wavelength and absorbance
"""
# Beer-Lambert law: A = ε * c * l
# Peak wavelength: 450 nm (hypothetical compound)
peak_wavelength = 450
peak_absorbance = concentration * 10 # εcl = 10 (assumed)
# Gaussian absorption spectrum
absorbance = peak_absorbance * np.exp(-((self.wavelengths - peak_wavelength) / 50)**2)
# Add noise
noise = np.random.normal(0, 0.01, len(self.wavelengths))
absorbance += noise
print(f"Sample {sample_id} measurement complete: peak wavelength {peak_wavelength}nm, absorbance {peak_absorbance:.3f}")
return self.wavelengths, absorbance
def kinetic_measurement(self, duration=60, interval=5):
"""
Reaction kinetics measurement (track the change over time)
Args:
duration: Measurement duration (seconds)
interval: Measurement interval (seconds)
Returns:
times, absorbance_at_450nm: Time and absorbance at 450 nm
"""
times = np.arange(0, duration, interval)
absorbance_450 = []
print("Starting reaction kinetics measurement...")
for t in times:
# Simulation of a first-order reaction: [A] = [A]0 * exp(-kt)
k = 0.02 # s^-1
concentration = 0.1 * np.exp(-k * t)
_, abs_spectrum = self.measure_absorbance(f't={t}s', concentration)
# Extract the absorbance at 450 nm
idx_450 = np.argmin(np.abs(self.wavelengths - 450))
absorbance_450.append(abs_spectrum[idx_450])
time.sleep(0.1) # In an actual measurement, wait in real time
return times, np.array(absorbance_450)
# Usage example
spectrometer = SpectrometerSimulator()
# Spectral measurement
wavelengths, absorbance = spectrometer.measure_absorbance('Sample_001', concentration=0.15)
plt.figure(figsize=(10, 6))
plt.plot(wavelengths, absorbance, linewidth=2)
plt.xlabel('Wavelength (nm)', fontsize=12)
plt.ylabel('Absorbance', fontsize=12)
plt.title('UV-Vis absorption spectrum', fontsize=14, fontweight='bold')
plt.grid(alpha=0.3)
plt.tight_layout()
plt.savefig('uv_vis_spectrum.png', dpi=300, bbox_inches='tight')
plt.show()
# Reaction kinetics measurement
times, abs_450 = spectrometer.kinetic_measurement(duration=100, interval=10)
plt.figure(figsize=(10, 6))
plt.plot(times, abs_450, 'o-', linewidth=2, markersize=8)
plt.xlabel('Time (s)', fontsize=12)
plt.ylabel('Absorbance (450 nm)', fontsize=12)
plt.title('Reaction kinetics measurement', fontsize=14, fontweight='bold')
plt.grid(alpha=0.3)
plt.tight_layout()
plt.savefig('kinetic_measurement.png', dpi=300, bbox_inches='tight')
plt.show()
# Fitting the first-order rate constant
from scipy.optimize import curve_fit
def first_order_kinetics(t, A0, k):
return A0 * np.exp(-k * t)
params, covariance = curve_fit(first_order_kinetics, times, abs_450, p0=[1.5, 0.02])
A0_fit, k_fit = params
print(f"\nFirst-order reaction fitting:")
print(f" Initial absorbance A0 = {A0_fit:.3f}")
print(f" Rate constant k = {k_fit:.4f} s^-1")
print(f" Half-life t1/2 = {np.log(2)/k_fit:.1f} s")
2.4.2 Image Analysis with a Camera
Automatically record and analyze crystal growth, precipitate formation, color changes, and more with a camera.
from PIL import Image, ImageDraw, ImageFont
import cv2
class LabCameraSimulator:
"""
Experimental camera simulator
Actual equipment: Basler ace, FLIR Blackfly
"""
def __init__(self, resolution=(1920, 1080)):
self.resolution = resolution
def capture_wellplate(self, plate_id='Plate_001'):
"""
Capture an image of a 96-well plate (simulation)
Returns:
image: PIL Image
"""
# Simulation: generate a grid image of a 96-well plate
img = Image.new('RGB', self.resolution, color='white')
draw = ImageDraw.Draw(img)
# 8 rows × 12 columns of wells
well_diameter = 50
spacing = 70
offset_x, offset_y = 200, 100
for row in range(8):
for col in range(12):
center_x = offset_x + col * spacing
center_y = offset_y + row * spacing
# Well color (simulate changes according to concentration)
intensity = int(255 * (1 - (row + col) / 20)) # Gradually darkens
color = (intensity, intensity, 255)
# Draw the well
draw.ellipse([center_x - well_diameter//2, center_y - well_diameter//2,
center_x + well_diameter//2, center_y + well_diameter//2],
fill=color, outline='black')
print(f"Image capture of plate {plate_id} complete")
return img
def analyze_well_color(self, image, well_position):
"""
Analyze the color of a specific well
Args:
image: PIL Image
well_position: (row, col) well position
Returns:
rgb_mean: Mean RGB value
"""
row, col = well_position
well_diameter = 50
spacing = 70
offset_x, offset_y = 200, 100
center_x = offset_x + col * spacing
center_y = offset_y + row * spacing
# Crop the well region
crop_box = (center_x - well_diameter//2, center_y - well_diameter//2,
center_x + well_diameter//2, center_y + well_diameter//2)
well_img = image.crop(crop_box)
# Compute the mean RGB value
well_array = np.array(well_img)
rgb_mean = well_array.mean(axis=(0, 1))
return rgb_mean
# Usage example
camera = LabCameraSimulator()
plate_image = camera.capture_wellplate('Plate_001')
plate_image.save('wellplate_image.png')
# Color analysis of well A1
rgb = camera.analyze_well_color(plate_image, well_position=(0, 0))
print(f"RGB value of well A1: R={rgb[0]:.1f}, G={rgb[1]:.1f}, B={rgb[2]:.1f}")
# Color analysis of all wells
print("\nBlue component (B value) map of all wells:")
blue_values = np.zeros((8, 12))
for row in range(8):
for col in range(12):
rgb = camera.analyze_well_color(plate_image, (row, col))
blue_values[row, col] = rgb[2]
plt.figure(figsize=(12, 6))
plt.imshow(blue_values, cmap='Blues', interpolation='nearest')
plt.colorbar(label='Blue component (B value)')
plt.xlabel('Column', fontsize=12)
plt.ylabel('Row', fontsize=12)
plt.title('Color distribution of the 96-well plate', fontsize=14, fontweight='bold')
plt.xticks(range(12), [f'{i+1}' for i in range(12)])
plt.yticks(range(8), ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H'])
plt.tight_layout()
plt.savefig('wellplate_color_map.png', dpi=300, bbox_inches='tight')
plt.show()
2.5 Safety and Error Handling
2.5.1 Emergency Stop Functionality
class SafetyController:
"""
Safety control for an experimental automation system
"""
def __init__(self):
self.emergency_stop = False
self.error_log = []
def check_safety(self, temperature, pressure, liquid_level):
"""
Check safety parameters
Args:
temperature: Temperature (℃)
pressure: Pressure (bar)
liquid_level: Liquid level (%)
Returns:
is_safe: Whether it is safe
"""
is_safe = True
warnings = []
# Temperature check
if temperature > 100:
warnings.append(f"Warning: temperature too high ({temperature}℃)")
is_safe = False
# Pressure check
if pressure > 5:
warnings.append(f"Warning: pressure too high ({pressure} bar)")
is_safe = False
# Liquid level check
if liquid_level < 10:
warnings.append(f"Warning: liquid level too low ({liquid_level}%)")
is_safe = False
if not is_safe:
for warning in warnings:
print(warning)
self.error_log.append({'time': time.strftime('%Y-%m-%d %H:%M:%S'), 'message': warning})
return is_safe
def emergency_stop_sequence(self):
"""Emergency stop sequence"""
print("\n🚨 Executing emergency stop 🚨")
self.emergency_stop = True
# Stop actions
actions = [
"1. Stop all robot motion",
"2. Turn off the heater",
"3. Circulate cooling water",
"4. Move valves to their safe positions",
"5. Notify the operator",
"6. Record the log"
]
for action in actions:
print(action)
time.sleep(0.5)
print("\nEmergency stop complete. The system is in a safe state.")
def print_error_log(self):
"""Display the error log"""
print("\n=== Error log ===")
for error in self.error_log:
print(f"[{error['time']}] {error['message']}")
# Usage example
safety = SafetyController()
# Normal state
print("=== Normal operation ===")
is_safe = safety.check_safety(temperature=80, pressure=2.0, liquid_level=50)
print(f"Safe state: {is_safe}\n")
# Abnormal state
print("=== Anomaly detection ===")
is_safe = safety.check_safety(temperature=120, pressure=6.5, liquid_level=5)
print(f"Safe state: {is_safe}")
if not is_safe:
safety.emergency_stop_sequence()
safety.print_error_log()
2.5.2 Error Recovery
def robust_pipetting_with_retry(pipette, source, dest, volume, max_retries=3):
"""
Pipetting with error recovery
Args:
pipette: OT-2 pipette
source: Aspiration source
dest: Dispensing destination
volume: Volume (µL)
max_retries: Maximum number of retries
Returns:
success: Whether it succeeded
"""
for attempt in range(max_retries):
try:
pipette.pick_up_tip()
pipette.aspirate(volume, source)
pipette.dispense(volume, dest)
pipette.drop_tip()
print(f"Pipetting succeeded: {volume} µL (attempt {attempt+1})")
return True
except Exception as e:
print(f"Error occurred (attempt {attempt+1}): {e}")
# Recovery action
if pipette.has_tip:
pipette.drop_tip()
if attempt < max_retries - 1:
print("Retrying...")
time.sleep(1)
else:
print("Reached the maximum number of retries. Aborting the operation.")
return False
return False
# Usage example with pseudocode
print("Demo of pipetting with error recovery:\n")
class MockPipette:
def __init__(self, fail_probability=0.3):
self.has_tip = False
self.fail_probability = fail_probability
def pick_up_tip(self):
if np.random.random() < self.fail_probability:
raise Exception("Tip pickup failed")
self.has_tip = True
def aspirate(self, volume, source):
if np.random.random() < self.fail_probability:
raise Exception("Aspiration failed")
def dispense(self, volume, dest):
if np.random.random() < self.fail_probability:
raise Exception("Dispensing failed")
def drop_tip(self):
self.has_tip = False
mock_pipette = MockPipette(fail_probability=0.4) # Fails with 40% probability
success = robust_pipetting_with_retry(mock_pipette, 'A1', 'B1', 100, max_retries=5)
print(f"\nFinal result: {'success' if success else 'failure'}")
2.6 Labware Standardization
2.6.1 SBS (Society for Biomolecular Screening) Standard
The dimensions of labware such as 96-well and 384-well plates are standardized by international standards.
class SBS_Labware:
"""
Specifications for SBS-standard labware
"""
@staticmethod
def plate_96_well():
"""Specifications of a 96-well plate"""
specs = {
'Name': '96-well plate',
'Footprint': '127.76 mm × 85.48 mm (SBS standard)',
'Well layout': '8 rows × 12 columns',
'Well spacing': '9.0 mm (center-to-center)',
'Well volume': 'Typically 300-360 µL',
'Well shape': 'Flat, U-shaped, V-shaped',
'Use': 'General screening, assays'
}
return specs
@staticmethod
def plate_384_well():
"""Specifications of a 384-well plate"""
specs = {
'Name': '384-well plate',
'Footprint': '127.76 mm × 85.48 mm (same as 96-well)',
'Well layout': '16 rows × 24 columns',
'Well spacing': '4.5 mm (center-to-center, half of the 96-well)',
'Well volume': 'Typically 50-100 µL',
'Use': 'High-density screening, drug discovery'
}
return specs
@staticmethod
def plot_wellplate_layout(n_wells=96):
"""Visualize the layout of a well plate"""
if n_wells == 96:
rows, cols = 8, 12
well_spacing = 9.0 # mm
elif n_wells == 384:
rows, cols = 16, 24
well_spacing = 4.5 # mm
else:
raise ValueError("Only 96 or 384 wells are supported")
fig, ax = plt.subplots(figsize=(12, 6))
# Draw the wells
for row in range(rows):
for col in range(cols):
x = col * well_spacing
y = row * well_spacing
circle = plt.Circle((x, y), radius=well_spacing*0.4, color='lightblue', ec='black')
ax.add_patch(circle)
# Display the well name (96-well only, for readability)
if n_wells == 96:
well_name = f"{chr(65+row)}{col+1}"
ax.text(x, y, well_name, ha='center', va='center', fontsize=8)
ax.set_xlim(-well_spacing, cols * well_spacing)
ax.set_ylim(-well_spacing, rows * well_spacing)
ax.set_aspect('equal')
ax.invert_yaxis() # Invert the Y axis (row A on top)
ax.set_xlabel('X direction (mm)', fontsize=12)
ax.set_ylabel('Y direction (mm)', fontsize=12)
ax.set_title(f'{n_wells}-well plate layout (SBS standard)', fontsize=14, fontweight='bold')
ax.grid(alpha=0.3)
plt.tight_layout()
plt.savefig(f'{n_wells}_wellplate_layout.png', dpi=300, bbox_inches='tight')
plt.show()
# Display the specifications
labware = SBS_Labware()
print("=== 96-well plate ===")
for key, value in labware.plate_96_well().items():
print(f"{key}: {value}")
print("\n=== 384-well plate ===")
for key, value in labware.plate_384_well().items():
print(f"{key}: {value}")
# Visualize the layout
labware.plot_wellplate_layout(n_wells=96)
2.7 Exercises
Exercise 1: Trajectory Optimization for a Robotic Arm (Difficulty: Medium)
Given two reagent bottles (position A: (0.35, 0.15), position B: (0.25, 0.35)) and a reaction vessel (position C: (0.4, 0.3)), plan a trajectory that visits them in the order A → B → C in the shortest time.
Hint
The travel time of a straight-line trajectory is proportional to distance. Compare the two routes A→B→C and A→C→B, and select the one with the shorter total travel distance.Sample Solution
def calculate_path_length(points):
"""Compute the total distance of a path"""
total_distance = 0
for i in range(len(points) - 1):
dx = points[i+1][0] - points[i][0]
dy = points[i+1][1] - points[i][1]
distance = np.sqrt(dx**2 + dy**2)
total_distance += distance
return total_distance
# Positions of the reagent bottles and the reaction vessel
A = (0.35, 0.15) # Reagent A
B = (0.25, 0.35) # Reagent B
C = (0.4, 0.3) # Reaction vessel
# Route 1: A → B → C
route1 = [A, B, C]
distance1 = calculate_path_length(route1)
# Route 2: A → C → B
route2 = [A, C, B]
distance2 = calculate_path_length(route2)
print("Route comparison:")
print(f" A → B → C: {distance1:.3f} m")
print(f" A → C → B: {distance2:.3f} m")
if distance1 < distance2:
print(f"\nOptimal route: A → B → C ({distance1:.3f} m)")
optimal_route = route1
else:
print(f"\nOptimal route: A → C → B ({distance2:.3f} m)")
optimal_route = route2
# Visualization
robot = SimpleRobotArm()
fig, ax = plt.subplots(figsize=(10, 8))
# Reachable range
theta = np.linspace(0, 2*np.pi, 100)
r_max = robot.L1 + robot.L2
ax.plot(r_max * np.cos(theta), r_max * np.sin(theta), 'k--', alpha=0.2, label='Maximum reach')
# Plot the positions
for point, label in zip([A, B, C], ['Reagent A', 'Reagent B', 'Reaction vessel']):
ax.plot(point[0], point[1], 'o', markersize=15, label=label)
# Draw the optimal route
route_x, route_y = zip(*optimal_route)
ax.plot(route_x, route_y, 'r-', linewidth=2, alpha=0.6, label='Optimal path')
ax.axis('equal')
ax.grid(alpha=0.3)
ax.set_xlabel('X position (m)', fontsize=12)
ax.set_ylabel('Y position (m)', fontsize=12)
ax.set_title('Path optimization', fontsize=14, fontweight='bold')
ax.legend()
plt.tight_layout()
plt.savefig('path_optimization.png', dpi=300, bbox_inches='tight')
plt.show()
Exercise 2: Dispensing Multiple Reagents into a 96-Well Plate (Difficulty: Medium)
Create an OpenTrons protocol that dispenses three reagents (A, B, C) into a 96-well plate according to the following pattern.
- Reagent A: 50 µL into columns 1-4 (wells A1-H4)
- Reagent B: 50 µL into columns 5-8 (wells A5-H8)
- Reagent C: 50 µL into columns 9-12 (wells A9-H12)
Hint
Using an 8-channel pipette, you can process each column with a single pipetting operation. You can select columns 1-4 with `plate.columns()[0:4]`.Sample Solution
from opentrons import protocol_api
metadata = {
'protocolName': '96-well 3-reagent dispensing',
'author': 'Materials Lab',
'description': 'Dispense three reagents column by column',
'apiLevel': '2.13'
}
def run(protocol: protocol_api.ProtocolContext):
# Deck layout
plate = protocol.load_labware('corning_96_wellplate_360ul_flat', location='1')
reservoir = protocol.load_labware('nest_12_reservoir_15ml', location='2')
tiprack = protocol.load_labware('opentrons_96_tiprack_300ul', location='3')
# 8-channel pipette
pipette = protocol.load_instrument('p300_multi_gen2', mount='left', tip_racks=[tiprack])
# Reagent positions
reagent_a = reservoir['A1']
reagent_b = reservoir['A2']
reagent_c = reservoir['A3']
# Dispense reagent A into columns 1-4
pipette.pick_up_tip()
for col in plate.columns()[0:4]: # Columns 1-4
pipette.aspirate(50, reagent_a)
pipette.dispense(50, col[0]) # First well of the column (A1, A2, A3, A4)
pipette.blow_out()
pipette.drop_tip()
protocol.comment("Reagent A dispensing complete")
# Dispense reagent B into columns 5-8
pipette.pick_up_tip()
for col in plate.columns()[4:8]: # Columns 5-8
pipette.aspirate(50, reagent_b)
pipette.dispense(50, col[0])
pipette.blow_out()
pipette.drop_tip()
protocol.comment("Reagent B dispensing complete")
# Dispense reagent C into columns 9-12
pipette.pick_up_tip()
for col in plate.columns()[8:12]: # Columns 9-12
pipette.aspirate(50, reagent_c)
pipette.dispense(50, col[0])
pipette.blow_out()
pipette.drop_tip()
protocol.comment("Reagent C dispensing complete")
protocol.comment("Protocol complete: dispensed reagents into all 96 wells")
# Run the simulation
print("Protocol creation complete")
print("Run command: opentrons_simulate three_reagent_protocol.py")
Exercise 3: Anomaly Detection in Sensor Data (Difficulty: Hard)
While monitoring a reaction with a UV-Vis spectrometer, implement a system that detects abnormal absorbance changes (sudden increases or decreases) and triggers an emergency stop.
Conditions: - Measurement interval: 10 seconds - Anomaly criterion: a change of 20% or more from the previous measurement - Emergency stop when anomalies occur twice in a row
Hint
At each measurement, compare with the previous value and compute the rate of change. Introduce an anomaly counter and execute the emergency stop when anomalies are detected twice in a row.Sample Solution
class ReactionMonitor:
"""
Reaction monitoring and anomaly detection
"""
def __init__(self, anomaly_threshold=0.2, consecutive_anomalies=2):
"""
Args:
anomaly_threshold: Threshold for the anomaly criterion (rate of change)
consecutive_anomalies: Number of consecutive anomalies before an emergency stop
"""
self.threshold = anomaly_threshold
self.consecutive_threshold = consecutive_anomalies
self.anomaly_count = 0
self.previous_value = None
self.measurements = []
def check_anomaly(self, current_value):
"""
Anomaly detection
Args:
current_value: Current measured value
Returns:
is_anomaly: Whether it is an anomaly
"""
if self.previous_value is None:
self.previous_value = current_value
return False
# Compute the rate of change
change_rate = abs((current_value - self.previous_value) / self.previous_value)
is_anomaly = change_rate > self.threshold
if is_anomaly:
self.anomaly_count += 1
print(f"⚠️ Anomaly detected: change rate {change_rate*100:.1f}% (threshold: {self.threshold*100:.1f}%)")
print(f" Previous value: {self.previous_value:.3f}, current value: {current_value:.3f}")
print(f" Consecutive anomaly count: {self.anomaly_count}/{self.consecutive_threshold}")
else:
self.anomaly_count = 0 # Reset the anomaly counter if normal
self.previous_value = current_value
return is_anomaly
def should_emergency_stop(self):
"""Determine whether an emergency stop is needed"""
return self.anomaly_count >= self.consecutive_threshold
# Simulation
print("=== Reaction monitoring started ===\n")
monitor = ReactionMonitor(anomaly_threshold=0.2, consecutive_anomalies=2)
safety = SafetyController()
# Measurement data (simulation)
# Normal → Normal → Anomaly 1 → Anomaly 2 (emergency stop)
simulated_absorbance = [1.0, 1.05, 1.08, 1.40, 1.75]
for i, absorbance in enumerate(simulated_absorbance):
print(f"--- Measurement {i+1} ---")
print(f"Absorbance: {absorbance:.3f}")
monitor.measurements.append(absorbance)
is_anomaly = monitor.check_anomaly(absorbance)
if monitor.should_emergency_stop():
print("\n🚨 Consecutive anomalies detected! Executing emergency stop.")
safety.emergency_stop_sequence()
break
print()
# Plot the measurement data
plt.figure(figsize=(10, 6))
plt.plot(range(1, len(monitor.measurements) + 1), monitor.measurements, 'o-', linewidth=2, markersize=10)
plt.axhline(y=monitor.measurements[0] * (1 + monitor.threshold), color='red', linestyle='--', label='Upper threshold')
plt.axhline(y=monitor.measurements[0] * (1 - monitor.threshold), color='red', linestyle='--', label='Lower threshold')
plt.xlabel('Measurement number', fontsize=12)
plt.ylabel('Absorbance', fontsize=12)
plt.title('Reaction monitoring and anomaly detection', fontsize=14, fontweight='bold')
plt.legend()
plt.grid(alpha=0.3)
plt.tight_layout()
plt.savefig('anomaly_detection.png', dpi=300, bbox_inches='tight')
plt.show()
Chapter Summary
In this chapter, we learned the foundational technologies of robotic experiments hands-on.
Key Points
-
Robotic arm control: - Forward kinematics: joint angles → position - Inverse kinematics: target position → joint angles (used in experiments) - Path planning: generating smooth and safe trajectories
-
Liquid handling (OpenTrons OT-2): - Basic pipetting: ±1 µL accuracy - Multi-channel: 8x speedup - Serial dilution: automated dilution series down to 10^-7 M
-
Solid handling: - Powder weighing: ±0.001 g accuracy - Multi-component blending: improved reproducibility through automation
-
Sensor integration: - UV-Vis spectrometer: real-time reaction monitoring - Camera: quantitative evaluation through image analysis
-
Safety design: - Emergency stop: automatic shutdown upon anomaly detection - Error recovery: retry functionality - Logging: ensuring traceability
-
Labware standardization: - SBS standard: international standard for 96/384-well plates - Compatibility: usable across different manufacturers
Preview of the Next Chapter
In Chapter 3, we learn closed-loop optimization that integrates Bayesian optimization with robotic experiments. We implement an automatic cycle of experiment → measurement → analysis → prediction → next experiment, dramatically accelerating materials discovery.
References
- OpenTrons. "OT-2 Robot Documentation." https://docs.opentrons.com/
- Lynch, K. M., & Park, F. C. (2017). Modern Robotics: Mechanics, Planning, and Control. Cambridge University Press.
- Granda, J. M. et al. (2018). "Controlling an organic synthesis robot with machine learning to search for new reactivity." Nature, 559, 377-381.
- SBS (Society for Laboratory Automation and Screening). "ANSI/SLAS Microplate Standards." https://www.slas.org/education/ansi-slas-microplate-standards/
To the next chapter: Chapter 3: Closed-Loop Optimization