FAST-VQA and FasterVQA

The project is designed with modular components to allow developers to modify backbone structures, sampling hyper-parameters, and loss functions.

Data Preprocessing

Core transformations are located in fastvqa/datasets/fusion_datasets.py. Specifically, examine the FusionDataset class and the get_spatial_and_temporal_samples function.

Spatial Sampling Approaches:

• fragments
• resize
• arp_resize (resizes while keeping the original Aspect Ratio)
• crop
• Note: These can be combined for multi-branch networks.

Temporal Sampling Approaches:

• SampleFrames (continuous frames, imported from MMAction2)
• FragmentSampleFrames (discontinuous, fragment-like frames)

Network Components

• Backbones: Supports Video Swin Transformer (with GRPB or vanilla) and ConvNext-I3D (vanilla).
• Heads: Supports the IP-NLR Head, which can generate local quality maps for videos.

To perform a quick benchmark of the video quality assessment models, follow these three steps:

1. Get Pretrained Weights: Download the weights corresponding to your desired model version (e.g., FAST-VQA-B, FasterVQA, or FAST-VQA-M). Ensure you use the correct .yml configuration file associated with the weights.
2. Download Datasets: Obtain the required datasets such as LSVQ, KoNViD-1k, or LIVE-VQC.
3. Execute Benchmark: Run the new_test.py script using your chosen options file.

Model Weight Reference Table:

python new_test.py -o [YOUR_OPTIONS]

For more granular control, you can manually instantiate a BaseEvaluator. The configuration depends on whether you are using the fast or faster architecture.

• fast: Uses a standard backbone.
• faster: Requires a backbone dictionary defining window_size and frag_biases.

When loading checkpoints, note that the weights may require a key transformation where 'cls' is replaced with 'vqa' in the state dictionary keys.

from fastvqa.models import BaseEvaluator
import torch
from collections import OrderedDict

model_type = 'fast' # or 'faster'
device = 'cuda'

# Configure backbone for 'faster' model
backbone = dict(window_size=(4,4,4), frag_biases=[False,False,True,False]) if model_type=='faster' else dict()
model = BaseEvaluator(backbone=backbone).to(device)

# Load and transform checkpoint
load_path = f'pretrained_weights/{model_type}_vqa_v0_3.pth'
state_dict = torch.load(load_path, map_location='cpu')

if 'state_dict' in state_dict:
    state_dict = state_dict['state_dict']
    i_state_dict = OrderedDict()
    for key, value in state_dict.items():
        if 'cls' in key:
            tkey = key.replace('cls', 'vqa')
            i_state_dict[tkey] = value
        else:
            i_state_dict[key] = value
    model.load_state_dict(i_state_dict)

The pyramid_ensemble_coefficients function evaluates the performance of an ensemble of varying sizes (from 1 to the total number of results) using a sliding window approach.

For each ensemble size i, it calculates the average of four metrics across all possible contiguous windows of that size:

1. Spearman Rank Correlation (spearmanr)
2. Pearson Correlation (pearsonr)
3. Kendall Tau (kendallr)
4. RMSE (Root Mean Square Error)

Input results should be a list of dictionaries where each dictionary contains 'pr_labels' (predicted labels) and 'gt_label' (ground truth label).

def pyramid_ensemble_coefficients(results):
    ens_result_len = results[0].get('pr_labels').shape[0]
    gt_labels = [r['gt_label'] for r in results]
    p_results = defaultdict()

    for i in range(1, ens_result_len + 1):
        scores = np.zeros(4)
        for j in range(ens_result_len - i + 1):
            pr_labels = [np.mean(r['pr_labels'][j:(j+i)]) for r in results]
            pr_labels = rescale(pr_labels, gt_labels)
            scores += np.array([
                spearmanr(gt_labels, pr_labels)[0], 
                pearsonr(gt_labels, pr_labels)[0], 
                kendallr(gt_labels, pr_labels)[0], 
                np.sqrt(((gt_labels - pr_labels) ** 2).mean())
            ])
        scores /= ens_result_len - i + 1
        p_results[i] = scores

    return p_results

The pyramid_ensemble_stds function measures the stability/variance of the ensemble predictions. It calculates the mean standard deviation of the predicted labels across different ensemble window sizes.

It returns a dictionary s_results where keys are the ensemble size i and values are an array containing the mean standard deviation across two different axes (typically representing different dimensions of the prediction array).

def pyramid_ensemble_stds(results):
    ens_result_len = results[0].get('pr_labels').shape[0]
    gt_labels = [r['gt_label'] for r in results]
    s_results = defaultdict()

    for i in range(1, ens_result_len + 1):
        all_pr_labels = []
        stds = np.zeros(2)

        for j in range(ens_result_len - i + 1):
            pr_labels = [np.mean(r['pr_labels'][j:(j+i)]) for r in results]
            pr_labels = rescale(pr_labels, gt_labels)
            all_pr_labels.append(pr_labels)

        all_pr_labels = np.stack(all_pr_labels, 1)
        for q in range(2):
            stds[q] = np.mean(np.std(all_pr_labels, q))
        s_results[i] = stds

    return s_results