教程 11:数值优化方法 (90分钟)


目标与简介

本教程将详细介绍如何在Webots中使用数值优化方法。数值优化是机器人学中的重要技术,可以用来优化机器人的步态、行为参数、路径规划等。你将学习不同的优化方法架构、机器人重置策略,以及如何实现各种优化算法。

目录

1. 选择正确的Supervisor controller方法

在Webots中使用优化算法有多种方法,大多数方法都依赖于Supervisor controller。数值优化通常可以分解为两个独立的任务:

优化任务分解

  • 运行优化算法:系统性搜索、随机搜索、遗传算法(GA)、粒子群优化(PSO)、模拟退火等
  • 运行机器人行为:使用优化算法指定的参数集执行机器人行为

一个重要的决策是这两个不同的任务应该在同一个控制器中实现,还是在两个独立的控制器中实现。让我们讨论这两种方法:

方法选择指南

  • 单控制器方法:适用于单机器人优化(如人形机器人步态优化)
  • 双控制器方法:适用于多机器人同时仿真(如群体机器人学)
  • 外部程序方法:适用于复杂优化算法或跨平台需求

2. 使用单个控制器

如果你的仿真一次只需要评估一个机器人,例如优化人形机器人的运动步态或单个机器人的行为,那么可以将两个任务都实现在同一个控制器中,这会使代码更简单。

📝 动手实践 #1: 单控制器优化示例 - 系统性优化两个参数a和b:
from controller import Robot, Supervisor
import math

TIME_STEP = 5

def reset_robot():
    """重置机器人到初始位置"""
    # 获取机器人节点和位置字段
    robot_node = supervisor.getFromDef("MY_ROBOT")
    trans_field = robot_node.getField("translation")
    
    # 重置位置
    initial_trans = [0, 0.5, 0]
    trans_field.setSFVec3f(initial_trans)
    supervisor.simulationResetPhysics()

def actuate_motors(a, b, time):
    """根据参数a、b和时间驱动电机"""
    # 这里实现具体的电机控制逻辑
    # 例如:正弦波运动模式
    position = a * math.sin(b * time)
    
    # 获取电机并设置位置
    motor = robot.getDevice("my_motor")
    motor.setPosition(position)

def compute_fitness():
    """计算适应度函数"""
    # 获取机器人当前位置
    robot_node = supervisor.getFromDef("MY_ROBOT")
    trans_field = robot_node.getField("translation")
    position = trans_field.getSFVec3f()
    
    # 计算移动距离作为适应度
    distance = math.sqrt(position[0]**2 + position[2]**2)
    return distance

# 主程序
robot = Robot()
supervisor = Supervisor()

for a in [x * 0.1 for x in range(5, 100)]:  # a从0.5到9.9
    for b in [x * 0.5 for x in range(1, 10)]:  # b从0.5到4.5
        reset_robot()  # 重置机器人到初始位置
        
        # 运行机器人仿真30秒
        time = 0.0
        while time < 30.0:
            actuate_motors(a, b, time)
            if robot.step(TIME_STEP) == -1:
                break
            time += TIME_STEP / 1000.0
        
        # 计算并打印适应度
        fitness = compute_fitness()
        print(f"参数: a={a:.1f}, b={b:.1f}, 适应度: {fitness:.3f}")

单控制器方法特点

  • 简单实现:所有逻辑都在一个控制器中,便于理解和调试
  • Supervisor权限:需要将Robot节点的supervisor字段设置为TRUE
  • 参数评估:机器人运行30秒,然后评估适应度并重置位置
  • 适用场景:单机器人优化、步态优化、行为参数调优

3. 使用双控制器架构

相反,如果你的仿真需要同时执行多个机器人,例如群体机器人学,建议使用两种不同类型的控制器:一个用于优化算法,一个用于机器人行为。优化算法应该在Supervisor controller中,而机器人行为可以在常规(非Supervisor)控制器中。

📝 动手实践 #2: Supervisor controller(优化算法):
from controller import Supervisor, Emitter
import random

