huangfu/MAGAIL4AutoDrive
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

回退了一些内容,增加了对于环境方法的描述

Browse Source
This commit is contained in:
huangfu
2025-10-24 21:39:31 +08:00
parent b626702cbb
commit 62e638c4d2
16 changed files with 844 additions and 1704 deletions
Download Patch File Download Diff File Expand all files Collapse all files

339
Env/DEBUG_GUIDE.md
View File

@@ -1,339 +0,0 @@
# 调试功能使用指南
## 📋 概述
已为车道过滤和红绿灯检测功能添加了详细的调试输出，帮助您诊断和理解代码行为。
---
## 🎛️ 调试开关
### 1. 配置参数
在创建环境时，可以通过 `config` 参数启用调试模式：
```python
env = MultiAgentScenarioEnv(
config={
# ... 其他配置 ...
# 🔥 调试开关
"debug_lane_filter": True, # 启用车道过滤调试
"debug_traffic_light": True, # 启用红绿灯检测调试
},
agent2policy=your_policy
)
```
### 2. 默认值
两个调试开关默认都是 `False`（关闭），避免正常运行时产生大量日志。
---
## 📊 车道过滤调试 (`debug_lane_filter=True`)
### 输出内容
```
📍 场景信息统计:
- 总车道数: 123
🔍 开始车道过滤: 共 51 辆车待检测
车辆 1/51: ID=128
🔍 检测位置 (-4.11, 46.76), 容差=3.0m
✅ 在车道上 (车道184, 检查了32条)
✅ 保留
车辆 7/51: ID=134
🔍 检测位置 (-51.34, -3.77), 容差=3.0m
❌ 不在任何车道上 (检查了123条车道)
❌ 过滤 (原因: 不在车道上)
... (所有车辆)
📊 过滤结果: 保留 45 辆, 过滤 6 辆
```
### 调试信息说明
| 信息 | 含义 |
|------|------|
| 📍 场景信息统计 | 场景的基本信息（车道数、红绿灯数） |
| 🔍 开始车道过滤 | 开始过滤，显示待检测车辆总数 |
| 🔍 检测位置 | 车辆的坐标和使用的容差值 |
| ✅ 在车道上 | 找到了车辆所在的车道，显示车道ID和检查次数 |
| ❌ 不在任何车道上 | 所有车道都检查完了，未找到匹配的车道 |
| 📊 过滤结果 | 最终统计：保留多少辆，过滤多少辆 |
### 典型输出案例
**情况1：车辆在正常车道上**
```
车辆 1/51: ID=128
🔍 检测位置 (-4.11, 46.76), 容差=3.0m
✅ 在车道上 (车道184, 检查了32条)
✅ 保留
```
→ 检查了32条车道后找到匹配的车道184
**情况2：车辆在草坪/停车场**
```
车辆 7/51: ID=134
🔍 检测位置 (-51.34, -3.77), 容差=3.0m
❌ 不在任何车道上 (检查了123条车道)
❌ 过滤 (原因: 不在车道上)
```
→ 检查了所有123条车道都不匹配，该车辆被过滤
---
## 🚦 红绿灯检测调试 (`debug_traffic_light=True`)
### 输出内容
```
📍 场景信息统计:
- 总车道数: 123
- 有红绿灯的车道数: 0
⚠️ 场景中没有红绿灯！
🚦 检测车辆红绿灯 - 位置: (-4.1, 46.8)
方法1-导航模块:
current_lane = <metadrive.component.lane.straight_lane.StraightLane object>
lane_index = 184
has_traffic_light = False
该车道没有红绿灯
方法2-遍历车道: 开始遍历 123 条车道
✓ 找到车辆所在车道: 184 (检查了32条)
has_traffic_light = False
该车道没有红绿灯
结果: 返回 0 (无红绿灯/未知)
```
### 调试信息说明
| 信息 | 含义 |
|------|------|
| 有红绿灯的车道数 | 统计场景中有多少个红绿灯 |
| ⚠️ 场景中没有红绿灯 | 如果数量为0，会特别提示 |
| 方法1-导航模块 | 尝试从导航系统获取 |
| current_lane | 导航系统返回的当前车道对象 |
| lane_index | 车道的唯一标识符 |
| has_traffic_light | 该车道是否有红绿灯 |
| status | 红绿灯的状态（GREEN/YELLOW/RED/None） |
| 方法2-遍历车道 | 兜底方案，遍历所有车道查找 |
| ✓ 找到车辆所在车道 | 遍历找到了匹配的车道 |
### 典型输出案例
**情况1：场景没有红绿灯**
```
📍 场景信息统计:
- 有红绿灯的车道数: 0
⚠️ 场景中没有红绿灯！
🚦 检测车辆红绿灯 - 位置: (-4.1, 46.8)
方法1-导航模块:
...
has_traffic_light = False
该车道没有红绿灯
结果: 返回 0 (无红绿灯/未知)
```
→ 所有车辆都会返回0，这是正常的
**情况2：有红绿灯且状态正常**
```
🚦 检测车辆红绿灯 - 位置: (10.5, 20.3)
方法1-导航模块:
current_lane = <...>
lane_index = 205
has_traffic_light = True
status = TRAFFIC_LIGHT_GREEN
✅ 方法1成功: 绿灯
```
→ 方法1直接成功，返回1（绿灯）
**情况3：红绿灯状态为None**
```
🚦 检测车辆红绿灯 - 位置: (10.5, 20.3)
方法1-导航模块:
current_lane = <...>
lane_index = 205
has_traffic_light = True
status = None
⚠️ 方法1: 红绿灯状态为None
```
→ 有红绿灯，但状态异常，返回0
**情况4：导航失败，方法2兜底**
```
🚦 检测车辆红绿灯 - 位置: (15.2, 30.5)
方法1-导航模块: 不可用 (hasattr=True, not_none=False)
方法2-遍历车道: 开始遍历 123 条车道
✓ 找到车辆所在车道: 210 (检查了45条)
has_traffic_light = True
status = TRAFFIC_LIGHT_RED
✅ 方法2成功: 红灯
```
→ 方法1失败，方法2兜底成功，返回3（红灯）
---
## 🧪 测试方法
### 方式1：使用测试脚本
```bash
# 标准测试（无详细调试）
python Env/test_lane_filter.py
# 调试模式（详细输出）
python Env/test_lane_filter.py --debug
```
### 方式2：在代码中直接启用
```python
from scenario_env import MultiAgentScenarioEnv
from simple_idm_policy import ConstantVelocityPolicy
env = MultiAgentScenarioEnv(
config={
"data_directory": "...",
"use_render": False,
# 启用调试
"debug_lane_filter": True,
"debug_traffic_light": True,
},
agent2policy=ConstantVelocityPolicy(target_speed=50)
)
obs = env.reset(0)
# 调试信息会自动输出
```
---
## 📝 调试输出控制
### 场景1：只想看车道过滤
```python
config = {
"debug_lane_filter": True,
"debug_traffic_light": False, # 关闭红绿灯调试
}
```
### 场景2：只想看红绿灯检测
```python
config = {
"debug_lane_filter": False,
"debug_traffic_light": True, # 只看红绿灯
}
```
### 场景3：生产环境（关闭所有调试）
```python
config = {
"debug_lane_filter": False,
"debug_traffic_light": False,
}
# 或者直接不设置这两个参数，默认就是False
```
---
## 💡 常见问题诊断
### 问题1：所有红绿灯状态都是0
**检查调试输出：**
```
📍 场景信息统计:
- 有红绿灯的车道数: 0
⚠️ 场景中没有红绿灯！
```
**结论：** 场景本身没有红绿灯，返回0是正常的
---
### 问题2：车辆被过滤但不应该过滤
**检查调试输出：**
```
车辆 X: ID=XXX
🔍 检测位置 (x, y), 容差=3.0m
❌ 不在任何车道上 (检查了123条车道)
❌ 过滤 (原因: 不在车道上)
```
**可能原因：**
1. 车辆确实在非车道区域（草坪/停车场）
2. 容差值太小，可以尝试增大 `lane_tolerance`
3. 车道数据有问题
**解决方案：**
```python
config = {
"lane_tolerance": 5.0, # 增大容差到5米
}
```
---
### 问题3：性能下降
启用调试模式会有大量输出，影响性能：
**解决方案：**
- 只在开发/调试时启用
- 生产环境关闭所有调试开关
- 或者只测试少量车辆：
```python
config = {
"max_controlled_vehicles": 5, # 只测试5辆车
"debug_traffic_light": True,
}
```
---
## 📌 最佳实践
1. **开发阶段**：启用调试，理解代码行为
2. **调试问题**：根据需要选择性启用调试
3. **性能测试**：关闭所有调试
4. **生产运行**：永久关闭调试
---
## 🔧 调试输出示例
完整的调试运行示例：
```bash
cd /home/huangfukk/MAGAIL4AutoDrive
python Env/test_lane_filter.py --debug
```
输出会包含：
- 场景统计信息
- 每辆车的详细检测过程
- 最终的过滤/检测结果
- 性能统计
---
## 📖 相关文档
- `README.md` - 项目总览和问题解决
- `CHANGELOG.md` - 更新日志
- `PERFORMANCE_OPTIMIZATION.md` - 性能优化指南

279
Env/ENVIRONMENT_INFO.md Normal file
View File

