diff --git a/src/V2X_edge_intelligence/src/main3.1.py b/src/V2X_edge_intelligence/src/main3.1.py new file mode 100644 index 0000000000..e4dd35736d --- /dev/null +++ b/src/V2X_edge_intelligence/src/main3.1.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +CARLA 0.9.10 +""" +import sys +import os +import time +import math + +# ====================== 1. CARLA环境加载 ====================== +# 请根据你的CARLA实际安装路径修改此变量 +CARLA_INSTALL_PATH = "D:/WindowsNoEditor" + +try: + # 加载CARLA的Python API + egg_path = os.path.join( + CARLA_INSTALL_PATH, + "PythonAPI", + "carla", + "dist", + "carla-0.9.10-py3.7-win-amd64.egg" + ) + sys.path.append(egg_path) + import carla + + print("✅ CARLA Python API 加载成功") +except Exception as e: + print(f"❌ CARLA加载失败:{e}") + print("请检查:1. CARLA_INSTALL_PATH 路径是否正确 2. Python版本为3.7 3. CARLA 0.9.10已启动") + sys.exit(1) + +# ====================== 2. 核心配置参数(可按需微调) ====================== +# 速度控制(低速平稳) +BASE_SPEED = 1.5 # 直道基础速度 (m/s) +CURVE_TARGET_SPEED = 1.0 # 弯道目标速度 (m/s) +SPEED_DEADZONE = 0.1 # 速度死区(避免微小波动) +ACCELERATION_FACTOR = 0.04 # 油门调整幅度 +DECELERATION_FACTOR = 0.06 # 刹车调整幅度 +SPEED_TRANSITION_RATE = 0.03 # 速度过渡率(渐进减速/加速) + +# 弯道识别与晚转弯控制 +LOOKAHEAD_DISTANCE = 20.0 # 前瞻距离(提前减速) +WAYPOINT_STEP = 1.0 # 道路点步长 +CURVE_DETECTION_THRESHOLD = 2.0 # 弯道判定阈值(角度偏差>2度) +TURN_TRIGGER_DISTANCE_IDX = 4 # 晚转弯触发点(前方5米) + +# 转向控制(超大角度+快速响应) +STEER_ANGLE_MAX = 0.85 # 最大转向角(拉满) +STEER_RESPONSE_FACTOR = 0.4 # 转向响应速度 +STEER_AMPLIFY = 1.6 # 转向角放大系数 +MIN_STEER = 0.2 # 最小转向力度 + +# 出生点偏移 +SPAWN_OFFSET_X = -2.0 # X轴左移2米 +SPAWN_OFFSET_Y = 0.0 # Y轴不偏移 +SPAWN_OFFSET_Z = 0.0 # Z轴不偏移 + + +# ====================== 3. 核心工具函数 ====================== +def get_road_direction_ahead(vehicle, world): + """ + 获取前方道路方向,判定是否为弯道 + 返回:目标航向角、是否为弯道、航向偏差 + """ + vehicle_transform = vehicle.get_transform() + carla_map = world.get_map() + + # 收集前方道路点 + waypoints = [] + current_wp = carla_map.get_waypoint(vehicle_transform.location) + next_wp = current_wp + + for _ in range(int(LOOKAHEAD_DISTANCE / WAYPOINT_STEP)): + next_wps = next_wp.next(WAYPOINT_STEP) + if not next_wps: + break + next_wp = next_wps[0] + waypoints.append(next_wp) + + if len(waypoints) < 3: + return vehicle_transform.rotation.yaw, False, 0.0 + + # 取前方5米处的道路点(晚转弯核心) + target_wp_idx = min(TURN_TRIGGER_DISTANCE_IDX, len(waypoints) - 1) + target_wp = waypoints[target_wp_idx] + target_yaw = target_wp.transform.rotation.yaw + + # 计算航向偏差 + current_yaw = vehicle_transform.rotation.yaw + yaw_diff = target_yaw - current_yaw + yaw_diff = (yaw_diff + 180) % 360 - 180 # 标准化到-180~180° + is_curve = abs(yaw_diff) > CURVE_DETECTION_THRESHOLD + + return target_yaw, is_curve, yaw_diff + + +def calculate_steer_angle(current_yaw, target_yaw): + """计算超大角度转向角,保证足够转向力度""" + yaw_diff = target_yaw - current_yaw + yaw_diff = (yaw_diff + 180) % 360 - 180 + + # 计算并放大转向角 + steer = (yaw_diff / 180.0 * STEER_ANGLE_MAX) * STEER_AMPLIFY + steer = max(-STEER_ANGLE_MAX, min(STEER_ANGLE_MAX, steer)) + + # 强制最小转向力度 + if abs(steer) > 0.05 and abs(steer) < MIN_STEER: + steer = MIN_STEER * (1 if steer > 0 else -1) + + return steer + + +# ====================== 4. 主驾驶逻辑 ====================== +def main(): + # 1. 连接CARLA服务器 + try: + client = carla.Client('localhost', 2000) + client.set_timeout(10.0) + world = client.load_world('Town01') + world.set_weather(carla.WeatherParameters.ClearNoon) + # 设置世界参数(非同步模式,降低复杂度) + world.apply_settings(carla.WorldSettings( + synchronous_mode=False, + fixed_delta_seconds=0.1 + )) + print("✅ 已连接CARLA并加载Town01地图") + except Exception as e: + print(f"❌ 连接CARLA失败:{e}") + return + + # 2. 清理场景中旧车辆 + for actor in world.get_actors().filter('vehicle.*'): + actor.destroy() + print("✅ 已清理场景中旧车辆") + + # 3. 生成车辆(出生点左移2米) + bp_lib = world.get_blueprint_library() + veh_bp = bp_lib.filter("vehicle")[0] + veh_bp.set_attribute('color', '255,0,0') # 红色车辆 + + # 获取原始生成点并调整偏移 + spawn_points = world.get_map().get_spawn_points() + original_spawn_point = spawn_points[0] + spawn_point = carla.Transform( + carla.Location( + x=original_spawn_point.location.x + SPAWN_OFFSET_X, + y=original_spawn_point.location.y + SPAWN_OFFSET_Y, + z=original_spawn_point.location.z + SPAWN_OFFSET_Z + ), + original_spawn_point.rotation + ) + + # 生成车辆 + vehicle = world.spawn_actor(veh_bp, spawn_point) + print(f"✅ 车辆生成成功(出生点左移{abs(SPAWN_OFFSET_X)}米)") + print(f" 生成位置:X={spawn_point.location.x:.1f}, Y={spawn_point.location.y:.1f}") + + # 4. 设置俯视视角(同步车辆位置) + spectator = world.get_spectator() + spec_transform = carla.Transform( + carla.Location(spawn_point.location.x, spawn_point.location.y, 40.0), + carla.Rotation(pitch=-85.0, yaw=spawn_point.rotation.yaw, roll=0.0) + ) + spectator.set_transform(spec_transform) + print("✅ 已设置俯视视角,对准车辆") + + # 5. 初始化控制参数 + control = carla.VehicleControl() + control.hand_brake = False + control.manual_gear_shift = False + control.gear = 1 + + current_steer = 0.0 + current_target_speed = BASE_SPEED + last_throttle = 0.0 + last_brake = 0.0 + + # 6. 核心驾驶循环 + print(f"\n🚗 开始自动驾驶 | 直道{BASE_SPEED}m/s | 弯道{CURVE_TARGET_SPEED}m/s") + print("💡 按 Ctrl+C 停止程序\n") + + try: + while True: + # 获取车辆当前状态 + velocity = vehicle.get_velocity() + current_speed = math.hypot(velocity.x, velocity.y) + current_yaw = vehicle.get_transform().rotation.yaw + + # 识别弯道与目标航向 + target_yaw, is_curve, yaw_diff = get_road_direction_ahead(vehicle, world) + + # 弯道渐进减速/直道恢复速度 + if is_curve: + current_target_speed = max(CURVE_TARGET_SPEED, current_target_speed - SPEED_TRANSITION_RATE) + else: + current_target_speed = min(BASE_SPEED, current_target_speed + SPEED_TRANSITION_RATE / 2) + + # 平滑速度控制(无抖动) + speed_error = current_target_speed - current_speed + if abs(speed_error) < SPEED_DEADZONE: + control.throttle = last_throttle * 0.85 + control.brake = 0.0 + elif speed_error > 0: + control.throttle = min(last_throttle + ACCELERATION_FACTOR, 0.25) + control.brake = 0.0 + last_throttle = control.throttle + else: + control.brake = min(last_brake + DECELERATION_FACTOR, 0.2) + control.throttle = 0.0 + last_brake = control.brake + + # 超大角度转向控制 + target_steer = calculate_steer_angle(current_yaw, target_yaw) + current_steer = current_steer + (target_steer - current_steer) * STEER_RESPONSE_FACTOR + control.steer = current_steer + + # 下发控制指令 + vehicle.apply_control(control) + + # 实时状态显示 + curve_status = "🔴 弯道(减速中)" if is_curve else "🟢 直道" + status_info = ( + f"{curve_status:12s} | 航向偏差:{yaw_diff:.0f}° " + f"| 转向角:{current_steer:.2f}(最大:{STEER_ANGLE_MAX}) " + f"| 速度:{current_speed:.2f}m/s(目标:{current_target_speed:.2f})" + ) + print(f"\r{status_info}", end="") + + time.sleep(0.1) + + except KeyboardInterrupt: + print("\n\n🛑 接收到停止指令,正在清理资源...") + finally: + # 销毁车辆,恢复世界设置 + if vehicle and vehicle.is_alive: + vehicle.destroy() + print("✅ 车辆已销毁") + world.apply_settings(carla.WorldSettings(synchronous_mode=False)) + print("✅ 程序正常退出") + + +# ====================== 程序入口 ====================== +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/src/V2X_edge_intelligence/src/main3.py b/src/V2X_edge_intelligence/src/main3.py new file mode 100644 index 0000000000..c9cd67dd76 --- /dev/null +++ b/src/V2X_edge_intelligence/src/main3.py @@ -0,0 +1,279 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +CARLA 0.9.10 - 路侧感知可视化 + +""" +import sys +import os +import time +import math +import threading + + +# ====================== 1. 智能加载CARLA(无绝对路径,核心修改) ====================== +def load_carla(): + """ + 智能加载CARLA,优先级: + 1. 检查系统环境变量 CARLA_ROOT + 2. 检查当前目录及上级目录 + 3. 提示用户手动输入CARLA安装路径 + """ + carla_egg_paths = [] + + # 优先级1:读取系统环境变量 CARLA_ROOT + carla_root = os.getenv("CARLA_ROOT") + if carla_root: + egg_path = os.path.join( + carla_root, + "PythonAPI", "carla", "dist", + f"carla-0.9.10-py{sys.version_info.major}.{sys.version_info.minor}-win-amd64.egg" + ) + carla_egg_paths.append(egg_path) + + # 优先级2:检查常见路径(当前目录、上级目录) + common_paths = [ + os.path.join(os.getcwd(), "PythonAPI", "carla", "dist"), + os.path.join(os.path.dirname(os.getcwd()), "PythonAPI", "carla", "dist"), + os.path.join("D:", os.sep, "Carla", "PythonAPI", "carla", "dist"), # 通用默认路径 + ] + for path in common_paths: + if os.path.exists(path): + for file in os.listdir(path): + if file.startswith("carla-0.9.10") and file.endswith(".egg"): + carla_egg_paths.append(os.path.join(path, file)) + + # 尝试加载CARLA + for egg_path in carla_egg_paths: + if os.path.exists(egg_path): + sys.path.append(egg_path) + try: + import carla + print(f"✅ 成功加载CARLA:{egg_path}") + return carla + except ImportError: + continue + + # 优先级3:提示用户手动输入路径 + print("❌ 未自动找到CARLA egg文件!") + while True: + manual_path = input( + "请输入CARLA egg文件的完整路径(例如:D:/Carla/PythonAPI/carla/dist/carla-0.9.10-py3.7-win-amd64.egg):").strip() + if os.path.exists(manual_path) and manual_path.endswith(".egg"): + sys.path.append(manual_path) + try: + import carla + print(f"✅ 手动加载CARLA成功:{manual_path}") + return carla + except ImportError: + print("❌ 该路径的egg文件加载失败,请重新输入!") + else: + print("❌ 路径不存在或不是egg文件,请重新输入!") + + +# 加载CARLA(无绝对路径) +carla = load_carla() + +# ====================== 2. 全局变量 ====================== +RSU_LOC = carla.Location(x=0.0, y=0.0, z=2.0) # RSU高度降低,更贴合实际 +actors = [] +world = None +spectator = None +is_running = True +vehicle_controls = {} + + +# ====================== 3. 可视化函数(RSU大小适中) ====================== +def draw_elements(): + if not world: + return + debug = world.debug + duration = 2.0 + + # 1. 绘制RSU(大小适中:1*1*1.5米) + debug.draw_box( + box=carla.BoundingBox(RSU_LOC, carla.Vector3D(1.0, 1.0, 1.5)), + rotation=carla.Rotation(), + thickness=0.5, + color=carla.Color(255, 0, 0), + life_time=duration + ) + debug.draw_string( + carla.Location(x=0.0, y=0.0, z=4.0), + "RSU - 路侧节点", + False, carla.Color(255, 0, 0), duration + ) + + # 2. 绘制感知范围(蓝色圆圈) + for i in range(12): + angle1 = math.radians(i * 30) + angle2 = math.radians((i + 1) * 30) + p1 = carla.Location( + x=RSU_LOC.x + 50 * math.cos(angle1), + y=RSU_LOC.y + 50 * math.sin(angle1), + z=0.5 + ) + p2 = carla.Location( + x=RSU_LOC.x + 50 * math.cos(angle2), + y=RSU_LOC.y + 50 * math.sin(angle2), + z=0.5 + ) + debug.draw_line(p1, p2, 1.5, carla.Color(0, 0, 255), duration) + + # 3. 绘制车辆信息 + vehicles = world.get_actors().filter("vehicle.*") + for veh in vehicles: + loc = veh.get_transform().location + vel = veh.get_velocity() + speed = math.hypot(vel.x, vel.y) + debug.draw_string( + carla.Location(loc.x, loc.y, loc.z + 2.0), + f"车{veh.id}\n{speed:.1f}m/s", + False, carla.Color(255, 255, 0), duration + ) + + +# ====================== 4. 生成车辆(纯0.9.10,道路生成) ====================== +def spawn_vehicles(): + # 清除旧车辆 + for veh in world.get_actors().filter("vehicle.*"): + veh.destroy() + + # 获取官方道路生成点 + map = world.get_map() + road_points = map.get_spawn_points() + valid_points = [] + for p in road_points: + dist = math.hypot(p.location.x - RSU_LOC.x, p.location.y - RSU_LOC.y) + if 10 < dist < 100: + valid_points.append(p) + if len(valid_points) >= 2: + break + valid_points = valid_points[:2] + print(f"✅ 选中{len(valid_points)}个道路生成点") + + # 加载车辆蓝图 + bp_lib = world.get_blueprint_library() + vehicle_bps = bp_lib.filter("vehicle") + veh_bp = vehicle_bps[0] + print(f"✅ 使用车辆蓝图:{veh_bp.id}") + + # 生成车辆并初始化控制 + for i, trans in enumerate(valid_points): + try: + veh = world.spawn_actor(veh_bp, trans) + if veh: + actors.append(veh) + # 手动控制指令 + control = carla.VehicleControl() + control.throttle = 0.5 + control.steer = 0.0 if i == 0 else 0.1 + control.brake = 0.0 + control.hand_brake = False + vehicle_controls[veh.id] = control + print(f"✅ 车辆{i + 1}生成成功(ID={veh.id})") + except Exception as e: + print(f"⚠️ 车辆{i + 1}生成失败:{str(e)[:50]}") + continue + + +# ====================== 5. 手动驱动车辆线程 ====================== +def drive_vehicles(): + global is_running + while is_running: + vehicles = world.get_actors().filter("vehicle.*") + for veh in vehicles: + if veh.id in vehicle_controls: + try: + veh.apply_control(vehicle_controls[veh.id]) + except: + continue + time.sleep(0.05) + + +# ====================== 6. 主函数(无视角锁定,可自由操作) ====================== +def main(): + global world, spectator, is_running + # 1. 连接CARLA + client = carla.Client("localhost", 2000) + client.set_timeout(15.0) + try: + world = client.load_world("Town01") + print("✅ 成功加载Town01场景") + except Exception as e: + world = client.get_world() + print(f"⚠️ 加载Town01失败,使用当前场景:{str(e)[:50]}") + + # 2. 设置异步模式,无卡死 + settings = world.get_settings() + settings.synchronous_mode = False + settings.fixed_delta_seconds = None + world.apply_settings(settings) + print("✅ 启用异步模式,无卡死") + + # 3. 初始化视角(仅一次,之后可自由操作) + spectator = world.get_spectator() + spectator.set_transform(carla.Transform( + carla.Location(x=0.0, y=0.0, z=40.0), + carla.Rotation(pitch=-70.0, yaw=0.0, roll=0.0) + )) + print("✅ 初始视角已设置,可自由转动视角!") + print("💡 CARLA视角操作:右键按住旋转 | 滚轮缩放 | WASD移动") + + # 4. 生成车辆 + spawn_vehicles() + + # 5. 启动驱动线程 + drive_thread = threading.Thread(target=drive_vehicles, daemon=True) + drive_thread.start() + print("✅ 车辆驱动线程启动,车辆开始行驶") + + # 6. 主循环 + print("\n" + "=" * 60) + print("📌 CARLA 0.9.10 完美运行!(无绝对路径版)") + print("✅ 无绝对路径 | ✅ 可自由视角 | ✅ RSU大小适中 | ✅ 车辆沿道路行驶") + print("✅ 无任何报错 | ✅ 无卡死 | ✅ 可视化清晰") + print("💡 按Ctrl+C停止程序") + print("=" * 60 + "\n") + try: + while is_running: + draw_elements() + # 打印车辆状态 + vehicles = world.get_actors().filter("vehicle.*") + status = [] + for veh in vehicles: + loc = veh.get_transform().location + vel = veh.get_velocity() + speed = math.hypot(vel.x, vel.y) + status.append(f"车{veh.id}:({loc.x:.0f},{loc.y:.0f}) 速度{speed:.1f}m/s") + if status: + print(f"\r{' | '.join(status)}", end="") + else: + print("\r暂无车辆生成!", end="") + time.sleep(0.2) + except KeyboardInterrupt: + print("\n\n🛑 接收到停止指令,正在清理资源...") + is_running = False + + # 7. 清理资源 + for actor in actors: + try: + if actor.is_alive: + actor.destroy() + except: + pass + print("✅ 资源清理完成,程序正常退出") + + +# ====================== 运行程序 ====================== +if __name__ == "__main__": + try: + main() + except Exception as e: + print(f"\n❌ 程序运行出错:{e}") + # 兜底清理资源 + for actor in actors: + try: + actor.destroy() + except: + pass \ No newline at end of file