class GeneticOptimizer:
    def __init__(self):
        self.supervisor = Supervisor()
        self.emitter = self.supervisor.getDevice("emitter")
        
        # 遗传算法参数
        self.population_size = 20
        self.genome_size = 10
        self.generation = 0
        self.population = []
        self.fitness_scores = []
        
        # 初始化种群
        self.initialize_population()
    
    def initialize_population():
        """初始化随机种群"""
        self.population = []
        for i in range(self.population_size):
            genome = [random.uniform(-1, 1) for _ in range(self.genome_size)]
            self.population.append(genome)
    
    def send_genome_to_robots(self, genome):
        """发送基因型到机器人控制器"""
        # 将基因型编码为字符串
        genome_str = ",".join([str(g) for g in genome])
        message = f"GENOME:{genome_str}"
        
        # 广播给所有机器人
        self.emitter.send(message.encode('utf-8'))
    
    def evaluate_population(self):
        """评估当前种群"""
        for i, genome in enumerate(self.population):
            print(f"评估个体 {i+1}/{self.population_size}")
            
            # 发送基因型给机器人
            self.send_genome_to_robots(genome)
            
            # 等待机器人完成评估
            self.wait_for_evaluation_complete()
            
            # 接收适应度分数
            fitness = self.receive_fitness_from_robots()
            self.fitness_scores.append(fitness)
    
    def wait_for_evaluation_complete(self):
        """等待机器人完成评估"""
        evaluation_time = 30.0  # 30秒评估时间
        start_time = self.supervisor.getTime()
        
        while (self.supervisor.getTime() - start_time) < evaluation_time:
            if self.supervisor.step(32) == -1:
                break
    
    def receive_fitness_from_robots(self):
        """从机器人接收适应度分数"""
        # 这里简化处理,实际应用中需要通过Receiver接收
        # 可以根据机器人的最终位置计算适应度
        robot_node = self.supervisor.getFromDef("ROBOT_1")
        trans_field = robot_node.getField("translation")
        position = trans_field.getSFVec3f()
        
        # 计算距离作为适应度
        fitness = (position[0]**2 + position[2]**2)**0.5
        return fitness
    
    def select_and_reproduce(self):
        """选择和繁殖操作"""
        # 简单的轮盘赌选择
        total_fitness = sum(self.fitness_scores)
        new_population = []
        
        for _ in range(self.population_size):
            # 选择父代
            r = random.uniform(0, total_fitness)
            cumsum = 0
            for i, fitness in enumerate(self.fitness_scores):
                cumsum += fitness
                if cumsum >= r:
                    parent = self.population[i][:]
                    
                    # 简单变异
                    for j in range(len(parent)):
                        if random.random() < 0.1:  # 10% 变异率
                            parent[j] += random.uniform(-0.1, 0.1)
                            parent[j] = max(-1, min(1, parent[j]))  # 限制范围
                    
                    new_population.append(parent)
                    break
        
        self.population = new_population
        self.fitness_scores = []
    
    def run_optimization(self):
        """运行优化过程"""
        max_generations = 50
        
        for generation in range(max_generations):
            print(f"第 {generation+1} 代")
            
            self.evaluate_population()
            
            # 打印最佳适应度
            best_fitness = max(self.fitness_scores)
            print(f"最佳适应度: {best_fitness:.3f}")
            
            self.select_and_reproduce()

# 运行优化器
if __name__ == "__main__":
    optimizer = GeneticOptimizer()
    optimizer.run_optimization()
📝 动手实践 #3: Robot controller(行为执行):
from controller import Robot, Receiver, Motor, DistanceSensor
import math

