Deduplicate Gaze Timestamps

deduplicate_gaze.py

import numpy as np

from plugin import Plugin
from player_methods import Bisector


class DeduplicateGaze(Plugin):
    @classmethod
    def parse_pretty_class_name(cls) -> str:
        return "(Custom) Deduplicate Gaze Timestamps"

    def __init__(self, g_pool):
        super().__init__(g_pool)
        self.remove_duplicate_gaze()

    def on_notify(self, notification):
        if notification["subject"] == "gaze_positions_changed":
            if notification.get("from_custom_deduplicator", False):
                return
            self.remove_duplicate_gaze()

    def remove_duplicate_gaze(self):
        data = self.g_pool.gaze_positions.data
        ts = self.g_pool.gaze_positions.data_ts

        unique_ts, unique_idxs = np.unique(ts, return_index=True)
        if len(unique_ts) == len(ts):
            return

        unique_data = [data[i] for i in unique_idxs]

        self.g_pool.gaze_positions = Bisector(unique_data, unique_ts)
        self.notify_all(
            {"subject": "gaze_positions_changed", "from_custom_deduplicator": True}
        )