Разработка моделей ИИ

Контракт модели, демо-модель, конвейеры модулей

Разработка модели ИИ для платформы

Руководство для разработчиков моделей ИИ: как устроен контракт модели с платформой и как реализовать правила демо-прототипа.

Контракт модели

Модель на платформе ARTEX — это функция «наблюдение → команда»:

       ┌──────────────── входы ────────────────┐            ┌────── выход ──────┐
       │ кадр(ы) видео: BGR, от камер робота   │            │ команда:          │
       │ телеметрия: словарь JSON + время      │ ─▶ МОДЕЛЬ ─▶│ "forward", "stop",│
       │ признаки от бортовых моделей          │            │ … или None        │
       └───────────────────────────────────────┘            └───────────────────┘
Элемент Формат Источник
Кадр numpy-массив BGR (после декодирования JPEG) Поток видео робота, актуальный кадр из кэша ХАБа
Телеметрия Словарь, например {"battery": 11.8, "pose": {...}} Актуальный вектор телеметрии с временем создания
Команда Строка или структура из словаря команд робота (см. «ПО для робота» → Обработка команд) Отправляется роботу через ХАБ от имени модели
None «Ничего не менять» —

Требования к модели:

Демо-модель: правила прототипа

Условие Команда
Напряжение АКБ ниже 11 В stop
Модель видит красный прямоугольник stop
Модель видит белый прямоугольник forward

Порядок проверки важен: безопасность (АКБ) → запрещающий сигнал (красный) → разрешающий (белый).

# policy.py — правила демо-прототипа
import cv2
import numpy as np

BATTERY_MIN_V = 11.0      # ниже — «стоп»
MIN_AREA = 0.02           # минимальная доля кадра, занимаемая прямоугольником

def find_rectangle(mask, frame_area):
    """True, если на бинарной маске есть крупный четырёхугольник."""
    contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
    for c in contours:
        if cv2.contourArea(c) < MIN_AREA * frame_area:
            continue
        approx = cv2.approxPolyDP(c, 0.04 * cv2.arcLength(c, True), True)
        if len(approx) == 4 and cv2.isContourConvex(approx):
            return True
    return False

def see(frame):
    """Возвращает {'white': bool, 'red': bool} для BGR-кадра."""
    hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)
    area = frame.shape[0] * frame.shape[1]
    white = cv2.inRange(hsv, (0, 0, 200), (180, 40, 255))
    red = cv2.inRange(hsv, (0, 120, 80), (10, 255, 255)) | \
          cv2.inRange(hsv, (170, 120, 80), (180, 255, 255))
    kernel = np.ones((5, 5), np.uint8)
    white = cv2.morphologyEx(white, cv2.MORPH_OPEN, kernel)
    red = cv2.morphologyEx(red, cv2.MORPH_OPEN, kernel)
    return {"white": find_rectangle(white, area), "red": find_rectangle(red, area)}

def decide(frame, telemetry):
    """Правила демо-прототипа. Возвращает команду или None (ничего не менять)."""
    battery = telemetry.get("battery")
    if battery is not None and battery < BATTERY_MIN_V:
        return "stop"                       # 1. разряжена АКБ
    if frame is None:
        return None
    seen = see(frame)
    if seen["red"]:
        return "stop"                       # 2. красный прямоугольник
    if seen["white"]:
        return "forward"                    # 3. белый прямоугольник
    return None

Проверка на синтетических кадрах:

import numpy as np, cv2
from policy import decide

def frame_with(color=None):
    f = np.full((480, 640, 3), 60, np.uint8)
    if color:
        cv2.rectangle(f, (200, 150), (440, 330), color, -1)
    return f

assert decide(frame_with((255, 255, 255)), {"battery": 12.0}) == "forward"
assert decide(frame_with((0, 0, 255)),     {"battery": 12.0}) == "stop"
assert decide(frame_with(),                {"battery": 12.0}) is None
assert decide(frame_with((255, 255, 255)), {"battery": 10.5}) == "stop"

Пороги цвета в HSV зависят от освещения полигона и камеры — откалибруйте их по записанным кадрам.

Вариант с YOLO

Вместо цветовых порогов можно использовать детектор. Обучите YOLO на классах white_sign и red_sign (или используйте результаты YoloModule из конвейера — порт yolo.detect):

from ultralytics import YOLO

model = YOLO("signs.pt")          # веса, обученные на знаках полигона
CONF = 0.5

def see_yolo(frame):
    result = model(frame, verbose=False)[0]
    names = {result.names[int(b.cls)] for b in result.boxes if float(b.conf) >= CONF}
    return {"white": "white_sign" in names, "red": "red_sign" in names}

Подключение модели

Вариант 1. Модель на борту робота

Самый простой способ запустить демо — выполнить модель прямо в основном цикле робота. Решение модели применяется без сетевой задержки, а в телеметрию добавляется, что «видит» модель — оператор увидит это в интерфейсе.

import cv2
from artex_robot import Robot
from policy import decide, see

robot = Robot({"type": "Robot", "name": "demo-01", "telem_can": True, "video_can": True})
robot.log.set_level("skip_TODO")
cam = cv2.VideoCapture(0)
robot["auto"] = False            # автопилот включается командой auto_run
robot["battery"] = 12.0

@robot.com_handler
def on_command(type_command, command, author=None):
    if command == "auto_run":
        robot["auto"] = True
    elif command == "auto_stop":
        robot["auto"] = False
        apply("stop")
    else:
        robot["auto"] = False    # ручная команда оператора отключает автопилот
        apply(command)

