"""
|
任务分配算法
|
"""
|
import time
|
import random
|
import logging
|
from typing import Dict, List, Tuple, Optional, Set, Any
|
from collections import defaultdict
|
from abc import ABC, abstractmethod
|
|
from common.data_models import TaskData, AGVStatus, TaskAssignment, BackpackData
|
from algorithm_system.models.agv_model import AGVModel, AGVModelManager
|
from common.utils import get_coordinate_from_path_id, calculate_distance, calculate_manhattan_distance
|
from dataclasses import dataclass
|
|
|
class TaskAllocation(ABC):
|
"""任务分配算法基类"""
|
|
def __init__(self, agv_manager: AGVModelManager):
|
"""
|
初始化任务分配算法
|
|
Args:
|
agv_manager: AGV模型管理器
|
"""
|
self.agv_manager = agv_manager
|
self.logger = logging.getLogger(__name__)
|
|
@abstractmethod
|
def allocate_tasks(self, tasks: List[TaskData]) -> List[TaskAssignment]:
|
"""
|
分配任务给AGV
|
|
Args:
|
tasks: 待分配的任务列表
|
|
Returns:
|
List[TaskAssignment]: 分配结果列表
|
"""
|
pass
|
|
def find_available_backpack_slot(self, agv_status: AGVStatus) -> Optional[int]:
|
"""
|
查找AGV的可用背篓位置
|
|
Args:
|
agv_status: AGV状态信息
|
|
Returns:
|
Optional[int]: 可用的背篓位置编号，如果没有可用位置则返回None
|
"""
|
if not agv_status.backpack:
|
# 如果没有背篓信息，假设从第一个位置开始
|
self.logger.warning(f"AGV {agv_status.agvId} 没有背篓信息，分配到位置0")
|
return 0
|
|
# 查找空闲且未执行任务的背篓位置
|
for backpack_item in agv_status.backpack:
|
if not backpack_item.loaded and not backpack_item.execute and not backpack_item.taskId:
|
self.logger.debug(f"AGV {agv_status.agvId} 找到可用背篓位置: {backpack_item.index}")
|
return backpack_item.index
|
|
# 如果所有位置都被占用，返回None
|
self.logger.debug(f"AGV {agv_status.agvId} 没有可用的背篓位置")
|
return None
|
|
def get_agv_available_capacity(self, agv_status: AGVStatus) -> int:
|
"""
|
获取AGV的可用背篓容量
|
|
Args:
|
agv_status: AGV状态信息
|
|
Returns:
|
int: 可用背篓数量
|
"""
|
if not agv_status.backpack:
|
return 1 # 假设至少有一个背篓位置
|
|
available_count = 0
|
for backpack_item in agv_status.backpack:
|
if not backpack_item.loaded and not backpack_item.execute and not backpack_item.taskId:
|
available_count += 1
|
|
return available_count
|
|
def assign_task_with_backpack(self, agv_model, task: TaskData, lev_id: int) -> bool:
|
"""
|
将任务分配给AGV的指定背篓位置
|
|
Args:
|
agv_model: AGV模型
|
task: 任务数据
|
lev_id: 背篓位置编号
|
|
Returns:
|
bool: 分配是否成功
|
"""
|
try:
|
# 使用AGV模型的assign_task方法
|
success = agv_model.assign_task(
|
task_id=task.taskId,
|
priority=task.priority,
|
start_code=task.start,
|
end_code=task.end
|
)
|
|
if success:
|
self.logger.info(f"任务 {task.taskId} 成功分配给AGV {agv_model.agvId} 的背篓位置 {lev_id}")
|
return True
|
else:
|
self.logger.warning(f"任务 {task.taskId} 分配给AGV {agv_model.agvId} 失败")
|
return False
|
|
except Exception as e:
|
self.logger.error(f"分配任务时发生异常: {e}")
|
return False
|
|
|
class NearestFirstAllocation(TaskAllocation):
|
"""最近优先分配算法"""
|
|
def allocate_tasks(self, tasks: List[TaskData]) -> List[TaskAssignment]:
|
"""
|
使用最近优先策略分配任务
|
|
Args:
|
tasks: 待分配的任务列表
|
|
Returns:
|
List[TaskAssignment]: 分配结果列表
|
"""
|
if not tasks:
|
return []
|
|
# 获取可用的AGV
|
available_agvs = self.agv_manager.get_available_agvs()
|
|
if not available_agvs:
|
self.logger.warning("没有可用的AGV进行任务分配")
|
return []
|
|
# 1. 首先检查任务是否已经分配，避免重复分配
|
already_assigned_tasks = set()
|
all_agvs = self.agv_manager.get_all_agvs()
|
for agv in all_agvs:
|
if agv.backpack:
|
for backpack_item in agv.backpack:
|
if backpack_item.taskId:
|
already_assigned_tasks.add(backpack_item.taskId)
|
self.logger.info(f"任务 {backpack_item.taskId} 已分配给 AGV {agv.agvId}，跳过重复分配")
|
|
# 2. 过滤掉已分配的任务
|
unassigned_tasks = [task for task in tasks if task.taskId not in already_assigned_tasks]
|
|
if not unassigned_tasks:
|
self.logger.info("所有任务都已分配，无需重新分配")
|
return []
|
|
self.logger.info(f"总任务数: {len(tasks)}, 已分配: {len(already_assigned_tasks)}, 待分配: {len(unassigned_tasks)}")
|
|
assignments = []
|
path_mapping = self.agv_manager.path_mapping
|
|
# 对每个任务找到最近的AGV
|
for task in unassigned_tasks:
|
if not available_agvs:
|
break
|
|
# 获取任务起点坐标
|
task_start_coord = get_coordinate_from_path_id(task.start, path_mapping)
|
if not task_start_coord:
|
self.logger.warning(f"无法获取任务 {task.taskId} 起点 {task.start} 的坐标")
|
continue
|
|
# 找到距离最近的AGV
|
nearest_agv = None
|
min_distance = float('inf')
|
|
for agv in available_agvs:
|
if agv.coordinates:
|
distance = calculate_manhattan_distance(agv.coordinates, task_start_coord)
|
|
# 如果AGV已有任务，计算完成当前任务后到新任务起点的距离
|
if agv.current_task_count > 0:
|
# 简化：假设AGV需要额外时间完成当前任务
|
distance += agv.current_task_count * 10
|
|
if distance < min_distance:
|
min_distance = distance
|
nearest_agv = agv
|
|
if nearest_agv and nearest_agv.can_accept_task(task.priority):
|
# 获取AGV的原始状态数据来查找可用背篓位置
|
agv_status = None
|
for agv_state in self.agv_manager.get_all_agv_status():
|
if agv_state.agvId == nearest_agv.agvId:
|
agv_status = agv_state
|
break
|
|
if agv_status:
|
# 查找可用的背篓位置
|
available_lev_id = self.find_available_backpack_slot(agv_status)
|
|
if available_lev_id is not None:
|
# 分配任务到指定背篓位置
|
success = self.assign_task_with_backpack(nearest_agv, task, available_lev_id)
|
if success:
|
assignments.append(TaskAssignment(
|
taskId=task.taskId,
|
agvId=nearest_agv.agvId,
|
lev_id=available_lev_id
|
))
|
|
self.logger.info(f"任务 {task.taskId} 分配给最近的AGV {nearest_agv.agvId}，背篓位置: {available_lev_id}，距离: {min_distance}")
|
|
# 检查AGV是否还有可用背篓位置
|
remaining_capacity = self.get_agv_available_capacity(agv_status) - 1
|
if remaining_capacity <= 0:
|
available_agvs.remove(nearest_agv)
|
else:
|
self.logger.warning(f"任务 {task.taskId} 分配给AGV {nearest_agv.agvId} 失败")
|
else:
|
self.logger.warning(f"AGV {nearest_agv.agvId} 没有可用的背篓位置")
|
available_agvs.remove(nearest_agv)
|
else:
|
self.logger.warning(f"无法获取AGV {nearest_agv.agvId} 的状态信息")
|
|
return assignments
|
|
|
class LoadBalancedAllocation(TaskAllocation):
|
"""负载均衡分配算法"""
|
|
def allocate_tasks(self, tasks: List[TaskData]) -> List[TaskAssignment]:
|
"""
|
使用负载均衡策略分配任务
|
|
Args:
|
tasks: 待分配的任务列表
|
|
Returns:
|
List[TaskAssignment]: 分配结果列表
|
"""
|
if not tasks:
|
return []
|
|
# 获取所有AGV
|
all_agvs = self.agv_manager.get_all_agvs()
|
|
if not all_agvs:
|
self.logger.warning("没有AGV进行任务分配")
|
return []
|
|
# 1. 首先检查任务是否已经分配，避免重复分配
|
already_assigned_tasks = set()
|
for agv in all_agvs:
|
if agv.backpack:
|
for backpack_item in agv.backpack:
|
if backpack_item.taskId:
|
already_assigned_tasks.add(backpack_item.taskId)
|
self.logger.info(f"任务 {backpack_item.taskId} 已分配给 AGV {agv.agvId}，跳过重复分配")
|
|
# 2. 过滤掉已分配的任务
|
unassigned_tasks = [task for task in tasks if task.taskId not in already_assigned_tasks]
|
|
if not unassigned_tasks:
|
self.logger.info("所有任务都已分配，无需重新分配")
|
return []
|
|
self.logger.info(f"总任务数: {len(tasks)}, 已分配: {len(already_assigned_tasks)}, 待分配: {len(unassigned_tasks)}")
|
|
assignments = []
|
path_mapping = self.agv_manager.path_mapping
|
|
# 按优先级排序任务
|
sorted_tasks = sorted(unassigned_tasks, key=lambda t: t.priority, reverse=True)
|
|
# 对每个任务分配给负载最低的AGV
|
for task in sorted_tasks:
|
# 获取任务起点坐标
|
task_start_coord = get_coordinate_from_path_id(task.start, path_mapping)
|
if not task_start_coord:
|
self.logger.warning(f"无法获取任务 {task.taskId} 起点 {task.start} 的坐标")
|
continue
|
|
# 按负载和距离排序AGV
|
agv_scores = []
|
for agv in all_agvs:
|
if not agv.can_accept_task(task.priority):
|
continue
|
|
# 计算负载得分（负载越低得分越高）
|
load_score = 1.0 - agv.get_workload_ratio()
|
|
# 计算距离得分（距离越近得分越高）
|
distance_score = 0.0
|
if agv.coordinates and task_start_coord:
|
distance = calculate_manhattan_distance(agv.coordinates, task_start_coord)
|
distance_score = 1.0 / (1.0 + distance / 100.0) # 归一化距离得分
|
|
# 计算效率得分
|
efficiency_score = agv.calculate_efficiency_score(path_mapping)
|
|
# 综合得分
|
total_score = 0.4 * load_score + 0.3 * distance_score + 0.3 * efficiency_score
|
agv_scores.append((agv, total_score))
|
|
if not agv_scores:
|
# 详细分析为什么没有AGV可以接受任务
|
total_agvs = len(all_agvs)
|
busy_count = 0
|
low_battery_count = 0
|
overloaded_count = 0
|
status_invalid_count = 0
|
|
for agv in all_agvs:
|
if agv.is_overloaded():
|
overloaded_count += 1
|
elif str(agv.status) not in ["0", "1", "2"]:
|
status_invalid_count += 1
|
elif agv.need_charging():
|
low_battery_count += 1
|
else:
|
busy_count += 1
|
|
self.logger.warning(f"没有AGV可以接受任务 {task.taskId} - 详细分析:")
|
self.logger.warning(f" 总AGV数: {total_agvs}")
|
self.logger.warning(f" 任务满载: {overloaded_count}")
|
self.logger.warning(f" 状态异常: {status_invalid_count}")
|
self.logger.warning(f" 电量过低: {low_battery_count}")
|
self.logger.warning(f" 其他原因: {busy_count}")
|
self.logger.warning(f" 任务优先级: {task.priority}")
|
continue
|
|
# 选择得分最高的AGV
|
agv_scores.sort(key=lambda x: x[1], reverse=True)
|
best_agv = agv_scores[0][0]
|
|
# 获取AGV的原始状态数据来查找可用背篓位置
|
agv_status = self.agv_manager.get_agv_status(best_agv.agvId)
|
|
if agv_status:
|
# 查找可用的背篓位置
|
available_lev_id = self.find_available_backpack_slot(agv_status)
|
|
if available_lev_id is not None:
|
# 分配任务到指定背篓位置
|
success = self.assign_task_with_backpack(best_agv, task, available_lev_id)
|
if success:
|
assignments.append(TaskAssignment(
|
taskId=task.taskId,
|
agvId=best_agv.agvId,
|
lev_id=available_lev_id
|
))
|
|
self.logger.info(f"任务 {task.taskId} 分配给负载均衡的AGV {best_agv.agvId}，背篓位置: {available_lev_id}，得分: {agv_scores[0][1]:.3f}")
|
else:
|
self.logger.warning(f"任务 {task.taskId} 分配给AGV {best_agv.agvId} 失败")
|
else:
|
self.logger.warning(f"AGV {best_agv.agvId} 没有可用的背篓位置")
|
else:
|
self.logger.warning(f"无法获取AGV {best_agv.agvId} 的状态信息")
|
|
return assignments
|
|
|
class PriorityFirstAllocation(TaskAllocation):
|
"""优先级优先分配算法"""
|
|
def allocate_tasks(self, tasks: List[TaskData]) -> List[TaskAssignment]:
|
"""
|
使用优先级优先策略分配任务
|
|
Args:
|
tasks: 待分配的任务列表
|
|
Returns:
|
List[TaskAssignment]: 分配结果列表
|
"""
|
if not tasks:
|
return []
|
|
# 获取可用的AGV
|
available_agvs = self.agv_manager.get_available_agvs()
|
|
if not available_agvs:
|
self.logger.warning("没有可用的AGV进行任务分配")
|
return []
|
|
# 1. 首先检查任务是否已经分配，避免重复分配
|
already_assigned_tasks = set()
|
all_agvs = self.agv_manager.get_all_agvs()
|
for agv in all_agvs:
|
if agv.backpack:
|
for backpack_item in agv.backpack:
|
if backpack_item.taskId:
|
already_assigned_tasks.add(backpack_item.taskId)
|
self.logger.info(f"任务 {backpack_item.taskId} 已分配给 AGV {agv.agvId}，跳过重复分配")
|
|
# 2. 过滤掉已分配的任务
|
unassigned_tasks = [task for task in tasks if task.taskId not in already_assigned_tasks]
|
|
if not unassigned_tasks:
|
self.logger.info("所有任务都已分配，无需重新分配")
|
return []
|
|
self.logger.info(f"总任务数: {len(tasks)}, 已分配: {len(already_assigned_tasks)}, 待分配: {len(unassigned_tasks)}")
|
|
# 按优先级排序任务（高优先级在前）
|
sorted_tasks = sorted(unassigned_tasks, key=lambda t: t.priority, reverse=True)
|
|
assignments = []
|
path_mapping = self.agv_manager.path_mapping
|
|
# 优先分配高优先级任务
|
for task in sorted_tasks:
|
if not available_agvs:
|
break
|
|
# 获取任务起点坐标
|
task_start_coord = get_coordinate_from_path_id(task.start, path_mapping)
|
if not task_start_coord:
|
self.logger.warning(f"无法获取任务 {task.taskId} 起点 {task.start} 的坐标")
|
continue
|
|
# 为高优先级任务选择最佳AGV
|
best_agv = None
|
best_score = -1
|
|
for agv in available_agvs:
|
if not agv.can_accept_task(task.priority):
|
continue
|
|
# 计算综合得分
|
distance_score = 0.0
|
if agv.coordinates and task_start_coord:
|
distance = calculate_manhattan_distance(agv.coordinates, task_start_coord)
|
distance_score = 1.0 / (1.0 + distance / 50.0)
|
|
efficiency_score = agv.calculate_efficiency_score(path_mapping)
|
capacity_score = agv.get_task_capacity() / agv.max_capacity
|
|
# 高优先级任务更注重效率和距离
|
total_score = 0.5 * distance_score + 0.3 * efficiency_score + 0.2 * capacity_score
|
|
if total_score > best_score:
|
best_score = total_score
|
best_agv = agv
|
|
if best_agv:
|
# 获取AGV的原始状态数据来查找可用背篓位置
|
agv_status = self.agv_manager.get_agv_status(best_agv.agvId)
|
|
if agv_status:
|
# 查找可用的背篓位置
|
available_lev_id = self.find_available_backpack_slot(agv_status)
|
|
if available_lev_id is not None:
|
# 分配任务到指定背篓位置
|
success = self.assign_task_with_backpack(best_agv, task, available_lev_id)
|
if success:
|
assignments.append(TaskAssignment(
|
taskId=task.taskId,
|
agvId=best_agv.agvId,
|
lev_id=available_lev_id
|
))
|
|
self.logger.info(f"高优先级任务 {task.taskId} (优先级: {task.priority}) 分配给AGV {best_agv.agvId}，背篓位置: {available_lev_id}")
|
|
# 检查AGV是否还有可用背篓位置
|
remaining_capacity = self.get_agv_available_capacity(agv_status) - 1
|
if remaining_capacity <= 0:
|
available_agvs.remove(best_agv)
|
else:
|
self.logger.warning(f"任务 {task.taskId} 分配给AGV {best_agv.agvId} 失败")
|
else:
|
self.logger.warning(f"AGV {best_agv.agvId} 没有可用的背篓位置")
|
available_agvs.remove(best_agv)
|
else:
|
self.logger.warning(f"无法获取AGV {best_agv.agvId} 的状态信息")
|
|
return assignments
|
|
|
class MultiObjectiveAllocation(TaskAllocation):
|
"""多目标优化分配算法"""
|
|
def __init__(self, agv_manager: AGVModelManager,
|
distance_weight: float = 0.4,
|
load_weight: float = 0.3,
|
efficiency_weight: float = 0.3):
|
"""
|
初始化多目标优化分配算法
|
|
Args:
|
agv_manager: AGV模型管理器
|
distance_weight: 距离权重
|
load_weight: 负载权重
|
efficiency_weight: 效率权重
|
"""
|
super().__init__(agv_manager)
|
self.distance_weight = distance_weight
|
self.load_weight = load_weight
|
self.efficiency_weight = efficiency_weight
|
|
def allocate_tasks(self, tasks: List[TaskData]) -> List[TaskAssignment]:
|
"""
|
使用多目标优化策略分配任务
|
|
Args:
|
tasks: 待分配的任务列表
|
|
Returns:
|
List[TaskAssignment]: 分配结果列表
|
"""
|
if not tasks:
|
return []
|
|
# 获取所有AGV
|
all_agvs = self.agv_manager.get_all_agvs()
|
|
if not all_agvs:
|
self.logger.warning("没有AGV进行任务分配")
|
return []
|
|
# 1. 首先检查任务是否已经分配，避免重复分配
|
already_assigned_tasks = set()
|
for agv in all_agvs:
|
if agv.backpack:
|
for backpack_item in agv.backpack:
|
if backpack_item.taskId:
|
already_assigned_tasks.add(backpack_item.taskId)
|
self.logger.info(f"任务 {backpack_item.taskId} 已分配给 AGV {agv.agvId}，跳过重复分配")
|
|
# 2. 过滤掉已分配的任务
|
unassigned_tasks = [task for task in tasks if task.taskId not in already_assigned_tasks]
|
|
if not unassigned_tasks:
|
self.logger.info("所有任务都已分配，无需重新分配")
|
return []
|
|
self.logger.info(f"总任务数: {len(tasks)}, 已分配: {len(already_assigned_tasks)}, 待分配: {len(unassigned_tasks)}")
|
|
assignments = []
|
path_mapping = self.agv_manager.path_mapping
|
|
# 对每个任务-AGV对计算得分
|
task_agv_scores = {}
|
|
for task in unassigned_tasks:
|
task_start_coord = get_coordinate_from_path_id(task.start, path_mapping)
|
if not task_start_coord:
|
continue
|
|
for agv in all_agvs:
|
if not agv.can_accept_task(task.priority):
|
continue
|
|
# 距离得分
|
distance_score = 0.0
|
if agv.coordinates:
|
distance = calculate_manhattan_distance(agv.coordinates, task_start_coord)
|
distance_score = 1.0 / (1.0 + distance / 100.0)
|
|
# 负载得分
|
load_score = 1.0 - agv.get_workload_ratio()
|
|
# 效率得分
|
efficiency_score = agv.calculate_efficiency_score(path_mapping)
|
|
# 计算综合得分
|
total_score = (
|
self.distance_weight * distance_score +
|
self.load_weight * load_score +
|
self.efficiency_weight * efficiency_score
|
)
|
|
task_agv_scores[(task.taskId, agv.agvId)] = total_score
|
|
# 使用贪心算法进行匹配
|
assignments = self._greedy_matching(unassigned_tasks, all_agvs, task_agv_scores)
|
|
return assignments
|
|
def _greedy_matching(self, tasks: List[TaskData], agvs: List[AGVModel],
|
scores: Dict[Tuple[str, str], float]) -> List[TaskAssignment]:
|
"""
|
使用贪心算法进行任务-AGV匹配
|
|
Args:
|
tasks: 任务列表
|
agvs: AGV列表
|
scores: 任务-AGV对的得分
|
|
Returns:
|
List[TaskAssignment]: 分配结果
|
"""
|
assignments = []
|
remaining_tasks = [task.taskId for task in tasks]
|
|
# 重复分配直到没有任务或没有可用AGV
|
while remaining_tasks:
|
# 找到得分最高的任务-AGV对
|
best_score = -1
|
best_task_id = None
|
best_agv = None
|
|
for task_id in remaining_tasks:
|
for agv in agvs:
|
if agv.is_overloaded():
|
continue
|
|
score = scores.get((task_id, agv.agvId), 0.0)
|
if score > best_score:
|
best_score = score
|
best_task_id = task_id
|
best_agv = agv
|
|
if best_task_id and best_agv:
|
# 获取AGV的原始状态数据来查找可用背篓位置
|
agv_status = self.agv_manager.get_agv_status(best_agv.agvId)
|
|
if agv_status:
|
# 查找可用的背篓位置
|
available_lev_id = self.find_available_backpack_slot(agv_status)
|
|
if available_lev_id is not None:
|
# 找到对应的任务对象
|
task = next((t for t in tasks if t.taskId == best_task_id), None)
|
if task:
|
# 分配任务到指定背篓位置
|
success = self.assign_task_with_backpack(best_agv, task, available_lev_id)
|
if success:
|
assignments.append(TaskAssignment(
|
taskId=best_task_id,
|
agvId=best_agv.agvId,
|
lev_id=available_lev_id
|
))
|
|
remaining_tasks.remove(best_task_id)
|
self.logger.info(f"多目标优化：任务 {best_task_id} 分配给AGV {best_agv.agvId}，背篓位置: {available_lev_id}，得分: {best_score:.3f}")
|
else:
|
self.logger.warning(f"任务 {best_task_id} 分配给AGV {best_agv.agvId} 失败")
|
break
|
else:
|
self.logger.error(f"找不到任务 {best_task_id} 的详细信息")
|
break
|
else:
|
self.logger.debug(f"AGV {best_agv.agvId} 没有可用的背篓位置，跳过")
|
break
|
else:
|
self.logger.warning(f"无法获取AGV {best_agv.agvId} 的状态信息")
|
break
|
else:
|
break
|
|
return assignments
|
|
|
class TaskAllocationFactory:
|
"""任务分配算法工厂类"""
|
|
@staticmethod
|
def create_allocator(algorithm_type: str, agv_manager: AGVModelManager) -> TaskAllocation:
|
"""
|
创建任务分配算法
|
|
Args:
|
algorithm_type: 算法类型
|
agv_manager: AGV模型管理器
|
|
Returns:
|
TaskAllocation: 任务分配算法对象
|
"""
|
if algorithm_type == "NEAREST_FIRST":
|
return NearestFirstAllocation(agv_manager)
|
elif algorithm_type == "LOAD_BALANCED":
|
return LoadBalancedAllocation(agv_manager)
|
elif algorithm_type == "PRIORITY_FIRST":
|
return PriorityFirstAllocation(agv_manager)
|
elif algorithm_type == "MULTI_OBJECTIVE":
|
return MultiObjectiveAllocation(agv_manager)
|
else:
|
# 默认使用负载均衡算法
|
return LoadBalancedAllocation(agv_manager)
|
