videomatch.py

import os
import logging
import faiss

from kats.detectors.cusum_detection import CUSUMDetector
from kats.detectors.robust_stat_detection import RobustStatDetector
from kats.consts import TimeSeriesData

from scipy import stats as st

import numpy as np
import pandas as pd

from videohash import compute_hashes, filepath_from_url
from config import FPS, ROLLING_WINDOW_SIZE

def index_hashes_for_video(url: str) -> faiss.IndexBinaryIVF:
    """ Compute hashes of a video and index the video using faiss indices and return the index. 
    
    Args:
        url (str): url to to compute hashes for and index.
    
    Returns:
        index (IndexBinaryIVF): an abstract structure for a FAISS-based binary index of the hashes.

    """
    # If the url already had indices created, fetch those.
    filepath = filepath_from_url(url)
    if os.path.exists(f'{filepath}.index'):
        logging.info(f"Loading indexed hashes from {filepath}.index")
        binary_index = faiss.read_index_binary(f'{filepath}.index') 
        logging.info(f"Index {filepath}.index has in total {binary_index.ntotal} frames")
        return binary_index

    # Create hash vectors for url by looping over hashes from the video.
    hash_vectors = np.array([x['hash'] for x in compute_hashes(url)])
    logging.info(f"Computed hashes for {hash_vectors.shape} frames.")

    # Initializing the quantizer.
    quantizer = faiss.IndexBinaryFlat(hash_vectors.shape[1]*8)

    # Initializing index.
    index = faiss.IndexBinaryIVF(quantizer, hash_vectors.shape[1]*8, min(16, hash_vectors.shape[0]))
    index.nprobe = 1 # Nr of nearest clusters to be searched per query. 

    # Training and write the quantizer.
    index.train(hash_vectors)
    index.add(hash_vectors)
    faiss.write_index_binary(index, f'{filepath}.index')
    logging.info(f"Indexed hashes for {index.ntotal} frames to {filepath}.index.")

    return index

def get_video_index(url: str):
    """" Builds up a FAISS index for a video.

    Args: 
        filepath (str): Location of the source video (video that is to be indexed)

    Returns:
        video_index (IndexBinaryIVF): an abstract structure for a FAISS-based binary index of the hashes.
        hash_vectors (ndarray): vector of the indexed frames that can be searched

    """
    video_index = index_hashes_for_video(url)
    
    # Make sure the index is indexable
    video_index.make_direct_map()
    
    # Retrieve original indices
    hash_vectors = np.array([video_index.reconstruct(i) for i in range(video_index.ntotal)]) 
    return video_index, hash_vectors

def compare_videos(hash_vectors, target_index, MIN_DISTANCE = 3):
    """ The comparison between the target and the original video will be plotted based
    on the matches between the target and the original video over time. The matches are determined
    based on the minimum distance between hashes (as computed by faiss-vectors) before they're considered a match.
    
    The results are returned as a triplet of 1D arrays:
    lims, D, I, where result for query i is in I[lims[i]:lims[i+1]] 
    (indices of neighbors), D[lims[i]:lims[i+1]] (distances).
    (See: https://github.com/facebookresearch/faiss/wiki/Special-operations-on-indexes)

    Args:
        hash_vectors (ndarray): vector of the indexed frames that can be searched.
        target_index (IndexBinaryIVF): an abstract structure for a FAISS-based binary index of the hashes.
        MIN_DISTANCE (int): minium distance for a match

    Returns:
        lims (ndarray): from where to where in I and D the result for query i is 
        D (ndarray): distances of the vectors within a radius around the query point
        I (ndarray): indices of the neighbours
        hash_vectors (ndarray): vector of the indexed frames that can be searched.

    """
    lims, D, I = target_index.range_search(hash_vectors, MIN_DISTANCE)
    return lims, D, I, hash_vectors

def get_decent_distance(video_index, hash_vectors, target_index, MIN_DISTANCE, MAX_DISTANCE):
    """ To get a decent heurstic for a base distance check every distance from MIN_DISTANCE to MAX_DISTANCE
    until the number of matches found is equal to or higher than the number of frames in the source video.
    If the number of matches with a certain video is larger than the amount of frames, we set the distance heuristic.
    This was emperically determined to be a decent heuristic to find the distance heuristic
    
    Args:
        video_index (IndexBinaryIVF): The index of the source video
        hash_vectors (ndarray): The hash vectors of the target video
        target_index (IndexBinaryIVF): The index of the target video 
        MIN_DISTANCE (int): Minimum distance between vectors to be considered a match.
        MAX_DISTANCE (int): Maximum distance between vectors to prevent bad matches.

    Returns:
        None if not distance is found, otherwise an integer representing the heuristic distance value.

    """
    # Go over every distance with a step size of 2, since the distance increases/decreases with that step size
    for distance in np.arange(start = MIN_DISTANCE - 2, stop = MAX_DISTANCE + 2, step = 2, dtype=int):
        distance = int(distance) # Cast for safety
        _, D, _, _ = compare_videos(hash_vectors, target_index, MIN_DISTANCE = distance)
        nr_source_frames = video_index.ntotal
        nr_matches = len(D)
        if nr_matches > 0:
            logging.info(f"{(nr_matches/nr_source_frames) * 100.0:.1f}% of frames have a match for distance '{distance}' ({nr_matches} matches for {nr_source_frames} frames)")
        if nr_matches >= nr_source_frames:
            return distance  
    logging.warning(f"No matches found for any distance between {MIN_DISTANCE} and {MAX_DISTANCE}")

    return None        
    
