🤖 Взаимодействие нескольких роботов

1        [ЦЕНТРАЛЬНЫЙ КОНТРОЛЛЕР]
2               /    |    \
3              /     |     \
4        [Робот 1] [Робот 2] [Робот 3]

1[Робот 1] ←→ [Робот 2] ←→ [Робот 3]
2    ↕           ↕           ↕
3[Робот 4] ←→ [Робот 5] ←→ [Робот 6]

1         [КООРДИНАТОР]
2        /       |       \
3   [Группа 1] [Группа 2] [Группа 3]
4    /    \     /    \     /    \
5  [Р1] [Р2] [Р3] [Р4] [Р5] [Р6]

 1class LeaderFollowerSystem:
 2    def __init__(self):
 3        self.leader = Robot("leader")
 4        self.followers = [Robot(f"follower_{i}") for i in range(3)]
 5    
 6    def execute_mission(self):
 7        # Лидер планирует маршрут
 8        path = self.leader.plan_path()
 9        
10        # Последователи следуют за лидером
11        for follower in self.followers:
12            follower.follow_leader(self.leader.position)
13        
14        # Адаптация при смене лидера
15        if self.leader.is_disabled():
16            self.elect_new_leader()

1{
2  "sender_id": "robot_01",
3  "timestamp": 1703520000,
4  "position": {"x": 12.5, "y": 8.3, "heading": 45},
5  "battery": 78,
6  "status": "active",
7  "current_task": "exploration"
8}

1{
2  "sender_id": "robot_02", 
3  "message_type": "help_request",
4  "priority": "high",
5  "task": "object_transport",
6  "required_robots": 2,
7  "location": {"x": 15.2, "y": 7.1}
8}

 1{
 2  "sender_id": "coordinator",
 3  "message_type": "task_assignment", 
 4  "target_robots": ["robot_03", "robot_04"],
 5  "task": {
 6    "type": "formation_flight",
 7    "leader": "robot_03",
 8    "formation": "triangle"
 9  }
10}

 1class ReliableCommunication:
 2    def send_message(self, message, recipient):
 3        # Добавление контрольной суммы
 4        message.checksum = self.calculate_checksum(message)
 5        
 6        # Повтор при отсутствии подтверждения
 7        for attempt in range(3):
 8            if self.transmit(message, recipient):
 9                if self.wait_for_ack(timeout=1000):
10                    return True
11            
12        return False  # Сообщение не доставлено
13    
14    def receive_message(self, message):
15        # Проверка целостности
16        if self.verify_checksum(message):
17            self.send_acknowledgment(message.sender)
18            return message
19        else:
20            return None  # Поврежденное сообщение

1    [R2]
2     |
3[R1]─[C]─[R3]  # C = центральный узел
4     |
5    [R4]

1[R1]─[R2]
2 |     |
3[R4]─[R3]

1[R1]─[R2]
2 |\ /|
3 | X |
4 |/ \|
5[R4]─[R3]

1[R1]─[R2]─[R3]
2 |\ /|\ /|
3 | X | X |
4 |/ \|/ \|
5[R4]─[R5]─[R6]

 1class TaskAuction:
 2    def __init__(self, robots):
 3        self.robots = robots
 4        self.tasks = []
 5    
 6    def assign_task(self, task):
 7        # Запрос ставок от всех роботов
 8        bids = {}
 9        for robot in self.robots:
10            if robot.can_perform(task):
11                bid = robot.calculate_bid(task)
12                bids[robot.id] = bid
13        
14        # Выбор лучшей ставки
15        if bids:
16            winner = min(bids.keys(), key=lambda x: bids[x])
17            self.robots[winner].assign_task(task)
18            return winner
19        
20        return None  # Задачу некому выполнить
21    
22    def calculate_bid(self, robot, task):
23        # Учитываем расстояние, загрузку, специализацию
24        distance_cost = robot.distance_to(task.location)
25        workload_cost = robot.current_workload * 2
26        specialization_bonus = -10 if robot.specialized_for(task) else 0
27        
28        return distance_cost + workload_cost + specialization_bonus

 1def balance_workload(robots):
 2    # Находим перегруженных и недогруженных роботов
 3    overloaded = [r for r in robots if r.workload > 0.8]
 4    underloaded = [r for r in robots if r.workload < 0.3]
 5    
 6    # Перераспределяем задачи
 7    for busy_robot in overloaded:
 8        for free_robot in underloaded:
 9            # Ищем задачи, которые может выполнить свободный робот
10            transferable_tasks = busy_robot.get_transferable_tasks()
11            
12            for task in transferable_tasks:
13                if free_robot.can_perform(task):
14                    busy_robot.transfer_task(task, free_robot)
15                    break

 1class FailureDetector:
 2    def __init__(self, robots):
 3        self.robots = robots
 4        self.heartbeat_timeout = 5.0  # секунд
 5    
 6    def monitor_robots(self):
 7        for robot in self.robots:
 8            time_since_heartbeat = time.now() - robot.last_heartbeat
 9            
10            if time_since_heartbeat > self.heartbeat_timeout:
11                self.handle_robot_failure(robot)
12    
13    def handle_robot_failure(self, failed_robot):
14        # Перераспределяем задачи отказавшего робота
15        for task in failed_robot.active_tasks:
16            self.reassign_task(task)
17        
18        # Уведомляем остальных роботов
19        self.broadcast_failure_notification(failed_robot.id)
20        
21        # Перестраиваем формацию если нужно
22        if failed_robot.role == "leader":
23            self.elect_new_leader()