Видеоаналитика: Разбор VideoMAE, ViViT и TimeSFormer

1. Patch Embedding:

• Видео разбивается на небольшие трехмерные патчи (например, размером 16×16 пикселей с временной глубиной в несколько кадров).

• Каждый патч затем преобразуется в эмбеддинг, используя линейную проекцию, представляющий собой векторное пространство.

• Для учета последовательности патчей добавляются позиционные эмбеддинги. Позиционные эмбеддинги – это способ «запомнить» порядок патчей для нейронной сети. Соответственно, мы получаем информацию о позиции патча в исходном изображении.

Важное дополнение: В ViViT позиционные эмбеддинги являются  обучаемыми параметрами.  Это означает, что сеть сама «учится»  оптимальному способу кодирования информации о положении патчей.

2. Transformer Encoder:

• Каждый патч (эмбеддинг и позиционный эмбеддинг) подаются на вход трансформера

• Трансформер состоит из нескольких слоев с механизмом внимания (attention), который помогает модели выучить зависимости между патчами.

3. Classification Head:

• После прохождения через энкодер модель выводит вектор признаков для классификации.

• Этот вектор подается в классификационный слой, который выдает вероятности каждого класса.

Для начала необходимо установить все необходимые библиотеки:

pip install transformers torch pillow opencv-python

Спасибо библиотеке transformers, что наши тесты будут максимально простыми и с минимальным количеством кода:

# Импортируем необходимые библиотеки from transformers import VivitForVideoClassification, VivitImageProcessor import torch import cv2  # Загружаем модель и процессинг кадров model_name = "google/vivit-b-16x2-kinetics400" processor = VivitImageProcessor.from_pretrained(model_name) model = VivitForVideoClassification.from_pretrained(model_name)  # Функция для загрузки и обработки видео def load_video(video_path, num_frames=32, frame_height=480, frame_width=480):     cap = cv2.VideoCapture(video_path)     frames = []     while len(frames) < num_frames:         ret, frame = cap.read()         if not ret:             break         frame = cv2.resize(frame, (frame_width, frame_height))         frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)         frames.append(frame)     cap.release()     if len(frames) < num_frames:         raise ValueError(f"Video is too short. Needed: {num_frames}, Found: {len(frames)}")     return frames  # Функция для загрузки меток def load_labels(label_path):     with open(label_path, 'r') as f:         labels = f.read().splitlines()     return labels   # Функция для получения предсказаний def predict_label(video_path, label_map_path):     frames = load_video(video_path)     inputs = processor(images=frames, return_tensors="pt")     outputs = model(**inputs)     # Загрузка меток     kinetics_labels = load_labels(label_map_path)     # Добавим вывод вероятностей     logits = outputs.logits     probabilities = torch.nn.functional.softmax(logits, dim=-1)     top_5_probs, top_5_indices = torch.topk(probabilities[0], 5)     top_5_predictions = [kinetics_labels[idx.item()] for idx in top_5_indices]      print("Топ 5 предсказаний:")     for i, (label, prob) in enumerate(zip(top_5_predictions, top_5_probs), 1):         print(f"{i}. {label}: {prob.item() * 100:.2f}%")  # Путь к видеофайлу video_path = 'snow.mp4'  # Загрузка label_map.txt происходит по след. ссылке: https://github.com/google-deepmind/kinetics-i3d/blob/master/data/label_map.txt # Получение предсказаний predict_label(video_path, "label_map.txt")

Давайте также коротко пробежимся по структуре:

1. Patch Embedding:

• Каждый фрейм делится на небольшие патчи (по аналогии с Vision Transformer, ViT).

• Патчи из каждого фрейма выравниваются по временной оси, образуя многослойную последовательность патчей.

• Патчи преобразуются в векторы фиксированной размерности через линейную проекцию.

• К этим векторным представлениям добавляется позиционное кодирование, чтобы сохранить информацию о позиции патчей в пространстве и времени.

2. Transformer Encoder:

• Энкодер состоит из нескольких слоев attention-механизмов и feed-forward сетей.

• В TimeSformer используется раздельное внимание (attention) к пространственным и временным аспектам.

• Space Attention (пространственное внимание) применяется ко всем патчам в каждом временном кадре независимо.

• Time Attention (временное внимание) применяется ко всем патчам из всех временных кадров для каждого пространственного патча.

3. Classification Head:

• После прохождения через энкодер модель выводит вектор признаков для классификации.

• Этот вектор подается в классификационный слой, который выдает вероятности каждого класса.

Код абсолютно идентичен приведенному в разделе ViViT, разница только в модели:

from transformers import AutoImageProcessor, TimesformerForVideoClassification  model_name = "facebook/timesformer-base-finetuned-k400" processor = AutoImageProcessor.from_pretrained(model_name) model = TimesformerForVideoClassification.from_pretrained(model_name)

Основной идеей является восстановление исходных данных из их частичной или искаженной версии. Для этого модель обучается понимать контекст и структуру данных, скрывая часть входных данных с помощью маскирования.

Процесс можно разделить на три этапа:
1) Маскирование: 

• Входное изображение разбивается на патчи (маленькие блоки).

• Применяется случайная или фиксированная маска, чтобы “спрятать” определенные части данных

• Оригинальные и замаскированные патчи передаются энкодеру.

2) Энкодер: 