class RobotController:
    def __init__(self):
        self.robot = Robot()
        self.receiver = self.robot.getDevice("receiver")
        self.receiver.enable(32)
        
        # 获取传感器和执行器
        self.left_sensor = self.robot.getDevice("left_sensor")
        self.right_sensor = self.robot.getDevice("right_sensor")
        self.left_sensor.enable(32)
        self.right_sensor.enable(32)
        
        self.left_motor = self.robot.getDevice("left_motor")
        self.right_motor = self.robot.getDevice("right_motor")
        self.left_motor.setPosition(float('inf'))
        self.right_motor.setPosition(float('inf'))
        
        self.genome = None
        self.evaluation_active = False
    
    def receive_genome(self):
        """接收来自Supervisor controller的基因型"""
        if self.receiver.getQueueLength() > 0:
            message = self.receiver.getData().decode('utf-8')
            self.receiver.nextPacket()
            
            if message.startswith("GENOME:"):
                genome_str = message[7:]  # 去掉"GENOME:"前缀
                self.genome = [float(x) for x in genome_str.split(",")]
                self.evaluation_active = True
                print(f"收到基因型: {self.genome[:3]}...")  # 只打印前3个基因
    
    def compute_motor_speeds(self, left_dist, right_dist):
        """根据基因型和传感器数据计算电机速度"""
        if self.genome is None:
            return 0.0, 0.0
        
        # 使用基因型作为神经网络权重
        # 简化的行为:避障 + 基因型影响
        
        # 归一化传感器值
        left_norm = left_dist / 1000.0
        right_norm = right_dist / 1000.0
        
        # 使用基因型计算行为
        left_speed = (self.genome[0] * left_norm + 
                     self.genome[1] * right_norm + 
                     self.genome[2]) * 6.0
        
        right_speed = (self.genome[3] * left_norm + 
                      self.genome[4] * right_norm + 
                      self.genome[5]) * 6.0
        
        # 限制速度范围
        left_speed = max(-6.0, min(6.0, left_speed))
        right_speed = max(-6.0, min(6.0, right_speed))
        
        return left_speed, right_speed
    
    def run(self):
        """主运行循环"""
        while self.robot.step(32) != -1:
            # 检查是否收到新的基因型
            self.receive_genome()
            
            if self.evaluation_active and self.genome is not None:
                # 读取传感器
                left_dist = self.left_sensor.getValue()
                right_dist = self.right_sensor.getValue()
                
                # 计算电机速度
                left_speed, right_speed = self.compute_motor_speeds(left_dist, right_dist)
                
                # 设置电机速度
                self.left_motor.setVelocity(left_speed)
                self.right_motor.setVelocity(right_speed)

# 运行Robot controller
if __name__ == "__main__":
    controller = RobotController()
    controller.run()

双控制器方法特点

  • 进程分离:控制器运行在独立的系统进程中,无法直接访问变量
  • 通信机制:使用Webots的Emitter和Receiver进行进程间通信
  • 参数传递:Supervisor controller发送基因型,机器人接收并执行行为
  • 适应度反馈:可在Robot controller或Supervisor controller中评估适应度
  • 适用场景:群体机器人、多机器人协作、复杂优化算法

4. 机器人重置方法

在使用优化算法时,你可能需要在每次适应度评估之前或之后重置机器人。有几种重置机器人的方法:

重置方法对比

  • 字段设置 + 物理重置:快速,但可能不完全
  • 世界重载:完全重置,但需要保存优化状态
  • 世界重置:平衡方案,可选择性重启控制器
  • 外部程序:完全隔离,易于管理

5. 使用字段设置和物理重置

你可以使用Supervisor controller字段设置函数和物理重置功能轻松重置机器人的位置、方向和物理状态:

📝 动手实践 #4: 字段设置重置方法:
from controller import Supervisor

def reset_robot_position(supervisor):
    """使用字段设置重置机器人位置和方向"""
    
    # 获取机器人节点的translation和rotation字段句柄
    robot_node = supervisor.getFromDef("MY_ROBOT")
    trans_field = robot_node.getField("translation")
    rot_field = robot_node.getField("rotation")
    
    # 重置机器人位置和方向
    initial_trans = [0, 0.5, 0]  # 初始位置
    initial_rot = [0, 1, 0, 1.5708]  # 初始方向 (90度绕Y轴)
    
    trans_field.setSFVec3f(initial_trans)
    rot_field.setSFRotation(initial_rot)
    
    # 重置物理状态
    supervisor.simulationResetPhysics()
    
    print("机器人已重置到初始位置")

def reset_motor_positions(robot):
    """重置电机位置"""
    # 获取所有电机
left_motor = robot.getDevice("left_motor")
right_motor = robot.getDevice("right_motor")
    
    # 重置电机位置
    left_motor.setPosition(0.0)
    right_motor.setPosition(0.0)
    
    print("电机位置已重置")

# 使用示例
supervisor = Supervisor()

# 优化循环
for parameter_set in parameter_combinations:
    # 重置机器人
    reset_robot_position(supervisor)
    reset_motor_positions(supervisor)
    
    # 运行评估
    run_evaluation(parameter_set)
    
    # 计算适应度
    fitness = compute_fitness()
    print(f"参数 {parameter_set}, 适应度: {fitness}")

字段重置方法的局限性

  • 部分重置:只重置机器人的主要位置和方向
  • 电机状态:需要手动重置电机位置
  • 控制器状态:机器人控制器的内部状态不会重置
  • 复杂场景:对于复杂的仿真可能需要重置更多参数

6. 使用世界重载功能

