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

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

"""
This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by the Free Software
Foundation, either version 3 of the License, or (at your option) any later
version.
This program is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
You should have received a copy of the GNU General Public License along with
this program. If not, see <https://www.gnu.org/licenses/>.
"""

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

import math

from argaze import GazeFeatures, DataFeatures

import cv2
import numpy

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

    def __init__(self, focus: tuple = (), deviation_max: float = math.nan, **kwargs):

        super().__init__(**kwargs)

        self._focus = focus
        self.__deviation_max = deviation_max

    @property
    def deviation_max(self) -> float:
        """Fixation's maximal deviation."""
        return self.__deviation_max

    def is_overlapping(self, fixation: GazeFeatures.Fixation) -> 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:
            image: where to draw
            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_positions:
        """

        # 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 = None, **kwargs):

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

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

        Parameters:
            image: where to draw
            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)

        # Draw positions if required
        if draw_positions is not None:

            self.draw_positions(image, **draw_positions)

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)
    """

    @DataFeatures.PipelineStepInit
    def __init__(self, **kwargs):

        # Init GazeMovementIdentifier class
        super().__init__()

        self.__deviation_max_threshold = 0
        self.__duration_min_threshold = 0

        self.__valid_positions = GazeFeatures.TimeStampedGazePositions()

        self.__centroid = ()
        self.__deviations = []

        self.__fixation = Fixation()
        self.__saccade = Saccade()

    @property
    def deviation_max_threshold(self) -> int|float:
        """Maximal distance allowed to consider a gaze movement as a fixation."""
        return self.__deviation_max_threshold

    @deviation_max_threshold.setter
    def deviation_max_threshold(self, deviation_max_threshold: int|float):

        self.__deviation_max_threshold = deviation_max_threshold

    @property
    def duration_min_threshold(self) -> int|float:
        """Minimal duration allowed to consider a gaze movement as a fixation.
        It is also used as maximal duration allowed to wait valid gaze positions."""
        return self.__duration_min_threshold

    @duration_min_threshold.setter
    def duration_min_threshold(self, duration_min_threshold: int|float):

        self.__duration_min_threshold = duration_min_threshold
    
    @DataFeatures.PipelineStepMethod
    def identify(self, gaze_position, terminate=False) -> GazeFeatures.GazeMovement:

        # When too much time elapsed since the last valid gaze position
        if self.__valid_positions:

            elapsed_time = gaze_position.timestamp - self.__valid_positions[-1].timestamp

            if elapsed_time > self.__duration_min_threshold:

                try:

                    # Finish and return current gaze movement
                    return self.current_gaze_movement().finish()

                finally:

                    # Reset valid gaze positions
                    self.__valid_positions = GazeFeatures.TimeStampedGazePositions()

                    # Reset centroid and deviations
                    self.__centroid = ()
                    self.__deviations = []

                    # Clear gaze movements
                    self.__fixation = Fixation()
                    self.__saccade = Saccade()
        
        # Consider only valid gaze position
        if gaze_position:

            # Update centroid and deviations
            if not self.__valid_positions:

                self.__centroid = gaze_position
                self.__deviations = []

            else:

                self.__centroid = self.__centroid + (gaze_position - self.__centroid) / (len(self.__valid_positions) + 1)
                self.__deviations.append(gaze_position.distance(self.__centroid))

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

            # Once the minimal duration is reached
            if self.__valid_positions.duration >= self.__duration_min_threshold:

                deviation_max = max(self.__deviations)

                # Maximal deviation small enough
                if deviation_max <= self.__deviation_max_threshold:

                    # Make valid gaze positions as current fixation
                    self.__fixation = Fixation(positions=self.__valid_positions, focus=self.__centroid, deviation_max=deviation_max)

                    # Is there a current saccade?
                    if self.__saccade:

                        try:

                            # Share first fixation position with current saccade
                            self.__saccade.append(self.__fixation[0])

                            # Finish and return the current saccade
                            return self.__saccade.finish()

                        # Clear saccade after the return
                        finally:

                            self.__saccade = Saccade()
                    
                # Maximal deviation too wide
                else:

                    # Is there a current fixation?
                    if self.__fixation:

                        try:

                            # Share last fixation position with current saccade
                            self.__saccade.append(self.__fixation[-1])

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

                            # Finish and return the current fixation
                            return self.__fixation.finish()

                        # Clear fixation after the return
                        finally:

                            self.__fixation = Fixation()

                    # No fixation case:

                    # Remove oldest valid position
                    old_gaze_position = self.__valid_positions.pop(0)

                    # Move oldest valid position into current saccade
                    self.__saccade.append(old_gaze_position)

                    # Update centroid and deviations
                    self.__centroid = self.__centroid - (old_gaze_position - self.__centroid) / (len(self.__valid_positions) + 1)
                    self.__deviations.pop(0)

        # Return current gaze movement
        return self.current_gaze_movement() if not terminate else self.current_gaze_movement().finish()

    def current_gaze_movement(self) -> GazeFeatures.GazeMovement:

        if self.__fixation:

            return self.__fixation

        if len(self.__saccade) > 1:

            return self.__saccade

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

    def current_fixation(self) -> GazeFeatures.GazeMovement:

        if self.__fixation:

            return self.__fixation

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

    def current_saccade(self) -> GazeFeatures.GazeMovement:

        if len(self.__saccade) > 1:
            
            return self.__saccade

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