title video classification with transformers sayak paulhttps twitter comrisingsayak date created 20210608 last modified 20232207 description training a video classifier with hybrid transformers accelerator gpu converted to keras 3 by soumik rakshithttp github comsoumik12345 this example is a followup to the video classification with a cnnrnn architecturehttps keras ioexamplesvisionvideoclassification example this time we will be using a transformerbased model vaswani et al https arxiv orgabs1706 03762 to classify videos you can follow this book chapterhttps livebook manning combookdeeplearningwithpythonsecondeditionchapter11 in case you need an introduction to transformers with code after reading this example you will know how to develop hybrid transformerbased models for video classification that operate on cnn feature maps shell pip install q githttps github comkerasteamkeras pip install q githttps github comtensorflowdocs data collection as done in the predecessorhttps keras ioexamplesvisionvideoclassification to this example we will be using a subsampled version of the ucf101 datasethttps www crcv ucf edudataucf101 php a wellknown benchmark dataset in case you want to operate on a larger subsample or even the entire dataset please refer to this notebookhttps colab research google comgithubsayakpaulactionrecognitionintensorflowblobmaindatapreparationucf101 ipynb shell wget q https github comsayakpaulactionrecognitionintensorflowreleasesdownloadv1 0 0ucf101top5 tar gz tar xf ucf101top5 tar gz setup define hyperparameters data preparation we will mostly be following the same data preparation steps in this example except for the following changes we reduce the image size to 128x128 instead of 224x224 to speed up computation instead of using a pretrained inceptionv3https arxiv orgabs1512 00567 network we use a pretrained densenet121http openaccess thecvf comcontentcvpr2017papershuangdenselyconnectedconvolutionalcvpr2017paper pdf for feature extraction we directly pad shorter videos to length maxseqlength first let s load up the dataframeshttps pandas pydata orgpandasdocsstablereferenceapipandas dataframe html following method is modified from this tutorial https www tensorflow orghubtutorialsactionrecognitionwithtfhub label preprocessing with stringlookup framefeatures are what we will feed to our sequence model for each video gather all its frames and add a batch dimension pad shorter videos initialize placeholder to store the features of the current video extract features from the frames of the current video calling prepareallvideos on traindf and testdf takes 20 minutes to complete for this reason to save time here we download already preprocessed numpy arrays shell wget q https git iojzmf4 o top5dataprepared tar gz tar xf top5dataprepared tar gz building the transformerbased model we will be building on top of the code shared in this book chapterhttps livebook manning combookdeeplearningwithpythonsecondeditionchapter11 of deep learning with python second ed https www manning combooksdeeplearningwithpython by franois chollet first selfattention layers that form the basic blocks of a transformer are orderagnostic since videos are ordered sequences of frames we need our transformer model to take into account order information we do this via positional encoding we simply embed the positions of the frames present inside videos with an embedding layerhttps keras ioapilayerscorelayersembedding we then add these positional embeddings to the precomputed cnn feature maps the inputs are of shape batchsize frames numfeatures now we can create a subclassed layer for the transformer utility functions for training model training and inference note this model has 4 23 million parameters which is way more than the sequence model 99918 parameters we used in the prequel of this example this kind of transformer model works best with a larger dataset and a longer pretraining schedule pad shorter videos extract features from the frames of the current video this utility is for visualization referenced from https www tensorflow orghubtutorialsactionrecognitionwithtfhub the performance of our model is far from optimal because it was trained on a small dataset this example is a follow up to the video classification with a cnn rnn architecture https keras io examples vision video_classification example this time we will be using a transformer based model vaswani et al https arxiv org abs 1706 03762 to classify videos you can follow this book chapter https livebook manning com book deep learning with python second edition chapter 11 in case you need an introduction to transformers with code after reading this example you will know how to develop hybrid transformer based models for video classification that operate on cnn feature maps shell pip install q git https github com keras team keras pip install q git https github com tensorflow docs data collection as done in the predecessor https keras io examples vision video_classification to this example we will be using a subsampled version of the ucf101 dataset https www crcv ucf edu data ucf101 php a well known benchmark dataset in case you want to operate on a larger subsample or even the entire dataset please refer to this notebook https colab research google com github sayakpaul action recognition in tensorflow blob main data_preparation_ucf101 ipynb shell wget q https github com sayakpaul action recognition in tensorflow releases download v1 0 0 ucf101_top5 tar gz tar xf ucf101_top5 tar gz setup define hyperparameters data preparation we will mostly be following the same data preparation steps in this example except for the following changes we reduce the image size to 128x128 instead of 224x224 to speed up computation instead of using a pre trained inceptionv3 https arxiv org abs 1512 00567 network we use a pre trained densenet121 http openaccess thecvf com content_cvpr_2017 papers huang_densely_connected_convolutional_cvpr_2017_paper pdf for feature extraction we directly pad shorter videos to length max_seq_length first let s load up the dataframes https pandas pydata org pandas docs stable reference api pandas dataframe html following method is modified from this tutorial https www tensorflow org hub tutorials action_recognition_with_tf_hub label preprocessing with stringlookup frame_features are what we will feed to our sequence model for each video gather all its frames and add a batch dimension pad shorter videos initialize placeholder to store the features of the current video extract features from the frames of the current video calling prepare_all_videos on train_df and test_df takes 20 minutes to complete for this reason to save time here we download already preprocessed numpy arrays shell wget q https git io jzmf4 o top5_data_prepared tar gz tar xf top5_data_prepared tar gz building the transformer based model we will be building on top of the code shared in this book chapter https livebook manning com book deep learning with python second edition chapter 11 of deep learning with python second ed https www manning com books deep learning with python by françois chollet first self attention layers that form the basic blocks of a transformer are order agnostic since videos are ordered sequences of frames we need our transformer model to take into account order information we do this via positional encoding we simply embed the positions of the frames present inside videos with an embedding layer https keras io api layers core_layers embedding we then add these positional embeddings to the precomputed cnn feature maps the inputs are of shape batch_size frames num_features now we can create a subclassed layer for the transformer utility functions for training model training and inference note this model has 4 23 million parameters which is way more than the sequence model 99918 parameters we used in the prequel of this example this kind of transformer model works best with a larger dataset and a longer pre training schedule pad shorter videos extract features from the frames of the current video this utility is for visualization referenced from https www tensorflow org hub tutorials action_recognition_with_tf_hub the performance of our model is far from optimal because it was trained on a small dataset
import os
import keras
from keras import layers
from keras.applications.densenet import DenseNet121

