Sim-to-Real Transfer Techniques
Learning Objectives
After completing this chapter, you will be able to:
- Understand the challenges and techniques involved in sim-to-real transfer
- Implement domain randomization and domain adaptation methods
- Evaluate and improve sim-to-real transfer success rates
- Apply sim-to-real techniques to humanoid robot control systems
Introduction
Sim-to-real transfer is the process of taking robot behaviors, controllers, or policies developed in simulation and successfully deploying them on physical robots. This is one of the most challenging aspects of robotics development, as the reality gap between simulated and real environments can cause policies that work perfectly in simulation to fail when deployed on real hardware.
For Physical AI & Humanoid Robotics applications, sim-to-real transfer is particularly challenging due to the complex dynamics of bipedal locomotion, the sensitivity of humanoid robots to modeling inaccuracies, and the need for precise balance and coordination. However, successfully bridging the sim-to-real gap is crucial for accelerating development and reducing the need for extensive physical testing.
Core Concepts
The sim-to-real problem stems from differences between simulation and reality in many aspects:
Reality Gap Sources
- Dynamics Differences: Friction, compliance, motor characteristics
- Sensor Noise: Different noise profiles than simulated sensors
- Visual Differences: Lighting, textures, and rendering differences
- Actuation Delays: Real hardware delays and response times
- Modeling Errors: Inaccuracies in robot dynamics modeling
Transfer Techniques
- Domain Randomization: Training policies with randomized simulation parameters
- Domain Adaptation: Adjusting policies between sim and real domains
- System Identification: Accurately modeling real robot dynamics
- Robust Control: Designing controllers insensitive to modeling errors
Domain Randomization
Domain randomization is a powerful technique that involves randomizing simulation parameters during training to make learned policies robust to domain differences. By training with a wide range of possible environments and robot parameters, the resulting policy is more likely to work when deployed on a real robot.
Hands-on Examples
Let's implement sim-to-real transfer techniques:
- Domain Randomization Implementation
- System Identification
- Transfer Validation
#!/usr/bin/env python3
"""
Domain Randomization for Sim-to-Real Transfer
"""
import rclpy
from rclpy.node import Node
from gazebo_msgs.srv import SetPhysicsProperties, GetPhysicsProperties
from gazebo_msgs.msg import ODEPhysics
from sensor_msgs.msg import JointState, Imu
from geometry_msgs.msg import Twist
import numpy as np
import random
import time
class DomainRandomizer(Node):
def __init__(self):
super().__init__('domain_randomizer')
# Service clients for physics configuration
self.get_physics_client = self.create_client(
GetPhysicsProperties,
'/gazebo/get_physics_properties'
)
self.set_physics_client = self.create_client(
SetPhysicsProperties,
'/gazebo/set_physics_properties'
)
# Joint state subscriber to monitor robot state
self.joint_state_subscription = self.create_subscription(
JointState,
'/joint_states',
self.joint_state_callback,
10
)
# IMU subscriber for balance control
self.imu_subscription = self.create_subscription(
Imu,
'/torso_imu',
self.imu_callback,
10
)
# Publisher to send commands for testing
self.cmd_vel_publisher = self.create_publisher(
Twist,
'/cmd_vel',
10
)
# Internal state
self.joint_states = None
self.imu_data = None
self.randomization_history = []
# Randomization parameters
self.randomization_interval = 30 # seconds
self.last_randomization_time = time.time()
# Physics parameter ranges for randomization
self.physics_params = {
'gravity_range': (-10.5, -8.5),
'time_step_range': (0.0005, 0.002),
'max_update_rate_range': (500, 2000),
'ode_sor_iters_range': (20, 80),
'ode_sor_w_range': (1.0, 1.9),
'ode_cfm_range': (0.0, 0.1),
'ode_erp_range': (0.1, 0.5),
'contact_surface_layer_range': (0.0005, 0.002),
'contact_max_correcting_vel_range': (50.0, 200.0)
}
# Robot parameter ranges for randomization
self.robot_params = {
'mass_variance': 0.1, # ±10% mass variation
'friction_variance': 0.3, # ±30% friction variation
'damping_variance': 0.2, # ±20% damping variation
'inertia_variance': 0.15 # ±15% inertia variation
}
# Timer for periodic randomization
self.randomization_timer = self.create_timer(1.0, self.randomize_parameters)
self.get_logger().info('Domain Randomizer initialized')
def joint_state_callback(self, msg):
"""Monitor joint states"""
self.joint_states = msg
def imu_callback(self, msg):
"""Monitor IMU data"""
self.imu_data = msg
def randomize_parameters(self):
"""Randomize physics and robot parameters"""
current_time = time.time()
# Randomize physics parameters periodically
if current_time - self.last_randomization_time > self.randomization_interval:
self.apply_randomization()
self.last_randomization_time = current_time
# Publish a test command to see how the robot responds to new parameters
self.test_robot_response()
def apply_randomization(self):
"""Apply randomization to simulation parameters"""
try:
# Get current physics properties
get_request = GetPhysicsProperties.Request()
get_future = self.get_physics_client.call_async(get_request)
rclpy.spin_until_future_complete(self, get_future)
current_props = get_future.result()
if not current_props:
self.get_logger().error('Failed to get physics properties')
return
# Create randomized properties
new_props = SetPhysicsProperties.Request()
# Randomize gravity
new_props.gravity = current_props.gravity
new_props.gravity.z = random.uniform(
self.physics_params['gravity_range'][0],
self.physics_params['gravity_range'][1]
)
# Randomize time step
new_props.time_step = random.uniform(
self.physics_params['time_step_range'][0],
self.physics_params['time_step_range'][1]
)
# Randomize max update rate
new_props.max_update_rate = random.uniform(
self.physics_params['max_update_rate_range'][0],
self.physics_params['max_update_rate_range'][1]
)
# Set ODE parameters
new_props.ode_config = ODEPhysics()
new_props.ode_config.auto_disable_bodies = False
new_props.ode_config.sor_pgs_precon_iters = 0
new_props.ode_config.sor_pgs_iters = int(random.uniform(
self.physics_params['ode_sor_iters_range'][0],
self.physics_params['ode_sor_iters_range'][1]
))
new_props.ode_config.sor_pgs_w = random.uniform(
self.physics_params['ode_sor_w_range'][0],
self.physics_params['ode_sor_w_range'][1]
)
new_props.ode_config.sor_pgs_rms_error_tol = 0.01
new_props.ode_config.contact_surface_layer = random.uniform(
self.physics_params['contact_surface_layer_range'][0],
self.physics_params['contact_surface_layer_range'][1]
)
new_props.ode_config.contact_max_correcting_vel = random.uniform(
self.physics_params['contact_max_correcting_vel_range'][0],
self.physics_params['contact_max_correcting_vel_range'][1]
)
new_props.ode_config.cfm = random.uniform(
self.physics_params['ode_cfm_range'][0],
self.physics_params['ode_cfm_range'][1]
)
new_props.ode_config.erp = random.uniform(
self.physics_params['ode_erp_range'][0],
self.physics_params['ode_erp_range'][1]
)
new_props.ode_config.max_contacts = 20
# Apply the new physics properties
set_future = self.set_physics_client.call_async(new_props)
rclpy.spin_until_future_complete(self, set_future)
if set_future.result() and set_future.result().success:
self.get_logger().info('Applied randomized physics parameters')
# Record the randomization for analysis
randomization_record = {
'time': time.time(),
'gravity_z': new_props.gravity.z,
'time_step': new_props.time_step,
'max_update_rate': new_props.max_update_rate,
'sor_pgs_iters': new_props.ode_config.sor_pgs_iters,
'sor_pgs_w': new_props.ode_config.sor_pgs_w,
'cfm': new_props.ode_config.cfm,
'erp': new_props.ode_config.erp
}
self.randomization_history.append(randomization_record)
# Log the changes
self.get_logger().info(f'Gravity: {new_props.gravity.z:.3f}')
self.get_logger().info(f'Time step: {new_props.time_step:.4f}')
self.get_logger().info(f'SOR iters: {new_props.ode_config.sor_pgs_iters}')
else:
self.get_logger().error('Failed to set randomized physics properties')
except Exception as e:
self.get_logger().error(f'Error applying randomization: {e}')
def test_robot_response(self):
"""Test robot response to new physics parameters"""
# Send a simple movement command to test robot response
twist = Twist()
twist.linear.x = 0.2 # Move forward slowly
twist.angular.z = 0.0 # No rotation
self.cmd_vel_publisher.publish(twist)
# Schedule stop command
self.create_timer(2.0, self.stop_robot)
def stop_robot(self):
"""Stop the robot"""
twist = Twist()
twist.linear.x = 0.0
twist.angular.z = 0.0
self.cmd_vel_publisher.publish(twist)
def main(args=None):
rclpy.init(args=args)
domain_randomizer = DomainRandomizer()
try:
rclpy.spin(domain_randomizer)
except KeyboardInterrupt:
domain_randomizer.get_logger().info('Domain randomizer stopped by user')
finally:
domain_randomizer.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
#!/usr/bin/env python3
"""
System Identification Node for Sim-to-Real Transfer
"""
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import JointState, Imu
from geometry_msgs.msg import Twist
from std_msgs.msg import Float64
import numpy as np
import scipy.optimize as opt
import matplotlib.pyplot as plt
from collections import deque
import time
class SystemIdentifier(Node):
def __init__(self):
super().__init__('system_identifier')
# Subscribers for system identification
self.joint_state_subscription = self.create_subscription(
JointState,
'/joint_states',
self.joint_state_callback,
10
)
self.imu_subscription = self.create_subscription(
Imu,
'/torso_imu',
self.imu_callback,
10
)
# Publisher for excitation signals
self.cmd_vel_publisher = self.create_publisher(
Twist,
'/cmd_vel',
10
)
# Publisher for identified parameters
self.params_publisher = self.create_publisher(
Float64,
'/identified_params',
10
)
# Internal state
self.joint_states_history = deque(maxlen=1000)
self.imu_history = deque(maxlen=1000)
self.command_history = deque(maxlen=1000)
self.time_history = deque(maxlen=1000)
# Identification parameters
self.excitation_signal = None
self.excitation_duration = 10.0 # seconds
self.excitation_start_time = None
self.is_exciting = False
# Robot dynamics parameters to identify
self.identified_params = {
'mass': 30.0, # kg - will be identified
'inertia': 5.0, # kg*m^2 - will be identified
'friction_coefficient': 0.1, # will be identified
'com_height': 0.8 # center of mass height
}
# Timer for data collection and analysis
self.collection_timer = self.create_timer(0.02, self.collect_data) # 50Hz
self.analysis_timer = self.create_timer(5.0, self.analyze_system) # Every 5s
self.get_logger().info('System Identifier initialized')
def joint_state_callback(self, msg):
"""Store joint state data"""
# Store only for the first few joints (e.g., hip joints)
if len(msg.name) > 0 and 'hip' in msg.name[0]:
self.joint_states_history.append({
'position': np.array(msg.position),
'velocity': np.array(msg.velocity),
'effort': np.array(msg.effort),
'timestamp': time.time()
})
def imu_callback(self, msg):
"""Store IMU data"""
self.imu_history.append({
'linear_acceleration': np.array([
msg.linear_acceleration.x,
msg.linear_acceleration.y,
msg.linear_acceleration.z
]),
'angular_velocity': np.array([
msg.angular_velocity.x,
msg.angular_velocity.y,
msg.angular_velocity.z
]),
'orientation': np.array([
msg.orientation.x,
msg.orientation.y,
msg.orientation.z,
msg.orientation.w
]),
'timestamp': time.time()
})
def collect_data(self):
"""Collect data for system identification"""
current_time = time.time()
# Add timestamp
self.time_history.append(current_time)
# Check if we should start excitation
if not self.is_exciting and len(self.time_history) > 100:
self.start_excitation()
def start_excitation(self):
"""Start robot excitation for system identification"""
if self.excitation_start_time is None:
self.excitation_start_time = time.time()
self.is_exciting = True
self.get_logger().info('Starting system identification excitation')
# Generate excitation signal (sine sweep, step, or PRBS)
elapsed_time = time.time() - self.excitation_start_time
if elapsed_time < self.excitation_duration:
# Sine sweep excitation
frequency = 0.1 + 1.9 * (elapsed_time / self.excitation_duration) # 0.1 to 2 Hz
amplitude = 0.3 # m/s
velocity_cmd = amplitude * np.sin(2 * np.pi * frequency * elapsed_time)
twist = Twist()
twist.linear.x = velocity_cmd
twist.angular.z = 0.1 * np.sin(0.5 * 2 * np.pi * elapsed_time) # Small rotation
self.cmd_vel_publisher.publish(twist)
# Store command
self.command_history.append({
'linear_velocity': velocity_cmd,
'angular_velocity': twist.angular.z,
'timestamp': time.time()
})
else:
# Stop excitation
self.stop_excitation()
def stop_excitation(self):
"""Stop excitation and perform identification"""
twist = Twist()
twist.linear.x = 0.0
twist.angular.z = 0.0
self.cmd_vel_publisher.publish(twist)
self.get_logger().info('Excitation stopped, performing system identification')
# Perform system identification
self.perform_identification()
# Reset for next identification cycle
self.excitation_start_time = None
self.is_exciting = False
def perform_identification(self):
"""Perform system identification to estimate robot parameters"""
try:
# Convert deque to numpy arrays
if len(self.joint_states_history) < 100:
self.get_logger().warn('Not enough data for identification')
return
# Extract relevant data
positions = np.array([state['position'] for state in self.joint_states_history])
velocities = np.array([state['velocity'] for state in self.joint_states_history])
efforts = np.array([state['effort'] for state in self.joint_states_history])
times = np.array(list(self.time_history))
# For this example, we'll identify parameters using a simple approach
# In practice, more sophisticated methods would be used
# Estimate mass and inertia using force-acceleration relationship
# F = ma, Torque = I*alpha
if len(velocities) > 1:
# Estimate acceleration (finite differences)
dt = times[1] - times[0]
accelerations = np.diff(velocities, axis=0) / dt
# For this example, we'll just average the identified values
avg_acceleration = np.mean(np.abs(accelerations), axis=0)
avg_force = np.mean(np.abs(efforts), axis=0)
# Calculate mass (approximately, for the simplest case)
if np.all(avg_acceleration > 0):
estimated_mass = avg_force / avg_acceleration
self.identified_params['mass'] = np.mean(estimated_mass[:3]) # Take first 3 joints as example
self.get_logger().info(f'Estimated mass: {self.identified_params["mass"]:.3f} kg')
# Publish the identified parameter
param_msg = Float64()
param_msg.data = self.identified_params['mass']
self.params_publisher.publish(param_msg)
except Exception as e:
self.get_logger().error(f'Error in system identification: {e}')
def analyze_system(self):
"""Periodically analyze the system for changes"""
self.get_logger().info('System analysis complete')
# In a real implementation, this would check for changes in dynamics
# and potentially trigger re-identification if significant changes detected
def main(args=None):
rclpy.init(args=args)
system_identifier = SystemIdentifier()
try:
rclpy.spin(system_identifier)
except KeyboardInterrupt:
system_identifier.get_logger().info('System identifier stopped by user')
finally:
system_identifier.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
#!/usr/bin/env python3
"""
Validation tools for sim-to-real transfer
"""
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import JointState, Imu
from geometry_msgs.msg import Twist, PoseStamped
from std_msgs.msg import Float64MultiArray, Bool
from nav_msgs.msg import Path
import numpy as np
import matplotlib.pyplot as plt
from collections import deque
import time
import math
class TransferValidator(Node):
def __init__(self):
super().__init__('transfer_validator')
# Subscribers for both simulated and real robot data
self.sim_joint_subscription = self.create_subscription(
JointState,
'/sim/joint_states',
self.sim_joint_callback,
10
)
self.real_joint_subscription = self.create_subscription(
JointState,
'/real/joint_states',
self.real_joint_callback,
10
)
self.sim_imu_subscription = self.create_subscription(
Imu,
'/sim/torso_imu',
self.sim_imu_callback,
10
)
self.real_imu_subscription = self.create_subscription(
Imu,
'/real/torso_imu',
self.real_imu_callback,
10
)
# Publisher for validation metrics
self.metrics_publisher = self.create_publisher(
Float64MultiArray,
'/transfer_metrics',
10
)
self.validation_status_publisher = self.create_publisher(
Bool,
'/transfer_validated',
10
)
# Internal state
self.sim_joint_history = deque(maxlen=500)
self.real_joint_history = deque(maxlen=500)
self.sim_imu_history = deque(maxlen=500)
self.real_imu_history = deque(maxlen=500)
# Validation metrics
self.validation_threshold = 0.2 # maximum acceptable difference
# Timer for validation
self.validation_timer = self.create_timer(2.0, self.validate_transfer)
self.get_logger().info('Transfer Validator initialized')
def sim_joint_callback(self, msg):
"""Store simulated joint data"""
self.sim_joint_history.append({
'position': np.array(msg.position),
'velocity': np.array(msg.velocity),
'effort': np.array(msg.effort),
'timestamp': time.time()
})
def real_joint_callback(self, msg):
"""Store real robot joint data"""
self.real_joint_history.append({
'position': np.array(msg.position),
'velocity': np.array(msg.velocity),
'effort': np.array(msg.effort),
'timestamp': time.time()
})
def sim_imu_callback(self, msg):
"""Store simulated IMU data"""
self.sim_imu_history.append({
'linear_acceleration': np.array([
msg.linear_acceleration.x,
msg.linear_acceleration.y,
msg.linear_acceleration.z
]),
'angular_velocity': np.array([
msg.angular_velocity.x,
msg.angular_velocity.y,
msg.angular_velocity.z
]),
'orientation': self.quaternion_to_euler(
msg.orientation.x,
msg.orientation.y,
msg.orientation.z,
msg.orientation.w
),
'timestamp': time.time()
})
def real_imu_callback(self, msg):
"""Store real robot IMU data"""
self.real_imu_history.append({
'linear_acceleration': np.array([
msg.linear_acceleration.x,
msg.linear_acceleration.y,
msg.linear_acceleration.z
]),
'angular_velocity': np.array([
msg.angular_velocity.x,
msg.angular_velocity.y,
msg.angular_velocity.z
]),
'orientation': self.quaternion_to_euler(
msg.orientation.x,
msg.orientation.y,
msg.orientation.z,
msg.orientation.w
),
'timestamp': time.time()
})
def quaternion_to_euler(self, x, y, z, w):
"""Convert quaternion to Euler angles (roll, pitch, yaw)"""
import math
# Roll (x-axis rotation)
sinr_cosp = 2 * (w * x + y * z)
cosr_cosp = 1 - 2 * (x * x + y * y)
roll = math.atan2(sinr_cosp, cosr_cosp)
# Pitch (y-axis rotation)
sinp = 2 * (w * y - z * x)
if abs(sinp) >= 1:
pitch = math.copysign(math.pi / 2, sinp)
else:
pitch = math.asin(sinp)
# Yaw (z-axis rotation)
siny_cosp = 2 * (w * z + x * y)
cosy_cosp = 1 - 2 * (y * y + z * z)
yaw = math.atan2(siny_cosp, cosy_cosp)
return np.array([roll, pitch, yaw])
def validate_transfer(self):
"""Validate sim-to-real transfer quality"""
# Check if we have enough data
if len(self.sim_joint_history) < 10 or len(self.real_joint_history) < 10:
self.get_logger().warn('Not enough data for validation')
return
# Calculate validation metrics
joint_position_error = self.calculate_joint_position_error()
orientation_error = self.calculate_orientation_error()
balance_error = self.calculate_balance_error()
# Create metrics array
metrics = Float64MultiArray()
metrics.data = [joint_position_error, orientation_error, balance_error]
self.metrics_publisher.publish(metrics)
# Determine if transfer is valid
max_error = max(joint_position_error, orientation_error, balance_error)
is_valid = max_error < self.validation_threshold
# Publish validation status
valid_msg = Bool()
valid_msg.data = is_valid
self.validation_status_publisher.publish(valid_msg)
self.get_logger().info(
f'Transfer validation: Joint error={joint_position_error:.3f}, '
f'Orientation error={orientation_error:.3f}, '
f'Balance error={balance_error:.3f}, '
f'Valid: {is_valid}'
)
if not is_valid:
self.get_logger().warn('Transfer validation failed - error exceeds threshold')
def calculate_joint_position_error(self):
"""Calculate average joint position error between sim and real"""
if len(self.sim_joint_history) == 0 or len(self.real_joint_history) == 0:
return float('inf')
# Take the most recent entries
sim_state = self.sim_joint_history[-1]
real_state = self.real_joint_history[-1]
# Calculate RMS error
pos_error = sim_state['position'][:min(len(sim_state['position']), len(real_state['position']))] - \
real_state['position'][:min(len(sim_state['position']), len(real_state['position']))]
rms_error = np.sqrt(np.mean(pos_error**2))
return rms_error
def calculate_orientation_error(self):
"""Calculate average orientation error between sim and real"""
if len(self.sim_imu_history) == 0 or len(self.real_imu_history) == 0:
return float('inf')
# Take the most recent entries
sim_orient = self.sim_imu_history[-1]['orientation']
real_orient = self.real_imu_history[-1]['orientation']
# Calculate orientation error (simple difference)
orient_diff = np.abs(sim_orient - real_orient)
avg_error = np.mean(orient_diff)
return avg_error
def calculate_balance_error(self):
"""Calculate balance error based on IMU data"""
if len(self.sim_imu_history) == 0 or len(self.real_imu_history) == 0:
return float('inf')
# Take the most recent entries
sim_imu = self.sim_imu_history[-1]
real_imu = self.real_imu_history[-1]
# Calculate the difference in vertical acceleration (gravity)
# In steady state, z-axis should be close to -9.81 m/s^2
sim_balance = abs(sim_imu['linear_acceleration'][2] + 9.81)
real_balance = abs(real_imu['linear_acceleration'][2] + 9.81)
balance_error = abs(sim_balance - real_balance)
return balance_error
def main(args=None):
rclpy.init(args=args)
transfer_validator = TransferValidator()
try:
rclpy.spin(transfer_validator)
except KeyboardInterrupt:
transfer_validator.get_logger().info('Transfer validator stopped by user')
finally:
transfer_validator.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
Expected Output:
[INFO] [1678882844.123456789] [domain_randomizer]: Domain Randomizer initialized
[INFO] [1678882844.123456789] [system_identifier]: System Identifier initialized
[INFO] [1678882844.123456789] [transfer_validator]: Transfer Validator initialized
[INFO] [1678882844.523456789] [domain_randomizer]: Applied randomized physics parameters
[INFO] [1678882844.523456789] [domain_randomizer]: Gravity: -9.213, Time step: 0.0012, SOR iters: 45
[INFO] [1678882850.523456789] [system_identifier]: Estimated mass: 28.45 kg
[INFO] [1678882855.023456789] [transfer_validator]: Transfer validation: Joint error=0.087, Orientation error=0.123, Balance error=0.045, Valid: True
Exercises
Complete the following exercises to reinforce your understanding:
-
Domain Randomization: Implement advanced domain randomization
- Randomize visual textures and lighting conditions
- Include sensor noise randomization
- Add actuator delay and response randomization
- Test policy robustness across randomization ranges
-
Transfer Validation: Develop comprehensive validation methods
- Create metrics for different aspects of robot behavior
- Implement automatic policy adjustment based on validation
- Design validation protocols for different robot tasks
- Establish confidence intervals for transfer success
Common Pitfalls and Solutions
- Pitfall 1: Over-randomization - Randomizing too much makes learning impossible
- Solution: Start with small randomization ranges and gradually increase
- Pitfall 2: Inconsistent real vs. sim - Different ROS message rates or timing
- Solution: Verify sensor rates, actuator commands, and timing synchronization
- Pitfall 3: Unmodeled dynamics - Real robot has unmodeled effects
- Solution: Use system identification to find model discrepancies
- Pitfall 4: Validation failure - No systematic approach to verification
- Solution: Develop comprehensive validation metrics and protocols
Summary
- Domain randomization makes policies robust to sim-real differences
- System identification helps match simulation to reality
- Transfer validation ensures policies work on real hardware
- Sim-to-real transfer requires careful attention to dynamics and sensors
- Proper validation is essential for safe real-world deployment