@@ -0,0 +1,279 @@
# 多智能体场景环境详细说明
## 1. 观测信息详解
### 观测向量结构（总维度：107维）
每个智能体的观测向量包含以下信息：
```python
观测向量 = [
# 1. 车辆状态信息 (5维)
position_x, # 车辆X坐标
position_y, # 车辆Y坐标
velocity_x, # X方向速度
velocity_y, # Y方向速度
heading_theta, # 朝向角度
# 2. 前向激光雷达 (80维)
lidar_1, # 第1个激光束的距离
lidar_2, # 第2个激光束的距离
...
lidar_80, # 第80个激光束的距离
# 范围：30米，用于前方障碍物检测
# 3. 侧向激光雷达 (10维)
side_lidar_1, # 第1个侧向激光束的距离
...
side_lidar_10, # 第10个侧向激光束的距离
# 范围：8米，用于侧方障碍物检测
# 4. 车道线检测 (10维)
lane_line_1, # 第1个车道线检测距离
...
lane_line_10, # 第10个车道线检测距离
# 范围：3米，用于车道线识别
# 5. 导航信息 (2维)
destination_x, # 目标点X坐标
destination_y, # 目标点Y坐标
]
```
### 观测信息说明
1. **车辆状态 (5维)**
- 位置：全局坐标系下的(x, y)
- 速度：车辆在全局坐标系下的速度分量
- 朝向：车辆的航向角（弧度）
2. **激光雷达 (100维)**
- 前向80束：覆盖前方视野，检测动态和静态障碍物
- 侧向10束：检测侧方物体，用于变道等操作
- 车道线10束：专门检测车道线位置
3. **导航信息 (2维)**
- 目标位置：从专家数据中提取的车辆最终位置
**总维度：5 + 80 + 10 + 10 + 2 = 107维**
---
## 2. 多场景加载逻辑
### 2.1 场景加载机制
MetaDrive的ScenarioEnv通过以下配置管理多场景：
```python
config = {
"data_directory": "path/to/dataset", # 包含dataset_mapping.pkl的目录
"num_scenarios": 3, # 场景总数（从mapping文件读取）
"start_scenario_index": 0, # 起始场景索引
"sequential_seed": True, # 是否顺序切换场景
}
```
### 2.2 场景切换逻辑
#### 场景索引管理
- **初始化**：环境启动时从`start_scenario_index`开始
- **顺序模式**（`sequential_seed=True`）：
- 每次`env.reset()`后自动切换到下一个场景
- 循环顺序：场景0 → 场景1 → 场景2 → 场景0 ...
#### 示例流程
```python
env = MultiAgentScenarioEnv(config={
"data_directory": "path/to/dataset", # 假设有3个场景
"start_scenario_index": 0,
"sequential_seed": True,
})
obs = env.reset(0) # 使用场景0
# ... 运行场景0 ...
obs = env.reset() # 自动切换到场景1
# ... 运行场景1 ...
obs = env.reset() # 自动切换到场景2
# ... 运行场景2 ...
obs = env.reset() # 循环回到场景0
```
### 2.3 当前场景信息获取
可以通过以下方式查看当前场景信息：
```python
# 获取当前场景索引
current_scenario = env.engine.current_seed
# 获取场景总数
total_scenarios = env.config["num_scenarios"]
# 查看场景ID
scenario_id = env.engine.traffic_manager.current_scenario_id
```
### 2.4 手动指定场景
如果需要固定使用某个场景：
```python
# 方法1：在reset时指定
obs = env.reset(seed=1) # 使用场景1
# 方法2：配置固定场景
config = {
"start_scenario_index": 2, # 从场景2开始
"sequential_seed": False, # 禁用自动切换
}
```
---
## 3. 车辆观测获取机制
### 3.1 观测获取流程
```
step() 被调用
↓
更新所有车辆物理状态
↓
_spawn_controlled_agents() # 生成新车辆（按时间步）
↓
_get_all_obs() # 获取所有车辆观测
↓
遍历 controlled_agents:
├─ 获取车辆状态 (position, velocity, heading)
├─ 调用激光雷达传感器
├─ 组装观测向量
└─ 添加到 obs_list
↓
返回 obs_list（包含所有车辆的观测）
```
### 3.2 观测返回格式
```python
obs = env.reset()
# obs 是一个列表，每个元素是一个车辆的观测向量
obs = [
[obs_vehicle_0], # 第1辆车的107维观测
[obs_vehicle_1], # 第2辆车的107维观测
...
]
# 在step中
obs, rewards, dones, infos = env.step(actions)
# obs格式相同，但只包含当前存活的车辆
```
### 3.3 观测一致性保证
1. **物理状态同步**
- 所有车辆在同一物理时间步获取观测
- 保证观测的时间一致性
2. **传感器独立性**
- 每个车辆有独立的传感器
- 激光雷达从各自位置发射
3. **动态车辆管理**
- 新车辆在生成时立即获取观测
- 观测列表动态更新
---
## 4. 常见问题解答
### Q1: 为什么观测维度是107而不是其他？
**A**:
- 车辆状态: 5维 (x, y, vx, vy, heading)
- 前向激光雷达: 80维
- 侧向激光雷达: 10维
- 车道线检测: 10维
- 目标位置: 2维
- **总计: 5 + 80 + 10 + 10 + 2 = 107维**
### Q2: 如何确保多场景都被使用？
**A**: 设置`sequential_seed=True`，环境会自动循环遍历所有场景。
### Q3: 车辆在不同时间步生成，如何获取观测？
**A**: 每次调用`step()`时：
1. 先检查是否有新车辆需要生成（`_spawn_controlled_agents`）
2. 为所有现存车辆（包括新生成的）获取观测
3. 返回的obs_list包含所有当前存活车辆的观测
### Q4: 如果场景中车辆数量不同怎么办？
**A**:
- 观测列表长度动态调整
- 使用`max_controlled_vehicles`可限制最大车辆数
- 使用`filter_offroad_vehicles`可过滤无效车辆
### Q5: 观测数据的坐标系是什么？
**A**:
- 位置/速度/目标：**全局坐标系**（世界坐标）
- 激光雷达：**车辆局部坐标系**（以车辆为中心）
---
## 5. 配置建议
### 5.1 训练配置
```python
config = {
"data_directory": "path/to/dataset",
"sequential_seed": True, # 循环使用所有场景
"filter_offroad_vehicles": True, # 过滤无效车辆
"max_controlled_vehicles": 20, # 限制车辆数防止过载
"inherit_expert_velocity": False, # 训练时不继承速度
"horizon": 300, # 每场景运行300步
}
```
### 5.2 评估配置
```python
config = {
"data_directory": "path/to/dataset",
"sequential_seed": False, # 固定场景
"start_scenario_index": 0, # 指定场景
"filter_offroad_vehicles": True,
"max_controlled_vehicles": None, # 不限制车辆数
"inherit_expert_velocity": True, # 评估时继承速度
"verbose_reset": True, # 输出详细信息
}
```
---
## 6. 调试技巧
### 6.1 查看当前场景信息
```python
# 在reset后查看
print(f"当前场景: {env.engine.current_seed}")
print(f"总场景数: {env.config['num_scenarios']}")
print(f"可控车辆数: {len(env.controlled_agents)}")
```
### 6.2 查看观测维度
```python
obs = env.reset()
print(f"车辆数量: {len(obs)}")
if len(obs) > 0:
print(f"单车辆观测维度: {len(obs[0])}")
```
### 6.3 启用详细日志
```python
config = {
"verbose_reset": True, # 重置时详细统计
"debug_lane_filter": True, # 车道过滤调试
}
```

221
Env/GPU_ACCELERATION.md
View File

@@ -1,221 +0,0 @@
# GPU加速指南
## 当前性能瓶颈分析
从测试结果看，即使关闭渲染，FPS仍然只有15-20左右，主要瓶颈是：
### 计算量分析（51辆车）
```
激光雷达计算：
- 前向雷达：80束 × 51车 = 4,080次射线检测
- 侧向雷达：10束 × 51车 = 510次射线检测
- 车道线雷达：10束 × 51车 = 510次射线检测
合计：5,100次射线检测/帧
红绿灯检测：
- 遍历所有车道 × 51车 = 数千次几何计算
```
**关键问题**：这些计算都是CPU单线程串行的，无法利用多核和GPU！
---
## GPU加速方案
### 方案1：优化激光雷达计算（已实现）✅
**优化内容：**
1. 减少激光束数量：100束 → 52束（减少48%）
2. 优化红绿灯检测：避免遍历所有车道
3. 激光雷达缓存：每N帧才重新计算一次
**预期提升：** 2-4倍（30-60 FPS）
**使用方法：**
```bash
python Env/run_multiagent_env_fast.py
```
---
### 方案2：MetaDrive GPU渲染（有限支持）
**说明：**
MetaDrive基于Panda3D引擎，理论上支持GPU渲染，但：
- GPU主要用于**图形渲染**，不是物理计算
- 激光雷达的射线检测仍在CPU上
- GPU渲染主要加速可视化，不加速训练
**启用方法：**
```python
config = {
"use_render": True,
"render_mode": "onscreen", # 或 "offscreen"
# Panda3D会自动尝试使用GPU
}
```
**限制：**
- 需要显示器或虚拟显示（Xvfb）
- WSL2环境需要配置X11转发
- 对无渲染训练无帮助
---
### 方案3：使用GPU加速的物理引擎（推荐但需要迁移）
**选项A：Isaac Gym (NVIDIA)**
- 完全在GPU上运行物理模拟和渲染
- 可同时模拟数千个环境
- **缺点**：需要完全重写环境代码，迁移成本高
**选项B：IsaacSim/Omniverse**
- NVIDIA的高级仿真平台
- 支持GPU加速的激光雷达
- **缺点**：学习曲线陡峭，环境配置复杂
**选项C：Brax (Google)**
- JAX驱动，完全在GPU/TPU上运行
- **缺点**：功能有限，不支持复杂场景
---
### 方案4：策略网络GPU加速（推荐）✅
虽然环境仿真在CPU，但可以让**策略网络在GPU上运行**：
```python
import torch
# 创建GPU上的策略模型
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
policy = PolicyNetwork().to(device)
# 批量处理观测
obs_batch = torch.tensor(obs_list).to(device)
with torch.no_grad():
actions = policy(obs_batch)
actions = actions.cpu().numpy()
```
**优势：**
- 51辆车的推理可以并行
- 如果使用RL训练，GPU加速训练过程
- 不需要修改环境代码
---
### 方案5：多进程并行（最实用）✅
既然单个环境受限于CPU单线程，可以**并行运行多个环境**：
```python
from multiprocessing import Pool
import os
def run_single_env(seed):
"""运行单个环境实例"""
env = MultiAgentScenarioEnv(config=...)
obs = env.reset(seed)
for step in range(1000):
actions = {...}
obs, rewards, dones, infos = env.step(actions)
if dones["__all__"]:
break
env.close()
return results
# 使用进程池并行运行
if __name__ == "__main__":
num_processes = os.cpu_count() # 12600KF有10核20线程
seeds = list(range(num_processes))
with Pool(processes=num_processes) as pool:
results = pool.map(run_single_env, seeds)
```
**预期提升：** 接近线性（10核 ≈ 10倍吞吐量）
**CPU利用率：** 可达80-100%
---
## 推荐的完整优化方案
### 1. 立即可用（已实现）
```bash
# 使用优化版本，激光束减少+缓存
python Env/run_multiagent_env_fast.py
```
**预期：** 30-60 FPS（2-4倍提升）
### 2. 短期优化（1-2小时）
- 实现多进程并行
- 策略网络迁移到GPU
**预期：** 300-600 FPS（总吞吐量）
### 3. 中期优化（1-2天）
- 使用NumPy矢量化批量处理观测
- 优化Python代码热点（用Cython/Numba）
**预期：** 额外20-30%提升
### 4. 长期方案（1-2周）
- 迁移到Isaac Gym等GPU加速仿真器
- 或使用分布式训练框架（Ray/RLlib）
**预期：** 10-100倍提升
---
## 为什么MetaDrive无法直接使用GPU？
### 架构限制：
1. **物理引擎**：使用Bullet/Panda3D的CPU物理引擎
2. **射线检测**：串行CPU计算，无法并行
3. **Python GIL**：全局解释器锁限制多线程
4. **设计目标**：MetaDrive设计时主要考虑灵活性而非极致性能
### GPU在仿真中的作用：
- ✅ **图形渲染**：绘制画面（但我们训练时不需要）
- ✅ **神经网络推理/训练**：策略模型计算
- ❌ **物理计算**：MetaDrive的物理引擎在CPU
- ❌ **传感器模拟**：激光雷达等在CPU
---
## 检查GPU是否可用
```bash
# 检查NVIDIA GPU
nvidia-smi
# 检查PyTorch GPU支持
python -c "import torch; print(f'CUDA available: {torch.cuda.is_available()}')"
# 检查MetaDrive渲染设备
python -c "from panda3d.core import GraphicsPipeSelection; print(GraphicsPipeSelection.get_global_ptr().get_default_pipe())"
```
---
## 总结
| 方案 | 实现难度 | 性能提升 | GPU使用 | 推荐度 |
|------|----------|----------|---------|--------|
| 减少激光束 | ⭐ | 2-4x | ❌ | ⭐⭐⭐⭐⭐ |
| 激光雷达缓存 | ⭐ | 1.5-3x | ❌ | ⭐⭐⭐⭐⭐ |
| 多进程并行 | ⭐⭐ | 5-10x | ❌ | ⭐⭐⭐⭐⭐ |
| 策略GPU加速 | ⭐⭐ | 2-5x | ✅ | ⭐⭐⭐⭐ |
| GPU渲染 | ⭐⭐⭐ | 1.2x | ✅ | ⭐⭐ |
| 迁移Isaac Gym | ⭐⭐⭐⭐⭐ | 10-100x | ✅ | ⭐⭐⭐ |
**结论：**
1. 先用已实现的优化（减少激光束+缓存）
2. 再实现多进程并行
3. 策略网络用GPU训练
4. 如果还不够，考虑迁移到GPU仿真器

