Estimación de Flujo Óptico Denso con Farneback y Raft en OpenCV y PyTorch

El flujo óptico es un concepto central en la visión artificial que describe el patrón de movimiento aparente de objetos, superficies y bordes en una secuencia de imágenes. Nos permite entender cómo los píxeles se mueven entre fotogramas consecutivos de un video, lo cual es crucial para aplicaciones como el seguimiento de objetos, la reconstrucción 3D, la estabilización de video y la navegación autónoma.

Existen dos tipos principales de flujo óptico: disperso y denso. El flujo óptico disperso calcula el movimiento de un pequeño subconjunto de puntos de interés, mientras que el flujo óptico denso, en el que nos centraremos en este tutorial, calcula el vector de movimiento para cada píxel de la imagen. Esto proporciona una comprensión mucho más rica y detallada del movimiento en la escena.

¿Qué es el Flujo Óptico Denso? 🚀

Imagina que tienes dos imágenes consecutivas de un video. El flujo óptico denso intenta responder a la pregunta: "¿A dónde se movió cada píxel de la primera imagen en la segunda imagen?". El resultado es un campo vectorial bidimensional del mismo tamaño que la imagen, donde cada vector (x, y) en una posición (i, j) indica el desplazamiento del píxel de la primera imagen en (i, j) a la segunda imagen.

Aplicaciones del Flujo Óptico Denso ✨

El flujo óptico denso es una herramienta increíblemente versátil con un sinfín de aplicaciones:

• Seguimiento de Objetos: Mantener un seguimiento preciso de múltiples objetos en movimiento.
• Estabilización de Video: Compensar el movimiento indeseado de la cámara.
• Reconocimiento de Actividades: Analizar patrones de movimiento para identificar acciones humanas (correr, saltar, etc.).
• Navegación Robótica: Ayudar a los robots a estimar su propio movimiento (odometría visual) y detectar obstáculos.
• Realidad Aumentada/Virtual: Integrar objetos virtuales en entornos reales de forma coherente con el movimiento de la cámara.
• Compresión de Video: Eliminar la redundancia temporal codificando solo los vectores de movimiento en lugar de píxeles completos.

Algoritmo de Farneback para Flujo Óptico Denso (OpenCV) 🛠️

El algoritmo de Farneback es un método clásico y robusto para la estimación de flujo óptico denso, propuesto por Gunnar Farneback en 2003. Se basa en la aproximación local de cada vecindad de píxeles con un polinomio cuadrático, y luego se estima el movimiento de estos polinomios entre fotogramas.

Principios de Farneback 💡

1. Aproximación Polinomial: Para cada vecindad de píxeles, el algoritmo ajusta un polinomio cuadrático. Este polinomio captura la estructura local de la intensidad de la imagen.
2. Movimiento Polinomial: Cuando la imagen se mueve, el polinomio también se mueve. El algoritmo estima el desplazamiento que minimiza la diferencia entre los polinomios ajustados en los dos fotogramas.
3. Pirámides de Imágenes: Para manejar movimientos grandes, Farneback utiliza una estrategia de pirámides de imágenes. El flujo se calcula inicialmente en una versión de baja resolución de las imágenes y luego se propaga y refina a resoluciones más altas. Esto permite capturar movimientos tanto grandes como pequeños de manera eficiente.

Implementación con OpenCV en Python 🐍

OpenCV proporciona una implementación muy sencilla y eficiente del algoritmo de Farneback a través de la función cv.calcOpticalFlowFarneback(). Necesitaremos dos fotogramas consecutivos (o imágenes) y algunas configuraciones.

import cv2
import numpy as np

# Función para visualizar el flujo óptico
def draw_flow(img, flow, step=16):
    h, w = img.shape[:2]
    y, x = np.mgrid[step/2:h:step, step/2:w:step].reshape(2,-1).astype(int)
    fx, fy = flow[y,x].T
    lines = np.vstack([x, y, x+fx, y+fy]).T.reshape(-1, 2, 2)
    lines = np.int32(lines + 0.5)
    vis = cv2.cvtColor(img, cv2.COLOR_GRAY2BGR)
    cv2.polylines(vis, lines, 0, (0, 255, 0))
    for (x1, y1), (_x2, _y2) in lines:
        cv2.circle(vis, (x1, y1), 1, (0, 255, 0), -1)
    return vis

