Transformers documentation

動画分類

Transformers

はじめに

Transformers インストールクイックスタート

ベースクラス

モデル

プリプロセッサ

推論

Pipeline API

Generate API

最適化

モデルとチャットする

サービング

トレーニング

はじめに

カスタマイズ

パラメータ効率の良いファインチューニング

パフォーマンス

分散学習

ハードウェア

量子化

エコシステム統合

リソース

API

You are viewing main version, which requires installation from source. If you'd like regular pip install, checkout the latest stable version (v5.22.2).

Join the Hugging Face community

and get access to the augmented documentation experience

Collaborate on models, datasets and Spaces

Faster examples with accelerated inference

Switch between documentation themes

to get started

動画分類

Video classificationは、動画全体にラベルまたはクラスを割り当てるタスクです。各動画には1つのクラスのみが割り当てられることが前提とされます。Video classificationモデルは動画をインプットとして受け取り、その動画がどのクラスに属するかについての予測を返します。これらのモデルは、動画の内容を分類するために使用できます。Video classificationの実世界での応用例として、フィットネスアプリケーションに有用なアクション/アクティビティ認識があります。また、視覚障害のある方々、特に通勤時にも役立ちます。

このガイドでは、次の方法について説明します：

UCF101 データセットの部分集合で、VideoMAE をファインチューニングします。
ファインチューニングしたモデルを推論に使用します。

このタスクと互換性のあるすべてのアーキテクチャとチェックポイントを確認するには、タスクページを参照することをお勧めします。

開始前に、必要なライブラリがすべてインストールされていることを確認してください。

pip install -q pytorchvideo transformers evaluate

PyTorchVideo（pytorchvideo）を使用して、動画の処理と準備を行います。

モデルをコミュニティにアップロードして共有できるよう、Hugging Face アカウントへのログインをお勧めします。プロンプトが表示されたら、ログイン用のトークンを入力してください:

>>> from huggingface_hub import notebook_login

>>> notebook_login()

UCF101データセットを読み込む

UCF-101 データセットのサブセットを読み込むことから始めます。これにより、フルデータセットでの学習により多くの時間を費やす前に、実験を行い、すべてが正常に動作することを確認できます。

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

サブセットのダウンロードが完了したら、圧縮アーカイブを展開する必要があります。

>>> import tarfile

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

大まかに言うと、データセットは次のように構成されています。

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

その後、動画の総数をカウントできます。

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

(sorted) された動画パスは次のようになります:

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

動画ファイルのパスでは、g がグループ（シーン）を示しており、同じグループ / シーンに属する複数の動画クリップがあることに気づくでしょう。例えば、v_ApplyEyeMakeup_g07_c04.avi と v_ApplyEyeMakeup_g07_c06.avi などです。

検証および評価スプリットでは、データリークを防ぐために、同じグループ/シーンからの動画クリップが含まれないようにする必要があります。このチュートリアルで使用しているサブセットでは、この情報が考慮されています。

次に、データセットに存在するラベルの集合を導出します。また、モデルの初期化時に役立つ2つの辞書を作成します。

label2id: クラス名を整数にマッピングします。
id2label: 整数をクラス名にマッピングします。

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

クラスは10種類あります。各クラスについて、トレーニングセットには30本の動画が含まれています。

ファインチューニング用にモデルを読み込む

事前学習済みチェックポイントとその関連する画像プロセッサーから、動画分類モデルをインスタンス化します。モデルのエンコーダーには事前学習済みのパラメータが組み込まれており、分類ヘッドはランダムに初期化されます。画像プロセッサーは、データセット用の前処理パイプラインを作成する際に役立ちます。

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

モデルの読み込み中に、次のような警告が表示されることがあります。

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.

この警告は、一部の重み（classifier層の重みとバイアスなど）を破棄し、他の一部（新しいclassifier層の重みとバイアス）をランダムに初期化することを示しています。これは、事前学習済みの重みを持っていない新しいヘッドを追加しているため、このケースでは想定内の動作です。したがって、ライブラリは、推論に使用する前にこのモデルをファインチューニングすべきだと警告しています。まさにこれから行う作業です。

注：このチェックポイントは、類似の下流タスクでファインチューニングして得られたため、このタスクでより良いパフォーマンスを発揮します。また、MCG-NJU/videomae-base-finetuned-kinetics をファインチューニングして得られたこのチェックポイントも確認できます。

学習用のデータセットを準備する

動画の前処理には、PyTorchVideo libraryを活用します。まず、必要な依存関係をインポートしてください。

>>> import pytorchvideo.data

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

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

