비디오 분류
비디오 분류 (Video classification)
비디오 분류는 비디오 전체에 라벨이나 클래스를 부여하는 작업이에요. 비디오마다 클래스가 하나만 있다고 가정해요. 비디오 분류 모델은 비디오를 입력으로 받아 그 비디오가 어떤 클래스에 속하는지 예측을 반환해요. 이런 모델은 비디오가 어떤 내용인지 분류하는 데 사용할 수 있죠. 비디오 분류의 실제 응용 사례로는 피트니스 앱에 유용한 동작/활동 인식(action/activity recognition)이 있어요. 또한 시각 장애인, 특히 통근 중인 시각 장애인에게도 도움이 돼요.
출처: 문서
본문
이 가이드에서는 다음을 배워요:
이 작업과 호환되는 모든 아키텍처와 체크포인트를 보려면 task-page를 확인하는 걸 추천해요.
시작하기 전에 필요한 라이브러리를 모두 설치했는지 확인해요:
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]
(정렬된) 비디오 경로는 이렇게 보여요:
...
'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가 그렇죠.
검증 및 평가 분할에서는 데이터 누수(data leakage)를 막기 위해 같은 그룹/장면의 비디오 클립을 두지 않는 게 좋아요. 이 튜토리얼에서 사용하는 하위 집합은 이런 정보를 감안해서 만들어졌어요.
다음으로 데이터셋에 있는 라벨 집합을 도출할 거예요. 또한 모델을 초기화할 때 도움이 되는 두 개의 딕셔너리를 만들어요:
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개의 비디오가 있어요.
파인튜닝할 모델 로드
사전 훈련된 체크포인트와 그에 연결된 이미지 프로세서에서 비디오 분류 모델을 인스턴스화해요. 모델의 인코더는 사전 훈련된 파라미터로 구성되고, 분류 헤드(classification head)는 무작위로 초기화돼요. 이미지 프로세서는 데이터셋용 전처리 파이프라인을 작성할 때 유용하게 쓰일 거예요.
>>> 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 라이브러리를 활용할 거예요. 먼저 필요한 의존성을 import 해요.
>>> import pytorchvideo.data
>>> from pytorchvideo.transforms import (
... ApplyTransformToKey,
... Normalize,
... RandomShortSideScale,
... RemoveKey,
... ShortSideScale,
... UniformTemporalSubsample,
... )
>>> from torchvision.transforms import (
... Compose,
... Lambda,
... RandomCrop,
... RandomHorizontalFlip,
... Resize,
... )
학습 데이터셋 변환에는 시간적 균일 서브샘플링(uniform temporal subsampling), 픽셀 정규화, 무작위 크롭, 무작위 수평 뒤집기의 조합을 사용해요. 검증 및 평가 데이터셋 변환에는 무작위 크롭과 수평 뒤집기를 제외한 같은 변환 체인을 유지해요. 이 변환들의 세부 사항은 PyTorchVideo 공식 문서를 참고하세요.
사전 훈련된 모델에 연결된 image_processor를 사용해 다음 정보를 얻어요:
- 비디오 프레임 픽셀을 정규화하는 데 사용할 이미지 평균(mean)과 표준편차(std).
- 비디오 프레임을 리사이즈할 공간 해상도(spatial resolution).
먼저 몇 가지 상수를 정의해요.
>>> 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라는 두 개의 키로 구성돼요.
>>> 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()
추론 (Inference)
좋아요, 이제 모델을 파인튜닝했으니 추론에 활용할 수 있어요!
추론할 비디오를 로드해요:
>>> 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