Clasificación de video

La clasificación de videos es la tarea de asignar una etiqueta o clase a un video completo. Se espera que los videos tengan una única clase por video. Los modelos de clasificación de videos toman un video como entrada y devuelven una predicción sobre a qué clase pertenece el video. Estos modelos se pueden utilizar para categorizar de qué trata un video. Una aplicación del mundo real de la clasificación de videos es el reconocimiento de acciones/actividades, que es útil para aplicaciones de fitness. También es útil para personas con discapacidad visual, especialmente cuando se desplazan.

Esta guía le mostrará cómo:

Ajusta VideoMAE a un subconjunto del conjunto de datos UCF101.
Usa tu modelo afinado para la inferencia.

Para ver todas las arquitecturas y puntos de control compatibles con esta tarea, recomendamos consultar la página de la tarea.

Antes de comenzar, asegúrate de tener instaladas todas las bibliotecas necesarias:

pip install -q pytorchvideo transformers evaluate

Utilizará PyTorchVideo (denominado pytorchvideo) para procesar y preparar los videos.

Te animamos a iniciar sesión en tu cuenta de Hugging Face para que puedas subir y compartir tu modelo con la comunidad. Cuando se te solicite, introduce tu token para iniciar sesión:

>>> from huggingface_hub import notebook_login

>>> notebook_login()

Cargar el conjunto de datos UCF101

Comience cargando un subconjunto del conjunto de datos UCF-101. Esto le dará la oportunidad de experimentar y asegurarse de que todo funcione correctamente antes de dedicar más tiempo a entrenar con el conjunto de datos completo.

>>> from huggingface_hub import hf_hub_download

>>> hf_dataset_identifier = "sayakpaul/ucf101-subset"
>>> filename = "UCF101_subset.tar.gz"
>>> file_path = hf_hub_download(repo_id=hf_dataset_identifier, filename=filename, repo_type="dataset")

Después de que se descargue el subconjunto, es necesario extraer el archivo comprimido:

>>> import tarfile

>>> with tarfile.open(file_path) as t:
...      t.extractall(".")

A alto nivel, el conjunto de datos se organiza de la siguiente manera:

UCF101_subset/
    train/
        BandMarching/
            video_1.mp4
            video_2.mp4
            ...
        Archery
            video_1.mp4
            video_2.mp4
            ...
        ...
    val/
        BandMarching/
            video_1.mp4
            video_2.mp4
            ...
        Archery
            video_1.mp4
            video_2.mp4
            ...
        ...
    test/
        BandMarching/
            video_1.mp4
            video_2.mp4
            ...
        Archery
            video_1.mp4
            video_2.mp4
            ...
        ...

A continuación, puede contar el número total de videos.

>>> import pathlib
>>> dataset_root_path = "UCF101_subset"
>>> dataset_root_path = pathlib.Path(dataset_root_path)

>>> video_count_train = len(list(dataset_root_path.glob("train/*/*.avi")))
>>> video_count_val = len(list(dataset_root_path.glob("val/*/*.avi")))
>>> video_count_test = len(list(dataset_root_path.glob("test/*/*.avi")))
>>> video_total = video_count_train + video_count_val + video_count_test
>>> print(f"Total videos: {video_total}")

>>> all_video_file_paths = (
...     list(dataset_root_path.glob("train/*/*.avi"))
...     + list(dataset_root_path.glob("val/*/*.avi"))
...     + list(dataset_root_path.glob("test/*/*.avi"))
...  )
>>> all_video_file_paths[:5]

Las rutas de video (sorted) se ven así:

...
'UCF101_subset/train/ApplyEyeMakeup/v_ApplyEyeMakeup_g07_c04.avi',
'UCF101_subset/train/ApplyEyeMakeup/v_ApplyEyeMakeup_g07_c06.avi',
'UCF101_subset/train/ApplyEyeMakeup/v_ApplyEyeMakeup_g08_c01.avi',
'UCF101_subset/train/ApplyEyeMakeup/v_ApplyEyeMakeup_g09_c02.avi',
'UCF101_subset/train/ApplyEyeMakeup/v_ApplyEyeMakeup_g09_c06.avi'
...

