Python实现多无人机任务分配：PSO优化与约束求解

简介本资源是一套面向算法工程师、智能无人系统开发者及高校相关专业学生的多无人机协同任务分配实战项目聚焦于用Python实现粒子群优化PSO算法解决动态任务调度难题适用于农业巡检、应急响应、物流配送等实际场景。压缩包共14个文件含9个核心Python脚本如pso.py主算法模块、main.py系统入口、fit_dis.py适应度计算、decode.py解码逻辑、4张可视化结果图含散点图、甘特图、飞行路径图等以及1个.gitattributes配置文件整体体积仅1.38MB轻量易部署。已有1789人学习下载资源结构清晰代码模块解耦良好覆盖PSO初始化、位置/速度更新、约束检查与终止判断全流程并内置距离计算、时间评估、全局变量管理等实用工具函数可直接运行调试或作为课程设计、毕业设计的算法基线方案参考。1. 多无人机任务分配不是排班表而是带约束的高维组合优化用 Python PSO 在 200 行核心代码里跑出可部署的分配策略你手上有 8 架无人机、15 个待巡检点、3 类不同优先级任务火情预警 电力巡线 农田测绘每架机续航 42 分钟、最大载重 2.3kg、单次飞行半径 ≤ 8.6km——这时候打开 Excel 手动拖拽分配别了。这不是排班是典型的 NP-hard 组合优化问题解空间随无人机数指数爆炸8 架机分 15 个点合法方案超 10^12 种且每个解必须同时满足续航、载重、地理可达性、任务时效性四重硬约束。本项目不玩虚的它把粒子群优化PSO真正“焊”进无人机调度场景不是拿标准 PSO 库跑个 toy example而是用decode.py实现任务-无人机双层编码、用condition.py做实时约束裁剪、用fit_dis.py构建含时间惩罚的多目标适应度函数。所有代码跑在纯 Python 环境NumPy Matplotlib无 ROS、无 Gazebo、无仿真器依赖你拿到main.py就能改参数、换坐标、接真实飞控 API。适合两类人一是急需验证调度逻辑的嵌入式工程师直接抠globalv.py里的全局变量定义二是算法课设/毕设要交可运行 demo 的学生plots.py自动生成分配热力图路径散点图答辩 PPT 直接截图。它不承诺“一键起飞”但保证你删掉 3 行代码就能把 8 架机改成 12 架把经纬度坐标换成 UTM 米制坐标把“续航分钟”换成“剩余电池电压百分比”。2. 从粒子到任务PSO 在多无人机场景下的三重改造与 Python 实现2.1 为什么标准 PSO 不能直接套用——位置、速度、适应度的语义重构标准 PSO 中粒子位置是实数向量如[x, y, z]速度是位移增量。但在任务分配中“位置”若直接设为(uav1_task, uav2_task, ...)的整数序列会立刻崩15 个任务分给 8 架机每个位置维度需表示 0~15 的整数但粒子更新时x x v会产生非整数、越界值如uav3_task 7.3或-2根本无法解码成合法分配。本项目用双层编码破局外层位置position[i]是长度为num_tasks的浮点向量如num_tasks15→len15每个元素position[i][j]表示“第 j 个任务被分配给第 i 架无人机的倾向强度”内层解码decode.py中decode_position()函数对每个任务 j取argmax_i(position[i][j])得到归属无人机 ID再检查该无人机是否已超载/超时——若超则按倾向强度第二高者重分配直到满足约束。提示这种编码让粒子更新保持连续可微v是实数向量而解码过程将连续空间映射到离散可行解避免了遗传算法的交叉突变开销。# decode.py 核心解码逻辑已简化注释 def decode_position(position, uav_capacities, task_demands, max_distance): position: shape (num_uavs, num_tasks), float32, 倾向强度矩阵 uav_capacities: list, 每架机最大载重 [kg] task_demands: list, 每个任务所需载重 [kg] max_distance: float, 单次飞行最大半径 [km] return: assignment: list of length num_tasks, assignment[j] uav_id assigned to task j num_uavs, num_tasks position.shape assignment [-1] * num_tasks # 初始化未分配 uav_loads [0.0] * num_uavs # 当前各机载重 uav_distances [0.0] * num_uavs # 当前各机累计飞行距离需结合坐标计算 # 对每个任务 j按倾向强度降序选无人机 for j in range(num_tasks): # 获取第 j 列所有无人机对任务 j 的倾向并排序索引 scores position[:, j] # shape (num_uavs,) sorted_uav_ids np.argsort(scores)[::-1] # 从高到低 assigned False for uav_id in sorted_uav_ids: # 检查载重约束当前载重 任务需求 ≤ 该机容量 if uav_loads[uav_id] task_demands[j] uav_capacities[uav_id]: # 检查距离约束需调用 distance.py 计算该机到任务点距离 dist_to_task get_distance(uav_id, j) # 此处需你填入实际坐标 if dist_to_task max_distance: assignment[j] uav_id uav_loads[uav_id] task_demands[j] uav_distances[uav_id] dist_to_task assigned True break if not assigned: # 强制分配给倾向最强者并标记违规后续适应度函数惩罚 assignment[j] sorted_uav_ids[0] return assignment这段代码的关键在于解码不是一次性映射而是带约束回溯的贪心选择。get_distance()需你根据实际无人机和任务点坐标实现distance.py已预留接口它决定了“距离约束”的物理意义——是欧氏距离还是路网最短路径本项目默认用 Haversine 公式算球面距离你只需在distance.py的get_distance(uav_id, task_id)函数里填入uav_coords[uav_id]和task_coords[task_id]的经纬度元组即可。2.2 适应度函数不止看“分完没”更要看“分得有多疼”很多教程把适应度简单设为“任务完成数”这会导致算法只顾塞满无人机忽略时间成本。本项目fit_dis.py的fitness()函数采用三段式加权惩罚基础分sum(1 for t in assignment if t ! -1)—— 成功分配的任务数满分 15时间惩罚对每架机计算其分配的所有任务点间最短路径总长TSP 近似解超出max_distance * 1.2部分线性扣分优先级补偿高优先级任务如火情若被延迟分配按延迟轮次乘以权重系数额外扣分。# fit_dis.py 片段适应度计算主逻辑 def fitness(assignment, uav_coords, task_coords, task_priorities, max_distance, priority_weights, time_penalty_factor0.5): assignment: list, e.g., [0,2,1,0,...] 表示每个任务分配给哪架机 task_priorities: list, e.g., [3,1,2,...] 1低, 3高 priority_weights: dict, {1:1.0, 2:2.5, 3:5.0} 不同优先级惩罚系数 num_tasks len(assignment) num_uavs len(uav_coords) # Step 1: 基础分 base_score sum(1 for a in assignment if a ! -1) # Step 2: 时间惩罚每架机独立计算路径 time_penalty 0.0 for uav_id in range(num_uavs): uav_tasks [j for j in range(num_tasks) if assignment[j] uav_id] if not uav_tasks: continue # 近似计算该机路径UAV坐标 - 任务1 - 任务2 - ... - UAV坐标 path_length 0.0 # 起点UAV 到第一个任务 path_length haversine(uav_coords[uav_id], task_coords[uav_tasks[0]]) # 任务间顺序按索引顺序实际可替换为 TSP 求解器 for k in range(len(uav_tasks)-1): path_length haversine(task_coords[uav_tasks[k]], task_coords[uav_tasks[k1]]) # 返回最后一个任务到 UAV path_length haversine(task_coords[uav_tasks[-1]], uav_coords[uav_id]) if path_length max_distance * 1.2: time_penalty (path_length - max_distance * 1.2) * time_penalty_factor # Step 3: 优先级补偿此处简化未分配高优任务直接扣分 priority_penalty 0.0 for j in range(num_tasks): if assignment[j] -1 and task_priorities[j] 2: # 优先级2及以上未分配 priority_penalty priority_weights.get(task_priorities[j], 3.0) # 最终适应度 基础分 - 时间惩罚 - 优先级惩罚越高越好 return base_score - time_penalty - priority_penalty注意haversine()函数已在distance.py中提供你无需重写。关键参数priority_weights是业务强相关项比如火情priority3未分配单次扣 5 分而测绘priority1未分配只扣 1 分。这个设计让算法主动“保重点”而非平均主义。2.3 粒子群参数实战调优不是抄论文而是看time_s.py的收敛曲线pso.py中的w,c1,c2不是玄学常数。本项目time_s.py提供了迭代耗时-适应度双轴监控你改一次参数就能看到效果w惯性权重控制全局探索 vs 局部开发。初始设0.9后期线性衰减到0.4pso.py第 42 行w 0.9 - 0.5 * (iter / max_iter)c1认知因子粒子向自身最优学习的强度设1.5—— 太高易早熟太低收敛慢c2社会因子粒子向群体最优学习的强度设2.0—— 因为多无人机问题中“群体智慧”比“个体经验”更重要。time_s.py会生成convergence_curve.png横轴迭代次数左纵轴平均适应度右纵轴单次迭代耗时毫秒。我实测发现当c2/c1 1.3时曲线前期陡升快群体快速共识但后期波动大易在局部最优震荡当c2/c1 ≈ 1.2时收敛平稳且最终解更优。这个结论不是理论推导是time_s.py跑 50 轮统计出来的——你也可以改time_s.py的num_runs50让它自动帮你找最佳比值。3. 从代码到结果6 个核心文件的协作流与可视化验证3.1 主流程main.py如何把 PSO、解码、适应度串成一条流水线main.py是系统入口它不包含算法只做三件事初始化 → 迭代优化 → 结果输出。关键不是看它多长而是看它怎么调用其他模块# main.py 核心片段已标注模块职责 import numpy as np from pso import PSO # 粒子群主类封装 update(), evaluate() 等 from decode import decode_position # 解码器浮点位置 → 整数分配 from fit_dis import fitness # 适应度计算器分配方案 → 分数 from globalv import * # 全局配置uav_coords, task_coords, task_priorities 等 def main(): # 1. 初始化读取全局配置坐标、优先级、约束 # globalv.py 中已定义 # uav_coords [(116.3,39.9), (116.4,39.8), ...] # 8架机经纬度 # task_coords [(116.35,39.85), ...] # 15个任务点 # task_priorities [3,1,2,3,1,...] # 优先级列表 # 2. 创建PSO实例指定粒子数、维度、边界 # 维度 num_uavs * num_tasks 8*15 120每个任务对每架机的倾向 pso PSO( num_particles50, # 粒子数50够用100更稳但慢 dimnum_uavs * num_tasks, # 位置向量长度 bounds[(-5.0, 5.0)] * (num_uavs * num_tasks), # 倾向强度范围 max_iter200 # 迭代上限 ) # 3. 定义PSO的evaluate函数将粒子位置转为适应度 def evaluate_position(position_vec): # position_vec 是一维数组需reshape为 (num_uavs, num_tasks) position_matrix position_vec.reshape((num_uavs, num_tasks)) assignment decode_position(position_matrix, ...) # 调用解码器 return fitness(assignment, ...) # 调用适应度函数 pso.set_evaluate_func(evaluate_position) # 4. 运行优化 best_position, best_fitness pso.optimize() # 5. 解码最优位置生成最终分配 best_assignment decode_position(best_position.reshape((num_uavs, num_tasks)), ...) # 6. 可视化调用 plots.py 画图 import plots plots.plot_assignment(uav_coords, task_coords, best_assignment) plots.plot_convergence(pso.history_best_fitness) print(f最优适应度: {best_fitness:.3f}) print(f任务分配: {best_assignment}) if __name__ __main__: main()这里的关键设计是PSO类完全不知道“无人机”或“任务”它只认position_vec和evaluate_position()函数。这种解耦让你未来想换算法比如换成差分进化 DE只需重写evaluate_position()PSO类甚至不用动——这就是globalv.py存放所有业务配置的价值它把领域知识坐标、优先级和算法框架PSO彻底隔离。3.2 可视化plots.py三张图看懂分配是否合理plots.py生成的三张图是验证结果的黄金标准不是装饰tu_scatter.png散点图显示 15 个任务点不同颜色和 8 架机三角形的地理分布连线表示分配关系。一眼看出是否出现“跨区调度”如北京朝阳的机去管海淀的任务而海淀有机闲置tu_diagram.png柱状图横轴是 8 架机 ID纵轴是每架机分配的任务数。理想状态是柱高接近均值15/8≈1.875若某机为 0 或 ≥4说明负载严重不均tu_fly.png路径图在tu_scatter.png基础上为每架机画出其任务点间的连接线按分配顺序线宽代表飞行距离。若某条线异常粗长或出现交叉缠绕说明distance.py的路径计算逻辑需优化当前用贪心顺序可升级为 Christofides 近似算法。# plots.py 中 plot_assignment() 关键逻辑 def plot_assignment(uav_coords, task_coords, assignment): plt.figure(figsize(12, 10)) # 绘制所有任务点按优先级着色 priorities np.array(task_priorities) colors [red, orange, green] # 优先级3/2/1对应色 for j, (lon, lat) in enumerate(task_coords): plt.scatter(lon, lat, ccolors[priorities[j]-1], s80, labelfTask{j}(P{priorities[j]})) # 绘制无人机三角形 for i, (lon, lat) in enumerate(uav_coords): plt.scatter(lon, lat, cblue, marker^, s120, labelfUAV{i}) # 绘制分配连线 for j, uav_id in enumerate(assignment): if uav_id -1: continue uav_lon, uav_lat uav_coords[uav_id] task_lon, task_lat task_coords[j] plt.plot([uav_lon, task_lon], [uav_lat, task_lat], k--, alpha0.6) plt.xlabel(Longitude) plt.ylabel(Latitude) plt.title(UAV-Task Assignment Scatter Plot) plt.legend(bbox_to_anchor(1.05, 1), locupper left) plt.savefig(tu_scatter.png, bbox_inchestight) plt.show()注意alpha0.6让连线半透明避免 15 条线叠成黑块。这张图在你调试condition.py的约束逻辑时尤其有用如果某任务点周围没有连线即assignment[j] -1说明decode.py在解码时因所有无人机都超载/超距而放弃分配此时你要检查uav_capacities或max_distance是否设得太严。3.3 约束检查condition.py让 PSO 不在违法边缘试探condition.py是系统的“交通警察”它不参与优化只在每次解码后强制校验。它的存在让PSO可以大胆探索即使产生超载粒子而最终解一定合法# condition.py 核心函数 def check_constraints(assignment, uav_capacities, task_demands, uav_coords, task_coords, max_distance): assignment: list, 任务分配结果 return: bool, True全部约束满足False存在违规 num_uavs len(uav_capacities) uav_loads [0.0] * num_uavs uav_distances [0.0] * num_uavs for j, uav_id in enumerate(assignment): if uav_id -1: continue # 检查载重 uav_loads[uav_id] task_demands[j] if uav_loads[uav_id] uav_capacities[uav_id] 1e-6: # 浮点容差 print(fConstraint Violation: UAV{uav_id} overload! {uav_loads[uav_id]:.2f} {uav_capacities[uav_id]}) return False # 检查距离单任务点到无人机距离 dist haversine(uav_coords[uav_id], task_coords[j]) if dist max_distance 1e-3: print(fConstraint Violation: Task{j} too far from UAV{uav_id}! {dist:.3f} {max_distance}) return False return True # 在 main.py 的最终输出前调用 if check_constraints(best_assignment, uav_capacities, task_demands, ...): print(✅ All constraints satisfied.) else: print(❌ Constraints violated! Check condition.py logic.)这个函数必须在main.py输出最终结果前执行。它打印的具体违规信息如UAV3 overload! 2.45 2.30比任何文档都管用——这是你调参的直接依据要么调大uav_capacities[3]要么在fit_dis.py中加大超载惩罚权重。4. 避坑指南我在复现时踩过的 4 个真实坑与血泪修复方案4.1 坑decode.py解码后assignment全是 -1PSO 收敛到 0 分现象运行main.py终端输出最优适应度: 0.000tu_scatter.png中所有任务点无连线。原因decode_position()中get_distance(uav_id, j)返回inf或极大值如1e8导致所有dist_to_task max_distance判断为真解码时跳过所有无人机。根源是distance.py里的uav_coords或task_coords坐标格式错误——例如把(lat, lon)误写成(lon, lat)Haversine 公式计算出荒谬距离。解决在distance.py的get_distance()开头加断言def get_distance(uav_id, task_id): uav_lat, uav_lon uav_coords[uav_id] # 确保是 (lat, lon) task_lat, task_lon task_coords[task_id] assert -90 uav_lat 90 and -180 uav_lon 180, fUAV{uav_id} coord invalid: {uav_coords[uav_id]} assert -90 task_lat 90 and -180 task_lon 180, fTask{task_id} coord invalid: {task_coords[task_id]} return haversine((uav_lat, uav_lon), (task_lat, task_lon))运行时会立即报错指出哪架机坐标越界而不是静默失败。4.2 坑plots.py报ValueError: x and y must be the same size散点图画不出来现象main.py运行到plots.plot_assignment()时崩溃提示坐标长度不匹配。原因globalv.py中uav_coords和task_coords的长度与main.py初始化 PSO 时用的num_uavs/num_tasks不一致。例如uav_coords只有 7 个元组但num_uavs8导致decode_position()里position[:, j]的num_uavs维度为 8而uav_coords索引越界。解决在main.py开头强制校验from globalv import uav_coords, task_coords assert len(uav_coords) num_uavs, fuav_coords length {len(uav_coords)} ! num_uavs {num_uavs} assert len(task_coords) num_tasks, ftask_coords length {len(task_coords)} ! num_tasks {num_tasks}这种校验应在任何调用前执行避免错误传导到绘图层。4.3 坑time_s.py显示单次迭代耗时飙升到 5000ms优化卡死现象convergence_curve.png中右纵轴耗时在迭代 50 次后突然跳到 5000ms后续迭代越来越慢。原因fit_dis.py的fitness()函数中haversine()被反复调用每架机对每个任务点都要算而haversine()若未向量化用纯 Python 循环计算复杂度 O(n²)。15 个任务 × 8 架机 120 次调用每次调用内部循环 100 行必然卡顿。解决用 NumPy 向量化重写haversine()distance.py中已提供def haversine_vectorized(lat1, lon1, lat2, lon2): # lat/lon in radians dlat lat2 - lat1 dlon lon2 - lon1 a np.sin(dlat/2)**2 np.cos(lat1) * np.cos(lat2) * np.sin(dlon/2)**2 c 2 * np.arcsin(np.sqrt(a)) return 6371 * c # Earth radius in km # 在 fitness() 中批量计算距离 uav_lats np.array([coord[0] for coord in uav_coords]) * np.pi/180 uav_lons np.array([coord[1] for coord in uav_coords]) * np.pi/180 task_lats np.array([coord[0] for coord in task_coords]) * np.pi/180 task_lons np.array([coord[1] for coord in task_coords]) * np.pi/180 # 批量计算所有 UAV-Task 距离矩阵 (num_uavs, num_tasks) dist_matrix haversine_vectorized( uav_lats[:, np.newaxis], uav_lons[:, np.newaxis], task_lats[np.newaxis, :], task_lons[np.newaxis, :] )向量化后120 次距离计算从 5000ms 降到 8ms这才是工业级性能。4.4 坑tu_diagram.png显示某架机分配 0 个任务但check_constraints()却通过现象柱状图中 UAV5 高度为 0但condition.py检查返回True无报错。原因check_constraints()只检查“已分配”的任务是否合规对assignment[j] -1的任务直接跳过。这意味着算法可能因过度保守如max_distance设太小而放弃分配却仍被判定为“合法”。解决修改condition.py增加“最低分配率”检查def check_constraints(..., min_assignment_rate0.8): # ... 原有检查 ... # 新增检查分配率 assigned_count sum(1 for a in assignment if a ! -1) assignment_rate assigned_count / len(assignment) if assignment_rate min_assignment_rate: print(fWarning: Assignment rate {assignment_rate:.2f} {min_assignment_rate}. Consider relaxing constraints.) # 注意这里不 return False因为业务允许部分任务延迟 return True这样既不破坏约束刚性又给你明确信号该调max_distance了。5. 进阶技巧用globalv.py快速切换 3 种真实业务场景与参数验证法5.1 场景一农业植保——从“任务点”到“作业区域”的坐标转换农业场景中任务不是离散点而是矩形地块如[(116.3,39.9), (116.32,39.9), (116.32,39.88), (116.3,39.88)]。直接套用点坐标会低估飞行距离。正确做法是在globalv.py中定义task_regions并在distance.py的get_distance()中将无人机到地块的距离改为到地块中心点的距离 地块半周长估算覆盖该区域所需最小飞行路径# globalv.py 新增 task_regions [ [(116.3,39.9), (116.32,39.9), (116.32,39.88), (116.3,39.88)], # 地块0 # ... 其他地块 ] # distance.py 修改 get_distance() def get_distance(uav_id, task_id): uav_lat, uav_lon uav_coords[uav_id] # 若 task_id 对应地块则计算到中心半周长 if task_id len(task_regions): region task_regions[task_id] # 计算中心点 lats [p[0] for p in region] lons [p[1] for p in region] center_lat np.mean(lats) center_lon np.mean(lons) # 计算半周长近似 perimeter 0 for i in range(len(region)): j (i1) % len(region) perimeter haversine(region[i], region[j]) extra_dist perimeter / 2 base_dist haversine((uav_lat, uav_lon), (center_lat, center_lon)) return base_dist extra_dist else: # 原有点坐标逻辑 return haversine((uav_lat, uav_lon), task_coords[task_id])这样PSO优化时会自然倾向于把大地块分配给续航强的无人机小地块给灵活型符合农业实际。5.2 场景二物流配送——动态加入“时间窗约束”物流任务有严格时间窗如task_time_windows [(9.0, 10.5), (14.0, 15.0), ...]单位小时。fit_dis.py的fitness()需升级在计算每架机路径时不仅要算距离还要模拟出发时间、到达时间、服务时间若到达早于窗口则等待浪费时间晚于窗口则惩罚。globalv.py中新增# globalv.py task_time_windows [(9.0, 10.5), (14.0, 15.0), (10.0, 11.0), ...] # (start, end) service_time_per_task 0.2 # 每个任务服务0.2小时12分钟然后在fitness()的路径计算循环中插入时间窗逻辑# fit_dis.py 中路径计算片段伪代码 for uav_id in range(num_uavs): uav_tasks [j for j in range(num_tasks) if assignment[j] uav_id] if not uav_tasks: continue current_time 8.0 # 假设8点起飞 for j in uav_tasks: # 计算到任务j的飞行时间距离 / 速度 flight_time dist_matrix[uav_id][j] / 60.0 # 60km/h → 小时 arrival_time current_time flight_time # 检查时间窗 start, end task_time_windows[j] if arrival_time start: wait_time start - arrival_time current_time start service_time_per_task elif arrival_time end: time_penalty 100.0 # 严重违约 current_time arrival_time service_time_per_task else: current_time arrival_time service_time_per_task这个改动让PSO学会“掐点”而不是只看距离。5.3 场景三应急响应——用condition.py实现“硬性优先级抢占”火情任务priority3必须 10 分钟内响应。condition.py可增加抢占逻辑若检测到高优任务未分配强制将其分配给最近的、未超载的无人机哪怕打乱原有分配# condition.py 新增函数 def enforce_priority_assignment(assignment, uav_capacities, task_demands, uav_coords, task_coords, priority_threshold3): 对 priority priority_threshold 的任务强制分配给最近可用无人机 new_assignment assignment.copy() for j, priority in enumerate(task_priorities): if priority priority_threshold or assignment[j] ! -1: continue # 找最近且未超载的无人机 min_dist float(inf) best_uav -1 for uav_id in range(len(uav_capacities)): if uav_capacities[uav_id] task_demands[j]: # 载重足够 dist haversine(uav_coords[uav_id], task_coords[j]) if dist min_dist: min_dist dist best_uav uav_id if best_uav ! -1: new_assignment[j] best_uav return new_assignment # 在 main.py 中解码后立即调用 best_assignment decode_position(...) best_assignment enforce_priority_assignment(best_assignment, ...) # 强制保障高优这相当于给 PSO 加了个“安全阀”确保核心业务不掉链子。5.4 参数验证法用time_s.py的 3 轮测试锁定最优配置不要凭感觉调w,c1,c2。我固定用time_s.py做三轮测试鲁棒性测试num_runs10固定max_iter200记录 10 次最优适应度的均值与标准差。标准差 0.5 表示稳定收敛速度测试max_iter100看第 100 次的适应度是否已达 95% 的 200 次最优值。若否说明c1/c2配比需调业务达标测试人工设定一个硬指标如“本文还有配套的精品资源点击获取