🚦 04. Seguimiento y Señales

En lugar de entrenar una inteligencia artificial desde cero (un proceso muy largo que haremos en portátiles o en servidores potentes), muchas veces nos basta con utilizar Visión Artificial Clásica. Esto se llama procesar los píxeles en crudo mediante algoritmos como los que ofrece la librería OpenCV.

Si entendemos qué estamos buscando (un color vivo, una forma de 8 lados), podemos darle órdenes directas al Jetracer.

Guía Paso a Paso

1. Cargar el "Cerebro" (Haar Cascade)

Un Cascade es un archivo .xml pequeñito que ya contiene la información algorítmica de alto contraste sobre cómo es una señal de tráfico (por ejemplo, una señal de Stop). Cargamos ese archivo primero:

import cv2
import os

cascade_path = os.path.expanduser("~/jetracer/notebooks/cascades/stop_cfizette_haar_19.xml")

if os.path.exists(cascade_path):
    stop_cascade = cv2.CascadeClassifier(cascade_path)
else:
    print("Archivo XML de cascade no encontrado.")
    stop_cascade = None

2. Configurar Interfaz y GStreamer

Como en cuadernos anteriores, no podemos usar .imshow() nativo sin bloquear el navegador, así que montamos un widget de imagen e incrustamos el pipeline de compresión GStreamer. Añadimos un botón de STOP de emergencia.

import ipywidgets as widgets
from IPython.display import display
import threading
import time

def gstreamer_pipeline():
    return ("nvarguscamerasrc ! video/x-raw(memory:NVMM), width=1280, height=720, format=NV12, framerate=30/1 ! nvvidconv flip-method=0 ! video/x-raw, width=640, height=360, format=BGRx ! videoconvert ! video/x-raw, format=BGR ! appsink max-buffers=1 drop=True")

cap = cv2.VideoCapture(gstreamer_pipeline(), cv2.CAP_GSTREAMER)

image_widget = widgets.Image(format='jpeg', width=640, height=360)
stop_button = widgets.Button(description='⏹ Parar Hardware', button_style='danger')

running = True
def on_stop(b):
    global running
    running = False
    cap.release()
    stop_button.disabled = True
    stop_button.description = 'Cámara Liberada'
stop_button.on_click(on_stop)

3. El Bucle de Inferencia (Detectar si hay objeto)

Este hilo toma un "frame", lo copia, lo pasa a Escala de Grises (los Cascades solo funcionan en blanco y negro por contraste de sombras) y ecualiza el brillo. Luego usa detectMultiScale. Si devuelve una "caja" de colisión, se dibuja un rectángulo rojo (0, 0, 255) sobre esa foto. Todo ocurre 30 veces por segundo.

def hilo_camara():
    while running:
        ret, frame = cap.read()
        if ret:
            result = frame.copy()

            # Pasar a gris para el clasificador XML
            gray = cv2.cvtColor(result, cv2.COLOR_BGR2GRAY)
            gray = cv2.equalizeHist(gray)

            if stop_cascade is not None and not stop_cascade.empty():
                hits = stop_cascade.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=4)

                # Por cada detección en la imagen: X, Y (Cordenadas) W, H (ancho y alto de la caja)
                for (x, y, w, h) in hits:
                     cv2.rectangle(result, (x, y), (x+w, y+h), (0, 0, 255), 3)
                     cv2.putText(result, "STOP DETECTADO", (x, y-10), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)

            _, jpeg = cv2.imencode('.jpg', result)
            image_widget.value = jpeg.tobytes()

        time.sleep(1/30) # Refrescar a ~30FPS

---

Código Completo

Con todo el bucle preparado, ejecuta este bloque global para probar la detección local. (Coloca una foto pequeña de móvil con una señal de STOP delante de la cámara del Jetracer para ver si lo pilla).

import cv2
import numpy as np
import ipywidgets as widgets
from IPython.display import display
import threading
import time
import os

# -- 1. Cargar el cascade de STOP --
cascade_path = os.path.expanduser("~/jetracer/notebooks/cascades/stop_cfizette_haar_19.xml")
if os.path.exists(cascade_path):
    stop_cascade = cv2.CascadeClassifier(cascade_path)
else:
    print("No se encontró el XML de la señal. Busca los cascades.")
    stop_cascade = None

# -- 2. Configurar la cámara virtual (Gstreamer Pipeline) --
def gstreamer_pipeline():
    return ("nvarguscamerasrc ! video/x-raw(memory:NVMM), width=1280, height=720, format=NV12, framerate=30/1 ! nvvidconv flip-method=0 ! video/x-raw, width=640, height=360, format=BGRx ! videoconvert ! video/x-raw, format=BGR ! appsink max-buffers=1 drop=True")

cap = cv2.VideoCapture(gstreamer_pipeline(), cv2.CAP_GSTREAMER)

# -- 3. Interfaz Visual --
image_widget = widgets.Image(format='jpeg', width=640, height=360)
stop_button = widgets.Button(description='⏹ Parar', button_style='danger')

running = True
def on_stop(b):
    global running
    running = False
    cap.release()
    stop_button.disabled = True
    stop_button.description = 'Liberada'
stop_button.on_click(on_stop)


def hilo_camara():
    while running:
        ret, frame = cap.read()
        if ret:
            result = frame.copy()

            # Filtro simple Gris y Cascade
            gray = cv2.cvtColor(result, cv2.COLOR_BGR2GRAY)
            gray = cv2.equalizeHist(gray)

            # Buscar el patrón si el archivo de IA ha cargado bien
            if stop_cascade is not None and not stop_cascade.empty():
                hits = stop_cascade.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=4)

                for (x, y, w, h) in hits:
                     cv2.rectangle(result, (x, y), (x+w, y+h), (0, 0, 255), 3)
                     cv2.putText(result, "STOP DETECTADO", (x, y-10), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)

            # Visualizar            
            _, jpeg = cv2.imencode('.jpg', result)
            image_widget.value = jpeg.tobytes()

        time.sleep(1/30)

# -- 4. Lanzar ejecución general --
display(widgets.VBox([image_widget, stop_button]))

thread = threading.Thread(target=hilo_camara, daemon=True)
thread.start()