Notará que hay clips de vídeo que pertenecen al mismo grupo / escena, donde el grupo se denota mediante g en las rutas de los archivos de vídeo. Por ejemplo, v_ApplyEyeMakeup_g07_c04.avi y v_ApplyEyeMakeup_g07_c06.avi.

Para las divisiones de validación y evaluación, no se desearía tener clips de video del mismo grupo / escena para evitar fugas de datos. El subconjunto que se utiliza en este tutorial tiene en cuenta esta información.

A continuación, derivará el conjunto de etiquetas presentes en el conjunto de datos. Además, creará dos diccionarios que serán útiles al inicializar el modelo:

label2id: asigna los nombres de las clases a enteros.
id2label: asigna los enteros a los nombres de las clases.

>>> class_labels = sorted({str(path).split("/")[2] for path in all_video_file_paths})
>>> label2id = {label: i for i, label in enumerate(class_labels)}
>>> id2label = {i: label for label, i in label2id.items()}

>>> print(f"Unique classes: {list(label2id.keys())}.")

# Unique classes: ['ApplyEyeMakeup', 'ApplyLipstick', 'Archery', 'BabyCrawling', 'BalanceBeam', 'BandMarching', 'BaseballPitch', 'Basketball', 'BasketballDunk', 'BenchPress'].

Hay 10 clases únicas. Para cada clase, hay 30 videos en el conjunto de entrenamiento.

Cargar un modelo para el ajuste fino

Instancie un modelo de clasificación de video desde un punto de control preentrenado y su procesador de imágenes asociado. El codificador del modelo viene con parámetros preentrenados, y la cabeza de clasificación se inicializa aleatoriamente. El procesador de imágenes le será útil al escribir el pipeline de preprocesamiento para nuestro conjunto de datos.

>>> from transformers import VideoMAEImageProcessor, VideoMAEForVideoClassification

>>> model_ckpt = "MCG-NJU/videomae-base"
>>> image_processor = VideoMAEImageProcessor.from_pretrained(model_ckpt)
>>> model = VideoMAEForVideoClassification.from_pretrained(
...     model_ckpt,
...     label2id=label2id,
...     id2label=id2label,
...     ignore_mismatched_sizes=True,  # provide this in case you're planning to fine-tune an already fine-tuned checkpoint
... )

Mientras el modelo se está cargando, es posible que observes la siguiente advertencia:

Some weights of the model checkpoint at MCG-NJU/videomae-base were not used when initializing VideoMAEForVideoClassification: [..., 'decoder.decoder_layers.1.attention.output.dense.bias', 'decoder.decoder_layers.2.attention.attention.key.weight']
- This IS expected if you are initializing VideoMAEForVideoClassification from the checkpoint of a model trained on another task or with another architecture (e.g. initializing a BertForSequenceClassification model from a BertForPreTraining model).
- This IS NOT expected if you are initializing VideoMAEForVideoClassification from the checkpoint of a model that you expect to be exactly identical (initializing a BertForSequenceClassification model from a BertForSequenceClassification model).
Some weights of VideoMAEForVideoClassification were not initialized from the model checkpoint at MCG-NJU/videomae-base and are newly initialized: ['classifier.bias', 'classifier.weight']
You should probably TRAIN this model on a down-stream task to be able to use it for predictions and inference.

La advertencia nos indica que estamos descartando algunos pesos (por ejemplo, los pesos y el sesgo de la capa classifier) e inicializando aleatoriamente otros (los pesos y el sesgo de una nueva capa classifier). Esto es esperado en este caso, ya que estamos añadiendo una nueva cabeza para la cual no disponemos de pesos preentrenados, por lo que la biblioteca nos advierte que debemos realizar un ajuste fino de este modelo antes de usarlo para inferencia, que es exactamente lo que vamos a hacer.

Nota: este checkpoint ofrece un mejor rendimiento en esta tarea, ya que se obtuvo mediante fine-tuning en una tarea de bajo nivel similar con una considerable superposición de dominio. Puede consultar este checkpoint, que se obtuvo mediante fine-tuning de MCG-NJU/videomae-base-finetuned-kinetics.

Preparar los conjuntos de datos para el entrenamiento

