1. Patch Embedding:

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

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

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

Важное дополнение: В 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()