这个函数会从头开始重启物理仿真和所有控制器。使用这种方法,包括物理、电机位置和控制器在内的所有内容都会被重置。但是这个函数也会重启调用它的控制器(通常是运行优化算法的控制器),因此优化状态会丢失。

📝 动手实践 #5: 世界重载优化示例:
from controller import Supervisor
import pickle
import os
import sys

class OptimizationState:
    def __init__(self):
        self.current_parameters = None
        self.current_generation = 0
        self.population = []
        self.fitness_scores = []
        self.best_fitness = 0.0
        self.evaluation_count = 0

def save_optimization_state(state, filename="optimization_state.pkl"):
    """保存完整的优化状态到文件"""
    try:
        with open(filename, 'wb') as f:
            pickle.dump(state, f)
        print(f"优化状态已保存到 {filename}")
    except Exception as e:
        print(f"保存状态失败: {e}")

def load_optimization_state(filename="optimization_state.pkl"):
    """从文件加载优化状态"""
    if os.path.exists(filename):
        try:
            with open(filename, 'rb') as f:
                state = pickle.load(f)
            print(f"优化状态已从 {filename} 加载")
            return state
        except Exception as e:
            print(f"加载状态失败: {e}")
    
    # 如果文件不存在或加载失败,返回新状态
    return OptimizationState()

def run_robot_evaluation(supervisor, parameters):
    """运行机器人评估"""
    print(f"评估参数: {parameters}")
    
    # 运行机器人30秒
    time = 0.0
    while time < 30.0:
        # 这里实现具体的机器人控制逻辑
        # 使用parameters影响机器人行为
        
        if supervisor.step(32) == -1:
            break
        time += 32 / 1000.0
    
    # 计算适应度
    robot_node = supervisor.getFromDef("MY_ROBOT")
    trans_field = robot_node.getField("translation")
    position = trans_field.getSFVec3f()
    
    fitness = (position[0]**2 + position[2]**2)**0.5
    return fitness

def get_next_parameters(state):
    """获取下一组评估参数"""
    # 这里实现你的优化算法逻辑
    # 例如:随机搜索、遗传算法、粒子群优化等
    
    if state.evaluation_count < 100:  # 评估100组参数
        # 简单随机搜索示例
        import random
        parameters = [random.uniform(-1, 1) for _ in range(5)]
        state.current_parameters = parameters
        state.evaluation_count += 1
        return parameters
    else:
        return None  # 优化完成

def evaluate_next_parameters(supervisor, state):
    """评估下一组参数"""
    parameters = get_next_parameters(state)
    if parameters is None:
        print("优化完成!")
        return False
    
    # 运行机器人评估
    fitness = run_robot_evaluation(supervisor, parameters)
    
    # 存储结果
    state.fitness_scores.append(fitness)
    if fitness > state.best_fitness:
        state.best_fitness = fitness
        print(f"新的最佳适应度: {fitness:.3f}")
    
    # 保存完整优化状态
    save_optimization_state(state)
    
    # 开始下一次评估
    supervisor.worldReload()
    supervisor.step(32)
    sys.exit(0)  # 终止当前进程

def main():
    supervisor = Supervisor()
    
    # 重新加载完整优化状态
    state = load_optimization_state()
    
    print(f"继续优化,当前评估次数: {state.evaluation_count}")
    
    if state.evaluation_count > 0:
        print(f"当前最佳适应度: {state.best_fitness:.3f}")
    
    # 评估下一组参数
    if not evaluate_next_parameters(supervisor, state):
        print("所有评估已完成")
        # 打印最终结果
        print(f"最终最佳适应度: {state.best_fitness:.3f}")
        print(f"总评估次数: {state.evaluation_count}")

if __name__ == "__main__":
    main()

世界重载方法要点

  • 完全重置:物理、电机位置、控制器状态全部重置
  • 状态保存:必须在调用worldReload()前保存优化状态
  • 状态恢复:Supervisor controller重启时需要恢复优化状态
  • 适用算法:遗传算法需保存种群和适应度,PSO需保存粒子位置和速度

7. 使用世界重置功能

与worldReload()函数类似,worldReset()函数也会重置物理仿真。但是,它不会重启控制器。这种方法的优势是可以使用nodeRestartController()函数选择性地重启所需的控制器。通常,你会重启Robot controller但不重启Supervisor controller,因此不需要Supervisor controller保存和恢复优化状态。

📝 动手实践 #6: 世界重置优化示例:
from controller import Supervisor
import random

