Skip to main content

Обработка команд

Команды приходят роботу от оператора (веб-интерфейс) или от модели ИИ через сервер. SDK разбирает сообщение и вызывает ваш обработчик @robot.com_handler.

Как команда доходит до робота

Оператор / модель ИИ ──▶ ХАБ ──▶ событие "command" ──▶ SDK ──▶ on_command(type_command, command, author)

Сообщение от сервера имеет вид:

{
  "author": "operator-7",
  "payload": { "com": "forward" }
}

Ключ внутри payload — тип команды, значение — сама команда. SDK передаёт их в обработчик как type_command="com", command="forward", author="operator-7".

Команды принимаются только после того, как сервер прислал сигнал готовности (Ready в heartbeat). Команды, пришедшие раньше, игнорируются.

Типы команд

Тип Кто обрабатывает Назначение
com Ваш обработчик Обычная команда управления
force_com Ваш обработчик Приоритетная команда — выполнить немедленно, прервав текущее действие (например, экстренная остановка)
sys_com Ваш обработчик Системная команда (перезагрузка, калибровка, смена режима)
file SDK автоматически Запрос журналов: log, error, clear (см. Протокол)
web_settings SDK (в версии 0.1.1 игнорируется) Настройки из веб-интерфейса — зарезервировано

Рекомендуемый словарь команд

Чтобы роботы разных команд одинаково управлялись из интерфейса и моделями ИИ, используйте единые имена команд:

Команда Действие
forward, back Движение вперёд / назад
left, right Поворот влево / вправо
stop Остановка двигателей
disarm Полное обесточивание силовых цепей
cam_left, cam_right, cam_up, cam_down, cam_center Поворот камеры
auto_run, auto_stop Запуск / остановка автоматического управления на основе видео
speed:<0..100> Установка скорости в процентах, например speed:60

Формат команд, которые отправляют кнопки конкретной страницы веб-интерфейса (Robot, LED, Drone), удобно проверить по журналу робота: SDK записывает каждую полученную команду сообщением recieved command from <author> with <payload>.

Шаблон обработчика

import threading, time

STOP_WORDS = {"stop", "disarm"}

def apply(command, ctx):
    motors = ctx["motors"]
    if command == "forward":   motors.drive(+1, +1)
    elif command == "back":    motors.drive(-1, -1)
    elif command == "left":    motors.drive(-1, +1)
    elif command == "right":   motors.drive(+1, -1)
    elif command == "stop":    motors.drive(0, 0)
    elif command == "disarm":  motors.power_off()
    elif isinstance(command, str) and command.startswith("speed:"):
        motors.set_speed(int(command.split(":", 1)[1]) / 100)
    else:
        robot.log("WARN", f"Неизвестная команда {command}")
        return
    ctx["last_command_time"] = time.monotonic()

@robot.com_handler
def on_command(type_command, command, author=None):
    robot.log("INFO", f"{type_command}:{command} от {author}")

    # Остановка — всегда и немедленно, от любого источника
    if command in STOP_WORDS or type_command == "force_com":
        apply(command, robot.context)
        return

    if type_command == "com":
        apply(command, robot.context)
    elif type_command == "sys_com":
        handle_system(command)

Безопасность: остановка при потере команд

Если связь прервалась в момент движения, робот продолжит ехать с последней командой. Добавьте в основной цикл «сторожевой таймер»: если команд не было дольше заданного времени — остановка.

COMMAND_TIMEOUT = 0.5   # секунд

@robot.mainloop
def loop(ctx):
    last = ctx.get("last_command_time", 0)
    if time.monotonic() - last > COMMAND_TIMEOUT:
        ctx["motors"].drive(0, 0)

Для режима, когда оператор удерживает кнопку «вперёд», интерфейс или модель должны повторять команду чаще, чем раз в COMMAND_TIMEOUT. Для дискретных команд (движение до следующей команды) увеличьте таймаут или привяжите остановку к событию отключения.

Приоритет оператора над моделью

Правило приоритета применяется на сервере. Если роботу нужно учитывать его и на борту (например, при прямом подключении модели), различайте источники по author:

OPERATOR_HOLD = 2.0  # секунд после команды оператора модель не управляет

@robot.com_handler
def on_command(type_command, command, author=None):
    now = time.monotonic()
    is_model = str(author).startswith("model")
    if is_model and now - robot.context.get("operator_time", 0) < OPERATOR_HOLD:
        return                                   # оператор недавно вмешался
    if not is_model:
        robot.context["operator_time"] = now
    apply(command, robot.context)

Префикс model в идентификаторе автора — соглашение для примера; используйте идентификаторы, которые выдаёт ваша платформа.

Back to top