Para preprocesar los videos, utilizará la librería PyTorchVideo. Comience importando las dependencias que necesitamos.

>>> import pytorchvideo.data

>>> from pytorchvideo.transforms import (
...     ApplyTransformToKey,
...     Normalize,
...     RandomShortSideScale,
...     RemoveKey,
...     ShortSideScale,
...     UniformTemporalSubsample,
... )

>>> from torchvision.transforms import (
...     Compose,
...     Lambda,
...     RandomCrop,
...     RandomHorizontalFlip,
...     Resize,
... )

Para las transformaciones del conjunto de datos de entrenamiento, utiliza una combinación de subsampling temporal uniforme, normalización de píxeles, recorte aleatorio y volteo horizontal aleatorio. Para las transformaciones de los conjuntos de datos de validación y evaluación, mantén la misma cadena de transformaciones, excepto por el recorte aleatorio y el volteo horizontal. Para conocer más detalles sobre estas transformaciones, consulta la documentación oficial de PyTorchVideo.

Use el image_processor asociado con el modelo preentrenado para obtener la siguiente información:

Media y desviación estándar de la imagen con la que se normalizarán los píxeles de los fotogramas del video.
Resolución espacial a la que se redimensionarán los fotogramas del video.

Comience definiendo algunas constantes.

>>> mean = image_processor.image_mean
>>> std = image_processor.image_std
>>> if "shortest_edge" in image_processor.size:
...     height = width = image_processor.size["shortest_edge"]
>>> else:
...     height = image_processor.size["height"]
...     width = image_processor.size["width"]
>>> resize_to = (height, width)

>>> num_frames_to_sample = model.config.num_frames
>>> sample_rate = 4
>>> fps = 30
>>> clip_duration = num_frames_to_sample * sample_rate / fps

Ahora, defina las transformaciones específicas del conjunto de datos y los conjuntos de datos, respectivamente. Comenzando con el conjunto de entrenamiento:

>>> train_transform = Compose(
...     [
...         ApplyTransformToKey(
...             key="video",
...             transform=Compose(
...                 [
...                     UniformTemporalSubsample(num_frames_to_sample),
...                     Lambda(lambda x: x / 255.0),
...                     Normalize(mean, std),
...                     RandomShortSideScale(min_size=256, max_size=320),
...                     RandomCrop(resize_to),
...                     RandomHorizontalFlip(p=0.5),
...                 ]
...             ),
...         ),
...     ]
... )

>>> train_dataset = pytorchvideo.data.Ucf101(
...     data_path=os.path.join(dataset_root_path, "train"),
...     clip_sampler=pytorchvideo.data.make_clip_sampler("random", clip_duration),
...     decode_audio=False,
...     transform=train_transform,
... )

La misma secuencia de flujo de trabajo se puede aplicar a los conjuntos de validación y evaluación:

>>> val_transform = Compose(
...     [
...         ApplyTransformToKey(
...             key="video",
...             transform=Compose(
...                 [
...                     UniformTemporalSubsample(num_frames_to_sample),
...                     Lambda(lambda x: x / 255.0),
...                     Normalize(mean, std),
...                     Resize(resize_to),
...                 ]
...             ),
...         ),
...     ]
... )

>>> val_dataset = pytorchvideo.data.Ucf101(
...     data_path=os.path.join(dataset_root_path, "val"),
...     clip_sampler=pytorchvideo.data.make_clip_sampler("uniform", clip_duration),
...     decode_audio=False,
...     transform=val_transform,
... )

>>> test_dataset = pytorchvideo.data.Ucf101(
...     data_path=os.path.join(dataset_root_path, "test"),
...     clip_sampler=pytorchvideo.data.make_clip_sampler("uniform", clip_duration),
...     decode_audio=False,
...     transform=val_transform,
... )