def main_farneback():
    # Captura de video (0 para webcam, o 'nombre_archivo.mp4')
    cap = cv2.VideoCapture(0)

    if not cap.isOpened():
        print("Error: No se pudo abrir la fuente de video.")
        return

    ret, frame1 = cap.read()
    if not ret:
        print("Error: No se pudo leer el primer fotograma.")
        return

    prev_gray = cv2.cvtColor(frame1, cv2.COLOR_BGR2GRAY)
    hsv = np.zeros_like(frame1)
    hsv[...,1] = 255

    while True:
        ret, frame2 = cap.read()
        if not ret:
            break

        next_gray = cv2.cvtColor(frame2, cv2.COLOR_BGR2GRAY)

        # Calcula el flujo óptico con Farneback
        # Parámetros:
        # prev: primera imagen en escala de grises
        # next: segunda imagen en escala de grises
        # flow: matriz de salida del flujo (inicialmente None)
        # pyr_scale: factor de reducción de la pirámide (0.5 = cada nivel es la mitad del anterior)
        # levels: número de niveles de la pirámide
        # winsize: tamaño de la ventana de promediado
        # iterations: número de iteraciones en cada nivel
        # poly_n: tamaño de la vecindad de píxeles para el ajuste polinomial (generalmente 5 o 7)
        # poly_sigma: desviación estándar del gaussiano para el ajuste polinomial
        # flags: opciones como OPTFLOW_FARNEBACK_GAUSSIAN
        flow = cv2.calcOpticalFlowFarneback(prev_gray, next_gray, None, 0.5, 3, 15, 3, 5, 1.2, 0)

        # Visualización del flujo
        mag, ang = cv2.cartToPolar(flow[...,0], flow[...,1])
        hsv[...,0] = ang*180/np.pi/2
        hsv[...,2] = cv2.normalize(mag, None, 0, 255, cv2.NORM_MINMAX)
        bgr = cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR)

        # Muestra el resultado
        cv2.imshow('Original', frame2)
        cv2.imshow('Flujo Optico Farneback', bgr)
        cv2.imshow('Flujo con vectores', draw_flow(next_gray, flow))

        k = cv2.waitKey(30) & 0xff
        if k == 27: # ESC para salir
            break

        prev_gray = next_gray

    cap.release()
    cv2.destroyAllWindows()

if __name__ == '__main__':
    main_farneback()

Explicación del código:

1. draw_flow: Esta función auxiliar toma una imagen en escala de grises y el campo de flujo calculado, y superpone flechas que representan los vectores de movimiento. Esto ayuda a visualizar la dirección y magnitud del movimiento.
2. Captura de Video: Iniciamos la captura desde una webcam (cv2.VideoCapture(0)) o un archivo de video.
3. prev_gray y next_gray: Leemos fotogramas consecutivos y los convertimos a escala de grises, ya que calcOpticalFlowFarneback trabaja con imágenes de un solo canal.
4. cv2.calcOpticalFlowFarneback(): Esta es la función central. Sus parámetros controlan la precisión y el rendimiento. Experimentar con pyr_scale, levels, winsize, poly_n, y poly_sigma puede tener un impacto significativo en los resultados.
5. Visualización HSV: El flujo óptico (vectores fx, fy) se convierte a coordenadas polares (magnitud, ángulo). La magnitud representa la velocidad del movimiento y el ángulo, la dirección. Estos se mapean a los canales H (tono), S (saturación) y V (valor) de un espacio de color HSV para crear una visualización intuitiva: el tono indica la dirección y la intensidad, la velocidad.

Estimación de Flujo Óptico con RAFT (PyTorch) 🔥

RAFT (Recurrent All-Pairs Field Transforms) es un enfoque de aprendizaje profundo de última generación para la estimación de flujo óptico, presentado por Zachary Teed y Jia Deng en 2020. A diferencia de los métodos clásicos como Farneback, RAFT utiliza una red neuronal convolucional recurrente para iterar sobre una representación de correlación multidimensional y refinar progresivamente la estimación del flujo.

Arquitectura de RAFT 🎯

La arquitectura de RAFT se puede dividir en varios componentes clave:

1. Extractores de Características: Dos redes siamesas extraen características de alto nivel de los dos fotogramas de entrada (imagen1 y imagen2).
2. Mapas de Correlación: Se construye un mapa de correlación de "todos los pares" (all-pairs correlation volume) entre las características de ambas imágenes. Este volumen codifica la similitud entre cada parche de la primera imagen y cada parche de la segunda, permitiendo buscar correspondencias.
3. Unidad de Actualización Recurrente (GRU): Una unidad de actualización recurrente tipo GRU (Gated Recurrent Unit) itera para refinar la estimación del flujo. En cada iteración, la GRU procesa la estimación de flujo actual, el contexto de la imagen y las características del mapa de correlación para producir una actualización del flujo.
4. Decodificador de Flujo: Finalmente, un decodificador convierte la representación de la GRU en el campo de flujo óptico denso.

Implementación con PyTorch 🐍

Para usar RAFT, generalmente se descarga un modelo preentrenado. Utilizaremos la implementación oficial de RAFT que suele estar disponible en GitHub.

Pasos previos:

1. Instalar PyTorch y CUDA (si tienes GPU):

pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 # Para CUDA 11.8
# O pip install torch torchvision torchaudio # Para CPU

2. Clonar el repositorio de RAFT:

git clone https://github.com/princeton-vl/RAFT.git
cd RAFT
pip install -r requirements.txt

3. Descargar un modelo preentrenado: Los modelos preentrenados suelen estar disponibles en los lanzamientos del repositorio o enlaces proporcionados. Por ejemplo, raft-things.pth o raft-small.pth.

import sys
import os
import argparse
import torch
import numpy as np
import cv2

# Añadir el directorio RAFT al path para poder importar el módulo
# Asegúrate de que este script esté en el mismo directorio que la carpeta RAFT
# O ajusta el path_to_raft si RAFT está en otro lugar
path_to_raft = './RAFT'
if path_to_raft not in sys.path:
    sys.path.append(path_to_raft)

from raft import RAFT
from raft import utils as raft_utils

def load_image(imfile):
    img = np.array(cv2.imread(imfile)).astype(np.uint8)
    img = torch.from_numpy(img).permute(2, 0, 1).float()
    return img[None].cuda() if torch.cuda.is_available() else img[None]

def show_flow(image1, image2, flow_up):
    # Convierte el flujo a una imagen para visualización (modo HSV)
    flow_np = flow_up[0].permute(1,2,0).cpu().numpy()
    
    # Calcula magnitud y ángulo
    mag, ang = cv2.cartToPolar(flow_np[...,0], flow_np[...,1])
    hsv = np.zeros_like(image1[0].permute(1,2,0).cpu().numpy(), dtype=np.uint8)
    hsv[...,1] = 255
    hsv[...,0] = ang * 180 / np.pi / 2
    hsv[...,2] = cv2.normalize(mag, None, 0, 255, cv2.NORM_MINMAX)
    bgr_flow = cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR)

    # Convierte las imágenes originales para mostrar
    img1_np = image1[0].permute(1,2,0).cpu().numpy().astype(np.uint8)
    img2_np = image2[0].permute(1,2,0).cpu().numpy().astype(np.uint8)
    img1_np = cv2.cvtColor(img1_np, cv2.COLOR_RGB2BGR)
    img2_np = cv2.cvtColor(img2_np, cv2.COLOR_RGB2BGR)

    cv2.imshow('Image 1', img1_np)
    cv2.imshow('Image 2', img2_np)
    cv2.imshow('RAFT Flow', bgr_flow)
    cv2.waitKey(0)
    cv2.destroyAllWindows()

def main_raft(args):
    model = torch.nn.DataParallel(RAFT(args))
    model.load_state_dict(torch.load(args.model))

    model = model.module
    model.eval()

    if torch.cuda.is_available():
        model = model.cuda()

    with torch.no_grad():
        images = args.images.split(',')
        for imfile1, imfile2 in zip(images[:-1], images[1:]):
            image1 = load_image(imfile1)
            image2 = load_image(imfile2)

            # Resize para que sea divisible por 8, requisito de RAFT
            padder = raft_utils.InputPadder(image1.shape)
            image1, image2 = padder.pad(image1, image2)

            # Calcula el flujo óptico
            flow_low, flow_up = model(image1, image2, iters=20, test_mode=True)
            
            # flow_up es el flujo de alta resolución
            print(f"Flujo calculado para {imfile1} y {imfile2}")
            show_flow(image1, image2, flow_up)

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--model', help="ruta al modelo preentrenado RAFT", default='./RAFT/models/raft-things.pth')
    parser.add_argument('--path_to_raft', help="Ruta al directorio RAFT", default='./RAFT')
    parser.add_argument('--images', help="Lista de rutas de imágenes separadas por comas (ej. 'img1.png,img2.png,img3.png')")
    parser.add_argument('--small', action='store_true', help='utilizar modelo RAFT pequeño')
    parser.add_argument('--mixed_precision', action='store_true', help='utilizar precisión mixta')
    parser.add_argument('--alternate_corr', action='store_true', help='utilizar correlación alternativa')
    
    # Parámetros del modelo (igual que en train.py de RAFT)
    parser.add_argument('--gpus', type=int, nargs='+', default=[0])
    parser.add_argument('--num_heads', default=1, type=int,
                        help='número de cabezas en el extractor de caracteristicas')
    parser.add_argument('--position_only', default=False, action='store_true',
                        help='solo correlación posicional')
    parser.add_argument('--position_and_content', default=False, action='store_true',
                        help='correlación posicional y de contenido')
    parser.add_argument('--shared_extractor', default=False, action='store_true',
                        help='compartir extractor de caracteristicas')
    parser.add_argument('--decoder_channels', default=128, type=int,
                        help='número de canales en el decodificador')
    parser.add_argument('--corr_range', default=4, type=int,
                        help='rango máximo para el volumen de correlación')
    parser.add_argument('--corr_levels', default=4, type=int,
                        help='número de niveles del volumen de correlación')
    parser.add_argument('--corr_radius', default=4, type=int,
                        help='radio para el volumen de correlación')
    parser.add_argument('--n_gru_blocks', default=1, type=int,
                        help='número de bloques GRU')
    parser.add_argument('--hidden_size', default=128, type=int,
                        help='tamaño de la capa oculta GRU')
    parser.add_argument('--dropout', default=0, type=float,
                        help='tasa de dropout')

    args = parser.parse_args()

    # Asegurarse de que el directorio RAFT está en el path
    if args.path_to_raft not in sys.path:
        sys.path.insert(0, args.path_to_raft)

    main_raft(args)

