🤖 第19课：多机调度

📌 多机调度概述

多台机器人协同工作需要智能调度——谁去哪里、避免冲突、负载均衡：

📌 任务分配调度

import math, random
from collections import defaultdict

class MultiRobotScheduler:
    """多机器人调度器"""
    def __init__(self):
        self.robots = {}
        self.tasks = []
        self.assignments = {}

    def add_robot(self, rid, position, battery=100, capabilities=None):
        self.robots[rid] = {
            "id": rid, "pos": position, "battery": battery,
            "capabilities": capabilities or ["navigate","deliver"],
            "status": "idle", "current_task": None
        }

    def add_task(self, tid, pickup, delivery, priority=0, capability="deliver"):
        self.tasks.append({
            "id": tid, "pickup": pickup, "delivery": delivery,
            "priority": priority, "capability": capability, "assigned": None
        })

    def schedule(self):
        """贪心调度算法"""
        unassigned = [t for t in self.tasks if t["assigned"] is None]
        idle_robots = [r for r in self.robots.values() if r["status"] == "idle"]
        
        assignments = []
        for task in sorted(unassigned, key=lambda t: -t["priority"]):
            best_robot = None; best_cost = float('inf')
            
            for robot in idle_robots:
                if task["capability"] not in robot["capabilities"]:
                    continue
                if robot["battery"] < 20:
                    continue
                
                dist = math.sqrt((robot["pos"][0]-task["pickup"][0])**2 + 
                                 (robot["pos"][1]-task["pickup"][1])**2)
                cost = dist - task["priority"] * 2
                
                if cost < best_cost:
                    best_cost = cost; best_robot = robot
            
            if best_robot:
                best_robot["status"] = "busy"
                best_robot["current_task"] = task["id"]
                task["assigned"] = best_robot["id"]
                idle_robots = [r for r in idle_robots if r["id"] != best_robot["id"]]
                assignments.append({"task": task["id"], "robot": best_robot["id"], "cost": round(best_cost, 2)})
        
        return assignments

sched = MultiRobotScheduler()
sched.add_robot("R01", (0, 0), 90, ["navigate","deliver"])
sched.add_robot("R02", (5, 3), 75, ["navigate","deliver","greet"])
sched.add_robot("R03", (2, 8), 85, ["navigate","deliver"])

sched.add_task("T01", (1, 1), (3, 5), 3)
sched.add_task("T02", (4, 2), (7, 1), 2)
sched.add_task("T03", (6, 6), (1, 3), 5)
sched.add_task("T04", (3, 7), (5, 4), 1)
sched.add_task("T05", (8, 2), (2, 1), 4)

print("多机器人调度")
print("=" * 55)
assignments = sched.schedule()
for a in assignments:
    print(f"  📋 任务{a['task']} → 机器人{a['robot']} (代价{a['cost']})")

unassigned = [t for t in sched.tasks if t["assigned"] is None]
print(f"\n已分配: {len(assignments)}, 未分配: {len(unassigned)}")
print("✅ 多机调度验证通过")

📌 多机冲突避免

class CollisionAvoidance:
    """多机冲突避免"""
    def __init__(self, grid_size=10):
        self.grid_size = grid_size
        self.reservations = {}  # (x,y,t) -> robot_id

    def reserve(self, robot_id, path, start_time=0):
        """预约路径"""
        reserved = []
        for i, (x, y) in enumerate(path):
            t = start_time + i
            cell = (int(x), int(y), t)
            if cell in self.reservations and self.reservations[cell] != robot_id:
                return {"success": False, "conflict": cell, "reserved_by": self.reservations[cell]}
            self.reservations[cell] = robot_id
            reserved.append(cell)
        return {"success": True, "reserved": len(reserved)}

    def plan_with_avoidance(self, robot_id, start, goal, start_time=0):
        """带冲突避免的路径规划"""
        path = self._simple_path(start, goal)
        t = start_time
        
        for wait in range(5):  # 最多等5个时间步
            conflict = False
            for i, pos in enumerate(path):
                cell = (int(pos[0]), int(pos[1]), t + i)
                if cell in self.reservations and self.reservations[cell] != robot_id:
                    conflict = True
                    break
            
            if not conflict:
                result = self.reserve(robot_id, path, t)
                if result["success"]:
                    return {"success": True, "path": path, "start_time": t, "wait": wait}
            t += 1
        
        return {"success": False, "reason": "无法找到无冲突时间窗口"}

    def _simple_path(self, start, goal):
        path = [start]
        x, y = start
        gx, gy = goal
        while (x, y) != (gx, gy):
            if x < gx: x += 1
            elif x > gx: x -= 1
            if y < gy: y += 1
            elif y > gy: y -= 1
            path.append((x, y))
        return path