from tensorflow_docs.vis import embed

import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
import imageio
import cv2



MAX_SEQ_LENGTH = 20
NUM_FEATURES = 1024
IMG_SIZE = 128

EPOCHS = 5



train_df = pd.read_csv("train.csv")
test_df = pd.read_csv("test.csv")

print(f"Total videos for training: {len(train_df)}")
print(f"Total videos for testing: {len(test_df)}")

center_crop_layer = layers.CenterCrop(IMG_SIZE, IMG_SIZE)


def crop_center(frame):
    cropped = center_crop_layer(frame[None, ...])
    cropped = keras.ops.convert_to_numpy(cropped)
    cropped = keras.ops.squeeze(cropped)
    return cropped




def load_video(path, max_frames=0, offload_to_cpu=False):
    cap = cv2.VideoCapture(path)
    frames = []
    try:
        while True:
            ret, frame = cap.read()
            if not ret:
                break
            frame = frame[:, :, [2, 1, 0]]
            frame = crop_center(frame)
            if offload_to_cpu and keras.backend.backend() == "torch":
                frame = frame.to("cpu")
            frames.append(frame)

            if len(frames) == max_frames:
                break
    finally:
        cap.release()
    if offload_to_cpu and keras.backend.backend() == "torch":
        return np.array([frame.to("cpu").numpy() for frame in frames])
    return np.array(frames)