class WorldResetOptimizer:
    def __init__(self):
        self.supervisor = Supervisor()
        self.robot_node = self.supervisor.getFromDef("MY_ROBOT")
        
        # 优化参数
        self.current_generation = 0
        self.max_generations = 50
        self.population_size = 20
        self.population = []
        self.fitness_scores = []
        
        # 初始化种群
        self.initialize_population()
    
    def initialize_population(self):
        """初始化随机种群"""
        self.population = []
        for _ in range(self.population_size):
            # 创建随机参数向量
            individual = [random.uniform(-1, 1) for _ in range(8)]
            self.population.append(individual)
        print(f"初始化了 {self.population_size} 个个体的种群")
    
    def reset_simulation(self):
        """重置仿真环境"""
        # 重置世界物理状态
        self.supervisor.worldReset()
        
        # 选择性重启机器人控制器(但不重启Supervisor controller)
        self.supervisor.nodeRestartController(self.robot_node)
        
        # 等待重置完成
        self.supervisor.step(32)
        print("仿真环境已重置")
    
    def send_parameters_to_robot(self, parameters):
        """发送参数给Robot controller"""
        # 这里可以使用Emitter发送参数
        # 或者写入文件供Robot controller读取
        # 为简化,我们假设通过某种方式传递了参数
        pass
    
    def evaluate_individual(self, individual):
        """评估单个个体"""
        print(f"评估个体: {individual[:3]}...")  # 只显示前3个参数
        
        # 重置仿真
        self.reset_simulation()
        
        # 发送参数给机器人
        self.send_parameters_to_robot(individual)
        
        # 让机器人运行30秒
        evaluation_time = 30.0
        start_time = self.supervisor.getTime()
        
        while (self.supervisor.getTime() - start_time) < evaluation_time:
            if self.supervisor.step(32) == -1:
                break
        
        # 计算适应度
        fitness = self.calculate_fitness()
        print(f"适应度: {fitness:.3f}")
        
        return fitness
    
    def calculate_fitness(self):
        """计算适应度函数"""
        # 获取机器人最终位置
        trans_field = self.robot_node.getField("translation")
        position = trans_field.getSFVec3f()
        
        # 距离起点的距离作为适应度
        distance = (position[0]**2 + position[2]**2)**0.5
        
        # 可以加入其他适应度指标
        # 例如:避免碰撞、能耗等
        
        return distance
    
    def selection(self):
        """选择操作"""
        # 找到最佳个体
        best_indices = sorted(range(len(self.fitness_scores)), 
                            key=lambda i: self.fitness_scores[i], 
                            reverse=True)
        
        # 选择前50%作为父代
        elite_count = self.population_size // 2
        elite_population = [self.population[i] for i in best_indices[:elite_count]]
        
        return elite_population
    
    def crossover_and_mutation(self, elite_population):
        """交叉和变异操作"""
        new_population = elite_population[:]  # 保留精英
        
        # 生成新个体直到达到种群大小
        while len(new_population) < self.population_size:
            # 随机选择两个父代
            parent1 = random.choice(elite_population)
            parent2 = random.choice(elite_population)
            
            # 简单的单点交叉
            crossover_point = random.randint(1, len(parent1) - 1)
            child = parent1[:crossover_point] + parent2[crossover_point:]
            
            # 变异
            for i in range(len(child)):
                if random.random() < 0.1:  # 10% 变异率
                    child[i] += random.uniform(-0.2, 0.2)
                    child[i] = max(-1, min(1, child[i]))  # 限制范围
            
            new_population.append(child)
        
        return new_population
    
    def run_optimization(self):
        """运行完整的优化过程"""
        print("开始遗传算法优化...")
        
        for generation in range(self.max_generations):
            print(f"\n=== 第 {generation + 1} 代 ===")
            
            # 评估当前种群
            self.fitness_scores = []
            for i, individual in enumerate(self.population):
                print(f"个体 {i + 1}/{self.population_size}")
                fitness = self.evaluate_individual(individual)
                self.fitness_scores.append(fitness)
            
            # 输出统计信息
            best_fitness = max(self.fitness_scores)
            avg_fitness = sum(self.fitness_scores) / len(self.fitness_scores)
            
            print(f"最佳适应度: {best_fitness:.3f}")
            print(f"平均适应度: {avg_fitness:.3f}")
            
            # 选择、交叉和变异
            elite_population = self.selection()
            self.population = self.crossover_and_mutation(elite_population)
        
        print(f"\n优化完成!最终最佳适应度: {max(self.fitness_scores):.3f}")

