Transformers documentation

Classificação de vídeo

You are viewing main version, which requires installation from source. If you'd like regular pip install, checkout the latest stable version (v5.22.1).
Hugging Face's logo
Join the Hugging Face community

and get access to the augmented documentation experience

to get started

Classificação de vídeo

A classificação de vídeo é a tarefa de atribuir um rótulo ou classe a um vídeo inteiro. Espera-se que os vídeos tenham apenas uma classe para cada vídeo. Os modelos de classificação de vídeo recebem um vídeo como entrada e retornam uma previsão sobre a classe à qual o vídeo pertence. Esses modelos podem ser usados para categorizar o conteúdo de um vídeo. Uma aplicação do mundo real da classificação de vídeo é o reconhecimento de ações/atividades, que é útil para aplicativos de fitness. Também é útil para pessoas com deficiência visual, especialmente quando estão se deslocando.

Este guia mostrará como:

  1. Ajuste fino de VideoMAE em um subconjunto do conjunto de dados UCF101.
  2. Use seu modelo ajustado para inferência.

Para ver todas as arquiteturas e checkpoints compatíveis com esta tarefa, recomendamos consultar a página da tarefa.

Antes de começar, verifique se você tem todas as bibliotecas necessárias instaladas:

pip install -q pytorchvideo transformers evaluate

Você usará o PyTorchVideo (denominado pytorchvideo) para processar e preparar os vídeos.

Recomendamos que você faça login na sua conta do Hugging Face para que possa enviar e compartilhar seu modelo com a comunidade. Quando solicitado, insira seu token para fazer login:

>>> from huggingface_hub import notebook_login

>>> notebook_login()

Carregar o conjunto de dados UCF101

Comece carregando um subconjunto do conjunto de dados UCF-101. Isso lhe dará a oportunidade de experimentar e garantir que tudo funcione antes de gastar mais tempo treinando no conjunto de dados 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")

Após o subconjunto ter sido baixado, você precisa extrair o arquivo compactado:

>>> import tarfile

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

Em linhas gerais, o conjunto de dados é organizado da seguinte forma:

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
            ...
        ...

Em seguida, você pode contar o número total de vídeos.

>>> 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]

Os caminhos de vídeo (sorted) aparecem da seguinte forma:

...
'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'
...

Você notará que há clipes de vídeo pertencentes ao mesmo grupo / cena, em que o grupo é indicado por g nos caminhos dos arquivos de vídeo. v_ApplyEyeMakeup_g07_c04.avi e v_ApplyEyeMakeup_g07_c06.avi, por exemplo.

Para as subdivisões de validação e avaliação, você não deve ter clipes de vídeo do mesmo grupo / cena, a fim de evitar vazamento de dados. O subconjunto que você está usando neste tutorial leva essa informação em consideração.

A seguir, você vai derivar o conjunto de rótulos presentes no conjunto de dados. Além disso, crie dois dicionários que serão úteis na inicialização do modelo:

  • label2id: mapeia os nomes das classes para inteiros.
  • id2label: mapeia os inteiros para os nomes das classes.
>>> 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'].

Há 10 classes únicas. Para cada classe, há 30 vídeos no conjunto de treinamento.

Carregar um modelo para ajuste fino

Instancie um modelo de classificação de vídeo a partir de um checkpoint pré-treinado e seu processador de imagens associado. O encoder do modelo vem com parâmetros pré-treinados, e a cabeça de classificação é inicializada aleatoriamente. O processador de imagens será útil ao escrever o pipeline de pré-processamento para o nosso conjunto de dados.

>>> 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
... )

Enquanto o modelo está sendo carregado, você pode notar o seguinte aviso:

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.

O aviso nos informa que estamos descartando alguns pesos (por exemplo, os pesos e o viés da camada classifier) e inicializando aleatoriamente outros (os pesos e o viés de uma nova camada classifier). Isso é esperado neste caso, pois estamos adicionando um novo head para o qual não temos pesos pré-treinados, então a biblioteca nos avisa que devemos fazer o fine-tuning deste modelo antes de usá-lo para inferência, o que é exatamente o que vamos fazer.

Observação: este checkpoint apresenta melhor desempenho nesta tarefa, pois foi obtido por meio de fine-tuning em uma tarefa de downstream semelhante, com sobreposição considerável de domínio. Você pode verificar este checkpoint, que foi obtido por meio de fine-tuning de MCG-NJU/videomae-base-finetuned-kinetics.

Prepare os conjuntos de dados para o treinamento

Para o pré-processamento dos vídeos, você utilizará a PyTorchVideo library. Comece importando as dependências de que precisamos.

>>> import pytorchvideo.data

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

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

Para as transformações do conjunto de dados de treinamento, use uma combinação de subsampling temporal uniforme, normalização de pixels, recorte aleatório e inversão horizontal aleatória. Para as transformações dos conjuntos de dados de validação e avaliação, mantenha a mesma cadeia de transformações, exceto o recorte aleatório e a inversão horizontal. Para saber mais sobre os detalhes dessas transformações, consulte a documentação oficial do PyTorchVideo.