131
Env/PERFORMANCE_OPTIMIZATION.md
View File

@@ -1,131 +0,0 @@
# MetaDrive 性能优化指南
## 为什么帧率只有15FPS且CPU利用率不高？
### 主要原因：
1. **渲染瓶颈（最主要）**
- `use_render: True` + 每帧调用 `env.render()` 会严重限制帧率
- MetaDrive 使用 Panda3D 渲染引擎，渲染是**同步阻塞**的
- 即使CPU有余力，也要等待渲染完成才能继续下一步
- 这就是为什么CPU利用率低但帧率也低的原因
2. **激光雷达计算开销**
- 每帧对每辆车进行3次激光雷达扫描（100个激光束）
- 需要进行物理射线检测，计算量较大
3. **物理引擎同步**
- 默认物理步长很小（0.02s），需要频繁计算
4. **Python GIL限制**
- Python全局解释器锁限制了多核并行
- 即使是多核CPU，Python单线程性能才是瓶颈
## 性能优化方案
### 方案1：关闭渲染（推荐用于训练）
**预期提升：10-20倍（150-300+ FPS）**
```python
config = {
"use_render": False, # 关闭渲染
"render_pipeline": False,
"image_observation": False,
"interface_panel": [],
"manual_control": False,
}
```
### 方案2：降低物理计算频率
**预期提升：2-3倍**
```python
config = {
"physics_world_step_size": 0.05, # 默认0.02，增大步长
"decision_repeat": 5, # 每5个物理步执行一次决策
}
```
### 方案3：优化激光雷达
**预期提升：1.5-2倍**
修改 `scenario_env.py` 中的 `_get_all_obs()` 函数：
```python
# 减少激光束数量
lidar = self.engine.get_sensor("lidar").perceive(
num_lasers=40, # 从80减到40
distance=30,
base_vehicle=vehicle,
physics_world=self.engine.physics_world.dynamic_world
)
# 或者降低扫描频率（每N步才扫描一次）
if self.round % 5 == 0:
lidar = self.engine.get_sensor("lidar").perceive(...)
else:
lidar = self.last_lidar[agent_id] # 使用缓存
```
### 方案4：间歇性渲染
**适用场景：既需要可视化又想提升性能**
```python
# 每10步渲染一次，而不是每步都渲染
if step % 10 == 0:
env.render(mode="topdown")
```
### 方案5：使用多进程并行（高级）
**预期提升：接近线性（取决于进程数）**
```python
from multiprocessing import Pool
def run_env(seed):
env = MultiAgentScenarioEnv(config=...)
# 运行仿真
return results
# 使用进程池并行运行多个环境
with Pool(processes=8) as pool:
results = pool.map(run_env, range(8))
```
## 文件说明
- `run_multiagent_env.py` - **标准版本**（无渲染，基础优化）
- `run_multiagent_env_fast.py` - **极速版本**（激光雷达优化+缓存）⭐推荐
- `run_multiagent_env_parallel.py` - **并行版本**（多进程，最高吞吐量）⭐⭐推荐
- `run_multiagent_env_visual.py` - **可视化版本**（有渲染，适合调试）
## 性能对比
| 配置 | 单环境FPS | 总吞吐量 | CPU利用率 | 文件 | 适用场景 |
|------|-----------|----------|-----------|------|----------|
| 原始配置（有渲染） | 15-20 | 15-20 | 15-20% | visual | 实时可视化调试 |
| 关闭渲染 | 20-25 | 20-25 | 20-30% | 标准版 | 基础训练 |
| 激光雷达优化+缓存 | 30-60 | 30-60 | 30-50% | fast | 快速训练⭐ |
| 多进程并行（10核） | 30-60 | 300-600 | 90-100% | parallel | 大规模训练⭐⭐ |
**说明：**
- **单环境FPS**：单个环境实例的帧率
- **总吞吐量**：所有进程合计的 steps/second
- 12600KF（10核20线程）推荐使用并行版本
## 建议
1. **训练时**：使用高性能版本（关闭渲染）
2. **调试时**：使用可视化版本，或间歇性渲染
3. **大规模实验**：使用多进程并行
4. **如果需要GPU加速**：考虑使用GPU渲染或将策略网络部署到GPU上
## 为什么CPU利用率低？
- **渲染阻塞**：CPU在等待渲染完成
- **Python GIL**：限制了多核利用
- **I/O等待**：可能在等待磁盘读取数据
- **单线程瓶颈**：MetaDrive主循环是单线程的
解决方法：关闭渲染 + 多进程并行

241
Env/QUICK_START.md
View File

@@ -1,241 +0,0 @@
# 快速使用指南
## 🚀 已实现的性能优化
根据您的测试结果，原始版本FPS只有15左右，现已进行了全面优化。
---
## 📊 性能瓶颈分析
您的CPU是12600KF（10核20线程），但利用率不到20%，原因是：
1. **激光雷达计算瓶颈**：51辆车 × 100个激光束 = 每帧5100次射线检测
2. **红绿灯检测低效**：遍历所有车道进行几何计算
3. **Python GIL限制**：单线程执行，无法利用多核
4. **计算串行化**：所有车辆依次处理，没有并行
---
## 🎯 推荐使用方案
### 方案1：极速单环境（推荐新手）⭐
```bash
python Env/run_multiagent_env_fast.py
```
**优化内容：**
- ✅ 激光束：100束 → 52束（减少48%计算量）
- ✅ 激光雷达缓存：每3帧才重新计算
- ✅ 红绿灯检测优化：避免遍历所有车道
- ✅ 关闭所有渲染和调试
**预期性能：** 30-60 FPS（2-4倍提升）
---
### 方案2：多进程并行（推荐训练）⭐⭐
```bash
python Env/run_multiagent_env_parallel.py
```
**优化内容：**
- ✅ 同时运行10个独立环境（充分利用10核CPU）
- ✅ 每个环境应用所有单环境优化
- ✅ CPU利用率可达90-100%
**预期性能：** 300-600 steps/s（20-40倍总吞吐量）
---
### 方案3：可视化调试
```bash
python Env/run_multiagent_env_visual.py
```
**说明：** 保留渲染功能，FPS约15，仅用于调试
---
## 🔧 关于GPU加速
### GPU能否加速MetaDrive？
**简短回答：有限支持，主要瓶颈不在GPU**
**详细说明：**
1. **物理计算（主要瓶颈）** ❌ 不支持GPU
- MetaDrive使用Bullet物理引擎，只在CPU运行
- 激光雷达射线检测也在CPU
- 这是FPS低的主要原因
2. **图形渲染** ✅ 支持GPU
- Panda3D会自动使用GPU渲染
- 但我们训练时关闭了渲染，所以GPU无用武之地
3. **策略网络** ✅ 支持GPU
- 可以把Policy模型放到GPU上
- 但环境本身仍在CPU
### GPU渲染配置（可选）
```python
config = {
"use_render": True,
# GPU会自动用于渲染
}
```
### 策略网络GPU加速（推荐）
```python
import torch
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
policy_model = PolicyNet().to(device)
# 批量推理
obs_tensor = torch.tensor(obs_list).to(device)
actions = policy_model(obs_tensor)
```
**详细说明请看：** `GPU_ACCELERATION.md`
---
## 📈 性能对比
| 版本 | FPS | CPU利用率 | 改进 |
|------|-----|-----------|------|
| 原始版本 | 15 | 20% | - |
| 极速版本 | 30-60 | 30-50% | 2-4x |
| 并行版本 | 30-60/env | 90-100% | 总吞吐20-40x |
---
## 💡 使用建议
### 场景1：快速测试环境
```bash
python Env/run_multiagent_env_fast.py
```
单环境，快速验证功能
### 场景2：大规模数据收集
```bash
python Env/run_multiagent_env_parallel.py
```
多进程，最大化数据收集速度
### 场景3：RL训练
```bash
# 推荐使用Ray RLlib等框架，它们内置了并行环境管理
# 或者修改parallel版本，保存经验到replay buffer
```
### 场景4：调试/可视化
```bash
python Env/run_multiagent_env_visual.py
```
带渲染，可以看到车辆运行
---
## 🔍 性能监控
所有版本都内置了性能统计，运行时会显示：
```
Step 100: FPS = 45.23, 车辆数 = 51, 平均步时间 = 22.10ms
```
---
## ⚙️ 高级优化选项
### 调整激光雷达缓存频率
编辑 `run_multiagent_env_fast.py`：
```python
env.lidar_cache_interval = 3 # 改为5可进一步提速（但观测会更旧）
```
### 调整并行进程数
编辑 `run_multiagent_env_parallel.py`：
```python
num_workers = 10 # 改为更少的进程数（如果内存不足）
```
### 进一步减少激光束
编辑 `scenario_env.py` 的 `_get_all_obs()` 函数：
```python
lidar = self.engine.get_sensor("lidar").perceive(
num_lasers=20, # 从40进一步减少到20
distance=20, # 从30减少到20米
...
)
```
---
## 🎓 为什么CPU利用率低？
### 原因分析：
1. **单线程瓶颈**
- Python GIL限制
- MetaDrive主循环是单线程的
- 即使有10个核心，也只用1个
2. **I/O等待**
- 等待渲染完成（如果开启）
- 等待磁盘读取数据
3. **计算不均衡**
- 某些计算很重（激光雷达），某些很轻
- CPU在重计算之间有空闲
### 解决方案：
✅ **已实现：** 多进程并行（`run_multiagent_env_parallel.py`）
- 每个进程占用1个核心
- 10个进程可充分利用10核CPU
- CPU利用率可达90-100%
---
## 📚 相关文档
- `PERFORMANCE_OPTIMIZATION.md` - 详细的性能优化指南
- `GPU_ACCELERATION.md` - GPU加速的完整说明
---
## ❓ 常见问题
### Q: 为什么关闭渲染后FPS还是只有20？
A: 主要瓶颈是激光雷达计算，不是渲染。请使用 `run_multiagent_env_fast.py`。
### Q: GPU能加速训练吗？
A: 环境模拟在CPU，但策略网络可以在GPU上训练。
### Q: 如何最大化CPU利用率？
A: 使用 `run_multiagent_env_parallel.py` 多进程版本。
### Q: 会影响观测精度吗？
A: 激光束减少会略微降低精度，但实践中影响很小。缓存会让观测滞后1-2帧。
### Q: 如何恢复原始配置？
A: 使用 `run_multiagent_env_visual.py` 或修改配置文件中的参数。
---
## 🚦 下一步
1. 先测试 `run_multiagent_env_fast.py`，验证性能提升
2. 如果满意，用于日常训练
3. 需要大规模训练时，使用 `run_multiagent_env_parallel.py`
4. 考虑将策略网络迁移到GPU
祝训练顺利！🎉