def build_feature_extractor():
    feature_extractor = DenseNet121(
        weights="imagenet",
        include_top=False,
        pooling="avg",
        input_shape=(IMG_SIZE, IMG_SIZE, 3),
    )
    preprocess_input = keras.applications.densenet.preprocess_input

    inputs = keras.Input((IMG_SIZE, IMG_SIZE, 3))
    preprocessed = preprocess_input(inputs)

    outputs = feature_extractor(preprocessed)
    return keras.Model(inputs, outputs, name="feature_extractor")


feature_extractor = build_feature_extractor()



label_processor = keras.layers.StringLookup(
    num_oov_indices=0, vocabulary=np.unique(train_df["tag"]), mask_token=None
)
print(label_processor.get_vocabulary())


def prepare_all_videos(df, root_dir):
    num_samples = len(df)
    video_paths = df["video_name"].values.tolist()
    labels = df["tag"].values
    labels = label_processor(labels[..., None]).numpy()

    
    frame_features = np.zeros(
        shape=(num_samples, MAX_SEQ_LENGTH, NUM_FEATURES), dtype="float32"
    )

    
    for idx, path in enumerate(video_paths):
        
        frames = load_video(os.path.join(root_dir, path))

        
        if len(frames) < MAX_SEQ_LENGTH:
            diff = MAX_SEQ_LENGTH - len(frames)
            padding = np.zeros((diff, IMG_SIZE, IMG_SIZE, 3))
            frames = np.concatenate(frames, padding)

        frames = frames[None, ...]

        
        temp_frame_features = np.zeros(
            shape=(1, MAX_SEQ_LENGTH, NUM_FEATURES), dtype="float32"
        )

        
        for i, batch in enumerate(frames):
            video_length = batch.shape[0]
            length = min(MAX_SEQ_LENGTH, video_length)
            for j in range(length):
                if np.mean(batch[j, :]) > 0.0:
                    temp_frame_features[i, j, :] = feature_extractor.predict(
                        batch[None, j, :]
                    )

                else:
                    temp_frame_features[i, j, :] = 0.0

        frame_features[idx,] = temp_frame_features.squeeze()

    return frame_features, labels






train_data, train_labels = np.load("train_data.npy"), np.load(
    "train_labels.npy"
)
test_data, test_labels = np.load("test_data.npy"), np.load("test_labels.npy")

print(f"Frame features in train set: {train_data.shape}")




class PositionalEmbedding(layers.Layer):
    def __init__(self, sequence_length, output_dim, **kwargs):
        super().__init__(**kwargs)
        self.position_embeddings = layers.Embedding(
            input_dim=sequence_length, output_dim=output_dim
        )
        self.sequence_length = sequence_length
        self.output_dim = output_dim

    def build(self, input_shape):
        self.position_embeddings.build(input_shape)

    def call(self, inputs):
        
        inputs = keras.ops.cast(inputs, self.compute_dtype)
        length = keras.ops.shape(inputs)[1]
        positions = keras.ops.arange(start=0, stop=length, step=1)
        embedded_positions = self.position_embeddings(positions)
        return inputs + embedded_positions





class TransformerEncoder(layers.Layer):
    def __init__(self, embed_dim, dense_dim, num_heads, **kwargs):
        super().__init__(**kwargs)
        self.embed_dim = embed_dim
        self.dense_dim = dense_dim
        self.num_heads = num_heads
        self.attention = layers.MultiHeadAttention(
            num_heads=num_heads, key_dim=embed_dim, dropout=0.3
        )
        self.dense_proj = keras.Sequential(
            [
                layers.Dense(dense_dim, activation=keras.activations.gelu),
                layers.Dense(embed_dim),
            ]
        )
        self.layernorm_1 = layers.LayerNormalization()
        self.layernorm_2 = layers.LayerNormalization()

    def call(self, inputs, mask=None):
        attention_output = self.attention(inputs, inputs, attention_mask=mask)
        proj_input = self.layernorm_1(inputs + attention_output)
        proj_output = self.dense_proj(proj_input)
        return self.layernorm_2(proj_input + proj_output)