@robot.telemetry("battery")
def battery():
    return robot["battery"]      # здесь — чтение АЦП

@robot.telemetry("ai")
def ai_state(decision, seen):
    return {"auto": robot["auto"], "decision": decision, **seen}

@robot.video("rgb")
def send_frame(frame):
    return frame

@robot.mainloop
def loop(ctx):
    battery()
    ok, frame = cam.read()
    if not ok:
        return
    send_frame(frame)
    decision = decide(frame, {"battery": ctx["battery"]})
    ai_state(decision, see(frame))
    if ctx["auto"] and decision and decision != ctx.get("last_decision"):
        apply(decision)
        ctx["last_decision"] = decision

def apply(command):
    robot.log("INFO", f"Выполняю {command}")
    # здесь — управление моторами (см. «ПО для робота» → «Примеры роботов»)

robot.init_connector()
robot.start()

Команды auto_run / auto_stop вводятся в поле команд на странице управления роботом в веб-интерфейсе.

Вариант 2. Модель в вычислительной среде платформы

Модель упаковывается в Docker-образ и запускается в вычислительной среде (AI Runtime). Платформа подаёт на вход модели актуальный кадр и вектор телеметрии выбранного робота и передаёт её команды роботу через ХАБ. Рекомендуемая структура модели — класс с явными методами входа и выхода, независимый от транспорта:

class DemoModel:
    name = "demo-rules"
    version = "1.0"

    def __init__(self):
        self.last = None

    def step(self, frame, telemetry):
        """Вызывается платформой на каждое обновление данных. Возвращает команду или None."""
        cmd = decide(frame, telemetry)
        if cmd == self.last:
            return None          # не повторять ту же команду
        self.last = cmd
        return cmd

FROM python:3.11-slim
RUN pip install --no-cache-dir opencv-python-headless numpy
WORKDIR /model
COPY policy.py model.py ./
CMD ["python", "model.py"]

Программный интерфейс подключения внешних моделей к ХАБу (методы get / push общей памяти, WebSocket/WebRTC-каналы) формируется на этапе прототипа «Альфа». Реализуйте логику модели как независимый класс, как в примере выше, — тогда перенос на итоговый транспорт сведётся к небольшому адаптеру.

Вариант 3. Модуль в конвейере

Для типовых задач (детекция YOLO) модель подключается как модуль в визуальном редакторе без программирования — см. Конвейеры модулей.

Чек-лист перед выпуском модели на полигон

Конвейеры модулей

Логика обработки данных на сервере собирается в визуальном редакторе модулей: модули перетаскиваются на рабочее поле, их порты соединяются линиями. Описание самих модулей и их портов — на странице Веб-интерфейс управления роботами; здесь — готовые схемы соединения для типовых задач.

Базовые правила

Схема 1. Ручное управление (минимальная)

Оператор видит видео и телеметрию робота и управляет им.

┌───────────────┐ robot.video      ┌────────────────┐
│ RobotWSModule │ ───────────────▶ │ ClientIOModule │
│               │ robot.telemetry  │  (Web UI)      │
│               │ ───────────────▶ │                │
│               │ robot.connected  │                │
│               │ ───────────────▶ │                │
│               │ robot.disconnected                │
│               │ ───────────────▶ │                │
│               │ ◀─────────────── │                │
└───────────────┘  robot.command   └────────────────┘

Схема 2. Детекция объектов поверх видео

Оператор управляет вручную, а YOLO подсвечивает найденные объекты.

                   robot.video   ┌────────────┐  yolo.detect   ┌────────────────┐
┌───────────────┐ ─────────────▶ │ YoloModule │ ─────────────▶ │                │
│ RobotWSModule │                └────────────┘                │ ClientIOModule │
│               │ robot.video, robot.telemetry, connected ...  │                │
│               │ ────────────────────────────────────────────▶│                │
└───────────────┘ ◀──────────────────────────────────────────── └────────────────┘
                                 robot.command

Схема 3. Автономное управление по видео

Конвейер полного автономного контура: захват → детекция → трекинг → решение → команда, с логированием.

 Захват видео ─▶ Детекция (YOLO) ─▶ Трекинг объектов ─▶ Принятие решения ─▶ Команда устройству
 (RobotWSModule)  (YoloModule)                              │                (robot.command)
                                                            └──▶ Логирование (S3 + DB)

Запуск и остановка автоматического управления — командами auto_run / auto_stop из поля ввода команд на странице робота. Оператор в любой момент может нажать «Стоп».

Схема 4. Запись датасета

Для сбора данных к источнику видео и телеметрии подключается модуль записи (VideoRecordModule). Запись включается командой video_start_record и останавливается командой video_stop_record.

┌───────────────┐ robot.video  ┌────────────────────┐
│ RobotWSModule │ ───────────▶ │ VideoRecordModule  │ ──▶ файл / S3
│               │              └────────────────────┘
│               │ robot.video, robot.telemetry ──────▶ ClientIOModule
└───────────────┘

Отладка конвейера

Проблема Проверка
В интерфейсе нет видео Соединён ли robot.video источника с ClientIOModule? Подключите ConsoleLogModule к robot.video
YOLO не выдаёт результатов Идёт ли видео на вход YoloModule; есть ли объекты известных модели классов
Команды не доходят до робота Соединён ли robot.command ClientIOModule с RobotWSModule; получил ли робот Ready
Высокая задержка Не подключено ли слишком много тяжёлых модулей к одному потоку; загрузка GPU в мониторинге