Встраиваем AI в ROS 2: от камеры до управления

Клиент пришёл с UGV на базе ROS 2 Humble: стереопара глубины, LiDAR Velodyne VLP-16 и Jetson Orin NX. Требование — object detection с частотой 30 Hz, но PyTorch YOLOv8 в лоб давал 12 FPS. Проблема: CPU-bound из-за OpenCV-конвертации и блокирующий callback executor. Рассказываем, как мы переписали архитектуру ноды и выжали 125 FPS.

Наша команда специализируется на интеграции AI-моделей в ROS 2 уже более 7 лет. Мы научились обходить типовые проблемы: latency, перегрузка шины, несовместимость фреймворков. Ниже — реальные кейсы и проверенные решения.

Архитектура ROS 2 + AI: как это работает?

AI-модель в ROS 2 — это Node, подписанный на sensor topics и публикующий результаты. Ключевое: не блокировать callback executor. Мы проектируем ноду так, что инференс выполняется в отдельном потоке, а публикация — через publisher.

import rclpy from rclpy.node import Node from sensor_msgs.msg import Image from vision_msgs.msg import Detection2DArray from cv_bridge import CvBridge import torch import numpy as np class ObjectDetectionNode(Node): def __init__(self): super().__init__('object_detection') # Загрузка модели self.model = torch.hub.load( 'ultralytics/yolov8', 'yolov8n', pretrained=True ) self.model.eval() if torch.cuda.is_available(): self.model.cuda() self.bridge = CvBridge() # Подписка на камеру self.subscription = self.create_subscription( Image, '/camera/color/image_raw', self.image_callback, 10 # QoS depth ) # Публикация результатов self.detection_pub = self.create_publisher( Detection2DArray, '/detections', 10 ) self.get_logger().info('Object detection node started') def image_callback(self, msg: Image): # Конвертация ROS Image → numpy cv_image = self.bridge.imgmsg_to_cv2(msg, desired_encoding='rgb8') # Инференс with torch.no_grad(): results = self.model(cv_image) # Конвертация в ROS Detection2DArray detections = self._to_detection_array(results, msg.header) self.detection_pub.publish(detections) def main(): rclpy.init() node = ObjectDetectionNode() rclpy.spin(node) rclpy.shutdown() 

LiDAR + PointNet для 3D восприятия: как это реализовать?

Для 3D сцен используем PointNet. LiDAR данные в формате PointCloud2 конвертируем в numpy, субсэмплируем до 1024 точек и прогоняем через модель.

from sensor_msgs.msg import PointCloud2 import sensor_msgs_py.point_cloud2 as pc2 class LidarPerceptionNode(Node): def __init__(self): super().__init__('lidar_perception') self.pointnet_model = load_pointnet_model('pointnet_weights.pth') self.sub = self.create_subscription( PointCloud2, '/velodyne_points', self.lidar_callback, 10 ) def lidar_callback(self, msg: PointCloud2): # Конвертация PointCloud2 → numpy (N, 3) points = np.array(list(pc2.read_points(msg, field_names=("x","y","z")))) # Субсэмплирование до 1024 точек для PointNet indices = np.random.choice(len(points), 1024, replace=False) sampled = points[indices] # Инференс tensor = torch.FloatTensor(sampled).unsqueeze(0).cuda() with torch.no_grad(): classes = self.pointnet_model(tensor) # Публикация классифицированных объектов 

Оптимизация latency: как добиться 30 Hz?

Latency — главный враг real-time. Стандартный PyTorch на Jetson Orin выдаёт ~45ms (22 Hz). Наш опыт: TensorRT (см. официальная документация NVIDIA) даёт 3-5x ускорение — до 8ms (125 Hz). Дополнительно:

• Параллельный инференс в отдельном потоке (не блокирует executor)
• CUDA streams для обработки нескольких кадров
• INT8 quantization с калибровкой на репрезентативном датасете

Сравнение подходов: точность vs скорость

TensorRT в INT8 даёт ускорение в 5,6 раза при потере точности всего 2% — идеально для real-time.

Навигация с RL policy: что входит?

from geometry_msgs.msg import Twist from nav_msgs.msg import Odometry class RLNavigationNode(Node): def __init__(self): super().__init__('rl_navigation') self.policy = load_stable_baselines3_model('navigation_policy.zip') self.cmd_vel_pub = self.create_publisher(Twist, '/cmd_vel', 10) self.odom_sub = self.create_subscription(Odometry, '/odom', self.step, 10) def step(self, odom_msg): state = self._odom_to_state(odom_msg) action, _ = self.policy.predict(state, deterministic=True) cmd = Twist() cmd.linear.x = float(action[0]) cmd.angular.z = float(action[1]) self.cmd_vel_pub.publish(cmd) 

Политика обучена в симуляторе (Isaac Gym) и перенесена на реального робота. Наш опыт: fine-tuning на реальных данных сокращает разрыв sim-to-real на 40%.

Мониторинг и наблюдаемость AI-нод

Производительность AI-ноды в ROS 2 необходимо отслеживать в реальном времени. Мы встраиваем следующие инструменты:

• Диагностические топики: /diagnostics публикует latency инференса, FPS и загрузку GPU каждые 1 секунду. Стандарт ROS 2 diagnostic_msgs.
• Prometheus + Grafana: экспортируем custom metrics через prometheus_client Python. Дашборд показывает перцентили p50/p95/p99 latency за скользящее окно 5 минут.
• Rosbag записи: критические инциденты (latency > 2× от baseline) автоматически записываются в rosbag для анализа.
• Data drift: входные распределения пикселей/точек сравниваются с обучающим датасетом через KS-тест. Алерт при p-value < 0.05 — сигнал о деградации качества детекции.

Такая observability позволяет обнаружить деградацию модели до того, как она повлияет на миссию робота.

Процесс работы: пошагово

1. Аналитика: разбор latency-бюджета, выбор модели, профилирование.
2. Проектирование: архитектура ноды, схема топиков, QoS.
3. Реализация: написание C++/Python кода, интеграция с TensorRT/ONNX.
4. Тестирование: симуляция (Gazebo + гонки пикселей) и реальное железо.
5. Деплой: контейнеризация Docker, ROS 2 Launch файлы, мониторинг.

Типичные ошибки при интеграции

• Забыть настроить QoS depth: при 30 FPS достаточно 10, иначе переполнение очереди.
• Использовать один поток для инференса и callbacks — ведёт к просадкам.
• Не профилировать конвертацию изображений: cv_bridge может занимать 30% времени.
• Игнорировать calibration dataset для INT8 — падает точность на 10+%.

Почему стоит работать с нами?

Мы — команда AI/ML инженеров с 7+ годами опыта в робототехнике. Более 30 успешных интеграций AI в ROS (от беспилотных дронов до промышленных манипуляторов). Гарантируем стабильную работу perception pipeline на целевой платформе. Оцените свой проект — получите консультацию. Интеграция занимает от 2 до 4 недель, стоимость рассчитывается индивидуально.

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