• Преобразует видимые части данных в сжатое скрытое представление. Видимые патчи кодируются в скрытые представления через несколько слоев трансформера.

Важная часть — энкодер не видит замаскированные части данных, он тренируется распознавать структуру и контекст только на основе доступных ему данных. 

3) Декодер: 

• Декодер получает скрытые представления из энкодера, которые содержат видимые патчи и токены маски. Его цель – восстановить полные данные, включая замаскированные части.

• Оценивается насколько восстановленные данные отличаются от оригинальных замаскированных данных. Чаще используется среднеквадратичная ошибка (MSE).

Код также аналогичен приведенным ранее, но с небольшими изменениями:

from transformers import VideoMAEForVideoClassification, VideoMAEImageProcessor   model_name = "MCG-NJU/videomae-base-finetuned-kinetics" model = VideoMAEForVideoClassification.from_pretrained(model_name) feature_extractor = VideoMAEImageProcessor.from_pretrained(model_name)  # Изменения также касаются num_frames def load_video(video_path, num_frames=16, frame_height=480, frame_width=480):     cap = cv2.VideoCapture(video_path)     frames = []     try:         while len(frames) < num_frames:             ret, frame = cap.read()             if not ret:                 break             frame = cv2.resize(frame, (frame_width, frame_height))             frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)             frames.append(frame)     except Exception as e:         print(f"Error reading video: {e}")      cap.release()      if len(frames) < num_frames:         raise ValueError(f"Video is too short. Needed: {num_frames}, Found: {len(frames)}")      return frames

Здесь всё абсолютно идентично стандартному transfer learning:

# Замораживаем все слои, кроме последнего for param in model.parameters():     param.requires_grad = False  model.classifier.requires_grad = True  # Создадим класс для загрузки данных class VideoDataset(torch.utils.data.Dataset):     def __init__(self, video_paths, labels, num_frames=16):         self.video_paths = video_paths         self.labels = labels         self.num_frames = num_frames         self.processor = VideoMAEImageProcessor.from_pretrained(model_name)      def __len__(self):         return len(self.video_paths)      def __getitem__(self, idx):         video_path = self.video_paths[idx]         label = self.labels[idx]         frames = load_video(video_path, num_frames=self.num_frames)         inputs = self.processor(frames, return_tensors="pt")         return inputs['pixel_values'].squeeze(0), label         ''' Далее настраиваем пути до наших данных Выбираем подходящую loss функцию и выполняем стандартный цикл обучения  ''' dataset = VideoDataset(video_paths, labels) dataloader = torch.utils.data.DataLoader(dataset, batch_size=2, shuffle=True)  for epoch in range(num_epochs):   ...   

На выходе получаем веса, которые мы можем использовать для инференса на новых видео данных.

Основная идея заключается в том, чтобы использовать трансформер в качестве фиче-экстрактора, сохранении полученных фичей и дальнейшем обучении модели классификатора.

Преимущество данного подхода в том, что нет необходимости постоянно переобучать модели трансформера: можно просто извлечь фичи для новых данных и обучить новый легковесный классификатор, соответственно, снижаются вычислительные и временные затраты.

Недостаток, очевидно, в том, что модель фиче-экстрактора должна соответствовать вашим ожиданиям, т.е. уметь доставать требуемые признаки для дальнейшего анализа.

Давайте очень приблизительно изобразим подход в коде:

# Изменим получение предсказаний на получение признаков def extract_features(video_path):     frames = load_video(video_path)     inputs = processor(frames, return_tensors="pt").to(device)     with torch.no_grad():         outputs = model(**inputs, output_hidden_states=True)     features = outputs.hidden_states[-1].squeeze(0).mean(dim=1)     return features.cpu()  # Добавим сохранение признаков def save_features(features, save_path):     torch.save(features, save_path)  # Добавим пути до наших видео (только для примера в таком формате) video_paths = {     'class1': ['video1.mp4'],     'class2': ['video2.mp4'], }  # Извлечем и сохраним наши признаки в отдельные файлы (опять же, только для примера) for class_name, paths in video_paths.items():     class_save_dir = os.path.join(save_dir, class_name)     os.makedirs(class_save_dir, exist_ok=True)     for video_path in paths:         features = extract_features(video_path)         video_name = os.path.basename(video_path).split('.')[0]         save_path = os.path.join(class_save_dir, f'{video_name}_features.pt')         save_features(features, save_path)

Дальше можем делать с полученными признаками все, что угодно. Один из вариантов — обучить классификатор с triplet loss:

from pytorch_metric_learning import losses from pytorch_metric_learning.miners import TripletMarginMiner from torch.utils.data import Dataset  # Создаем класс для загрузки полученных ранее фичей class FeaturesLoad(Dataset):     def __init__(self, features_dir):       ...     def __getitem__(self, idx):       ...       return feature, label  # Описываем модель классификатора или используем готовый class SimpleCls(nn.Module):   ...  model = SimpleСls(...).to(device) criterion = losses.TripletMarginLoss(margin=0.1) triplet_miner = TripletMarginMiner(margin=0.2, type_of_triplets="hard")  # Обучение for epoch in range(num_epochs):     for features, labels in dataloader:         features, labels = features.to(device), labels.to(device)                  optimizer.zero_grad()         embeds = model(features)         triplets = triplet_miner(embeds, labels)         loss = criterion(embeds, labels, triplets)         loss.backward()         optimizer.step()