ca = CollisionAvoidance()
print("多机冲突避免(时空A*)")
print("=" * 55)

# 机器人1先规划
r1_path = ca._simple_path((0, 0), (5, 5))
r1_result = ca.reserve("R01", r1_path, 0)
print(f"  R01 路径(0,0)→(5,5): {'✅' if r1_result['success'] else '❌'}")

# 机器人2规划（路径可能冲突）
r2_result = ca.plan_with_avoidance("R02", (5, 0), (0, 5), 0)
if r2_result["success"]:
    print(f"  R02 路径(5,0)→(0,5): ✅ 开始时间t={r2_result['start_time']}, 等待{r2_result['wait']}步")
else:
    print(f"  R02 路径(5,0)→(0,5): ❌ {r2_result['reason']}")

# 机器人3规划
r3_result = ca.plan_with_avoidance("R03", (2, 0), (2, 8), 0)
if r3_result["success"]:
    print(f"  R03 路径(2,0)→(2,8): ✅ 开始时间t={r3_result['start_time']}, 等待{r3_result['wait']}步")

print(f"\n预约总数: {len(ca.reservations)}")
print("✅ 冲突避免验证通过")

📌 机群管理与负载均衡

class FleetManager:
    """机群管理"""
    def __init__(self):
        self.robots = {}
        self.zones = {}  # 区域定义
        self.zone_capacity = {}  # 区域容量

    def add_robot(self, rid, zone):
        self.robots[rid] = {"zone": zone, "status": "idle", "battery": 100}

    def add_zone(self, name, capacity):
        self.zones[name] = capacity
        self.zone_capacity[name] = capacity

    def get_zone_load(self, zone):
        count = sum(1 for r in self.robots.values() if r["zone"] == zone)
        return count

    def rebalance(self):
        """负载均衡重分配"""
        moves = []
        for zone, capacity in self.zone_capacity.items():
            load = self.get_zone_load(zone)
            if load > capacity:
                excess = load - capacity
                idle_in_zone = [rid for rid, r in self.robots.items() 
                               if r["zone"] == zone and r["status"] == "idle"]
                for rid in idle_in_zone[:excess]:
                    # 找最空闲的区域
                    best_zone = min(self.zone_capacity.keys(),
                                    key=lambda z: self.get_zone_load(z) / self.zone_capacity[z])
                    moves.append({"robot": rid, "from": zone, "to": best_zone})
                    self.robots[rid]["zone"] = best_zone
        return moves

    def get_fleet_status(self):
        status = {"total": len(self.robots), "idle": 0, "busy": 0, "low_battery": 0}
        for r in self.robots.values():
            if r["status"] == "idle": status["idle"] += 1
            else: status["busy"] += 1
            if r["battery"] < 20: status["low_battery"] += 1
        return status

fm = FleetManager()
fm.add_zone("1F大厅", 3)
fm.add_zone("3F办公", 2)
fm.add_zone("5F行政", 2)

# 初始分配（大厅过载）
for i in range(5):
    fm.add_robot(f"R{i+1:02d}", "1F大厅")
for i in range(5, 7):
    fm.add_robot(f"R{i+1:02d}", "3F办公")
fm.robots["R03"]["battery"] = 15  # 低电量
fm.robots["R05"]["status"] = "busy"

print("机群管理与负载均衡")
print("=" * 55)

print("\n初始状态:")
for zone in fm.zones:
    load = fm.get_zone_load(zone)
    cap = fm.zone_capacity[zone]
    print(f"  {zone}: {load}/{cap} ({'⚠️过载' if load>cap else '✅'})")

moves = fm.rebalance()
print(f"\n负载均衡调整:")
for m in moves:
    print(f"  🔄 {m['robot']}: {m['from']} → {m['to']}")

status = fm.get_fleet_status()
print(f"\n机群状态: 总{status['total']} 空闲{status['idle']} 忙碌{status['busy']} 低电量{status['low_battery']}")
print("✅ 机群管理验证通过")

📌 调度算法对比

📌 练习

📌 成就