비디오 분류

Video classification은 전체 비디오에 라벨 또는 클래스를 할당하는 작업입니다. 각 비디오에는 하나의 클래스만 할당되는 것이 기대됩니다. Video classification 모델은 비디오를 입력으로 받아 해당 비디오가 속하는 클래스에 대한 예측을 반환합니다. 이러한 모델은 비디오의 주제를 분류하는 데 사용할 수 있습니다. Video classification의 실제 응용 사례로는 동작/활동 인식이 있으며, 이는 피트니스 애플리케이션에 유용합니다. 또한, 특히 출퇴근 시 시각 장애가 있는 개인에게도 도움이 됩니다.

이 가이드에서는 다음 내용을 보여 드립니다:

UCF101 데이터셋의 하위 집합에서 VideoMAE를 fine-tune합니다.
파인튜닝된 모델을 추론에 사용하세요.

이 작업과 호환되는 모든 아키텍처와 체크포인트를 확인하려면 task-page를 참고하는 것을 권장합니다.

시작하기 전에 필요한 라이브러리가 모두 설치되어 있는지 확인하세요:

pip install -q pytorchvideo transformers evaluate

PyTorchVideo(pytorchvideo로 불림)를 사용하여 비디오를 처리하고 준비합니다.

모델을 업로드하여 커뮤니티와 공유할 수 있도록 Hugging Face 계정에 로그인하는 것을 권장합니다. 프롬프트가 표시되면 토큰을 입력하여 로그인하세요:

>>> from huggingface_hub import notebook_login

>>> notebook_login()

UCF101 데이터셋 로드

UCF-101 dataset의 하위 집합을 로드하는 것으로 시작합니다. 이렇게 하면 전체 데이터셋으로 더 많은 시간을 들여 학습하기 전에 실험을 해보고 모든 것이 정상적으로 작동하는지 확인할 수 있습니다.

>>> 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가 있습니다.

검증 및 평가 분할의 경우, 데이터 누출을 방지하기 위해 같은 그룹/장면의 비디오 클립이 포함되지 않도록 해야 합니다. 이 튜토리얼에서 사용하는 하위 집합은 이러한 정보를 고려하고 있습니다.

다음으로, 데이터셋에 존재하는 라벨 집합을 도출합니다. 또한, 모델을 초기화할 때 유용한 두 개의 사전(dictionary)을 생성합니다.

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 레이어의 가중치와 bias)를 버리고, 다른 일부(새로운 classifier 레이어의 가중치와 bias)는 무작위로 초기화한다는 것을 알려 줍니다. 이 경우 새로운 헤드를 추가하는 데 사전 학습된 가중치가 없으므로, 라이브러리가 추론에 사용하기 전에 이 모델을 fine-tune해야 한다고 경고하는 것은 예상된 동작입니다. 바로 이 작업을 수행할 것입니다.

참고할 점은, 이 체크포인트는 상당한 도메인 중복이 있는 유사한 다운스트림 태스크로 파인튜닝하여 획득되었기 때문에 해당 태스크에서 더 나은 성능을 발휘한다는 것입니다. 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 함수에서 사용되지 않는 모든 특징을 제거합니다. 기본적으로 True로 설정되어 있는데, 이는 사용되지 않는 특징 열을 제거하는 것이 모델의 call 함수에 입력을 전달하는 과정을 단순화하므로 이상적이기 때문입니다. 그러나 이 경우에는 pixel_values를 생성하기 위해 사용되지 않는 특징(특히 ‘video’)이 필요합니다. pixel_values는 모델이 입력으로 기대하는 필수 키입니다.

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

추론

좋습니다. 이제 모델을 fine-tuning 했으므로 추론에 사용할 수 있습니다!

추론을 위해 비디오를 로드합니다:

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