Alineación de Frecuencia de Inferencia en Máquinas Industriales Linux para Modelos Isaac Gym (Sim-to-Real) con Aprendizaje por Refuerzo

La migración de modelos de simulación a entornos físicos, especialmente en el ámbito de la robótica industrial y el control en tiempo real, presenta desafíos significativos en la sincronización temporal. Este artículo explora cómo lograr una alineación precisa entre la frecuencia de inferencia de un modelo de aprendizaje por refuerzo, entrenado en NVIDIA Isaac Gym, y los requisitos de tiempo real de una máquina controladora industrial bajo Linux.

Los sistemas de control industrial a menudo operan bajo estrictos requisitos de latencia y jitter. Una discrepancia en los ciclos de tiempo entre la simulación y la ejecución en la máquina real puede resultar en comportamientos erráticos del robot, fallos en la percepción y, en última instancia, incumplimiento de las normativas de seguridad. Para abordar esto, se propone una solución que combina hardware de temporización de precisión con capacidades de Linux en tiempo real (PREEMPT_RT).

Conceptos Clave:

  • Sim-to-Real: Transferencia de una política de control aprendida en simulación al mundo físico.
  • Frecuencia de Inferencia: El período con el que se ejecuta la red neuronal para generar acciones. El objetivo es igualar o mejorar la frecuencia de la simulación (ej. 1 kHz o 500 Hz).
  • Temporizadores de Hardware: Mecanismos en la CPU (como TSC/HPET) que generan señales de tiempo con precisión de nanosegundos. Se utiliza la interfaz timerfd\_settime() para interactuar con ellos.
  • PREEMPT_RT: Un parche para el kernel de Linux que reduce drásticamente la latencia de conmutación de tareas a menos de 50 microsegundos.
  • Paso de Tiempo: La consistencia entre el intervalo de tiempo simulado (delta t) y el intervalo de tiempo real de ejecución es crucial para la estabilidad.

Preparación del Entorno:

Se requiere una máquina industrial (x86_64), un robot compatible (como un cuadrúpedo) con interfaces de comunicación (CAN-FD), y un sistema operativo Linux con el kernel PREEMPT_RT. La instalación del kernel RT se puede automatizar con un script sencillo.

#!/bin/bash
# install_rt_kernel.sh
RT_KERNEL_VERSION="5.15.71-rt53"
# Descarga e instalación de paquetes del kernel RT (ejemplo para Ubuntu)
# Se deben ajustar las URLs y nombres de archivo según la distribución y versión
wget https://cdn.kernel.org/pub/linux/kernel/v5.x/patch-5.15.71-rt53.patch.xz
# ... (comandos para descargar e instalar .deb del kernel) ...
sudo dpkg -i linux-image-*.deb linux-headers-*.deb
sudo update-grub
sudo reboot
# Verificar la instalación
uname -r

Tras reiniciar y seleccionar el kernel RT, se verifica la versión con uname -r. Se recomienda el uso de Isaac Gym y PyTorch.

Caso de Uso: Robot Cuadrúpedo en Entorno Industrial

Un robot cuadrúpedo entrenado para caminar a 500 Hz (paso de tiempo de 2 ms) en Isaac Gym. Al desplegarlo en una máquina industrial con un Linux estándar, la fluctuación en el ciclo de ejecución (jitter) del hilo de infeerncia, debido a la programación del sistema operativo, puede variar entre 1.8 y 3.2 ms. Esto provoca:

  • Movimientos bruscos y temblores en las articulaciones.
  • Pérdida de fotogramas en sistemas de SLAM visual, afectando la localización.
  • Activación de sistemas de seguridad por exceder los límites de tiempo del ciclo, resultando en paradas de emergencia.

Mediante la combinación de temporizadores de hardware y el kernel PREEMPT_RT, se puede fijar el ciclo de inferencia a 2 ms con una desviación máxima de 0.1 ms. Esto asegura que el comportamiento del robot en el mundo real sea consistente con la simulación, eliminando la necesidad de reajustar parámetros y reduciendo significativamente el tiempo de depuración. Además, cumple con normativas de seguridda industrial como IEC 61508 SIL 2.

