🔗 ROS2: middleware para robótica real

ROS2 reescribió el middleware sobre DDS (Data Distribution Service): comunicación descentralizada (sin roscore único), QoS configurable, compatible con tiempo real y seguridad (SROS2). Los bloques son nodos que se comunican mediante tópicos (pub/sub), servicios (req/resp) y acciones (objetivos largos con feedback).

Un sistema ROS2 descompone el robot en nodos con una sola responsabilidad: drivers de cámara/RGB-D, nodo de percepción, planificador, controlador de articulaciones. Esta modularidad permite cambiar la "policy" sin modificar el driver del brazo —exactamente donde entra el VLA.

# inspecionar o sistema vivo
ros2 node list
ros2 topic list
ros2 topic hz /camera/color/image_raw     # checar frequência
ros2 topic echo /joint_states --once
ros2 run rqt_graph rqt_graph              # visualizar o grafo

El VLA se convierte en un nodo rclpy que conecta la cámara y el estado, ejecuta el forward y publica el comando. La imagen y el estado deben venir del mismo instante: usa message_filters (ApproximateTimeSynchronizer) y elige la QoS cierta — BEST_EFFORT para imagen (descarta las atrasadas), RELIABLE para comandos.

import rclpy
from rclpy.node import Node
from message_filters import Subscriber, ApproximateTimeSynchronizer
from sensor_msgs.msg import Image, JointState
from rclpy.qos import QoSProfile, ReliabilityPolicy

class VLANode(Node):
    def __init__(self):
        super().__init__("vla_node")
        qos = QoSProfile(depth=1, reliability=ReliabilityPolicy.BEST_EFFORT)
        img = Subscriber(self, Image, "/camera/color/image_raw", qos_profile=qos)
        st  = Subscriber(self, JointState, "/joint_states")
        sync = ApproximateTimeSynchronizer([img, st], queue_size=2, slop=0.02)
        sync.registerCallback(self.on_obs)

    def on_obs(self, img, state):
        action = self.policy(preprocess(img), state.position)  # forward VLA
        self.pub.publish(to_cmd(action))                       # ação chunked

rclpy.init(); rclpy.spin(VLANode())

El VLA predice una pose en un frame, generalmente de la cámara. El brazo espera comandos en el frame de la base. tf2 mantiene el árbol de transformaciones temporizado (base_link → camera_link → tool0) y convierte poses entre frames respetando la marca de tiempo.

from tf2_ros import Buffer, TransformListener
import tf2_geometry_msgs

buf = Buffer(); TransformListener(buf, node)
# transformar a pose prevista (frame da câmera) -> frame base do braço
pose_base = buf.transform(pose_cam, "base_link", timeout=Duration(seconds=0.1))
# agora a pose está no frame que o controlador entende

MoveIt2 planifica el movimiento con verificación de colisiones, cinemática (IK/FK) y parametrización temporal respetando los límites de las articulaciones. En VLA reactivo de alta frecuencia, muchas veces ignoras el planner y envías setpoints directamente vía ros2_control — pero MoveIt2 se incorpora para una aproximación segura y la detección de colisiones.

La cadena cámara → inferencia → tf2 → controlador → torque tiene un presupuesto de latencia. Peor que una latencia alta es el jitter (varianza): un bucle irregular desestabiliza el control. Usa el kernel PREEMPT_RT, prioridad de tiempo real y watchdogs que detienen el brazo si el comando se retrasa.

# watchdog: se não chegar comando em 100ms, freia
self.create_timer(0.1, self.watchdog)
def watchdog(self):
    if (self.now() - self.last_cmd) > 0.1:
        self.hold_position()   # fail-safe

# medir latência fim-a-fim do pipeline
ros2 topic delay /arm_command