Use o image_processor associado ao modelo pré-treinado para obter as seguintes informações:

  • Média e desvio padrão da imagem com os quais os pixels dos quadros do vídeo serão normalizados.
  • Resolução espacial para a qual os quadros do vídeo serão redimensionados.

Comece definindo algumas 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

Agora, defina as transformações específicas do conjunto de dados e os conjuntos de dados, respectivamente. Começando pelo conjunto de treinamento:

>>> 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,
... )

A mesma sequência de fluxo de trabalho pode ser aplicada aos conjuntos de validação e avaliação:

>>> 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: Os pipelines de dataset acima são extraídos do exemplo oficial do PyTorchVideo. Estamos usando a função pytorchvideo.data.Ucf101() porque ela é específica para o dataset UCF-101. Internamente, ela retorna um objeto pytorchvideo.data.labeled_video_dataset.LabeledVideoDataset. A classe LabeledVideoDataset é a classe base para tudo relacionado a vídeo no dataset do PyTorchVideo. Portanto, se você quiser usar um dataset personalizado não suportado nativamente pelo PyTorchVideo, pode estender a classe LabeledVideoDataset de forma adequada. Consulte a documentação da API data para saber mais. Além disso, se o seu dataset seguir uma estrutura semelhante (como mostrado acima), o uso de pytorchvideo.data.Ucf101() deve funcionar perfeitamente.

Você pode acessar o argumento num_videos para saber o número de vídeos no conjunto de dados.

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

Visualize o vídeo pré-processado para depuração mais eficaz

>>> 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)
Person playing basketball

Treinar o modelo

Utilize o Trainer do 🤗 Transformers para treinar o modelo. Para instanciar um Trainer, é necessário definir a configuração de treinamento e uma métrica de avaliação. O mais importante é o TrainingArguments, uma classe que contém todos os atributos para configurar o treinamento. Ele requer um nome de pasta de saída, que será usado para salvar os checkpoints do modelo. Ele também ajuda a sincronizar todas as informações no repositório do modelo no 🤗 Hub.

A maioria dos argumentos de treinamento é autoexplicativa, mas um que é bastante importante aqui é remove_unused_columns=False. Este argumento remove quaisquer recursos não utilizados pela função de chamada do modelo. Por padrão, ele é True, pois geralmente é ideal descartar colunas de recursos não utilizados, facilitando a descompactação das entradas para a função de chamada do modelo. No entanto, neste caso, você precisa dos recursos não utilizados (especificamente ‘video’) para criar pixel_values (que é uma chave obrigatória que nosso modelo espera em suas 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,
... )

O conjunto de dados retornado por pytorchvideo.data.Ucf101() não implementa o método __len__. Por isso, é necessário definir max_steps ao instanciar TrainingArguments.

Em seguida, você precisa definir uma função para calcular as métricas a partir das previsões, que usará o metric que você carregará agora. A única pré-processamento que você precisa fazer é obter o argmax dos logits previstos:

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)

Uma nota sobre avaliação:

No artigo do VideoMAE, os autores utilizam a seguinte estratégia de avaliação. Eles avaliam o modelo em vários clipes de vídeos de teste, aplicam diferentes recortes a esses clipes e relatam a pontuação agregada. No entanto, em nome da simplicidade e da concisão, não consideramos isso neste tutorial.

Além disso, defina uma collate_fn, que será usada para agrupar exemplos em lotes. Cada lote consiste em 2 chaves, a saber, pixel_values e 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}

Em seguida, basta passar tudo isso junto com os conjuntos de dados para o 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,
... )

Você pode se perguntar por que passou o image_processor como um tokenizer, já que os dados já foram pré-processados. Isso é feito apenas para garantir que o arquivo de configuração do processador de imagens (armazenado em JSON) também seja enviado para o repositório no Hub.

Agora, ajuste fino do nosso modelo chamando o método train:

>>> train_results = trainer.train()

Quando o treinamento for concluído, compartilhe seu modelo no Hub com o método push_to_hub() para que todos possam usá-lo:

>>> trainer.push_to_hub()

Inferência

Ótimo, agora que você ajustou um modelo, pode usá-lo para inferência!

Carregue um vídeo para inferência:

>>> sample_test_video = next(iter(test_dataset))
Teams playing basketball

A maneira mais simples de testar seu modelo ajustado para inferência é usá-lo em um VideoClassificationPipeline. Instancie um pipeline para classificação de vídeo com seu modelo e passe seu vídeo para ele:

>>> 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'}]

Você também pode replicar manualmente os resultados do pipeline, se desejar.

>>> 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

Agora, passe sua entrada para o modelo e retorne os logits:

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

Ao decodificar os logits, obtemos:

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