Implementación Práctica: Desde el Modelo a la Ejecución en Tiempo Real

1. Exportación del Modelo

El modelo entrenado en PyTorch se exporta a TorchScript para una inferencia más eficiente en C++.

# export_model.py
import torch

# Cargar el modelo entrenado
policy_model = torch.load("policy.pt")
policy_model.eval()

# Convertir a TorchScript
scripted_model = torch.jit.script(policy_model)
scripted_model.save("policy_ts.pt")

2. Encapsulación del Temporizador de Tiempo Real

Se crea una clase simple para manejar la temporización usando timerfd.

// rt_timer.hpp
#pragma once
#include <sys>
#include <unistd.h>
#include <cstdint>

class RealTimeTimer {
public:
    explicit RealTimeTimer(uint64_t nanoseconds) : interval_ns(nanoseconds) {
        timer_fd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timer_fd == -1) {
            perror("timerfd_create");
            // Manejar error
        }
        set_timer();
    }

    ~RealTimeTimer() {
        close(timer_fd);
    }

    void wait_for_tick() {
        uint64_t expirations;
        ssize_t ret = read(timer_fd, &expirations, sizeof(expirations));
        if (ret == -1) {
            perror("read timerfd");
            // Manejar error
        }
    }

private:
    void set_timer() {
        struct itimerspec ts = {};
        ts.it_value.tv_nsec = interval_ns; // Tiempo hasta la primera expiración
        ts.it_interval.tv_nsec = interval_ns; // Intervalo entre expiraciones
        if (timerfd_settime(timer_fd, 0, &ts, nullptr) == -1) {
            perror("timerfd_settime");
            // Manejar error
        }
    }

    int timer_fd;
    uint64_t interval_ns;
};
</cstdint></unistd.h></sys>

3. Programa Principal de Inferencia en Tiempo Real

El programa principal configura el hilo para ejecución en tiempo real, carga el modelo y ejecuta el ciclo de inferencia sincronizado con el temporizador.

// main_inference.cpp
#include <torch>
#include <iostream>
#include <vector>
#include <pthread.h>
#include <sched.h>
#include <sys> // Para mlockall
#include "rt_timer.hpp"

// Constantes
const int OBSERVATION_DIM = 12;
const uint64_t CYCLE_TIME_NS = 2000000; // 2 ms para 500 Hz

int main() {
    // Bloquear memoria para evitar paginación
    if (mlockall(MCL_CURRENT | MCL_FUTURE) == -1) {
        perror("mlockall");
        // Manejar error
    }

    // Configurar política de planificación SCHED_FIFO
    struct sched_param scheduling_param;
    scheduling_param.sched_priority = 95; // Prioridad alta (ajustable)
    if (pthread_setschedparam(pthread_self(), SCHED_FIFO, &scheduling_param) != 0) {
        perror("pthread_setschedparam");
        // Manejar error
    }

    // Cargar el modelo TorchScript
    torch::jit::script::Module inference_model;
    try {
        inference_model = torch::jit::load("policy_ts.pt");
        inference_model.eval();
    } catch (const c10::Error& e) {
        std::cerr << "Error loading the model: " << e.what() << std::endl;
        return -1;
    }

    // Inicializar el temporizador en tiempo real
    RealTimeTimer cycle_timer(CYCLE_TIME_NS);

    // Buffer de observación (ejemplo)
    std::vector<float> current_observation(OBSERVATION_DIM, 0.0f);

    // Bucle principal de inferencia
    while (true) {
        cycle_timer.wait_for_tick(); // Esperar al próximo ciclo

        // Preparar la entrada para el modelo
        // El tensor se crea directamente desde los datos del vector sin copiar
        torch::Tensor observation_tensor = torch::from_blob(
            current_observation.data(),
            {1, OBSERVATION_DIM}, // Forma del tensor (batch_size, features)
            torch::kFloat32);

        // Ejecutar la inferencia
        torch::Tensor output_tensor;
        try {
             output_tensor = inference_model.forward({observation_tensor}).toTensor();
        } catch (const c10::Error& e) {
            std::cerr << "Error during model forward pass: " << e.what() << std::endl;
            // Manejar error o continuar
            continue;
        }


        // Procesar la salida (ej. comandos para actuadores)
        float* actuator_commands = output_tensor.data_ptr<float>();
        // En una aplicación real, estos comandos se enviarían a los actuadores
        // Aquí, se imprimen como ejemplo
        for (int i = 0; i < output_tensor.size(1); ++i) {
            printf("%.4f ", actuator_commands[i]);
        }
        printf("\n");

        // Aquí se actualizaría la observación para el siguiente ciclo
        // current_observation = read_sensor_data();
    }

    return 0;
}
</float></float></sys></sched.h></pthread.h></vector></iostream></torch>

