path: root/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py

#!/usr/bin/env python

"""Dispersion threshold identification (I-DT) module.
"""

__author__ = "Théo de la Hogue"
__credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "BSD"

from typing import TypeVar, Tuple
from dataclasses import dataclass, field
import math

from argaze import GazeFeatures, DataFeatures

import numpy
import cv2

GazeMovementType = TypeVar('GazeMovement', bound="GazeMovement")
# Type definition for type annotation convenience

FixationType = TypeVar('Fixation', bound="Fixation")
# Type definition for type annotation convenience

SaccadeType = TypeVar('Saccade', bound="Saccade")
# Type definition for type annotation convenience

class Fixation(GazeFeatures.Fixation):
    """Define dispersion based fixation."""

    def __init__(self, positions: GazeFeatures.TimeStampedGazePositions = (), finished: bool = False, message: str = None, **kwargs):

        super().__init__(positions, finished, message, **kwargs)

        if positions:

            positions_array = numpy.asarray(self.values())
            centroid = numpy.mean(positions_array, axis=0)
            deviations_array = numpy.sqrt(numpy.sum((positions_array - centroid)**2, axis=1))

            # Set focus as positions centroid
            self.focus = (centroid[0], centroid[1])

            # Set deviation_max attribute
            self.__deviation_max = deviations_array.max()

    @property
    def deviation_max(self):
        """Get fixation's maximal deviation."""
        return self.__deviation_max

    def overlap(self, fixation: FixationType) -> bool:
        """Does a gaze position from another fixation having a deviation to this fixation centroïd smaller than maximal deviation?"""
        
        positions_array = numpy.asarray(fixation.values())
        centroid = numpy.mean(self.focus, axis=0)
        deviations_array = numpy.sqrt(numpy.sum((positions_array - centroid)**2, axis=1))

        return min(deviations_array) <= self.deviation_max
    
    def draw(self, image: numpy.array, deviation_circle_color: tuple = None, duration_border_color: tuple = None, duration_factor: float = 1., draw_positions: dict = None):
        """Draw fixation into image.
        
        Parameters:
            deviation_circle_color: color of circle representing fixation's deviation
            duration_border_color: color of border representing fixation's duration
            duration_factor: how many pixels per duration unit
        """

        # Draw duration border if required
        if duration_border_color is not None:

            cv2.circle(image, (int(self.focus[0]), int(self.focus[1])), int(self.deviation_max), duration_border_color, int(self.duration * duration_factor))

        # Draw deviation circle if required
        if deviation_circle_color is not None:

            cv2.circle(image, (int(self.focus[0]), int(self.focus[1])), int(self.deviation_max), deviation_circle_color, -1)

        # Draw positions if required
        if draw_positions is not None:

            self.draw_positions(image, **draw_positions)

class Saccade(GazeFeatures.Saccade):
    """Define dispersion based saccade."""

    def __init__(self, positions: GazeFeatures.TimeStampedGazePositions = (), finished: bool = False, message: str = None, **kwargs):

        super().__init__(positions, finished, message, **kwargs)

    def draw(self, image: numpy.array, line_color: tuple = None):
        """Draw saccade into image.

        Parameters:
            line_color: color of line from first position to last position
        """

        # Draw line if required
        if line_color is not None:

            start_position = self[0]
            last_position = self[-1]

            cv2.line(image, (int(start_position[0]), int(start_position[1])), (int(last_position[0]), int(last_position[1])), line_color, 2)