BIN
Env/__pycache__/logger_utils.cpython-310.pyc
View File

Binary file not shown.

BIN
Env/__pycache__/scenario_env.cpython-310.pyc
View File

Binary file not shown.

BIN
Env/__pycache__/scenario_env.cpython-313.pyc
View File

Binary file not shown.

BIN
Env/__pycache__/simple_idm_policy.cpython-310.pyc
View File

Binary file not shown.

13
Env/run_multiagent_env.py
View File

@@ -5,7 +5,7 @@ from logger_utils import setup_logger
import sys
import os
WAYMO_DATA_DIR = r"/home/huangfukk/MAGAIL4AutoDrive/Env"
WAYMO_DATA_DIR = r"/home/huangfukk/mdsn"
def main(enable_logging=False, log_file=None):
"""
@@ -18,7 +18,7 @@ def main(enable_logging=False, log_file=None):
env = MultiAgentScenarioEnv(
config={
# "data_directory": AssetLoader.file_path(AssetLoader.asset_path, "waymo", unix_style=False),
"data_directory": AssetLoader.file_path(WAYMO_DATA_DIR, "exp_converted", unix_style=False),
"data_directory": AssetLoader.file_path(WAYMO_DATA_DIR, "exp_filtered", unix_style=False),
"is_multi_agent": True,
"num_controlled_agents": 3,
"horizon": 300,
@@ -30,9 +30,12 @@ def main(enable_logging=False, log_file=None):
# 车道检测与过滤配置
"filter_offroad_vehicles": True, # 启用车道区域过滤，过滤草坪等非车道区域的车辆
"lane_tolerance": 3.0, # 车道检测容差（米），可根据需要调整
"max_controlled_vehicles": 2, # 限制最大车辆数（可选，None表示不限制）
"debug_lane_filter": True,
"debug_traffic_light": True,
"max_controlled_vehicles": None, # 限制最大车辆数（可选，None表示不限制）
# 调试配置（可选）
# "debug_lane_filter": True, # 启用车道过滤详细调试
# "verbose_reset": True, # 启用重置详细统计
# "inherit_expert_velocity": True, # 继承专家速度
},
agent2policy=ConstantVelocityPolicy(target_speed=50)
)

115
Env/run_multiagent_env_fast.py
View File

@@ -1,115 +0,0 @@
from scenario_env import MultiAgentScenarioEnv
from simple_idm_policy import ConstantVelocityPolicy
from metadrive.engine.asset_loader import AssetLoader
from logger_utils import setup_logger
import time
import sys
import os
WAYMO_DATA_DIR = r"/home/huangfukk/MAGAIL4AutoDrive/Env"
def main(enable_logging=False):
"""极致性能优化版本 - 启用所有优化选项"""
env = MultiAgentScenarioEnv(
config={
"data_directory": AssetLoader.file_path(WAYMO_DATA_DIR, "exp_converted", unix_style=False),
"is_multi_agent": True,
"num_controlled_agents": 3,
"horizon": 300,
# 关闭所有渲染
"use_render": False,
"render_pipeline": False,
"image_observation": False,
"interface_panel": [],
"manual_control": False,
"show_fps": False,
"debug": False,
# 物理引擎优化
"physics_world_step_size": 0.02,
"decision_repeat": 5,
"sequential_seed": True,
"reactive_traffic": True,
# 车道检测与过滤配置
"filter_offroad_vehicles": True, # 过滤非车道区域的车辆
"lane_tolerance": 3.0,
"max_controlled_vehicles": 15, # 限制车辆数以提升性能
},
agent2policy=ConstantVelocityPolicy(target_speed=50)
)
# 【关键优化】启用激光雷达缓存
# 每3帧才重新计算激光雷达，其余帧使用缓存
# 可将激光雷达计算量减少到原来的1/3
env.lidar_cache_interval = 3
obs = env.reset(0)
# 性能统计
start_time = time.time()
total_steps = 0
print("=" * 60)
print("极致性能模式")
print("激光雷达优化：80→40束 (前向), 10→6束 (侧向+车道线)")
print("激光雷达缓存：每3帧计算一次，中间帧使用缓存")
print("预期性能提升：3-5倍")
print("=" * 60)
for step in range(10000):
actions = {
aid: env.controlled_agents[aid].policy.act()
for aid in env.controlled_agents
}
obs, rewards, dones, infos = env.step(actions)
total_steps += 1
# 每100步输出一次性能统计
if step % 100 == 0 and step > 0:
elapsed = time.time() - start_time
fps = total_steps / elapsed
print(f"Step {step:4d}: FPS = {fps:6.2f}, 车辆数 = {len(env.controlled_agents):3d}, "
f"平均步时间 = {1000/fps:.2f}ms")
if dones["__all__"]:
break
# 最终统计
elapsed = time.time() - start_time
fps = total_steps / elapsed
print("\n" + "=" * 60)
print(f"总计: {total_steps} 步")
print(f"耗时: {elapsed:.2f}s")
print(f"平均FPS: {fps:.2f}")
print(f"单步平均耗时: {1000/fps:.2f}ms")
print("=" * 60)
env.close()
if __name__ == "__main__":
# 解析命令行参数
enable_logging = "--log" in sys.argv or "-l" in sys.argv
# 提取自定义日志文件名
log_file = None
for arg in sys.argv:
if arg.startswith("--log-file="):
log_file = arg.split("=")[1]
break
if enable_logging:
# 使用日志记录
log_dir = os.path.join(os.path.dirname(__file__), "logs")
with setup_logger(log_file=log_file or "run_fast.log", log_dir=log_dir):
main(enable_logging=True)
else:
# 普通运行（只输出到终端）
print("💡 提示: 使用 --log 或 -l 参数启用日志记录")
print("-" * 60)
main(enable_logging=False)

156
Env/run_multiagent_env_parallel.py
View File

@@ -1,156 +0,0 @@
"""
多进程并行版本 - 充分利用多核CPU
适合大规模数据收集和训练
"""
from scenario_env import MultiAgentScenarioEnv
from simple_idm_policy import ConstantVelocityPolicy
from metadrive.engine.asset_loader import AssetLoader
import time
import os
from multiprocessing import Pool, cpu_count
WAYMO_DATA_DIR = r"/home/huangfukk/MAGAIL4AutoDrive/Env"
def run_single_env(args):
"""在单个进程中运行一个环境实例"""
seed, num_steps, worker_id = args
# 创建环境（每个进程独立）
env = MultiAgentScenarioEnv(
config={
"data_directory": AssetLoader.file_path(WAYMO_DATA_DIR, "exp_converted", unix_style=False),
"is_multi_agent": True,
"num_controlled_agents": 3,
"horizon": 300,
# 性能优化
"use_render": False,
"render_pipeline": False,
"image_observation": False,
"interface_panel": [],
"manual_control": False,
"show_fps": False,
"debug": False,
"physics_world_step_size": 0.02,
"decision_repeat": 5,
"sequential_seed": True,
"reactive_traffic": True,
# 车道检测与过滤配置
"filter_offroad_vehicles": True,
"lane_tolerance": 3.0,
"max_controlled_vehicles": 15,
},
agent2policy=ConstantVelocityPolicy(target_speed=50)
)
# 启用激光雷达缓存
env.lidar_cache_interval = 3
# 运行仿真
start_time = time.time()
obs = env.reset(seed)
total_steps = 0
total_agents = 0
for step in range(num_steps):
actions = {
aid: env.controlled_agents[aid].policy.act()
for aid in env.controlled_agents
}
obs, rewards, dones, infos = env.step(actions)
total_steps += 1
total_agents += len(env.controlled_agents)
if dones["__all__"]:
break
elapsed = time.time() - start_time
fps = total_steps / elapsed if elapsed > 0 else 0
avg_agents = total_agents / total_steps if total_steps > 0 else 0
env.close()
return {
'worker_id': worker_id,
'seed': seed,
'steps': total_steps,
'elapsed': elapsed,
'fps': fps,
'avg_agents': avg_agents,
}
def main():
"""主函数：协调多个并行环境"""
# 获取CPU核心数
num_cores = cpu_count()
# 建议使用物理核心数（12600KF是10核20线程，使用10个进程）
num_workers = min(10, num_cores)
print("=" * 80)
print(f"多进程并行模式")
print(f"CPU核心数: {num_cores}")
print(f"并行进程数: {num_workers}")
print(f"每个环境运行: 1000步")
print("=" * 80)
# 准备任务参数
num_steps_per_env = 1000
tasks = [(seed, num_steps_per_env, worker_id)
for worker_id, seed in enumerate(range(num_workers))]
# 启动多进程池
start_time = time.time()
with Pool(processes=num_workers) as pool:
results = pool.map(run_single_env, tasks)
total_elapsed = time.time() - start_time
# 统计结果
print("\n" + "=" * 80)
print("各进程执行结果：")
print("-" * 80)
print(f"{'Worker':<8} {'Seed':<6} {'Steps':<8} {'Time(s)':<10} {'FPS':<8} {'平均车辆数':<12}")
print("-" * 80)
total_steps = 0
total_fps = 0
for result in results:
print(f"{result['worker_id']:<8} "
f"{result['seed']:<6} "
f"{result['steps']:<8} "
f"{result['elapsed']:<10.2f} "
f"{result['fps']:<8.2f} "
f"{result['avg_agents']:<12.1f}")
total_steps += result['steps']
total_fps += result['fps']
print("-" * 80)
avg_fps_per_env = total_fps / len(results)
total_throughput = total_steps / total_elapsed
print(f"\n总体统计：")
print(f" 总步数: {total_steps}")
print(f" 总耗时: {total_elapsed:.2f}s")
print(f" 单环境平均FPS: {avg_fps_per_env:.2f}")
print(f" 总吞吐量: {total_throughput:.2f} steps/s")
print(f" 并行效率: {total_throughput / avg_fps_per_env:.1f}x")
print("=" * 80)
# 与单进程对比
print(f"\n性能对比：")
print(f" 单进程FPS (预估): ~30 FPS")
print(f" 多进程吞吐量: {total_throughput:.2f} steps/s")
print(f" 性能提升: {total_throughput / 30:.1f}x")
print("=" * 80)
if __name__ == "__main__":
main()

60
Env/run_multiagent_env_visual.py
View File

@@ -1,60 +0,0 @@
from scenario_env import MultiAgentScenarioEnv
from simple_idm_policy import ConstantVelocityPolicy
from metadrive.engine.asset_loader import AssetLoader
import time
WAYMO_DATA_DIR = r"/home/huangfukk/MAGAIL4AutoDrive/Env"
def main():
"""带可视化的版本（低FPS，约15帧）"""
env = MultiAgentScenarioEnv(
config={
"data_directory": AssetLoader.file_path(WAYMO_DATA_DIR, "exp_converted", unix_style=False),
"is_multi_agent": True,
"num_controlled_agents": 3,
"horizon": 300,
# 可视化设置（牺牲性能）
"use_render": True,
"manual_control": False,
"sequential_seed": True,
"reactive_traffic": True,
},
agent2policy=ConstantVelocityPolicy(target_speed=50)
)
obs = env.reset(0)
start_time = time.time()
total_steps = 0
for step in range(10000):
actions = {
aid: env.controlled_agents[aid].policy.act()
for aid in env.controlled_agents
}
obs, rewards, dones, infos = env.step(actions)
env.render(mode="topdown") # 实时渲染
total_steps += 1
if step % 100 == 0 and step > 0:
elapsed = time.time() - start_time
fps = total_steps / elapsed
print(f"Step {step}: FPS = {fps:.2f}, 车辆数 = {len(env.controlled_agents)}")
if dones["__all__"]:
break
elapsed = time.time() - start_time
fps = total_steps / elapsed
print(f"\n总计: {total_steps} 步，耗时 {elapsed:.2f}s，平均FPS = {fps:.2f}")
env.close()
if __name__ == "__main__":
main()

698
Env/scenario_env.py
View File

@@ -1,3 +1,91 @@
"""
多智能体场景环境 (MultiAgentScenarioEnv)
==================================
配置参数说明 (写在最前面)
==================================
基础配置：
data_directory (str): 专家数据目录路径
num_controlled_agents (int): 默认可控智能体数量，默认3
horizon (int): 每个回合的最大步数，默认1000
车道检测与过滤配置：
filter_offroad_vehicles (bool): 是否过滤非车道区域的车辆，默认True
- True: 过滤掉在草坪、停车场等非车道区域生成的车辆
- False: 保留所有车辆
lane_tolerance (float): 车道检测容差（米），默认3.0
- 用于放宽车道检测的边界条件
max_controlled_vehicles (int|None): 最大可控车辆数限制，默认None
- None: 不限制车辆数量
- int: 限制最多生成的车辆数
场景对象配置：
no_traffic_lights (bool): 是否禁用红绿灯渲染和逻辑，默认False
- True: 完全移除场景中的红绿灯
- False: 保留红绿灯（按数据集原样生成）
专家数据继承配置：
inherit_expert_velocity (bool): 是否继承专家数据中车辆的初始速度，默认False
- True: 车辆生成时使用专家数据中的速度
- False: 车辆生成时速度为0（由策略控制）
调试模式配置：
debug_lane_filter (bool): 车道过滤详细调试输出，默认False
- 输出每个车辆位置的车道检测详细过程
verbose_reset (bool): 重置时输出详细统计信息，默认False
- 输出场景统计、过滤详情等
使用示例：
env = MultiAgentScenarioEnv(
config={
"data_directory": "path/to/data",
"max_controlled_vehicles": 10,
"inherit_expert_velocity": True, # 继承专家速度
"no_traffic_lights": True, # 禁用红绿灯
"verbose_reset": True, # 详细输出
},
agent2policy=your_policy
)
==================================
整体逻辑和处理流程
==================================
1. 初始化阶段：
- 继承MetaDrive的ScenarioEnv基类
- 配置多智能体参数（车辆数量、调试模式等）
- 接收策略映射字典(agent2policy)
2. 环境重置阶段 (reset方法)：
- 解析专家数据，提取车辆生成信息(car_birth_info_list)
- 清理原始交通数据，只保留车辆位置、朝向、目的地、速度（可选）
- 禁用红绿灯（如果配置）
- 初始化地图和车道信息
- 执行车道过滤(_filter_valid_spawn_positions)，移除非车道区域的车辆
- 限制最大车辆数量
- 生成可控智能体(_spawn_controlled_agents)
3. 观测获取阶段 (_get_all_obs方法)：
- 遍历所有可控车辆
- 获取车辆状态（位置、速度、朝向）
- 检测红绿灯状态(_get_traffic_light_state)
- 获取激光雷达数据（前向、侧向、车道线检测）
- 组装完整观测向量
4. 环境步进阶段 (step方法)：
- 执行所有智能体的动作
- 更新物理引擎状态
- 生成新的智能体（按时间步）
- 返回新的观测、奖励、完成状态
核心功能模块：
- PolicyVehicle: 可控制策略的车辆类
- 车道检测与过滤: 确保车辆只在有效车道上生成
- 多智能体管理: 动态生成和管理可控车辆
- 专家速度继承: 可选地继承专家数据中的初始速度
"""
import numpy as np
from metadrive.component.navigation_module.node_network_navigation import NodeNetworkNavigation
from metadrive.envs.scenario_env import ScenarioEnv
@@ -11,172 +99,347 @@ from metadrive.type import MetaDriveType
class PolicyVehicle(DefaultVehicle):
"""
可控制策略的车辆类
继承自MetaDrive的DefaultVehicle，增加了策略控制和目标设置功能。
用于多智能体环境中的可控车辆，支持自定义策略和目的地。
"""
def __init__(self, *args, **kwargs):
"""
初始化策略车辆
Args:
*args: 传递给父类的位置参数
**kwargs: 传递给父类的关键字参数
"""
super().__init__(*args, **kwargs)
self.policy = None
self.destination = None
self.policy = None # 车辆的控制策略
self.destination = None # 车辆的目标目的地
def set_policy(self, policy):
"""
设置车辆的控制策略
Args:
policy: 控制策略对象，必须实现act(observation)方法
"""
self.policy = policy
def set_destination(self, des):
"""
设置车辆的目标目的地
Args:
des: 目标位置坐标 (x, y)
"""
self.destination = des
def act(self, observation, policy=None):
"""
根据观测获取动作
Args:
observation: 环境观测数据
policy: 可选的外部策略，如果提供则优先使用
Returns:
动作向量，如果无策略则返回随机动作
"""
if self.policy is not None:
return self.policy.act(observation)
else:
return self.action_space.sample()
def before_step(self, action):
self.last_position = self.position # 2D vector
self.last_velocity = self.velocity # 2D vector
self.last_speed = self.speed # Scalar
self.last_heading_dir = self.heading
"""
执行动作前的状态记录
在每步执行前记录当前状态，用于后续的状态追踪和分析。
Args:
action: 即将执行的动作
"""
self.last_position = self.position # 记录当前位置 (2D向量)
self.last_velocity = self.velocity # 记录当前速度 (2D向量)
self.last_speed = self.speed # 记录当前速度大小 (标量)
self.last_heading_dir = self.heading # 记录当前朝向
if action is not None:
self.last_current_action.append(action)
self._set_action(action)
self.last_current_action.append(action) # 记录动作历史
self._set_action(action) # 设置动作到车辆
def is_done(self):
"""
检查车辆是否完成任务
目前为空实现，可根据需要添加到达目的地或碰撞检测逻辑
Returns:
bool: True表示任务完成，False表示继续执行
"""
# arrive or crash
pass
# 将PolicyVehicle注册为默认车辆类型
vehicle_class_to_type[PolicyVehicle] = "default"
class MultiAgentScenarioEnv(ScenarioEnv):
"""
多智能体场景环境
基于MetaDrive的ScenarioEnv扩展，支持多智能体强化学习训练。
主要功能包括：
1. 从专家数据中提取车辆信息并生成可控智能体
2. 车道检测与过滤，确保车辆在有效区域生成
3. 红绿灯状态检测，为智能体提供交通信号信息
4. 多智能体观测、动作和奖励管理
"""
@classmethod
def default_config(cls):
"""
获取环境的默认配置
继承父类配置并添加多智能体相关的配置参数
配置参数说明：
- data_directory: 专家数据目录路径
- num_controlled_agents: 默认可控智能体数量
- horizon: 每个回合的最大步数
车道检测与过滤配置：
- filter_offroad_vehicles: 是否过滤非车道区域的车辆
- lane_tolerance: 车道检测容差（米）
- max_controlled_vehicles: 最大可控车辆数限制（None表示不限制）
场景对象配置：
- no_traffic_lights: 禁用红绿灯渲染和逻辑
专家数据继承配置：
- inherit_expert_velocity: 是否继承专家数据中车辆的初始速度（默认False）
调试模式配置：
- debug_traffic_light: 红绿灯检测详细调试输出
- debug_lane_filter: 车道过滤详细调试输出
- verbose_reset: 重置时输出详细统计信息
Returns:
dict: 包含所有配置参数的字典
"""
config = super().default_config()
config.update(dict(
# 基础配置
data_directory=None,
num_controlled_agents=3,
horizon=1000,
# 车道检测与过滤配置
filter_offroad_vehicles=True, # 是否过滤非车道区域的车辆
lane_tolerance=3.0, # 车道检测容差（米），用于放宽边界条件
max_controlled_vehicles=None, # 最大可控车辆数限制（None表示不限制）
filter_offroad_vehicles=True,
lane_tolerance=3.0,
max_controlled_vehicles=None,
# 场景对象配置
no_traffic_lights=False,
# 专家数据继承配置
inherit_expert_velocity=False,
# 调试模式配置
debug_traffic_light=False, # 是否启用红绿灯检测调试输出
debug_lane_filter=False, # 是否启用车道过滤调试输出
debug_lane_filter=False,
verbose_reset=False,
))
return config
def __init__(self, config, agent2policy):
self.policy = agent2policy
self.controlled_agents = {}
self.controlled_agent_ids = []
self.obs_list = []
self.round = 0
"""
初始化多智能体场景环境
Args:
config: 环境配置字典，包含各种参数设置
agent2policy: 智能体ID到策略的映射字典
"""
self.policy = agent2policy # 智能体策略映射
self.controlled_agents = {} # 可控智能体字典 {agent_id: vehicle}
self.controlled_agent_ids = [] # 可控智能体ID列表
self.obs_list = [] # 观测数据列表
self.round = 0 # 当前时间步
# 调试模式配置
self.debug_traffic_light = config.get("debug_traffic_light", False)
self.debug_lane_filter = config.get("debug_lane_filter", False)
# 调用父类初始化
super().__init__(config)
def reset(self, seed: Union[None, int] = None):
self.round = 0
"""
重置环境到初始状态
这是环境的核心重置方法，执行以下步骤：
1. 初始化日志系统
2. 解析专家数据，提取车辆生成信息
3. 清理原始交通数据
4. 初始化地图和车道信息
5. 执行车道过滤和车辆数量限制
6. 生成可控智能体
7. 返回初始观测
Args:
seed: 随机种子，用于环境重置时的随机性控制
Returns:
list: 所有智能体的初始观测数据
"""
self.round = 0 # 重置时间步计数器
# 初始化日志系统
if self.logger is None:
self.logger = get_logger()
log_level = self.config.get("log_level", logging.DEBUG if self.config.get("debug", False) else logging.INFO)
set_log_level(log_level)
# 延迟初始化MetaDrive引擎
self.lazy_init()
self._reset_global_seed(seed)
if self.engine is None:
raise ValueError("Broken MetaDrive instance.")
# 在场景加载前禁用红绿灯（如果配置中启用了no_traffic_lights选项）
if self.config.get("no_traffic_lights", False):
# 重写红绿灯管理器的方法，阻止创建和使用红绿灯
if hasattr(self.engine, 'light_manager') and self.engine.light_manager is not None:
self.engine.light_manager.before_reset = lambda *args, **kwargs: None
self.engine.light_manager.after_reset = lambda *args, **kwargs: None
self.engine.light_manager.before_step = lambda *args, **kwargs: None
self.engine.light_manager.get_traffic_light = lambda *args, **kwargs: None
self.logger.info("已禁用红绿灯管理器")
# 步骤1：解析专家数据，提取车辆生成信息
# 记录专家数据中每辆车的位置，接着全部清除，只保留位置等信息，用于后续生成
_obj_to_clean_this_frame = []
self.car_birth_info_list = []
_obj_to_clean_this_frame = [] # 需要清理的对象ID列表
self.car_birth_info_list = [] # 车辆生成信息列表
for scenario_id, track in self.engine.traffic_manager.current_traffic_data.items():
# 跳过自车（SDC - Self Driving Car）
if scenario_id == self.engine.traffic_manager.sdc_scenario_id:
continue
else:
if track["type"] == MetaDriveType.VEHICLE:
_obj_to_clean_this_frame.append(scenario_id)
valid = track['state']['valid']
first_show = np.argmax(valid) if valid.any() else -1
last_show = len(valid) - 1 - np.argmax(valid[::-1]) if valid.any() else -1
# id，出现时间，出生点坐标，出生朝向，目的地
self.car_birth_info_list.append({
'id': track['metadata']['object_id'],
'show_time': first_show,
'begin': (track['state']['position'][first_show, 0], track['state']['position'][first_show, 1]),
'heading': track['state']['heading'][first_show],
'end': (track['state']['position'][last_show, 0], track['state']['position'][last_show, 1])
})
# 只处理车辆类型的对象
if track["type"] == MetaDriveType.VEHICLE:
_obj_to_clean_this_frame.append(scenario_id)
valid = track['state']['valid'] # 车辆有效性标记
# 找到车辆首次出现和最后出现的时间步
first_show = np.argmax(valid) if valid.any() else -1
last_show = len(valid) - 1 - np.argmax(valid[::-1]) if valid.any() else -1
# 提取车辆关键信息
car_info = {
'id': track['metadata']['object_id'],
'show_time': first_show,
'begin': (track['state']['position'][first_show, 0], track['state']['position'][first_show, 1]),
'heading': track['state']['heading'][first_show],
'end': (track['state']['position'][last_show, 0], track['state']['position'][last_show, 1])
}
# 如果配置要求继承专家速度，则提取初始速度
if self.config.get("inherit_expert_velocity", False):
velocity = track['state']['velocity'][first_show]
car_info['velocity'] = (velocity[0], velocity[1])
self.car_birth_info_list.append(car_info)
# 非车辆对象（如红绿灯、行人等）保留，不清理
# 清理车辆原始交通数据，释放内存（保留红绿灯等其他对象）
for scenario_id in _obj_to_clean_this_frame:
self.engine.traffic_manager.current_traffic_data.pop(scenario_id)
# 步骤2：重置MetaDrive引擎和传感器
self.engine.reset()
self.reset_sensors()
self.engine.taskMgr.step()
# 步骤3：获取地图车道信息
self.lanes = self.engine.map_manager.current_map.road_network.graph
# 调试：场景信息统计
if self.debug_lane_filter or self.debug_traffic_light:
# 调试：场景信息统计（仅在verbose_reset模式下输出）
if self.config.get("verbose_reset", False):
print(f"\n📍 场景信息统计:")
print(f" - 总车道数: {len(self.lanes)}")
# 统计红绿灯数量
if self.debug_traffic_light:
# 统计红绿灯数量（如果未禁用红绿灯）
if not self.config.get("no_traffic_lights", False):
traffic_light_lanes = []
for lane in self.lanes.values():
if self.engine.light_manager.has_traffic_light(lane.lane.index):
traffic_light_lanes.append(lane.lane.index)
print(f" - 有红绿灯的车道数: {len(traffic_light_lanes)}")
if len(traffic_light_lanes) > 0:
print(f" 车道索引: {traffic_light_lanes[:5]}" +
(f" ... 共{len(traffic_light_lanes)}个" if len(traffic_light_lanes) > 5 else ""))
else:
print(f" ⚠️ 场景中没有红绿灯！")
if len(traffic_light_lanes) > 5:
print(f" 车道索引示例: {traffic_light_lanes[:5]} ...")
else:
print(f" - 红绿灯: 已禁用")
# 在获取车道信息后，进行车道区域过滤
# 步骤4：执行车道区域过滤
total_cars_before = len(self.car_birth_info_list)
valid_count, filtered_count, filtered_list = self._filter_valid_spawn_positions()
# 输出过滤信息
# 输出过滤信息（仅在有过滤时输出）
if filtered_count > 0:
self.logger.warning(f"车辆生成位置过滤: 原始 {total_cars_before} 辆, "
f"有效 {valid_count} 辆, 过滤 {filtered_count} 辆")
for filtered_car in filtered_list[:5]: # 只显示前5个
self.logger.debug(f" - 过滤车辆 ID={filtered_car['id']}, "
f"位置={filtered_car['position']}, "
f"原因={filtered_car['reason']}")
if filtered_count > 5:
self.logger.debug(f" - ... 还有 {filtered_count - 5} 辆车被过滤")
if self.config.get("verbose_reset", False):
self.logger.warning(f"车辆生成位置过滤: 原始 {total_cars_before} 辆, "
f"有效 {valid_count} 辆, 过滤 {filtered_count} 辆")
for filtered_car in filtered_list[:3]:
self.logger.debug(f" 过滤车辆 ID={filtered_car['id']}, "
f"位置={filtered_car['position']}, "
f"原因={filtered_car['reason']}")
if filtered_count > 3:
self.logger.debug(f" ... 还有 {filtered_count - 3} 辆车被过滤")
else:
self.logger.info(f"车辆过滤: {total_cars_before} 辆 -> {valid_count} 辆 (过滤 {filtered_count} 辆)")
# 限制最大车辆数（在过滤后应用）
# 步骤5：限制最大车辆数（在过滤后应用）
max_vehicles = self.config.get("max_controlled_vehicles", None)
if max_vehicles is not None and len(self.car_birth_info_list) > max_vehicles:
original_count = len(self.car_birth_info_list)
self.car_birth_info_list = self.car_birth_info_list[:max_vehicles]
self.logger.info(f"限制最大车辆数为 {max_vehicles} 辆")
if self.config.get("verbose_reset", False):
self.logger.info(f"限制最大车辆数: {original_count} 辆 -> {max_vehicles} 辆")
self.logger.info(f"最终生成 {len(self.car_birth_info_list)} 辆可控车辆")
# 最终统计
if self.config.get("verbose_reset", False):
self.logger.info(f"✓ 最终生成 {len(self.car_birth_info_list)} 辆可控车辆")
# 清理渲染器
if self.top_down_renderer is not None:
self.top_down_renderer.clear()
self.engine.top_down_renderer = None
self.dones = {}
self.episode_rewards = defaultdict(float)
self.episode_lengths = defaultdict(int)
# 初始化回合相关变量
self.dones = {} # 智能体完成状态
self.episode_rewards = defaultdict(float) # 回合奖励累积
self.episode_lengths = defaultdict(int) # 回合长度累积
# 清空可控智能体
self.controlled_agents.clear()
self.controlled_agent_ids.clear()
# 步骤6：调用父类重置并生成可控智能体
super().reset(seed) # 初始化场景
self._spawn_controlled_agents()
# 步骤7：返回初始观测
return self._get_all_obs()
def _is_position_on_lane(self, position, tolerance=None):
"""
检测给定位置是否在有效车道范围内
这个函数用于验证车辆生成位置是否在合法的车道上，避免在草坪、停车场等
非车道区域生成车辆。支持两种检测方法：
1. 严格检测：直接使用MetaDrive的point_on_lane方法
2. 容差检测：考虑车道边缘的容差范围（当前已禁用）
Args:
position: (x, y) 车辆位置坐标
tolerance: 容差范围（米），用于放宽检测条件。None时使用配置中的默认值
@@ -184,11 +447,13 @@ class MultiAgentScenarioEnv(ScenarioEnv):
Returns:
bool: True表示在车道上，False表示在非车道区域（如草坪、停车场等）
"""
# 检查车道信息是否已初始化
if not hasattr(self, 'lanes') or self.lanes is None:
if self.debug_lane_filter:
print(f" ⚠️ 车道信息未初始化，默认允许")
return True # 如果车道信息未初始化，默认允许生成
# 设置容差参数
if tolerance is None:
tolerance = self.config.get("lane_tolerance", 3.0)
@@ -198,6 +463,7 @@ class MultiAgentScenarioEnv(ScenarioEnv):
print(f" 🔍 检测位置 ({position_2d[0]:.2f}, {position_2d[1]:.2f}), 容差={tolerance}m")
# 方法1：直接检测是否在任一车道上
# 遍历所有车道，使用MetaDrive的point_on_lane方法进行精确检测
checked_lanes = 0
for lane in self.lanes.values():
try:
@@ -207,6 +473,7 @@ class MultiAgentScenarioEnv(ScenarioEnv):
print(f" ✅ 在车道上 (车道{lane.lane.index}, 检查了{checked_lanes}条)")
return True
except:
# 如果检测过程中出现异常，继续检查下一条车道
continue
if self.debug_lane_filter:
@@ -214,6 +481,7 @@ class MultiAgentScenarioEnv(ScenarioEnv):
# 方法2：如果严格检测失败，使用容差范围检测（考虑车道边缘）
# 注释：此方法已被禁用，如需启用请取消注释
# 该方法通过计算点到车道中心线的横向距离来判断是否在容差范围内
# if tolerance > 0:
# for lane in self.lanes.values():
# try:
@@ -233,12 +501,22 @@ class MultiAgentScenarioEnv(ScenarioEnv):
def _filter_valid_spawn_positions(self):
"""
过滤掉生成位置不在有效车道上的车辆信息
根据配置决定是否执行过滤
这个函数是车道过滤的核心实现，用于确保所有生成的车辆都在合法的车道上。
它会遍历所有车辆生成信息，使用_is_position_on_lane方法检测每个位置，
过滤掉在草坪、停车场等非车道区域的车辆。
过滤过程包括：
1. 检查配置是否启用过滤
2. 遍历所有车辆生成信息
3. 对每个车辆位置进行车道检测
4. 分离有效和无效的车辆
5. 更新车辆生成列表
Returns:
tuple: (有效车辆数量, 被过滤车辆数量, 被过滤车辆ID列表)
tuple: (有效车辆数量, 被过滤车辆数量, 被过滤车辆详细信息列表)
"""
# 如果配置中禁用了过滤，直接返回
# 检查配置是否启用车道过滤
if not self.config.get("filter_offroad_vehicles", True):
if self.debug_lane_filter:
print(f"🚫 车道过滤已禁用")
@@ -247,19 +525,23 @@ class MultiAgentScenarioEnv(ScenarioEnv):
if self.debug_lane_filter:
print(f"\n🔍 开始车道过滤: 共 {len(self.car_birth_info_list)} 辆车待检测")
valid_cars = []
filtered_cars = []
# 初始化过滤结果
valid_cars = [] # 有效车辆列表
filtered_cars = [] # 被过滤车辆列表
tolerance = self.config.get("lane_tolerance", 3.0)
# 遍历所有车辆生成信息进行检测
for idx, car in enumerate(self.car_birth_info_list):
if self.debug_lane_filter:
print(f"\n车辆 {idx+1}/{len(self.car_birth_info_list)}: ID={car['id']}")
# 检测车辆生成位置是否在有效车道上
if self._is_position_on_lane(car['begin'], tolerance=tolerance):
valid_cars.append(car)
if self.debug_lane_filter:
print(f" ✅ 保留")
else:
# 记录被过滤的车辆信息
filtered_cars.append({
'id': car['id'],
'position': car['begin'],
@@ -268,6 +550,7 @@ class MultiAgentScenarioEnv(ScenarioEnv):
if self.debug_lane_filter:
print(f" ❌ 过滤 (原因: 不在车道上)")
# 更新车辆生成列表为过滤后的结果
self.car_birth_info_list = valid_cars
if self.debug_lane_filter:
@@ -276,204 +559,185 @@ class MultiAgentScenarioEnv(ScenarioEnv):
return len(valid_cars), len(filtered_cars), filtered_cars
def _spawn_controlled_agents(self):
"""
生成可控智能体车辆
根据当前时间步和车辆生成信息，动态生成需要出现的可控车辆。
每个车辆都会被分配策略和目标目的地，并注册到MetaDrive引擎中
参与物理仿真。
生成过程：
1. 遍历所有车辆生成信息
2. 检查车辆是否应该在当前时间步出现
3. 创建PolicyVehicle实例
4. 设置车辆策略和目的地
5. 注册到环境管理和引擎中
"""
# 注释：可以获取自车位置用于相对位置计算（当前未使用）
# ego_vehicle = self.engine.agent_manager.active_agents.get("default_agent")
# ego_position = ego_vehicle.position if ego_vehicle else np.array([0, 0])
# 遍历所有车辆生成信息
for car in self.car_birth_info_list:
# 检查车辆是否应该在当前时间步出现
if car['show_time'] == self.round:
# 生成智能体ID
agent_id = f"controlled_{car['id']}"
# 在MetaDrive引擎中生成车辆对象
vehicle = self.engine.spawn_object(
PolicyVehicle,
vehicle_config={},
position=car['begin'],
heading=car['heading']
PolicyVehicle, # 使用自定义的策略车辆类
vehicle_config={}, # 车辆配置（使用默认）
position=car['begin'], # 车辆生成位置
heading=car['heading'] # 车辆生成朝向
)
# 重置车辆状态到指定位置和朝向
vehicle.reset(position=car['begin'], heading=car['heading'])
# 如果配置要求继承专家速度，则设置初始速度
if 'velocity' in car and self.config.get("inherit_expert_velocity", False):
vehicle.set_velocity(car['velocity'])
vehicle.set_policy(self.policy)
vehicle.set_destination(car['end'])
# 设置车辆的控制策略和目标
vehicle.set_policy(self.policy) # 设置策略
vehicle.set_destination(car['end']) # 设置目的地
# 注册到环境管理
self.controlled_agents[agent_id] = vehicle
self.controlled_agent_ids.append(agent_id)
# ✅ 关键：注册到引擎的 active_agents，才能参与物理更新
# 这是MetaDrive引擎识别和管理智能体的关键步骤
self.engine.agent_manager.active_agents[agent_id] = vehicle
def _get_traffic_light_state(self, vehicle):
def _get_all_obs(self):
"""
获取车辆当前位置的红绿灯状态（优化版）
获取所有可控智能体的观测数据
解决问题：
1. 部分红绿灯状态为None的问题 - 添加异常处理和默认值
2. 车道分段导致无法获取红绿灯的问题 - 优先使用导航模块，失败时回退到遍历
这是环境的核心观测函数，为每个可控智能体组装完整的观测向量。
观测数据包括：
1. 车辆状态信息：位置、速度、朝向
2. 传感器数据：激光雷达（前向、侧向、车道线检测）
3. 导航信息：目标目的地
观测向量结构：
- position[2]: 车辆位置 (x, y)
- velocity[2]: 车辆速度 (vx, vy)
- heading[1]: 车辆朝向角度
- lidar[80]: 前向激光雷达数据 (80个激光束，30米范围)
- side_lidar[10]: 侧向激光雷达数据 (10个激光束，8米范围)
- lane_line_lidar[10]: 车道线检测数据 (10个激光束，3米范围)
- destination[2]: 目标目的地 (x, y)
Returns:
int: 0=无红绿灯, 1=绿灯, 2=黄灯, 3=红灯
list: 所有智能体的观测数据列表
"""
traffic_light = 0
state = vehicle.get_state()
position_2d = state['position'][:2]
self.obs_list = [] # 清空观测列表
if self.debug_traffic_light:
print(f"\n🚦 检测车辆红绿灯 - 位置: ({position_2d[0]:.1f}, {position_2d[1]:.1f})")
try:
# 方法1：优先尝试从车辆导航模块获取当前车道（更高效）
if hasattr(vehicle, 'navigation') and vehicle.navigation is not None:
current_lane = vehicle.navigation.current_lane
if self.debug_traffic_light:
print(f" 方法1-导航模块:")
print(f" current_lane = {current_lane}")
print(f" lane_index = {current_lane.index if current_lane else 'None'}")
if current_lane:
has_light = self.engine.light_manager.has_traffic_light(current_lane.index)
if self.debug_traffic_light:
print(f" has_traffic_light = {has_light}")
if has_light:
status = self.engine.light_manager._lane_index_to_obj[current_lane.index].status
if self.debug_traffic_light:
print(f" status = {status}")
if status == 'TRAFFIC_LIGHT_GREEN':
if self.debug_traffic_light:
print(f" ✅ 方法1成功: 绿灯")
return 1
elif status == 'TRAFFIC_LIGHT_YELLOW':
if self.debug_traffic_light:
print(f" ✅ 方法1成功: 黄灯")
return 2
elif status == 'TRAFFIC_LIGHT_RED':
if self.debug_traffic_light:
print(f" ✅ 方法1成功: 红灯")
return 3
elif status is None:
if self.debug_traffic_light:
print(f" ⚠️ 方法1: 红绿灯状态为None")
return 0
else:
if self.debug_traffic_light:
print(f" 该车道没有红绿灯")
else:
if self.debug_traffic_light:
print(f" 导航模块current_lane为None")
else:
if self.debug_traffic_light:
has_nav = hasattr(vehicle, 'navigation')
nav_not_none = vehicle.navigation is not None if has_nav else False
print(f" 方法1-导航模块: 不可用 (hasattr={has_nav}, not_none={nav_not_none})")
except Exception as e:
if self.debug_traffic_light:
print(f" ❌ 方法1异常: {type(e).__name__}: {e}")
pass
try:
# 方法2：遍历所有车道查找（兜底方案，处理车道分段问题）
if self.debug_traffic_light:
print(f" 方法2-遍历车道: 开始遍历 {len(self.lanes)} 条车道")
found_lane = False
checked_lanes = 0
for lane in self.lanes.values():
try:
checked_lanes += 1
if lane.lane.point_on_lane(position_2d):
found_lane = True
if self.debug_traffic_light:
print(f" ✓ 找到车辆所在车道: {lane.lane.index} (检查了{checked_lanes}条)")
has_light = self.engine.light_manager.has_traffic_light(lane.lane.index)
if self.debug_traffic_light:
print(f" has_traffic_light = {has_light}")
if has_light:
status = self.engine.light_manager._lane_index_to_obj[lane.lane.index].status
if self.debug_traffic_light:
print(f" status = {status}")
if status == 'TRAFFIC_LIGHT_GREEN':
if self.debug_traffic_light:
print(f" ✅ 方法2成功: 绿灯")
return 1
elif status == 'TRAFFIC_LIGHT_YELLOW':
if self.debug_traffic_light:
print(f" ✅ 方法2成功: 黄灯")
return 2
elif status == 'TRAFFIC_LIGHT_RED':
if self.debug_traffic_light:
print(f" ✅ 方法2成功: 红灯")
return 3
elif status is None:
if self.debug_traffic_light:
print(f" ⚠️ 方法2: 红绿灯状态为None")
return 0
else:
if self.debug_traffic_light:
print(f" 该车道没有红绿灯")
break
except:
continue
if self.debug_traffic_light and not found_lane:
print(f" ⚠️ 未找到车辆所在车道 (检查了{checked_lanes}条)")
except Exception as e:
if self.debug_traffic_light:
print(f" ❌ 方法2异常: {type(e).__name__}: {e}")
pass
if self.debug_traffic_light:
print(f" 结果: 返回 {traffic_light} (无红绿灯/未知)")
return traffic_light
def _get_all_obs(self):
# position, velocity, heading, lidar, navigation, TODO: trafficlight -> list
self.obs_list = []
# 遍历所有可控智能体
for agent_id, vehicle in self.controlled_agents.items():
# 获取车辆基本状态信息
state = vehicle.get_state()
# 使用优化后的红绿灯检测方法
traffic_light = self._get_traffic_light_state(vehicle)
# 获取激光雷达传感器数据
# 前向激光雷达：80个激光束，30米检测距离，用于障碍物检测
lidar = self.engine.get_sensor("lidar").perceive(
num_lasers=80,
distance=30,
base_vehicle=vehicle,
physics_world=self.engine.physics_world.dynamic_world
)
# 侧向激光雷达：10个激光束，8米检测距离，用于侧向障碍物检测
side_lidar = self.engine.get_sensor("side_detector").perceive(
num_lasers=10,
distance=8,
base_vehicle=vehicle,
physics_world=self.engine.physics_world.static_world
)
# 车道线检测激光雷达：10个激光束，3米检测距离，用于车道线识别
lane_line_lidar = self.engine.get_sensor("lane_line_detector").perceive(
num_lasers=10,
distance=3,
base_vehicle=vehicle,
physics_world=self.engine.physics_world.static_world
)
lidar = self.engine.get_sensor("lidar").perceive(num_lasers=80, distance=30, base_vehicle=vehicle,
physics_world=self.engine.physics_world.dynamic_world)
side_lidar = self.engine.get_sensor("side_detector").perceive(num_lasers=10, distance=8,
base_vehicle=vehicle,
physics_world=self.engine.physics_world.static_world)
lane_line_lidar = self.engine.get_sensor("lane_line_detector").perceive(num_lasers=10, distance=3,
base_vehicle=vehicle,
physics_world=self.engine.physics_world.static_world)
obs = (state['position'][:2] + list(state['velocity']) + [state['heading_theta']]
+ lidar[0] + side_lidar[0] + lane_line_lidar[0] + [traffic_light]
+ list(vehicle.destination))
# 组装完整的观测向量
obs = (state['position'][:2] + # 位置 (x, y)
list(state['velocity']) + # 速度 (vx, vy)
[state['heading_theta']] + # 朝向角度
lidar[0] + # 前向激光雷达数据
side_lidar[0] + # 侧向激光雷达数据
lane_line_lidar[0] + # 车道线检测数据
list(vehicle.destination)) # 目标目的地 (x, y)
self.obs_list.append(obs)
return self.obs_list
def step(self, action_dict: Dict[AnyStr, Union[list, np.ndarray]]):
"""
执行环境的一个时间步
这是环境的核心步进函数，执行以下操作序列：
1. 更新时间步计数器
2. 执行所有智能体的动作（before_step）
3. 更新MetaDrive物理引擎状态
4. 执行智能体动作后的处理（after_step）
5. 生成新的智能体（按时间步）
6. 获取新的观测数据
7. 计算奖励和完成状态
8. 返回环境状态
Args:
action_dict: 智能体动作字典 {agent_id: action}
Returns:
tuple: (观测数据, 奖励字典, 完成状态字典, 信息字典)
- obs: 所有智能体的观测数据列表
- rewards: 每个智能体的奖励 {agent_id: reward}
- dones: 每个智能体的完成状态 {agent_id: done, "__all__": episode_done}
- infos: 每个智能体的额外信息 {agent_id: info}
"""
# 步骤1：更新时间步计数器
self.round += 1
# 步骤2：执行所有智能体的动作（动作执行前处理）
for agent_id, action in action_dict.items():
if agent_id in self.controlled_agents:
# 记录车辆状态并设置动作
self.controlled_agents[agent_id].before_step(action)
# 步骤3：更新MetaDrive物理引擎状态
# 这是核心的物理仿真步骤，所有车辆状态都会根据动作更新
self.engine.step()
# 步骤4：执行智能体动作后的处理
for agent_id in action_dict:
if agent_id in self.controlled_agents:
# 执行动作后的状态更新（如果有after_step方法）
self.controlled_agents[agent_id].after_step()
# 步骤5：生成新的智能体（按时间步动态生成）
self._spawn_controlled_agents()
# 步骤6：获取新的观测数据
obs = self._get_all_obs()
# 步骤7：计算奖励和完成状态
# 初始化所有智能体的奖励为0（可根据需要实现奖励计算逻辑）
rewards = {aid: 0.0 for aid in self.controlled_agents}
# 初始化所有智能体的完成状态为False（可根据需要实现完成条件）
dones = {aid: False for aid in self.controlled_agents}
# 检查整个回合是否结束（达到最大步数）
dones["__all__"] = self.episode_step >= self.config["horizon"]
# 初始化所有智能体的额外信息为空字典
infos = {aid: {} for aid in self.controlled_agents}
# 步骤8：返回环境状态
return obs, rewards, dones, infos

219
Env/test_lane_filter.py
View File

@@ -1,219 +0,0 @@
"""
测试车道过滤和红绿灯检测功能
"""
from scenario_env import MultiAgentScenarioEnv
from simple_idm_policy import ConstantVelocityPolicy
from metadrive.engine.asset_loader import AssetLoader
from logger_utils import setup_logger
import os
WAYMO_DATA_DIR = r"/home/huangfukk/MAGAIL4AutoDrive/Env"
def test_lane_filter():
"""测试车道过滤功能（基础版）"""
print("=" * 60)
print("测试1：车道过滤功能（基础）")
print("=" * 60)
# 创建启用过滤的环境
env = MultiAgentScenarioEnv(
config={
"data_directory": AssetLoader.file_path(WAYMO_DATA_DIR, "exp_converted", unix_style=False),
"is_multi_agent": True,
"num_controlled_agents": 3,
"horizon": 100,
"use_render": False,
# 车道过滤配置
"filter_offroad_vehicles": True,
"lane_tolerance": 3.0,
"max_controlled_vehicles": 10,
},
agent2policy=ConstantVelocityPolicy(target_speed=50)
)
print("\n启用车道过滤...")
obs = env.reset(0)
print(f"生成车辆数: {len(env.controlled_agents)}")
print(f"观测数据长度: {len(obs)}")
# 运行几步
for step in range(5):
actions = {aid: env.controlled_agents[aid].policy.act()
for aid in env.controlled_agents}
obs, rewards, dones, infos = env.step(actions)
env.close()
print("✓ 车道过滤测试通过\n")
def test_lane_filter_debug():
"""测试车道过滤功能（详细调试）"""
print("=" * 60)
print("测试1b：车道过滤功能（详细调试模式）")
print("=" * 60)
env = MultiAgentScenarioEnv(
config={
"data_directory": AssetLoader.file_path(WAYMO_DATA_DIR, "exp_converted", unix_style=False),
"is_multi_agent": True,
"num_controlled_agents": 3,
"horizon": 100,
"use_render": False,
# 车道过滤配置
"filter_offroad_vehicles": True,
"lane_tolerance": 3.0,
"max_controlled_vehicles": 5, # 只看前5辆车
# 🔥 启用调试模式
"debug_lane_filter": True, # 启用车道过滤调试
},
agent2policy=ConstantVelocityPolicy(target_speed=50)
)
print("\n启用车道过滤调试...")
obs = env.reset(0)
env.close()
print("\n✓ 车道过滤调试测试完成\n")
def test_traffic_light():
"""测试红绿灯检测功能"""
print("=" * 60)
print("测试2：红绿灯检测功能（启用详细调试）")
print("=" * 60)
env = MultiAgentScenarioEnv(
config={
"data_directory": AssetLoader.file_path(WAYMO_DATA_DIR, "exp_converted", unix_style=False),
"is_multi_agent": True,
"num_controlled_agents": 3,
"horizon": 100,
"use_render": False,
"filter_offroad_vehicles": True,
"max_controlled_vehicles": 3, # 只测试3辆车
# 🔥 启用调试模式
"debug_traffic_light": True, # 启用红绿灯调试
},
agent2policy=ConstantVelocityPolicy(target_speed=50)
)
obs = env.reset(0)
# 测试红绿灯检测（调试模式会自动输出详细信息）
print(f"\n" + "="*60)
print(f"开始逐车检测红绿灯状态（共 {len(env.controlled_agents)} 辆车）")
print("="*60)
for idx, (aid, vehicle) in enumerate(list(env.controlled_agents.items())[:3]): # 只测试前3辆
print(f"\n【车辆 {idx+1}/3】 ID={aid}")
traffic_light = env._get_traffic_light_state(vehicle)
state = vehicle.get_state()
status_text = {0: '无/未知', 1: '绿灯', 2: '黄灯', 3: '红灯'}[traffic_light]
print(f"最终结果: 红绿灯状态={traffic_light} ({status_text})\n")
env.close()
print("="*60)
print("✓ 红绿灯检测测试完成")
print("="*60 + "\n")
def test_without_filter():
"""测试禁用过滤的情况"""
print("=" * 60)
print("测试3：禁用过滤（对比测试）")
print("=" * 60)
env = MultiAgentScenarioEnv(
config={
"data_directory": AssetLoader.file_path(WAYMO_DATA_DIR, "exp_converted", unix_style=False),
"is_multi_agent": True,
"num_controlled_agents": 3,
"horizon": 100,
"use_render": False,
# 禁用过滤
"filter_offroad_vehicles": False,
"max_controlled_vehicles": None,
},
agent2policy=ConstantVelocityPolicy(target_speed=50)
)
print("\n禁用车道过滤...")
obs = env.reset(0)
print(f"生成车辆数（未过滤）: {len(env.controlled_agents)}")
env.close()
print("✓ 禁用过滤测试通过\n")
def run_tests(debug_mode=False):
"""运行测试的主函数"""
try:
if debug_mode:
print("🐛 调试模式启用")
print("=" * 60 + "\n")
test_lane_filter_debug()
test_traffic_light()
else:
print("⚡ 标准测试模式（使用 --debug 参数启用详细调试）")
print("=" * 60 + "\n")
test_lane_filter()
test_traffic_light()
test_without_filter()
print("\n" + "=" * 60)
print("✅ 所有测试通过！")
print("=" * 60)
print("\n功能说明：")
print("1. 车道过滤功能已启用，自动过滤非车道区域车辆")
print("2. 红绿灯检测采用双重策略，确保稳定获取状态")
print("3. 可通过配置参数灵活启用/禁用功能")
print("\n使用方法：")
print(" python Env/test_lane_filter.py # 标准测试")
print(" python Env/test_lane_filter.py --debug # 详细调试")
print(" python Env/test_lane_filter.py --log # 保存日志")
print(" python Env/test_lane_filter.py --debug --log # 调试+日志")
print("\n请运行 run_multiagent_env.py 查看完整效果")
except Exception as e:
print(f"\n❌ 测试失败: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
import sys
# 解析命令行参数
debug_mode = "--debug" in sys.argv or "-d" in sys.argv
enable_logging = "--log" in sys.argv or "-l" in sys.argv
# 提取自定义日志文件名
log_file = None
for arg in sys.argv:
if arg.startswith("--log-file="):
log_file = arg.split("=")[1]
break
if enable_logging:
# 启用日志记录
log_dir = os.path.join(os.path.dirname(__file__), "logs")
# 生成默认日志文件名
if log_file is None:
mode_suffix = "debug" if debug_mode else "standard"
from datetime import datetime
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
log_file = f"test_{mode_suffix}_{timestamp}.log"
with setup_logger(log_file=log_file, log_dir=log_dir):
run_tests(debug_mode=debug_mode)
else:
# 不启用日志，直接运行
run_tests(debug_mode=debug_mode)

76
Env/test_no_traffic_lights.py Normal file
View File

@@ -0,0 +1,76 @@
"""测试禁用红绿灯功能"""
from scenario_env import MultiAgentScenarioEnv
from simple_idm_policy import ConstantVelocityPolicy
from metadrive.engine.asset_loader import AssetLoader
WAYMO_DATA_DIR = r"/home/huangfukk/MAGAIL4AutoDrive/Env"
def test_no_traffic_lights():
"""测试禁用红绿灯"""
print("=" * 60)
print("测试：禁用红绿灯功能")
print("=" * 60)
env = MultiAgentScenarioEnv(
config={
"data_directory": AssetLoader.file_path(WAYMO_DATA_DIR, "exp_converted", unix_style=False),
"is_multi_agent": True,
"num_controlled_agents": 3,
"horizon": 300,
"use_render": True,
"sequential_seed": True,
"reactive_traffic": True,
"manual_control": True,
# 车道检测与过滤配置
"filter_offroad_vehicles": True,
"lane_tolerance": 3.0,
"max_controlled_vehicles": 2,
# 禁用红绿灯
"no_traffic_lights": True, # 关键配置
# 调试模式
"debug_lane_filter": False,
"verbose_reset": False,
},
agent2policy=ConstantVelocityPolicy(target_speed=50)
)
print("\n重置环境...")
obs = env.reset(0)
# 检查红绿灯管理器状态
if hasattr(env.engine, 'light_manager') and env.engine.light_manager is not None:
num_lights = len(env.engine.light_manager._lane_index_to_obj)
print(f"✓ 红绿灯管理器中的红绿灯数量: {num_lights}")
if num_lights == 0:
print("✅ 成功：所有红绿灯已被移除！")
else:
print(f"⚠️ 警告：仍有 {num_lights} 个红绿灯")
print("\n运行几步测试...")
for step in range(100):
actions = {
aid: env.controlled_agents[aid].policy.act()
for aid in env.controlled_agents
}
obs, rewards, dones, infos = env.step(actions)
env.render(mode="topdown")
if step == 0:
print(f"步骤 {step}: 环境运行正常")
if dones["__all__"]:
break
print(f"\n测试完成，共运行 {step+1} 步")
print("请检查渲染窗口中是否还有红绿灯显示")
env.close()
print("=" * 60)
if __name__ == "__main__":
test_no_traffic_lights()