def get_change_points(df, smoothing_window_size=10, method='ROBUST', metric="ROLL_OFFSET_MODE"):
    """Using https://github.com/facebookresearch/Kats to analyze the data to find points where the metric 
    changes. 

    Args:
        df (DataFrame): Dataframe holding the information between the matching of two videos 
        smoothing_window_size (int): Smoothing window for the timeseries analysis. Defaults to 10.
        method (str): Method for the timeseries analyis. Defaults to 'ROBUST'.
        metric (str): Main reporting metric for the timeseries analysis. Defaults to "ROLL_OFFSET_MODE".

    Returns:
        change_points [TimeSeriesChangePoint]: Array of time series change point objects.

    """
    # Convert the df to how kats wants it 
    tsd = TimeSeriesData(df.loc[:,['time', metric]])

    # Depending on the method get the change points
    if method.upper() == "CUSUM":
        detector = CUSUMDetector(tsd)
    elif method.upper() == "ROBUST":
        detector = RobustStatDetector(tsd)
    change_points =  detector.detector(smoothing_window_size=smoothing_window_size, comparison_window=-2)

    # Log some statistics
    if method.upper() == "CUSUM" and change_points != []:
        mean_offset_prechange = change_points[0].mu0 
        mean_offset_postchange = change_points[0].mu1 
        jump_s = mean_offset_postchange - mean_offset_prechange
        logging.info(f"Video jumps {jump_s:.1f}s in time at {mean_offset_prechange:.1f} seconds")

    return change_points

def get_videomatch_df(lims, D, I, hash_vectors, distance, window_size=ROLLING_WINDOW_SIZE, vanilla_df=False):
    """Get the dataframe holding all information of the comparison between two videos.

    Args:
        lims (ndarray): from where to where in I and D the result for query i is 
        D (ndarray): distances of the vectors within a radius around the query point
        I (ndarray): indices of the neighbours
        hash_vectors (ndarray): vector of the indexed frames that can be searched.
        distance (int): heuristic distance to use for the search for most accurate matches.
        window_size (int): Rolling window size that is used when calculating the mode. Defaults to ROLLING_WINDOW_SIZE.
        vanilla_df: Toggle for returning other baseline dataframe. Defaults to False.

    Returns:
        df (DataFrame): Dataframe with extra information added about decision making regarding the match between videos.

    """
    # Get match locations in seconds
    target = [(lims[i+1]-lims[i]) * [i] for i in range(hash_vectors.shape[0])]
    target_s = [i/FPS for j in target for i in j]
    source_s = [i/FPS for i in I]

    # Make dataframe
    df = pd.DataFrame(zip(target_s, source_s, D, I), columns = ['TARGET_S', 'SOURCE_S', 'DISTANCE', 'INDICES'])
    if vanilla_df:
        return df

    # Weight values by distance of their match
    df['TARGET_WEIGHT'] = 1 - df['DISTANCE']/distance # Higher value means a better match    
    df['SOURCE_WEIGHTED_VALUE'] = df['SOURCE_S'] * df['TARGET_WEIGHT'] # Multiply the weight (which indicates a better match) with the value for Y and aggregate to get a less noisy estimate of Y

    # Group by X so for every second/x there will be 1 source value in the end
    grouped_X = df.groupby('TARGET_S').agg({'SOURCE_WEIGHTED_VALUE' : 'sum', 'TARGET_WEIGHT' : 'sum'})
    grouped_X['FINAL_SOURCE_VALUE'] = grouped_X['SOURCE_WEIGHTED_VALUE'] / grouped_X['TARGET_WEIGHT'] 

    # Remake the dataframe
    df = grouped_X.reset_index()
    df = df.drop(columns=['SOURCE_WEIGHTED_VALUE', 'TARGET_WEIGHT'])
    df = df.rename({'FINAL_SOURCE_VALUE' : 'SOURCE_S'}, axis='columns')

    # Add NAN to "missing" x values
    step_size = 1/FPS
    x_complete =  np.round(np.arange(start=0.0, stop = max(df['TARGET_S'])+step_size, step = step_size), 1) # More robust    
    df['TARGET_S'] = np.round(df['TARGET_S'], 1)
    df_complete = pd.DataFrame(x_complete, columns=['TARGET_S'])

    # Merge dataframes to get NAN values for every missing SOURCE_S
    df = df_complete.merge(df, on='TARGET_S', how='left')

    # Interpolate between frames since there are missing values
    df['SOURCE_LIP_S'] = df['SOURCE_S'].interpolate(method='linear', limit_direction='both', axis=0)
   
    # Add timeshift col and timeshift col with Linearly Interpolated Values (LIP)
    df['TIMESHIFT'] = df['SOURCE_S'].shift(1) - df['SOURCE_S']
    df['TIMESHIFT_LIP'] = df['SOURCE_LIP_S'].shift(1) - df['SOURCE_LIP_S']

    # Add offset col that assumes the video is played at the same speed as the other to do a "timeshift"
    df['OFFSET'] = df['SOURCE_S'] - df['TARGET_S'] - np.min(df['SOURCE_S'])
    df['OFFSET_LIP'] = df['SOURCE_LIP_S'] - df['TARGET_S'] - np.min(df['SOURCE_LIP_S'])
    
    # Add rolling window mode
    df['ROLL_OFFSET_MODE'] = np.round(df['OFFSET_LIP'], 0).rolling(window_size, center=True, min_periods=1).apply(lambda x: st.mode(x)[0])

    # Add time column for plotting
    df['time'] = pd.to_datetime(df["TARGET_S"], unit='s') # Needs a datetime as input

    return df