Workaround for blink detection ZeroDivisionError in Pupil Core software <v3.6, see https://github.com/pupil-labs/pupil/pull/2234

blink_detection_fixed.py

import logging

import numpy as np
from blink_detection import Offline_Blink_Detection
from scipy.signal import fftconvolve

logger = logging.getLogger(__name__)


class Fixed_Offline_Blink_Detection(Offline_Blink_Detection):
    @classmethod
    def parse_pretty_class_name(cls) -> str:
        return "Blink Detector (Fixed)"

    def recalculate(self):
        import time

        t0 = time.perf_counter()
        all_pp = self._pupil_data()
        if not all_pp:
            self.filter_response = []
            self.response_classification = []
            self.timestamps = []
            self.consolidate_classifications()
            return

        self.timestamps = all_pp.timestamps
        total_time = self.timestamps[-1] - self.timestamps[0]

        conf_iter = (pp["confidence"] for pp in all_pp)
        activity = np.fromiter(conf_iter, dtype=float, count=len(all_pp))
        total_time = all_pp[-1]["timestamp"] - all_pp[0]["timestamp"]
        filter_size = 2 * round(len(all_pp) * self.history_length / total_time / 2.0)
        blink_filter = np.ones(filter_size) / filter_size

        # This is different from the online filter. Convolution will flip
        # the filter and result in a reverse filter response. Therefore
        # we set the first half of the filter to -1 instead of the second
        # half such that we get the expected result.
        blink_filter[: filter_size // 2] *= -1

        # The theoretical response maximum is +-0.5
        # Response of +-0.45 seems sufficient for a confidence of 1.
        self.filter_response = fftconvolve(activity, blink_filter, "same") / 0.45

        onsets = self.filter_response > self.onset_confidence_threshold
        offsets = self.filter_response < -self.offset_confidence_threshold

        self.response_classification = np.zeros(self.filter_response.shape)
        self.response_classification[onsets] = 1.0
        self.response_classification[offsets] = -1.0

        self.consolidate_classifications()

        tm1 = time.perf_counter()
        logger.debug(
            "Recalculating took\n\t{:.4f}sec for {} pp\n\t{} pp/sec\n\tsize: {}".format(
                tm1 - t0, len(all_pp), len(all_pp) / (tm1 - t0), filter_size
            )
        )

    def on_notify(self, notification):
        if notification["subject"] == "blink_detection.should_recalculate":
            self.recalculate()
        elif notification["subject"] == "blinks_changed":
            self.cache_activation()
            try:
                self.timeline.refresh()
            except AttributeError:
                pass
        elif notification["subject"] == "should_export":
            self.export(notification["ts_window"], notification["export_dir"])