class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier):
    """Implementation of the I-DT algorithm as described in:
            
        **Dario D. Salvucci and Joseph H. Goldberg (2000).**  
        *Identifying fixations and saccades in eye-tracking protocols.*  
        Proceedings of the 2000 symposium on Eye tracking research & applications (ETRA'00, 71-78).  
        [https://doi.org/10.1145/355017.355028](https://doi.org/10.1145/355017.355028)
    
    Parameters:
        deviation_max_threshold: Maximal distance allowed to consider a gaze movement as a fixation.
        duration_min_threshold: Minimal duration allowed to consider a gaze movement as a fixation. \
            It is also used as maximal duration allowed to wait valid gaze positions.
    """

    def __init__(self, deviation_max_threshold: int|float, duration_min_threshold: int|float):

        super().__init__()

        self.__deviation_max_threshold = deviation_max_threshold
        self.__duration_min_threshold = duration_min_threshold

        self.__valid_positions = GazeFeatures.TimeStampedGazePositions()
        self.__fixation_positions = GazeFeatures.TimeStampedGazePositions()
        self.__saccade_positions = GazeFeatures.TimeStampedGazePositions()

    @property
    def deviation_max_threshold(self):
        """Get identifier's deviation max threshold."""
        return self.__deviation_max_threshold

    @property
    def duration_min_threshold(self):
        """Get identifier duration min threshold."""
        return self.__duration_min_threshold
    
    @DataFeatures.PipelineStepMethod
    def identify(self, timestamp: int|float, gaze_position, terminate=False) -> GazeMovementType:
        
        # Ignore empty gaze position
        if not gaze_position:

            return GazeFeatures.GazeMovement() if not terminate else self.current_fixation.finish()
        
        # Check if too much time elapsed since last valid gaze position
        if self.__valid_positions:

            ts_last = self.__valid_positions[-1].timestamp

            if (timestamp - ts_last) > self.__duration_min_threshold:

                # Get last movement
                last_movement = self.current_gaze_movement.finish()

                # Clear all former gaze positions
                self.__valid_positions = GazeFeatures.TimeStampedGazePositions()
                self.__fixation_positions = GazeFeatures.TimeStampedGazePositions()
                self.__saccade_positions = GazeFeatures.TimeStampedGazePositions()

                # Store valid gaze position
                self.__valid_positions.append(gaze_position)

                # Return last valid movement if exist
                return last_movement

        # Store gaze positions until a minimal duration 
        self.__valid_positions.append(gaze_position)
        
        # Once the minimal duration is reached
        if self.__valid_positions.duration >= self.__duration_min_threshold:

            # Calculate the deviation of valid gaze positions
            deviation = Fixation(self.__valid_positions).deviation_max

            # Valid gaze positions deviation small enough
            if deviation <= self.__deviation_max_threshold:

                last_saccade = GazeFeatures.GazeMovement()

                # Is there saccade positions?
                if self.__saccade_positions:

                    # Copy oldest valid position into saccade positions
                    self.__saccade_positions.append(self.__valid_positions[0])

                    # Finish last saccade
                    last_saccade = self.current_saccade.finish()

                    # Clear saccade positions
                    self.__saccade_positions = GazeFeatures.TimeStampedGazePositions()

                # Copy valid gaze positions into fixation positions
                self.__fixation_positions = self.__valid_positions.copy()

                # Output last saccade
                return last_saccade if not terminate else self.current_fixation.finish()
                
            # Valid gaze positions deviation too wide
            else:

                last_fixation = GazeFeatures.GazeMovement()

                # Is there fixation positions?
                if self.__fixation_positions:

                    # Copy most recent fixation position into saccade positions
                    self.__saccade_positions.append(self.__fixation_positions[-1])

                    # Finish last fixation
                    last_fixation = self.current_fixation.finish()

                    # Clear fixation positions
                    self.__fixation_positions = GazeFeatures.TimeStampedGazePositions()

                    # Clear valid positions
                    self.__valid_positions = GazeFeatures.TimeStampedGazePositions()

                    # Store current gaze position
                    self.__valid_positions.append(gaze_position)

                    # Output last fixation
                    return last_fixation if not terminate else self.current_saccade.finish()

                # Move oldest valid position into saccade positions
                self.__saccade_positions.append(self.__valid_positions.pop(0))

        # Always return unvalid gaze movement at least
        return GazeFeatures.GazeMovement()

    @property
    def current_gaze_movement(self) -> GazeMovementType:

        # It shouldn't have a current fixation and a current saccade at the same time
        assert(not (self.__fixation_positions and len(self.__saccade_positions) > 1))

        if self.__fixation_positions:

            return Fixation(self.__fixation_positions)

        if len(self.__saccade_positions) > 1:
            
            return Saccade(self.__saccade_positions)

        # Always return unvalid gaze movement at least
        return GazeFeatures.GazeMovement()
    @property
    def current_fixation(self) -> FixationType:

        if self.__fixation_positions:

            return Fixation(self.__fixation_positions)

        # Always return unvalid gaze movement at least
        return GazeFeatures.GazeMovement()

    @property
    def current_saccade(self) -> SaccadeType:

        if len(self.__saccade_positions) > 1:

            return Saccade(self.__saccade_positions)

        # Always return unvalid gaze movement at least
        return GazeFeatures.GazeMovement()