Nota: Los pipelines de conjuntos de datos anteriores se toman del ejemplo oficial de PyTorchVideo. Utilizamos la función pytorchvideo.data.Ucf101() porque está diseñada específicamente para el conjunto de datos UCF-101. Internamente, devuelve un objeto pytorchvideo.data.labeled_video_dataset.LabeledVideoDataset. La clase LabeledVideoDataset es la clase base para todo lo relacionado con video en el conjunto de datos de PyTorchVideo. Por lo tanto, si deseas utilizar un conjunto de datos personalizado no soportado de forma nativa por PyTorchVideo, puedes extender la clase LabeledVideoDataset en consecuencia. Consulta la documentación de la API data para obtener más información. Además, si tu conjunto de datos sigue una estructura similar (como se muestra arriba), entonces el uso de pytorchvideo.data.Ucf101() debería funcionar perfectamente.

Puede acceder al argumento num_videos para conocer el número de videos en el conjunto de datos.

>>> print(train_dataset.num_videos, val_dataset.num_videos, test_dataset.num_videos)
# (300, 30, 75)

Visualiza el video preprocesado para depurar mejor

>>> import imageio
>>> import numpy as np
>>> from IPython.display import Image

>>> def unnormalize_img(img):
...     """Un-normalizes the image pixels."""
...     img = (img * std) + mean
...     img = (img * 255).astype("uint8")
...     return img.clip(0, 255)

>>> def create_gif(video_tensor, filename="sample.gif"):
...     """Prepares a GIF from a video tensor.
...
...     The video tensor is expected to have the following shape:
...     (num_frames, num_channels, height, width).
...     """
...     frames = []
...     for video_frame in video_tensor:
...         frame_unnormalized = unnormalize_img(video_frame.permute(1, 2, 0).numpy())
...         frames.append(frame_unnormalized)
...     kargs = {"duration": 0.25}
...     imageio.mimsave(filename, frames, "GIF", **kargs)
...     return filename

>>> def display_gif(video_tensor, gif_name="sample.gif"):
...     """Prepares and displays a GIF from a video tensor."""
...     video_tensor = video_tensor.permute(1, 0, 2, 3)
...     gif_filename = create_gif(video_tensor, gif_name)
...     return Image(filename=gif_filename)

>>> sample_video = next(iter(train_dataset))
>>> video_tensor = sample_video["video"]
>>> display_gif(video_tensor)

Entrenar el modelo

Aprovecha Trainer de 🤗 Transformers para entrenar el modelo. Para instanciar un Trainer, debes definir la configuración de entrenamiento y una métrica de evaluación. Lo más importante es TrainingArguments, una clase que contiene todos los atributos para configurar el entrenamiento. Requiere un nombre de carpeta de salida, que se utilizará para guardar los puntos de control del modelo. También ayuda a sincronizar toda la información en el repositorio del modelo en 🤗 Hub.

La mayoría de los argumentos de entrenamiento son autoexplicativos, pero uno que es bastante importante aquí es remove_unused_columns=False. Este parámetro elimina cualquier característica que no se utilice en la función de llamada del modelo. Por defecto, su valor es True, ya que generalmente es ideal eliminar las columnas de características no utilizadas, lo que facilita la extracción de las entradas para la función de llamada del modelo. Sin embargo, en este caso, necesitas las características no utilizadas (en particular, ‘video’) para crear pixel_values (que es una clave obligatoria que nuestro modelo espera en sus entradas).

>>> from transformers import TrainingArguments, Trainer

>>> model_name = model_ckpt.split("/")[-1]
>>> new_model_name = f"{model_name}-finetuned-ucf101-subset"
>>> num_epochs = 4