# 运行优化器
if __name__ == "__main__":
    optimizer = WorldResetOptimizer()
    optimizer.run_optimization()

世界重置方法优势

  • 选择性重启:可以只重启Robot controller,保留Supervisor controller状态
  • 状态保持:不需要保存和恢复优化算法状态
  • 清洁重置:物理状态完全重置,确保每次评估的一致性
  • 灵活控制:可以控制哪些控制器需要重启

8. 外部程序优化方法

最后一种方法是为每次参数评估启动和退出Webots程序。虽然这听起来像是开销,但实际上Webots的启动时间通常比评估控制器所需的时间短得多,所以这种方法是完全合理的。

📝 动手实践 #7: 外部优化脚本(Python):
#!/usr/bin/env python3
"""
外部优化程序 - 通过启动和停止Webots进行优化
"""

import subprocess
import random
import json
import os
import time

class ExternalOptimizer:
    def __init__(self):
        self.webots_path = "webots"  # Webots可执行文件路径
        self.world_file = "my_optimization_world.wbt"
        self.genotype_file = "genotype.json"
        self.fitness_file = "fitness.json"
        
        # 优化参数
        self.population_size = 20
        self.generations = 30
        self.genome_size = 8
        
        # 种群和适应度
        self.population = []
        self.fitness_scores = []
        
        # 初始化种群
        self.initialize_population()
    
    def initialize_population(self):
        """初始化随机种群"""
        self.population = []
        for _ in range(self.population_size):
            genome = [random.uniform(-1, 1) for _ in range(self.genome_size)]
            self.population.append(genome)
        print(f"初始化了 {self.population_size} 个个体的种群")
    
    def write_genotype_file(self, genotype):
        """写入基因型到文件"""
        with open(self.genotype_file, 'w') as f:
            json.dump(genotype, f)
    
    def read_fitness_file(self):
        """从文件读取适应度"""
        if os.path.exists(self.fitness_file):
            with open(self.fitness_file, 'r') as f:
                return json.load(f)
        return 0.0
    
    def run_webots_evaluation(self, genotype):
        """运行Webots评估单个基因型"""
        print(f"评估基因型: {genotype[:3]}...")
        
        # 写入基因型文件
        self.write_genotype_file(genotype)
        
        # 删除旧的适应度文件
        if os.path.exists(self.fitness_file):
            os.remove(self.fitness_file)
        
        # 启动Webots
        cmd = [self.webots_path, "--mode=fast", self.world_file]
        
        try:
            # 运行Webots并等待完成
            result = subprocess.run(cmd, 
                                  timeout=60,  # 60秒超时
                                  capture_output=True, 
                                  text=True)
            
            if result.returncode == 0:
                # 成功完成,读取适应度
                fitness = self.read_fitness_file()
                print(f"适应度: {fitness:.3f}")
                return fitness
            else:
                print(f"Webots执行失败: {result.stderr}")
                return 0.0
                
        except subprocess.TimeoutExpired:
            print("Webots执行超时")
            return 0.0
        except Exception as e:
            print(f"运行Webots时出错: {e}")
            return 0.0
    
    def evaluate_population(self):
        """评估整个种群"""
        self.fitness_scores = []
        
        for i, individual in enumerate(self.population):
            print(f"个体 {i + 1}/{self.population_size}")
            fitness = self.run_webots_evaluation(individual)
            self.fitness_scores.append(fitness)
            
            # 短暂延迟避免文件系统问题
            time.sleep(0.5)
    
    def genetic_operations(self):
        """遗传算法操作:选择、交叉、变异"""
        # 选择:保留最佳个体
        fitness_pairs = list(zip(self.fitness_scores, self.population))
        fitness_pairs.sort(key=lambda x: x[0], reverse=True)
        
        # 保留前50%
        elite_count = self.population_size // 2
        elite = [individual for _, individual in fitness_pairs[:elite_count]]
        
        # 生成新种群
        new_population = elite[:]  # 精英保留
        
        while len(new_population) < self.population_size:
            # 选择父代
            parent1 = random.choice(elite)
            parent2 = random.choice(elite)
            
            # 交叉
            crossover_point = random.randint(1, self.genome_size - 1)
            child = parent1[:crossover_point] + parent2[crossover_point:]
            
            # 变异
            for j in range(len(child)):
                if random.random() < 0.15:  # 15% 变异率
                    child[j] += random.uniform(-0.3, 0.3)
                    child[j] = max(-1, min(1, child[j]))  # 限制范围
            
            new_population.append(child)
        
        self.population = new_population
    
    def run_optimization(self):
        """运行完整的优化过程"""
        print("开始外部优化过程...")
        print(f"种群大小: {self.population_size}")
        print(f"世代数: {self.generations}")
        print(f"基因组大小: {self.genome_size}")
        
        best_fitness_history = []
        
        for generation in range(self.generations):
            print(f"\n=== 第 {generation + 1} 代 ===")
            
            # 评估种群
            self.evaluate_population()
            
            # 统计信息
            best_fitness = max(self.fitness_scores)
            avg_fitness = sum(self.fitness_scores) / len(self.fitness_scores)
            
            best_fitness_history.append(best_fitness)
            
            print(f"最佳适应度: {best_fitness:.3f}")
            print(f"平均适应度: {avg_fitness:.3f}")
            
            # 遗传操作(最后一代不需要)
            if generation < self.generations - 1:
                self.genetic_operations()
        
        # 输出最终结果
        print(f"\n优化完成!")
        print(f"最终最佳适应度: {max(best_fitness_history):.3f}")
        
        # 保存结果
        self.save_results(best_fitness_history)
    
    def save_results(self, fitness_history):
        """保存优化结果"""
        results = {
            "best_fitness_history": fitness_history,
            "final_population": self.population,
            "final_fitness_scores": self.fitness_scores
        }
        
        with open("optimization_results.json", 'w') as f:
            json.dump(results, f, indent=2)
        
        print("结果已保存到 optimization_results.json")