4. Compilación y Ejecución

Se utiliza CMake para la compilación, asegurando la vinculación con la biblioteca de PyTorch.

# CMakeLists.txt
cmake_minimum_required(VERSION 3.16)
project(RealTimeInference)

set(CMAKE_CXX_STANDARD 17)
set(CMAKE_CXX_STANDARD_REQUIRED True)

# Buscar y vincular con PyTorch
find_package(Torch REQUIRED)
message(STATUS "Torch include dirs: ${TORCH_INCLUDE_DIRS}")
message(STATUS "Torch libraries: ${TORCH_LIBRARIES}")

add_executable(main_inference main_inference.cpp rt_timer.cpp) # Asumiendo rt_timer.cpp si no es header-only
target_link_libraries(main_inference PRIVATE ${TORCH_LIBRARIES} pthread)

# Incluir directorios de cabecera de PyTorch
target_include_directories(main_inference PRIVATE ${TORCH_INCLUDE_DIRS})

Un script de inicio simplifica el proceso de ejecución con los permisos necesarios.

#!/bin/bash
# start_inference.sh

# Bloquear memoria de forma anticipada para minimizar page faults iniciales
sudo mlockall -M 4G # Ejemplo: Bloquear hasta 4GB, ajustar según necesidad

# Ejecutar la aplicación con alta prioridad en tiempo real
# El número '95' debe coincidir con la prioridad definida en el código C++
sudo chrt -f 95 ./main_inference

Se compila y ejecuta:

mkdir build && cd build
cmake ..
make
chmod +x ../start_inference.sh
sudo ../start_inference.sh

Optimización y Mejores Prácticas

  • Aislamiento de CPU: Dedicar núcleos específicos para el hilo de inferencia mediante la opción isolcpus en el arranque del kernel y pthread\_setaffinity\_np.
  • Pre-asignación de Memoria: Utilizar contenedores como std::array o pools de memoria para evitar asignaciones dinámicas en el bucle crítico.
  • Doble Buffer: Implementar colas sin bloqueo (lock-free queues) para la comunicación entre el hilo de inferencia y otros hilos (ej. lectura de sensores).
  • Monitorización de Jitter: Integrar mediciones de latencia dentro del bucle de ejecución para obtener métricas de jitter en tiempo real.
  • Integración Continua: Automatizar pruebas de latencia (ej. con cyclictest) en pipelines CI/CD para garantizar la estabilidad a lo largo del tiempo.

Conclusión:

La alineación precisa de la frecuencia de inferencia entre la simulación y el entorno real es fundamental para la implementación exitosa de sistemas de IA en robótica industrial. La combinación de herramientas de software como Isaac Gym y PyTorch con las capacidades de tiempo real de Linux y hardware de temporización permite alcanzar la sincronización necesaria, logrando un rendimiento robótico predecible y fiable.

Etiquetas: linux Tiempo Real RTOS Isaac Gym nvidia

Publicado el 7-6 04:38

Etiquetas populares

©2026 Friki Work