>>> args = TrainingArguments(
...     new_model_name,
...     remove_unused_columns=False,
...     eval_strategy="epoch",
...     save_strategy="epoch",
...     learning_rate=5e-5,
...     per_device_train_batch_size=batch_size,
...     per_device_eval_batch_size=batch_size,
...     warmup_steps=0.1,
...     logging_steps=10,
...     load_best_model_at_end=True,
...     metric_for_best_model="accuracy",
...     push_to_hub=True,
...     max_steps=(train_dataset.num_videos // batch_size) * num_epochs,
... )

El conjunto de datos devuelto por pytorchvideo.data.Ucf101() no implementa el método __len__. Por lo tanto, debemos definir max_steps al instanciar TrainingArguments.

A continuación, debe definir una función para calcular las métricas a partir de las predicciones, que utilizará la metric que cargará ahora. El único preprocesamiento que debe realizar es tomar el argmax de los logits predichos:

import evaluate

metric = evaluate.load("accuracy")


def compute_metrics(eval_pred):
    predictions = np.argmax(eval_pred.predictions, axis=1)
    return metric.compute(predictions=predictions, references=eval_pred.label_ids)

Una nota sobre la evaluación:

En el artículo de VideoMAE, los autores utilizan la siguiente estrategia de evaluación. Evalúan el modelo en varios clips de los videos de prueba, aplican diferentes recortes a dichos clips e informan la puntuación agregada. Sin embargo, por simplicidad y brevedad, no lo consideramos en este tutorial.

Además, defina una collate_fn, que se utilizará para agrupar los ejemplos en lotes. Cada lote consta de 2 claves, a saber, pixel_values y labels.

>>> def collate_fn(examples):
...     # permute to (num_frames, num_channels, height, width)
...     pixel_values = torch.stack(
...         [example["video"].permute(1, 0, 2, 3) for example in examples]
...     )
...     labels = torch.tensor([example["label"] for example in examples])
...     return {"pixel_values": pixel_values, "labels": labels}

Luego, simplemente pasas todo esto junto con los conjuntos de datos a Trainer:

>>> trainer = Trainer(
...     model,
...     args,
...     train_dataset=train_dataset,
...     eval_dataset=val_dataset,
...     processing_class=image_processor,
...     compute_metrics=compute_metrics,
...     data_collator=collate_fn,
... )

Es posible que te preguntes por qué pasaste image_processor como un tokenizador si ya habías preprocesado los datos. Esto se hace únicamente para asegurarse de que el archivo de configuración del procesador de imágenes (guardado como JSON) también se suba al repositorio en el Hub.

Ahora, ajusta el modelo a fin de llamar al método train:

>>> train_results = trainer.train()

Una vez completado el entrenamiento, comparte tu modelo en el Hub con el método push_to_hub() para que todo el mundo pueda usarlo:

>>> trainer.push_to_hub()

Inferencia

Genial, ahora que has ajustado un modelo, ¡puedes usarlo para inferencia!

Cargar un video para inferencia:

>>> sample_test_video = next(iter(test_dataset))

La forma más sencilla de probar tu modelo afinado para inferencia es usarlo en un VideoClassificationPipeline. Instancia un pipeline para clasificación de video con tu modelo y pásale tu video:

>>> from transformers import pipeline
from accelerate import Accelerator

>>> video_cls = pipeline(model="my_awesome_video_cls_model")
>>> video_cls("https://huggingface.co/datasets/sayakpaul/ucf101-subset/resolve/main/v_BasketballDunk_g14_c06.avi")
[{'score': 0.9272987842559814, 'label': 'BasketballDunk'},
 {'score': 0.017777055501937866, 'label': 'BabyCrawling'},
 {'score': 0.01663011871278286, 'label': 'BalanceBeam'},
 {'score': 0.009560945443809032, 'label': 'BandMarching'},
 {'score': 0.0068979403004050255, 'label': 'BaseballPitch'}]

También puedes replicar manualmente los resultados de pipeline si lo deseas.

>>> def run_inference(model, video):
...     # (num_frames, num_channels, height, width)
...     permuted_sample_test_video = video.permute(1, 0, 2, 3)
...     inputs = {
...         "pixel_values": perumuted_sample_test_video.unsqueeze(0),
...         "labels": torch.tensor(
...             [sample_test_video["label"]]
...         ),  # this can be skipped if you don't have labels available.
...     }

...     device = Accelerator().device
...     inputs = {k: v.to(device) for k, v in inputs.items()}
...     model = model.to(device)

...     # forward pass
...     with torch.no_grad():
...         outputs = model(**inputs)
...         logits = outputs.logits

...     return logits

Ahora, pasa tu entrada al modelo y devuelve los logits:

>>> logits = run_inference(trained_model, sample_test_video["video"])

Al decodificar los logits, obtenemos:

>>> predicted_class_idx = logits.argmax(-1).item()
>>> print("Predicted class:", model.config.id2label[predicted_class_idx])
# Predicted class: BasketballDunk

Update on GitHub