diff --git a/src/Unmanned_vehicle_AD_DQN/Environment.py b/src/Unmanned_vehicle_AD_DQN/Environment.py index 092655362d..5929383d5f 100644 --- a/src/Unmanned_vehicle_AD_DQN/Environment.py +++ b/src/Unmanned_vehicle_AD_DQN/Environment.py @@ -24,6 +24,14 @@ def __init__(self): self.client = carla.Client("localhost", 2000) # CARLA客户端 self.client.set_timeout(20.0) # 连接超时设置 self.front_camera = None # 前置摄像头图像 + + # 新增变量 + self.last_action = 1 # 上一个动作，默认保持 + self.same_steer_counter = 0 # 连续同向转向计数器 + self.suggested_action = None # 建议的避让动作 + self.episode_start_time = None # 每轮开始时间 + self.last_ped_distance = float('inf') # 上次最近行人距离 + self.current_episode = 1 # 当前episode编号 # 加载世界和蓝图 self.world = self.client.load_world('Town03') @@ -63,40 +71,36 @@ def setup_observer_view(self): except Exception as e: print(f"设置观察者视角时出错: {e}") - def setup_observer_view(self): - """设置观察者视角，让用户可以在CARLA窗口中看到场景""" - try: - # 获取当前地图的生成点 - spawn_points = self.world.get_map().get_spawn_points() - if spawn_points: - # 选择一个合适的观察者位置 - spectator = self.world.get_spectator() - - # 设置观察者位置在车辆起始位置附近 - transform = carla.Transform() - transform.location.x = -81.0 - transform.location.y = -195.0 - transform.location.z = 15.0 # 提高视角高度 - transform.rotation.pitch = -45.0 # 向下倾斜视角 - transform.rotation.yaw = 0.0 - transform.rotation.roll = 0.0 - - spectator.set_transform(transform) - print("观察者视角已设置") - except Exception as e: - print(f"设置观察者视角时出错: {e}") - def spawn_pedestrians_general(self, number, isCross): - """生成指定数量的行人 - 大幅减少数量""" - # 限制最大生成数量 - number = min(number, 8) # 最多8个行人 + """生成指定数量的行人""" + # 记录要生成的数量 + target_number = number + + # 记录成功生成的数量 + success_count = 0 - for i in range(number): + # 尝试生成指定数量的行人 + attempts = 0 + max_attempts = number * 3 # 最多尝试3倍次数 + + while success_count < target_number and attempts < max_attempts: + attempts += 1 isLeft = random.choice([True, False]) # 随机选择左右侧 - if isLeft: - self.spawn_pedestrians_left(isCross) - else: - self.spawn_pedestrians_right(isCross) + + try: + if isLeft: + if self.spawn_pedestrians_left(isCross): + success_count += 1 + else: + if self.spawn_pedestrians_right(isCross): + success_count += 1 + except Exception as e: + # 如果生成失败，继续尝试 + print(f"生成行人失败: {e}") + continue + + print(f"成功生成 {success_count}/{target_number} 个行人 (isCross={isCross})") + return success_count def spawn_pedestrians_right(self, isCross): """在右侧生成行人""" @@ -118,31 +122,40 @@ def spawn_pedestrians_right(self, isCross): min_x = 17 max_x = 20.5 - # 随机生成位置 - x = random.uniform(min_x, max_x) - y = random.uniform(min_y, max_y) - - spawn_point = carla.Transform(carla.Location(x, y, 2.0)) - - # 避免在特定区域生成 - while (-10 < spawn_point.location.x < 17) or (70 < spawn_point.location.x < 100): + # 尝试多次生成直到成功 + for attempt in range(3): # 尝试3次 + # 随机生成位置 x = random.uniform(min_x, max_x) y = random.uniform(min_y, max_y) + spawn_point = carla.Transform(carla.Location(x, y, 2.0)) - # 尝试生成行人 - walker_bp = random.choice(blueprints_walkers) - npc = self.world.try_spawn_actor(walker_bp, spawn_point) + # 避免在特定区域生成 + while (-10 < spawn_point.location.x < 17) or (70 < spawn_point.location.x < 100): + x = random.uniform(min_x, max_x) + y = random.uniform(min_y, max_y) + spawn_point = carla.Transform(carla.Location(x, y, 2.0)) - if npc is not None: - # 设置行人控制参数 - ped_control = carla.WalkerControl() - ped_control.speed = random.uniform(0.5, 1.0) # 随机速度 - ped_control.direction.y = -1 # 主要移动方向 - ped_control.direction.x = 0.15 # 轻微横向移动 - npc.apply_control(ped_control) - npc.set_simulate_physics(True) # 启用物理模拟 - self.walker_list.append(npc) # 添加到行人列表 + # 尝试生成行人 + try: + walker_bp = random.choice(blueprints_walkers) + npc = self.world.try_spawn_actor(walker_bp, spawn_point) + + if npc is not None: + # 设置行人控制参数 + ped_control = carla.WalkerControl() + ped_control.speed = random.uniform(0.5, 1.0) # 随机速度 + ped_control.direction.y = -1 # 主要移动方向 + ped_control.direction.x = 0.15 # 轻微横向移动 + npc.apply_control(ped_control) + npc.set_simulate_physics(True) # 启用物理模拟 + self.walker_list.append(npc) # 添加到行人列表 + return True # 生成成功 + except Exception as e: + print(f"生成右侧行人失败 (尝试 {attempt+1}): {e}") + continue + + return False # 生成失败 def spawn_pedestrians_left(self, isCross): """在左侧生成行人""" @@ -164,49 +177,75 @@ def spawn_pedestrians_left(self, isCross): min_x = 17 max_x = 20.5 - # 随机生成位置 - x = random.uniform(min_x, max_x) - y = random.uniform(min_y, max_y) - - spawn_point = carla.Transform(carla.Location(x, y, 2.0)) - - # 避免在特定区域生成 - while (-10 < spawn_point.location.x < 17) or (70 < spawn_point.location.x < 100): + # 尝试多次生成直到成功 + for attempt in range(3): # 尝试3次 + # 随机生成位置 x = random.uniform(min_x, max_x) y = random.uniform(min_y, max_y) + spawn_point = carla.Transform(carla.Location(x, y, 2.0)) - # 尝试生成行人 - walker_bp = random.choice(blueprints_walkers) - npc = self.world.try_spawn_actor(walker_bp, spawn_point) - - if npc is not None: - # 设置行人控制参数 - ped_control = carla.WalkerControl() - ped_control.speed = random.uniform(0.7, 1.3) # 随机速度 - ped_control.direction.y = 1 # 主要移动方向 - ped_control.direction.x = -0.05 # 轻微横向移动 - npc.apply_control(ped_control) - npc.set_simulate_physics(True) # 启用物理模拟 - self.walker_list.append(npc) # 添加到行人列表 - - def reset(self): - """重置环境""" + # 避免在特定区域生成 + while (-10 < spawn_point.location.x < 17) or (70 < spawn_point.location.x < 100): + x = random.uniform(min_x, max_x) + y = random.uniform(min_y, max_y) + spawn_point = carla.Transform(carla.Location(x, y, 2.0)) + + # 尝试生成行人 + try: + walker_bp = random.choice(blueprints_walkers) + npc = self.world.try_spawn_actor(walker_bp, spawn_point) + + if npc is not None: + # 设置行人控制参数 + ped_control = carla.WalkerControl() + ped_control.speed = random.uniform(0.7, 1.3) # 随机速度 + ped_control.direction.y = 1 # 主要移动方向 + ped_control.direction.x = -0.05 # 轻微横向移动 + npc.apply_control(ped_control) + npc.set_simulate_physics(True) # 启用物理模拟 + self.walker_list.append(npc) # 添加到行人列表 + return True # 生成成功 + except Exception as e: + print(f"生成左侧行人失败 (尝试 {attempt+1}): {e}") + continue + + return False # 生成失败 + + def reset(self, episode=1): + """重置环境，根据episode参数生成不同数量的行人""" + self.current_episode = episode + # 清理现有的行人和车辆 self.cleanup_actors() # 重置行人列表 self.walker_list = [] - # 大幅减少行人数量 - 从30+10减少到8+4 - self.spawn_pedestrians_general(8, True) - self.spawn_pedestrians_general(4, False) + # 根据训练阶段生成不同数量的行人 + if episode < 100: # 第一阶段：少量行人 + print(f"Episode {episode}: 生成少量行人 (4十字路口 + 2非十字路口)") + self.spawn_pedestrians_general(4, True) # 十字路口行人 + self.spawn_pedestrians_general(2, False) # 非十字路口行人 + elif episode < 400: # 第二阶段：中等数量行人 + print(f"Episode {episode}: 生成中等数量行人 (6十字路口 + 3非十字路口)") + self.spawn_pedestrians_general(6, True) + self.spawn_pedestrians_general(3, False) + else: # 第三阶段：正常难度 + print(f"Episode {episode}: 生成正常数量行人 (8十字路口 + 4非十字路口)") + self.spawn_pedestrians_general(8, True) + self.spawn_pedestrians_general(4, False) # 重置状态变量 self.collision_history = [] self.actor_list = [] self.slow_counter = 0 self.steer_counter = 0 + self.same_steer_counter = 0 + self.suggested_action = None + self.last_action = 1 + self.episode_start_time = time.time() + self.last_ped_distance = float('inf') # 设置车辆生成点 spawn_point = carla.Transform() @@ -276,6 +315,7 @@ def cleanup_actors(self): actor.destroy() self.actor_list = [] + self.walker_list = [] # 清空行人列表 def setup_follow_camera(self): """设置跟随车辆的相机，用于在CARLA窗口中观察""" @@ -314,152 +354,182 @@ def process_img(self, image): self.front_camera = processed_image # 更新前置摄像头图像 - def reward(self): - """计算奖励函数 - 增强方向控制奖励""" + def reward(self, speed_kmh, current_steer): + """增强的奖励函数 - 特别强调行人避障""" reward = 0 done = False - - # 计算车辆速度 - velocity = self.vehicle.get_velocity() - velocity_kmh = int(3.6 * math.sqrt(velocity.x ** 2 + velocity.y ** 2 + velocity.z ** 2)) - # 获取车辆位置和方向 + # 获取车辆状态 vehicle_location = self.vehicle.get_location() vehicle_rotation = self.vehicle.get_transform().rotation.yaw - # 计算距离终点的进度奖励 - progress_reward = (vehicle_location.x + 81) / 236.0 # 从-81到155，总共236单位 + # 1. 道路保持奖励（重要的基础） + heading_error = abs(vehicle_rotation) + + if heading_error < 5: # 完美保持方向 + reward += 0.8 # 略微降低权重，给行人避障更多空间 + elif heading_error < 15: # 良好保持 + reward += 0.4 + elif heading_error < 30: # 可接受 + reward += 0.1 + else: # 方向偏差过大 + reward -= 0.3 * (heading_error / 30.0) # 降低惩罚强度 + + # 2. 行人避障（最高优先级） - 增强奖励机制 + min_ped_distance = float('inf') + closest_pedestrian = None + + # 统计有效行人数量 + active_pedestrians = 0 - # 速度奖励 - 更加平滑 - if velocity_kmh == 0: - reward -= 0.5 # 停车惩罚减少 - elif 20 <= velocity_kmh <= 40: # 理想速度区间 - reward += 0.8 - elif 10 <= velocity_kmh < 20 or 40 < velocity_kmh <= 50: - reward += 0.3 # 可接受速度区间 - else: - reward -= 0.2 # 不理想速度 - - # 增强方向奖励 - 确保车辆朝正确方向行驶 - if -20 <= vehicle_rotation <= 20: # 严格限制在正东方向附近 - reward += 0.5 # 增加直行奖励 - self.steer_counter = max(0, self.steer_counter - 1) # 减少转向计数 - elif -45 <= vehicle_rotation <= 45: - reward += 0.1 # 较小奖励 - else: - reward -= 1.0 # 严重偏离惩罚 - self.steer_counter += 2 # 增加转向计数 - - # 行人距离检测 - min_dist = float('inf') # 最小距离初始化为无穷大 for walker in self.walker_list: if not walker.is_alive: continue + active_pedestrians += 1 ped_location = walker.get_location() dx = vehicle_location.x - ped_location.x dy = vehicle_location.y - ped_location.y distance = math.sqrt(dx**2 + dy**2) - min_dist = min(min_dist, distance) # 更新最小距离 - # 清理边界外的行人 - try: - player_direction = walker.get_control().direction - if (ped_location.y < -214 and player_direction.y == -1) or \ - (ped_location.y > -191 and player_direction.y == 1): - if walker.is_alive: - walker.destroy() - except: - pass # 如果无法获取控制信息，跳过 - - # 基于行人距离的奖励 - if min_dist < 3.0: # 非常危险距离 - reward -= 3.0 - done = True - elif min_dist < 5.0: # 危险距离 - reward -= 1.0 - elif min_dist < 8.0: # 警告距离 - reward -= 0.3 - elif min_dist > 15.0: # 安全距离 + if distance < min_ped_distance: + min_ped_distance = distance + closest_pedestrian = walker + + # 如果当前没有有效行人，重置距离 + if active_pedestrians == 0: + min_ped_distance = float('inf') + + # 行人距离分级奖励 - 增强避障激励 + if min_ped_distance < 100: # 只考虑100米内的行人 + if min_ped_distance < 3.0: # 紧急避让距离 + reward -= 8.0 # 增加惩罚 + done = True + print(f"Episode {self.current_episode}: 与行人距离过近 ({min_ped_distance:.1f}m)!") + elif min_ped_distance < 5.0: # 危险距离 + reward -= 3.0 # 增加惩罚 + # 计算避让方向 + if closest_pedestrian: + ped_y = closest_pedestrian.get_location().y + veh_y = vehicle_location.y + # 如果行人在车辆左侧，鼓励右转；反之鼓励左转 + if ped_y < veh_y: # 行人在左侧 + self.suggested_action = 4 # 右转 + else: # 行人在右侧 + self.suggested_action = 3 # 左转 + elif min_ped_distance < 8.0: # 预警距离 + reward -= 0.8 + elif min_ped_distance < 12.0: # 安全距离 + reward += 0.5 # 增加安全距离奖励 + else: # 非常安全 + reward += 0.2 + + # 3. 成功避障奖励 - 当成功避开行人后给予额外奖励 + if self.last_ped_distance < 8.0 and min_ped_distance > self.last_ped_distance: + # 如果上次距离危险，这次距离更远了，说明成功避让 + reward += 0.3 + + self.last_ped_distance = min_ped_distance # 保存当前距离 + + # 4. 速度奖励（平衡避障和前进） + if 15 <= speed_kmh <= 35: # 理想速度区间 + reward += 0.4 + elif 5 <= speed_kmh < 15: # 较慢但安全（避障时可能需要减速） reward += 0.2 - - # 碰撞检测 + elif 35 < speed_kmh <= 45: # 稍快 + reward += 0.1 + elif speed_kmh > 45: # 过快，在行人环境中危险 + reward -= 0.5 # 增加惩罚 + else: # 停车或极慢 + reward -= 0.05 # 轻微惩罚 + + # 5. 转向平滑性奖励（避障时可能需要适当转向） + steer_penalty = abs(current_steer) * 0.3 # 降低惩罚，给避障转向更多空间 + reward -= steer_penalty + + # 6. 碰撞检测 if len(self.collision_history) != 0: - reward = -10 # 碰撞惩罚 + reward = -15 # 增加碰撞惩罚 done = True - - # 进度奖励 - reward += progress_reward * 0.5 + print(f"Episode {self.current_episode}: 发生碰撞!") + + # 7. 进度奖励（次要于避障） + progress = (vehicle_location.x + 81) / 236.0 # 从-81到155 + reward += progress * 0.2 # 降低进度权重 - # 完成条件判断 + # 8. 边界检查 if vehicle_location.x > 155: # 成功到达终点 - reward += 10 # 成功到达奖励 + reward += 20 # 增加到达终点的奖励 done = True - elif vehicle_location.x < -90: # 倒退太多 - reward -= 5 + print(f"Episode {self.current_episode}: 成功到达终点!") + elif vehicle_location.x < -90 or abs(vehicle_location.y + 195) > 30: # 偏离道路 + reward -= 3 # 降低偏离惩罚 done = True - + print(f"Episode {self.current_episode}: 偏离道路!") + return reward, done def step(self, action): - """执行动作并返回新状态 - 扩展为5个动作包含转向""" - # 扩展的动作空间: 0-减速, 1-保持, 2-加速, 3-左转, 4-右转 + """执行动作并返回新状态 - 增强平滑性和安全性""" + # 获取当前速度 + velocity = self.vehicle.get_velocity() + speed_kmh = 3.6 * math.sqrt(velocity.x**2 + velocity.y**2 + velocity.z**2) - # 限制连续转向次数，避免过度转向 - max_continuous_steer = 3 - current_steer = 0.0 + # 5个动作: 0-减速, 1-保持, 2-加速, 3-左转, 4-右转 - if action == 3: # 左转 - if self.steer_counter < max_continuous_steer: - current_steer = -0.3 # 小角度左转 - self.steer_counter += 1 - else: - # 强制直行一段时间 - current_steer = 0.0 - action = 1 # 改为保持动作 - elif action == 4: # 右转 - if self.steer_counter < max_continuous_steer: - current_steer = 0.3 # 小角度右转 - self.steer_counter += 1 - else: - # 强制直行一段时间 - current_steer = 0.0 - action = 1 # 改为保持动作 - else: - # 非转向动作时逐渐减少转向计数 - self.steer_counter = max(0, self.steer_counter - 0.5) + # 根据速度调整转向幅度 - 速度越快，转向越平滑 + speed_factor = max(0.5, min(1.0, 30.0 / max(1.0, speed_kmh))) - # 速度控制 + # 基础控制参数 throttle = 0.0 brake = 0.0 + steer = 0.0 + # 速度控制 if action == 0: # 减速 throttle = 0.0 - brake = 0.3 + brake = 0.5 elif action == 1: # 保持/轻微加速 throttle = 0.3 brake = 0.0 elif action == 2: # 加速 throttle = 0.7 brake = 0.0 - elif action == 3 or action == 4: # 转向时保持适中速度 + elif action == 3: # 左转 throttle = 0.4 brake = 0.0 - + steer = -0.2 * speed_factor # 速度相关的转向幅度 + elif action == 4: # 右转 + throttle = 0.4 + brake = 0.0 + steer = 0.2 * speed_factor # 速度相关的转向幅度 + + # 限制连续同向转向 - 防止过度转向 + if (action == 3 and self.last_action == 3) or (action == 4 and self.last_action == 4): + self.same_steer_counter += 1 + if self.same_steer_counter > 3: # 连续3次同向转向后强制回正 + steer *= 0.5 # 减小转向幅度 + throttle *= 0.8 # 减速 + else: + self.same_steer_counter = 0 + + # 记录上一个动作 + self.last_action = action + # 应用控制 self.vehicle.apply_control(carla.VehicleControl( throttle=throttle, brake=brake, - steer=current_steer + steer=steer )) - + # 等待物理更新 time.sleep(0.05) # 计算奖励和完成状态 - reward, done = self.reward() + reward, done = self.reward(speed_kmh, steer) # 限制极端奖励值 - reward = np.clip(reward, -10, 10) + reward = np.clip(reward, -15, 20) return self.front_camera, reward, done, None \ No newline at end of file diff --git a/src/Unmanned_vehicle_AD_DQN/Hyperparameters.py b/src/Unmanned_vehicle_AD_DQN/Hyperparameters.py index 270b43a586..717228cf84 100644 --- a/src/Unmanned_vehicle_AD_DQN/Hyperparameters.py +++ b/src/Unmanned_vehicle_AD_DQN/Hyperparameters.py @@ -1,8 +1,8 @@ # Hyperparameters.py # 深度强化学习超参数配置 -DISCOUNT = 0.95 -# 未来奖励的折扣因子 +DISCOUNT = 0.97 +# 未来奖励的折扣因子 - 提高未来奖励的重要性 FPS = 60 # 模拟环境的帧率 @@ -13,14 +13,14 @@ REWARD_OFFSET = -100 # 停止模拟的奖励阈值 -MIN_REPLAY_MEMORY_SIZE = 2_000 -# 开始训练前经验回放缓冲区的最小大小 +MIN_REPLAY_MEMORY_SIZE = 3_000 +# 开始训练前经验回放缓冲区的最小大小 - 增加以获得更稳定训练 REPLAY_MEMORY_SIZE = 10_000 # 经验回放缓冲区的最大容量 -MINIBATCH_SIZE = 32 -# 每次训练从经验回放中采样的经验数量 +MINIBATCH_SIZE = 64 +# 每次训练从经验回放中采样的经验数量 - 增加批次大小 PREDICTION_BATCH_SIZE = 1 # 预测阶段使用的批次大小 @@ -28,7 +28,7 @@ TRAINING_BATCH_SIZE = MINIBATCH_SIZE // 4 # 训练阶段使用的批次大小 -EPISODES = 1000 +EPISODES = 800 # 从1000减少到800，因为减少了无行人阶段 # 智能体训练的总轮次数 SECONDS_PER_EPISODE = 60 @@ -40,8 +40,8 @@ EPSILON = 1.0 # 初始探索率 -EPSILON_DECAY = 0.995 -# 探索率的衰减率 +EPSILON_DECAY = 0.998 +# 探索率的衰减率 - 减缓衰减速度 MODEL_NAME = "YY_Optimized" # 训练模型的名称标识 @@ -49,8 +49,8 @@ MIN_REWARD = 5 # 被认为是"良好"或"积极"经验的最小奖励值 -UPDATE_TARGET_EVERY = 10 -# 目标网络更新的频率 +UPDATE_TARGET_EVERY = 20 +# 目标网络更新的频率 - 增加以获得更稳定训练 AGGREGATE_STATS_EVERY = 10 # 计算和聚合统计信息（如平均得分、奖励）的频率 @@ -73,5 +73,5 @@ SUCCESSFUL_THRESHOLD = 3 # 成功阈值 -LEARNING_RATE = 0.0001 -# 优化器的学习率 \ No newline at end of file +LEARNING_RATE = 0.00005 +# 优化器的学习率 - 降低以获得更稳定训练 \ No newline at end of file diff --git a/src/Unmanned_vehicle_AD_DQN/Model.py b/src/Unmanned_vehicle_AD_DQN/Model.py index f62147b7e2..5adcb42262 100644 --- a/src/Unmanned_vehicle_AD_DQN/Model.py +++ b/src/Unmanned_vehicle_AD_DQN/Model.py @@ -10,12 +10,10 @@ import matplotlib.pyplot as plt from collections import deque from tensorflow.keras.applications.xception import Xception -from tensorflow.keras.layers import Dense, GlobalAveragePooling2D -from tensorflow.keras.models import Sequential, Model from tensorflow.keras.layers import Dense, GlobalAveragePooling2D, Input, Concatenate, Conv2D, AveragePooling2D, Activation, \ - Flatten, Dropout, BatchNormalization, MaxPooling2D + Flatten, Dropout, BatchNormalization, MaxPooling2D, Multiply from tensorflow.keras.optimizers import Adam -from tensorflow.keras.models import Model +from tensorflow.keras.models import Sequential, Model from tensorflow.keras.callbacks import TensorBoard import tensorflow as tf import tensorflow.keras.backend as backend @@ -78,47 +76,52 @@ def __init__(self): self.training_initialized = False def create_model(self): - """创建深度Q网络模型 - 扩展为5个输出""" - model = Sequential() + """创建深度Q网络模型 - 优化网络结构""" + # 使用函数式API以支持注意力机制 + inputs = Input(shape=(IM_HEIGHT, IM_WIDTH, 3)) # 第一卷积块 - model.add(Conv2D(32, (5, 5), strides=(2, 2), input_shape=(IM_HEIGHT, IM_WIDTH, 3), padding='same')) - model.add(Activation('relu')) - model.add(BatchNormalization()) # 批归一化 - model.add(MaxPooling2D(pool_size=(2, 2))) + x = Conv2D(32, (5, 5), strides=(2, 2), padding='same')(inputs) + x = Activation('relu')(x) + x = BatchNormalization()(x) + x = MaxPooling2D(pool_size=(2, 2))(x) # 第二卷积块 - model.add(Conv2D(64, (3, 3), padding='same')) - model.add(Activation('relu')) - model.add(BatchNormalization()) - model.add(MaxPooling2D(pool_size=(2, 2))) + x = Conv2D(64, (3, 3), padding='same')(x) + x = Activation('relu')(x) + x = BatchNormalization()(x) + x = MaxPooling2D(pool_size=(2, 2))(x) # 第三卷积块 - model.add(Conv2D(128, (3, 3), padding='same')) - model.add(Activation('relu')) - model.add(BatchNormalization()) - model.add(MaxPooling2D(pool_size=(2, 2))) + x = Conv2D(128, (3, 3), padding='same')(x) + x = Activation('relu')(x) + x = BatchNormalization()(x) + x = MaxPooling2D(pool_size=(2, 2))(x) - # 第四卷积块 - model.add(Conv2D(256, (3, 3), padding='same')) - model.add(Activation('relu')) - model.add(BatchNormalization()) + # 空间注意力机制 + attention = Conv2D(1, (1, 1), padding='same', activation='sigmoid')(x) + x = Multiply()([x, attention]) # 展平层 - model.add(Flatten()) + x = Flatten()(x) + + # 全连接层 - 增加层深度 + x = Dense(512, activation='relu')(x) + x = Dropout(0.3)(x) + x = Dense(256, activation='relu')(x) + x = Dropout(0.3)(x) + x = Dense(128, activation='relu')(x) + x = Dropout(0.2)(x) + x = Dense(64, activation='relu')(x) + x = Dropout(0.1)(x) - # 全连接层 - model.add(Dense(512, activation='relu')) - model.add(Dropout(0.3)) # 防止过拟合 - model.add(Dense(256, activation='relu')) - model.add(Dropout(0.3)) - model.add(Dense(128, activation='relu')) - model.add(Dropout(0.2)) + # 输出层 - 5个动作 + outputs = Dense(5, activation='linear')(x) - # 输出层 - 扩展为5个动作: 0-减速, 1-保持, 2-加速, 3-左转, 4-右转 - model.add(Dense(5, activation='linear')) + # 创建模型 + model = Model(inputs=inputs, outputs=outputs) - # 编译模型，使用Huber损失和Adam优化器 + # 编译模型 model.compile(loss="huber", optimizer=Adam(lr=LEARNING_RATE), metrics=["mae"]) return model diff --git a/src/Unmanned_vehicle_AD_DQN/Test.py b/src/Unmanned_vehicle_AD_DQN/Test.py index 7e00fe6435..9ee3f92184 100644 --- a/src/Unmanned_vehicle_AD_DQN/Test.py +++ b/src/Unmanned_vehicle_AD_DQN/Test.py @@ -11,7 +11,26 @@ from Hyperparameters import * -MODEL_PATH = r'D:\Work\T_Unmanned_vehicle_AD_DQN\models\YY_Optimized___290.14max___97.16avg___13.42min__1764553908.model' # 请替换为实际的最佳模型路径 +MODEL_PATH = r'D:\Work\T_Unmanned_vehicle_AD_DQN\models\YY_Optimized_best_363.53.model' # 请替换为实际的最佳模型路径 + +def get_safe_action(model, state, env, previous_action): + """结合模型预测和安全规则的混合动作选择""" + # 模型预测 + qs = model.predict(np.array(state).reshape(-1, *state.shape)/255)[0] + + # 如果有建议的避让动作（来自环境），优先考虑 + if hasattr(env, 'suggested_action') and env.suggested_action is not None: + suggested_q = qs[env.suggested_action] + qs[env.suggested_action] += 1.0 # 提高建议动作的Q值 + env.suggested_action = None # 重置 + + # 避免频繁切换动作（平滑性） + if previous_action in [3, 4]: # 如果是转向动作 + qs[previous_action] += 0.3 # 稍微提高继续当前转向的倾向 + + # 选择动作 + action = np.argmax(qs) + return action, qs if __name__ == '__main__': @@ -37,13 +56,16 @@ # 循环测试多个episode episode_count = 0 + previous_action = 1 # 初始动作为保持 + try: while True: episode_count += 1 print(f'\n开始第 {episode_count} 个测试轮次') # 重置环境并获取初始状态 - current_state = env.reset() + # 测试时使用正常难度（相当于训练的第3阶段） + current_state = env.reset(401) # 401表示使用正常难度 env.collision_hist = [] # 重置碰撞历史 done = False @@ -56,9 +78,9 @@ # FPS计数开始 step_start = time.time() - # 基于当前观察空间预测动作 - qs = model.predict(np.array(current_state).reshape(-1, *current_state.shape)/255)[0] - action = np.argmax(qs) # 选择Q值最大的动作 + # 基于当前观察空间预测动作（使用安全版本） + action, qs = get_safe_action(model, current_state, env, previous_action) + previous_action = action # 执行环境步进 new_state, reward, done, _ = env.step(action) @@ -92,4 +114,4 @@ finally: # 清理环境 print("清理环境...") - env.cleanup_actors() + env.cleanup_actors() \ No newline at end of file diff --git a/src/Unmanned_vehicle_AD_DQN/main.py b/src/Unmanned_vehicle_AD_DQN/main.py index 9bf9ca442f..3dadcf5b39 100644 --- a/src/Unmanned_vehicle_AD_DQN/main.py +++ b/src/Unmanned_vehicle_AD_DQN/main.py @@ -71,22 +71,13 @@ env.collision_hist = [] # 重置碰撞历史 agent.tensorboard.step = episode # 设置TensorBoard步数 - # 课程学习 - 随训练进度调整难度 - if episode > EPISODES // 2: - # 训练后期增加行人数量以提高难度 - env.spawn_pedestrians_general(8, True) # 减少数量 - env.spawn_pedestrians_general(4, False) # 减少数量 - else: - # 训练前期减少行人数量以降低难度 - env.spawn_pedestrians_general(6, True) # 减少数量 - env.spawn_pedestrians_general(3, False) # 减少数量 - # 重置每轮统计 - 重置得分和步数 score = 0 step = 1 # 重置环境并获取初始状态 - current_state = env.reset() + # 将episode编号传递给环境，环境会根据episode生成相应数量的行人 + current_state = env.reset(episode) # 重置完成标志并开始迭代直到本轮结束 done = False @@ -153,7 +144,7 @@ f'models/{MODEL_NAME}__{max_reward:_>7.2f}max_{average_reward:_>7.2f}avg_{min_reward:_>7.2f}min__{int(time.time())}.model') epds.append(episode) - print('轮次: ', episode, '得分 %.2f' % score, '成功次数:', success_count) + print(f'轮次: {episode}, 得分: {score:.2f}, 成功次数: {success_count}') # 衰减探索率 if Hyperparameters.EPSILON > Hyperparameters.MIN_EPSILON: @@ -163,15 +154,23 @@ # 设置训练线程终止标志并等待其结束 agent.terminate = True trainer_thread.join() + # 保存最终模型 - agent.model.save( - f'models/{MODEL_NAME}__{max_reward:_>7.2f}max_{average_reward:_>7.2f}avg_{min_reward:_>7.2f}min__{int(time.time())}.model') + if len(scores) > 0: + final_max_reward = max(scores[-AGGREGATE_STATS_EVERY:] if len(scores) >= AGGREGATE_STATS_EVERY else scores) + final_avg_reward = np.mean(scores[-AGGREGATE_STATS_EVERY:] if len(scores) >= AGGREGATE_STATS_EVERY else scores) + final_min_reward = min(scores[-AGGREGATE_STATS_EVERY:] if len(scores) >= AGGREGATE_STATS_EVERY else scores) + agent.model.save( + f'models/{MODEL_NAME}__{final_max_reward:_>7.2f}max_{final_avg_reward:_>7.2f}avg_{final_min_reward:_>7.2f}min__{int(time.time())}.model') # 绘制训练曲线 fig = plt.figure() ax = fig.add_subplot(111) - plt.plot(scores) # 得分曲线 - plt.plot(avg_scores) # 平均得分曲线 + plt.plot(scores, label='每轮得分') + plt.plot(avg_scores, label='平均得分(最近10轮)', linewidth=2) plt.ylabel('得分') plt.xlabel('训练轮次') + plt.title('训练进度') + plt.legend() + plt.grid(True, alpha=0.3) plt.show() \ No newline at end of file