def get_compiled_model(shape):
    sequence_length = MAX_SEQ_LENGTH
    embed_dim = NUM_FEATURES
    dense_dim = 4
    num_heads = 1
    classes = len(label_processor.get_vocabulary())

    inputs = keras.Input(shape=shape)
    x = PositionalEmbedding(
        sequence_length, embed_dim, name="frame_position_embedding"
    )(inputs)
    x = TransformerEncoder(
        embed_dim, dense_dim, num_heads, name="transformer_layer"
    )(x)
    x = layers.GlobalMaxPooling1D()(x)
    x = layers.Dropout(0.5)(x)
    outputs = layers.Dense(classes, activation="softmax")(x)
    model = keras.Model(inputs, outputs)

    model.compile(
        optimizer="adam",
        loss="sparse_categorical_crossentropy",
        metrics=["accuracy"],
    )
    return model


def run_experiment():
    filepath = "/tmp/video_classifier.weights.h5"
    checkpoint = keras.callbacks.ModelCheckpoint(
        filepath, save_weights_only=True, save_best_only=True, verbose=1
    )

    model = get_compiled_model(train_data.shape[1:])
    history = model.fit(
        train_data,
        train_labels,
        validation_split=0.15,
        epochs=EPOCHS,
        callbacks=[checkpoint],
    )

    model.load_weights(filepath)
    _, accuracy = model.evaluate(test_data, test_labels)
    print(f"Test accuracy: {round(accuracy * 100, 2)}%")

    return model




trained_model = run_experiment()




def prepare_single_video(frames):
    frame_features = np.zeros(
        shape=(1, MAX_SEQ_LENGTH, NUM_FEATURES), dtype="float32"
    )

    
    if len(frames) < MAX_SEQ_LENGTH:
        diff = MAX_SEQ_LENGTH - len(frames)
        padding = np.zeros((diff, IMG_SIZE, IMG_SIZE, 3))
        frames = np.concatenate(frames, padding)

    frames = frames[None, ...]

    
    for i, batch in enumerate(frames):
        video_length = batch.shape[0]
        length = min(MAX_SEQ_LENGTH, video_length)
        for j in range(length):
            if np.mean(batch[j, :]) > 0.0:
                frame_features[i, j, :] = feature_extractor.predict(
                    batch[None, j, :]
                )
            else:
                frame_features[i, j, :] = 0.0

    return frame_features


def predict_action(path):
    class_vocab = label_processor.get_vocabulary()

    frames = load_video(os.path.join("test", path), offload_to_cpu=True)
    frame_features = prepare_single_video(frames)
    probabilities = trained_model.predict(frame_features)[0]

    plot_x_axis, plot_y_axis = [], []

    for i in np.argsort(probabilities)[::-1]:
        plot_x_axis.append(class_vocab[i])
        plot_y_axis.append(probabilities[i])
        print(f"  {class_vocab[i]}: {probabilities[i] * 100:5.2f}%")

    plt.bar(plot_x_axis, plot_y_axis, label=plot_x_axis)
    plt.xlabel("class_label")
    plt.xlabel("Probability")
    plt.show()

    return frames





def to_gif(images):
    converted_images = images.astype(np.uint8)
    imageio.mimsave("animation.gif", converted_images, fps=10)
    return embed.embed_file("animation.gif")


test_video = np.random.choice(test_df["video_name"].values.tolist())
print(f"Test video path: {test_video}")
test_frames = predict_action(test_video)
to_gif(test_frames[:MAX_SEQ_LENGTH])