if __name__ == "__main__":
    optimizer = ExternalOptimizer()
    optimizer.run_optimization()
📝 动手实践 #8: Webots Robot controller(机器人评估):
from controller import Robot, Supervisor, DistanceSensor, Motor
import json
import sys
import math

def load_genotype(filename="genotype.json"):
    """从文件加载基因型"""
    try:
        with open(filename, 'r') as f:
            return json.load(f)
    except Exception as e:
        print(f"加载基因型失败: {e}")
        return [0.0] * 8  # 默认基因型

def save_fitness(fitness, filename="fitness.json"):
    """保存适应度到文件"""
    try:
        with open(filename, 'w') as f:
            json.dump(fitness, f)
        print(f"适应度已保存: {fitness:.3f}")
    except Exception as e:
        print(f"保存适应度失败: {e}")

def compute_fitness(supervisor):
    """计算适应度函数"""
    # 获取机器人最终位置
    robot_node = supervisor.getFromDef("MY_ROBOT")
    trans_field = robot_node.getField("translation")
    position = trans_field.getSFVec3f()
    
    # 计算移动距离
    distance = math.sqrt(position[0]**2 + position[2]**2)
    
    # 可以添加更复杂的适应度计算
    # 例如:避免碰撞、路径效率等
    
    return distance

def run_robot_behavior(robot, genotype):
    """使用基因型控制机器人行为"""
    # 获取传感器和电机
    left_sensor = robot.getDevice("left_sensor")
    right_sensor = robot.getDevice("right_sensor")
    left_sensor.enable(32)
    right_sensor.enable(32)
    
    left_motor = robot.getDevice("left_motor")
    right_motor = robot.getDevice("right_motor")
    left_motor.setPosition(float('inf'))
    right_motor.setPosition(float('inf'))
    
    # 使用基因型作为神经网络权重
    def compute_motor_speeds(left_dist, right_dist):
        # 归一化传感器输入
        left_norm = left_dist / 1000.0
        right_norm = right_dist / 1000.0
        
        # 使用基因型计算输出
        left_speed = (genotype[0] * left_norm + 
                     genotype[1] * right_norm + 
                     genotype[2]) * 6.0
        
        right_speed = (genotype[3] * left_norm + 
                      genotype[4] * right_norm + 
                      genotype[5]) * 6.0
        
        # 限制速度范围
        left_speed = max(-6.0, min(6.0, left_speed))
        right_speed = max(-6.0, min(6.0, right_speed))
        
        return left_speed, right_speed
    
    # 运行机器人30秒
    TIME_STEP = 32
    evaluation_time = 30.0
    current_time = 0.0
    
    while current_time < evaluation_time:
        # 读取传感器
        left_dist = left_sensor.getValue()
        right_dist = right_sensor.getValue()
        
        # 计算电机速度
        left_speed, right_speed = compute_motor_speeds(left_dist, right_dist)
        
        # 设置电机速度
        left_motor.setVelocity(left_speed)
        right_motor.setVelocity(right_speed)
        
        # 推进仿真
        if robot.step(TIME_STEP) == -1:
                break
        
        current_time += TIME_STEP / 1000.0