Explicación del código RAFT:

1. Configuración del Entorno: Es crucial clonar el repositorio de RAFT y asegurarse de que sus módulos son accesibles para Python. El script ajusta sys.path para esto.
2. load_image(): Una función auxiliar para cargar imágenes, convertirlas a tensores de PyTorch, y moverlas a la GPU si está disponible. RAFT espera imágenes en formato [B, C, H, W] (Batch, Channels, Height, Width) y en float.
3. show_flow(): Similar a draw_flow para Farneback, esta función visualiza el campo de flujo calculado por RAFT, convirtiéndolo a un formato HSV para una representación a color.
4. Carga del Modelo: Se instancia el modelo RAFT y se carga un estado preentrenado (.pth file). Se recomienda usar raft-things.pth para buenos resultados generales.
5. Preprocesamiento: Las imágenes de entrada deben tener dimensiones divisibles por 8 (o 16, dependiendo de la configuración del modelo) para evitar problemas en las capas convolucionales. raft_utils.InputPadder se encarga de esto.
6. Inferencia: model(image1, image2, iters=20, test_mode=True) realiza la estimación del flujo. El parámetro iters controla el número de iteraciones de la GRU; más iteraciones generalmente resultan en mayor precisión pero mayor tiempo de cómputo.

Para ejecutar el script RAFT:

Necesitarás un par de imágenes para probar. Crea un directorio images y guarda image1.png y image2.png dentro.

python tu_script_raft.py --images images/image1.png,images/image2.png --model RAFT/models/raft-things.pth

Asegúrate de que RAFT/models/raft-things.pth exista, si no, ajusta la ruta.

Comparativa: Farneback vs. RAFT 🆚

Ambos algoritmos son válidos para la estimación de flujo óptico denso, pero tienen diferencias significativas en su enfoque, rendimiento y requisitos.

Consideraciones Prácticas y Consejos 📌

Elección del Algoritmo

• Si la velocidad en CPU es crítica y los movimientos son relativamente pequeños, o si no tienes acceso a una GPU, Farneback es una excelente opción. Es fácil de integrar y no requiere dependencias de aprendizaje profundo.
• Si la precisión es la prioridad absoluta (especialmente con movimientos complejos, oclusiones y escenas dinámicas), y tienes acceso a una GPU, RAFT es la elección superior. Es el estado del arte y produce resultados mucho más suaves y precisos.

Optimización

• Farneback: Experimenta con los parámetros. pyr_scale y levels afectan la capacidad de manejar movimientos grandes. winsize, poly_n y poly_sigma influyen en la suavidad y el detalle del flujo. Un winsize más pequeño captura más detalle pero puede ser más ruidoso.
• RAFT: Reducir iters (número de iteraciones en la GRU) puede acelerar la inferencia a costa de un poco de precisión. Considera usar el modelo raft-small para aplicaciones con recursos limitados.

Preprocesamiento de Imágenes

• Escala de grises: Ambos algoritmos suelen trabajar mejor con imágenes en escala de grises para reducir la complejidad. Farneback lo requiere explícitamente en OpenCV.
• Normalización: Para RAFT, es importante que los valores de los píxeles estén en el rango esperado por el modelo (generalmente [0, 255] o [0, 1] flotante, dependiendo de cómo se entrenó el modelo).

Conclusión ✅

La estimación de flujo óptico denso es una técnica poderosa y fundamental en visión artificial. Hemos explorado dos enfoques clave: el método clásico de Farneback, robusto y eficiente en CPU, y el enfoque de aprendizaje profundo RAFT, que ofrece una precisión sin precedentes en GPU.

Comprender cómo y cuándo aplicar cada uno de estos algoritmos te permitirá abordar una amplia gama de problemas de visión artificial que requieren analizar el movimiento. Ya sea para seguir la trayectoria de un balón en un partido de fútbol, ayudar a un robot a navegar por un almacén, o estabilizar tus videos caseros, el flujo óptico denso es tu aliado.

¡Experimenta con los códigos proporcionados, ajusta los parámetros y observa el mundo en movimiento a través de los ojos de la visión artificial!