学習データセットの変換には、均一な時間方向サブサンプリング、ピクセル正規化、ランダムクロッピング、ランダムな水平方向フリップの組み合わせを使用します。検証および評価データセットの変換には、ランダムクロッピングと水平方向フリップを除き、同じ変換チェーンを維持します。これらの変換の詳細については、PyTorchVideoの公式ドキュメントを参照してください。

事前学習済みモデルに関連付けられたimage_processorを使用して、以下の情報を取得します。

動画フレームのピクセルを正規化するために使用される画像の平均値と標準偏差。
動画フレームがリサイズされる空間解像度。

いくつかの定数を定義することから始めます。

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

次に、データセット固有の変換とデータセットをそれぞれ定義します。まず、トレーニングセットから始めます。

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

同じワークフローのシーケンスを、検証セットおよび評価セットにも適用できます:

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

注記: 上記のデータセットパイプラインは、PyTorchVideoの公式サンプルから引用しています。UCF-101データセットに特化しているため、pytorchvideo.data.Ucf101()関数を使用しています。内部では、pytorchvideo.data.labeled_video_dataset.LabeledVideoDatasetオブジェクトを返します。LabeledVideoDatasetクラスは、PyTorchVideoのデータセットにおける動画関連のすべての処理の基底クラスです。したがって、PyTorchVideoが標準でサポートしていないカスタムデータセットを使用したい場合は、LabeledVideoDatasetクラスを適切に拡張できます。詳細については、data APIのドキュメントを参照してください。また、データセットが上記と同様の構造（ディレクトリ構造）を備えている場合、pytorchvideo.data.Ucf101()を使用しても問題なく動作します。

num_videos 引数にアクセスすることで、データセット内の動画の数を把握できます。

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

デバッグを容易にするために、前処理済みの動画を可視化します

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

モデルを学習

🤗 Transformers の Trainer を活用してモデルを学習します。Trainer をインスタンス化するには、学習設定と評価指標を定義する必要があります。最も重要なのは TrainingArguments で、これは学習を設定するためのすべての属性を含むクラスです。モデルのチェックポイントを保存するために使用される出力フォルダ名を指定する必要があります。また、🤗 Hub 上のモデルリポジトリ内のすべての情報を同期するのにも役立ちます。

ほとんどの学習引数はその名前から意味がわかりますが、ここではremove_unused_columns=Falseが特に重要です。この引数は、モデルのcall関数で使用されない特徴量を削除します。通常、未使用の特徴量カラムを削除するのが理想的であり、モデルのcall関数への入力のアンパックを容易にするため、デフォルトではTrueになっています。しかし、このケースでは、pixel_values（モデルの入力に必須のキー）を作成するために、未使用の特徴量（特に‘video’）が必要です。

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

pytorchvideo.data.Ucf101() が返すデータセットは、__len__ メソッドを実装していません。そのため、TrainingArguments をインスタンス化する際に max_steps を定義する必要があります。

次に、予測値からメトリクスを計算する関数を定義する必要があります。この関数では、これから読み込むmetricを使用します。必要な前処理は、予測されたlogitsのargmaxを取得することだけです。

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)

評価に関する補足:

VideoMAE論文では、著者たちは次のような評価戦略を採用しています。テスト動画から複数のクリップを抽出し、それらに異なるクロップを適用して評価を行い、集計スコアを報告します。ただし、本チュートリアルでは簡潔さと簡略化を目的として、この手法は考慮していません。

また、collate_fnを定義します。これは、サンプルをバッチにまとめるために使用されます。各バッチはpixel_valuesとlabelsという2つのキーで構成されます。

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

次に、これらすべてとデータセットを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,
... )

データの前処理をすでに実施しているにもかかわらず、なぜimage_processorをトークナイザーとして渡したのかと疑問に思うかもしれません。これは、画像プロセッサの設定ファイル（JSONとして保存されるもの）もHub上のリポジトリにアップロードされることを確認するためです。

trainメソッドを呼び出して、モデルをファインチューニングします:

>>> train_results = trainer.train()

トレーニングが完了したら、push_to_hub() メソッドでモデルを Hub に共有し、誰でもそのモデルを利用できるようにしてください。

>>> trainer.push_to_hub()

推論

素晴らしいですね。モデルのファインチューニングが完了したので、推論に使用できます！

推論用の動画を読み込みます:

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

ファインチューニングしたモデルを推論で試す最も簡単な方法は、VideoClassificationPipelineでそれを使用することです。モデルを指定して動画分類用のpipelineをインスタンス化し、動画を入力します:

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

必要に応じて、pipelineの結果を手動で再現することもできます。

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

次に、入力をモデルに渡し、logitsを返します。

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

logits をデコードすると、次のようになります:

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

Update on GitHub

←画像セグメンテーションオブジェクト検出→