def main():
    # 初始化
    robot = Robot()
    supervisor = Supervisor()
    
    print("开始机器人评估...")
    
    # 加载基因型
    genotype = load_genotype()
    print(f"加载的基因型: {genotype}")
    
    # 运行机器人行为
    run_robot_behavior(robot, genotype)
    
    # 计算并保存适应度
    fitness = compute_fitness(supervisor)
    save_fitness(fitness)
    
    # 退出Webots
    supervisor.simulationQuit()
    robot.step(32)
    
    print("评估完成")

if __name__ == "__main__":
    main()

外部程序方法特点

  • 完全隔离:优化算法和Robot controller完全分离
  • 语言灵活:外部程序可以用任何编程语言编写
  • 文件通信:通过文件交换参数和适应度数据
  • 清洁重启:每次评估都是完全的新开始
  • 易于调试:可以独立测试优化算法和机器人行为

9. 实际应用示例

Webots发行版中包含了使用优化技术的完整示例。你可以在WEBOTS_HOME/projects/samples/curriculum/worlds目录下找到这些世界文件:

官方示例

  • advanced_particle_swarm_optimization.wbt:粒子群优化算法示例
  • advanced_genetic_algorithm.wbt:遗传算法优化示例
  • 这些示例在Cyberbotics机器人课程的高级编程练习中有详细描述

10. 最佳实践与性能优化

优化方法选择建议

  • 单机器人简单优化:使用单控制器方法
  • 多机器人或复杂算法:使用双控制器架构
  • 长期运行或分布式:使用外部程序方法
  • 需要完全重置:考虑世界重载或外部程序

性能优化技巧

  • 时间步设置:根据需要调整基本时间步和控制时间步
  • 仿真模式:使用--mode=fast提高评估速度
  • 渲染控制:在优化过程中禁用不必要的渲染
  • 并行评估:对于独立评估,考虑并行运行多个Webots实例
  • 结果缓存:避免重复评估相同的参数组合

调试和监控

  • 进度跟踪:定期保存优化状态和中间结果
  • 可视化:绘制适应度曲线监控收敛情况
  • 参数边界:确保参数在合理范围内,防止发散
  • 错误处理:实现robust的错误处理和恢复机制
  • 日志记录:详细记录优化过程以便后续分析

11. 总结

恭喜你完成了Webots数值优化方法教程!你已经学会了在Webots中实现优化算法的多种方法和技术。

学到的核心技能

  • 优化架构设计:理解单控制器、双控制器和外部程序三种方法的优缺点
  • 机器人重置策略:掌握字段重置、世界重载、世界重置等不同重置方法
  • 参数传递机制:学会使用文件通信、Emitter/Receiver通信等方式
  • 适应度函数设计:能够根据任务需求设计合适的评估指标
  • 优化算法实现:理解遗传算法、随机搜索等优化算法的实现

应用场景总结

  • 步态优化:优化人形机器人或四足机器人的行走模式
  • 路径规划:优化机器人的导航和避障策略
  • 控制参数调优:自动调整PID控制器参数
  • 群体行为优化:优化多机器人协作策略
  • 神经网络训练:使用进化算法训练机器人控制网络

方法选择指南

方法 适用场景 优点 缺点
单控制器 单机器人优化 代码简单,易于实现 不适合多机器人
双控制器 多机器人,复杂算法 支持并行,功能强大 通信复杂
外部程序 分布式,长期运行 完全隔离,语言灵活 启动开销

下一步建议

  • 实践探索:尝试实现不同的优化算法(PSO、差分进化、模拟退火等)
  • 性能优化:学习并行优化、GPU加速等高级技术
  • 多目标优化:探索Pareto最优解和多目标优化方法
  • 强化学习:结合强化学习算法进行机器人控制优化
  • 实际应用:将学到的技术应用到实际的机器人项目中

数值优化是机器人学中的强大工具,能够自动发现最优的机器人行为和控制参数。通过本教程的学习,你已经掌握了在Webots中实现各种优化方法的技能。继续实践和探索,你将能够解决更加复杂和有趣的机器人优化问题!