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

#!/usr/bin/env python

""" """

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

import unittest
import random
import time
import math

from argaze import GazeFeatures
from argaze.GazeAnalysis import DispersionThresholdIdentification

import numpy

def build_gaze_fixation(size: int, center: tuple, deviation_max: float, min_time: float, max_time: float, start_ts: float = 0., validity: list = []):
    """ Generate N TimeStampedGazePsoitions dispersed around a center point for testing purpose.
        Timestamps are current time after random sleep (second).
        GazePositions are random values.
    """
    ts_gaze_positions = GazeFeatures.TimeStampedGazePositions()

    start_time = time.time()

    for i in range(0, size):

        # Sleep a random time
        sleep_time = random.random() * (max_time - min_time) + min_time
        time.sleep(sleep_time)

        # Check position validity
        valid = True
        if len(validity) > i:

             valid = validity[i]

        if valid:

            # Edit gaze position
            random_x = center[0] + deviation_max * (random.random() - 0.5) / math.sqrt(2)
            random_y = center[1] + deviation_max * (random.random() - 0.5) / math.sqrt(2)
            gaze_position = GazeFeatures.GazePosition((random_x, random_y))

        else:

            gaze_position = GazeFeatures.UnvalidGazePosition()

        # Store gaze position
        ts = time.time() - start_time + start_ts
        ts_gaze_positions[ts] = gaze_position

    return ts_gaze_positions

def build_gaze_saccade(size: int, center_A: tuple, center_B: tuple, min_time: float, max_time: float, start_ts: float = 0., validity: list = []):
    """ Generate N TimeStampedGazePsoitions between 2 center points for testing purpose.
        Timestamps are current time after random sleep (second).
        GazePositions are random values.
    """
    ts_gaze_positions = GazeFeatures.TimeStampedGazePositions()

    start_time = time.time()

    for i in range(0, size):

        # Sleep a random time
        sleep_time = random.random() * (max_time - min_time) + min_time
        time.sleep(sleep_time)

        # Check position validity
        valid = True
        if len(validity) > i:

             valid = validity[i]

        if valid:

            # Edit gaze position
            move_x = center_A[0] + (center_B[0] - center_A[0]) * (i / size)
            move_y = center_A[1] + (center_B[1] - center_A[1]) * (i / size)
            gaze_position = GazeFeatures.GazePosition((move_x, move_y))

        else:

            gaze_position = GazeFeatures.UnvalidGazePosition()

        # Store gaze position
        ts = time.time() - start_time + start_ts
        ts_gaze_positions[ts] = gaze_position

    return ts_gaze_positions

class TestDispersionThresholdIdentificationClass(unittest.TestCase):
    """Test DispersionThresholdIdentification class."""

    def test_fixation_identification(self):
        """Test DispersionThresholdIdentification fixation identification."""

        size = 10
        center = (0, 0)
        deviation_max = 10
        min_time = 0.01
        max_time = 0.1
        
        ts_gaze_positions = build_gaze_fixation(size, center, deviation_max, min_time, max_time)
        gaze_movement_identifier = DispersionThresholdIdentification.GazeMovementIdentifier(deviation_max_threshold=deviation_max, duration_min_threshold=max_time*2)
        ts_fixations, ts_saccades, ts_status = gaze_movement_identifier.browse(ts_gaze_positions)

        # Check result size
        self.assertEqual(len(ts_fixations), 1)
        self.assertEqual(len(ts_saccades), 0)
        self.assertEqual(len(ts_status), size)

        # Check fixation
        ts, fixation = ts_fixations.pop_first()

        self.assertEqual(len(fixation.positions.keys()), size)
        self.assertLessEqual(fixation.deviation_max, deviation_max)
        self.assertGreaterEqual(fixation.duration, (size - 1) * min_time)
        self.assertLessEqual(fixation.duration, (size - 1) * max_time)
        self.assertLessEqual(fixation.finished, True)

    def test_fixation_and_direct_saccade_identification(self):
        """Test DispersionThresholdIdentification fixation and saccade identification."""

        size = 10
        center_A = (0, 0)
        center_B = (50, 50)
        deviation_max = 10
        min_time = 0.01
        max_time = 0.1
        
        ts_gaze_positions_A = build_gaze_fixation(size, center_A, deviation_max, min_time, max_time)
        ts_gaze_positions_B = build_gaze_fixation(size, center_B, deviation_max, min_time, max_time, start_ts=ts_gaze_positions_A.last[0])

        ts_gaze_positions = ts_gaze_positions_A.append(ts_gaze_positions_B)

        gaze_movement_identifier = DispersionThresholdIdentification.GazeMovementIdentifier(deviation_max_threshold=deviation_max, duration_min_threshold=max_time*2)
        ts_fixations, ts_saccades, ts_status = gaze_movement_identifier.browse(ts_gaze_positions)

        # Check result size
        self.assertEqual(len(ts_fixations), 2)
        self.assertEqual(len(ts_saccades), 1)
        self.assertEqual(len(ts_status), size * 2)

        # Check first fixation
        ts, fixation = ts_fixations.pop_first()

        self.assertEqual(len(fixation.positions.keys()), size)
        self.assertLessEqual(fixation.deviation_max, deviation_max)
        self.assertGreaterEqual(fixation.duration, (size - 1) * min_time)
        self.assertLessEqual(fixation.duration, (size - 1) * max_time)
        self.assertLessEqual(fixation.finished, True)

        # Check first saccade
        ts, saccade = ts_saccades.pop_first()

        self.assertEqual(len(saccade.positions.keys()), 2)
        self.assertGreaterEqual(saccade.duration, min_time)
        self.assertLessEqual(saccade.duration, max_time)
        self.assertLessEqual(saccade.finished, True)

        # Check that last position of a movement is equal to first position of next movement
        last_ts, last_position = fixation.positions.last
        first_ts, first_position = saccade.positions.first

        self.assertEqual(last_ts, first_ts)
        self.assertEqual(last_position.value, first_position.value)

        # Check second fixation
        ts, fixation = ts_fixations.pop_first()

        self.assertEqual(len(fixation.positions.keys()), size)
        self.assertLessEqual(fixation.deviation_max, deviation_max)
        self.assertGreaterEqual(fixation.duration, (size - 1) * min_time)
        self.assertLessEqual(fixation.duration, (size - 1) * max_time)
        self.assertLessEqual(fixation.finished, True)

        # Check that last position of a movement is equal to first position of next movement
        last_ts, last_position = saccade.positions.last
        first_ts, first_position = fixation.positions.first

        self.assertEqual(last_ts, first_ts)
        self.assertEqual(last_position.value, first_position.value)

    def test_fixation_and_short_saccade_identification(self):
        """Test DispersionThresholdIdentification fixation and saccade identification."""

        size = 10
        move = 2
        center_A = (0, 0)
        out_A = (20, 20)
        center_B = (50, 50)
        deviation_max = 10
        min_time = 0.01
        max_time = 0.1
        
        ts_gaze_positions_A = build_gaze_fixation(size, center_A, deviation_max, min_time, max_time)
        ts_move_positions = build_gaze_saccade(move, out_A, center_B, min_time, min_time, start_ts=ts_gaze_positions_A.last[0])
        ts_gaze_positions_B = build_gaze_fixation(size, center_B, deviation_max, min_time, max_time, start_ts=ts_move_positions.last[0])

        ts_gaze_positions = ts_gaze_positions_A.append(ts_move_positions).append(ts_gaze_positions_B)

        gaze_movement_identifier = DispersionThresholdIdentification.GazeMovementIdentifier(deviation_max_threshold=deviation_max, duration_min_threshold=max_time*2)
        ts_fixations, ts_saccades, ts_status = gaze_movement_identifier.browse(ts_gaze_positions)

        # Check result size
        self.assertEqual(len(ts_fixations), 2)
        self.assertEqual(len(ts_saccades), 1)
        self.assertEqual(len(ts_status), size * 2 + move)

        # Check first fixation
        ts, fixation = ts_fixations.pop_first()

        self.assertEqual(len(fixation.positions.keys()), size)
        self.assertLessEqual(fixation.deviation_max, deviation_max)
        self.assertGreaterEqual(fixation.duration, (size - 1) * min_time)
        self.assertLessEqual(fixation.duration, (size - 1) * max_time)
        self.assertLessEqual(fixation.finished, True)

        # Check first saccade
        ts, saccade = ts_saccades.pop_first()

        self.assertEqual(len(saccade.positions.keys()), move + 2)
        self.assertGreaterEqual(saccade.duration, (move + 1) * min_time)
        self.assertLessEqual(saccade.duration, (move + 1) * max_time)
        self.assertLessEqual(saccade.finished, True)

        # Check that last position of a movement is equal to first position of next movement
        last_ts, last_position = fixation.positions.last
        first_ts, first_position = saccade.positions.first

        self.assertEqual(last_ts, first_ts)
        self.assertEqual(last_position.value, first_position.value)

        # Check second fixation
        ts, fixation = ts_fixations.pop_first()

        self.assertEqual(len(fixation.positions.keys()), size)
        self.assertLessEqual(fixation.deviation_max, deviation_max)
        self.assertGreaterEqual(fixation.duration, (size - 1) * min_time)
        self.assertLessEqual(fixation.duration, (size - 1) * max_time)
        self.assertLessEqual(fixation.finished, True)

        # Check that last position of a movement is equal to first position of next movement
        last_ts, last_position = saccade.positions.last
        first_ts, first_position = fixation.positions.first

        self.assertEqual(last_ts, first_ts)
        self.assertEqual(last_position.value, first_position.value)

    def test_invalid_gaze_position(self):
        """Test DispersionThresholdIdentification fixation and saccade identification with invalid gaze position."""

        size = 15
        center = (0, 0)
        deviation_max = 10
        min_time = 0.05
        max_time = 0.1
        validity = [True, True, True, True, True, True, True, False, False, False, True, True, True, True, True]
        
        ts_gaze_positions = build_gaze_fixation(size, center, deviation_max, min_time, max_time, validity=validity)

        gaze_movement_identifier = DispersionThresholdIdentification.GazeMovementIdentifier(deviation_max_threshold=deviation_max, duration_min_threshold=max_time*2)
        ts_fixations, ts_saccades, ts_status = gaze_movement_identifier.browse(ts_gaze_positions)

        # Check result size
        self.assertEqual(len(ts_fixations), 2)
        self.assertEqual(len(ts_saccades), 0)
        self.assertEqual(len(ts_status), size-3)

        # Check first fixation
        ts, fixation = ts_fixations.pop_first()

        self.assertEqual(len(fixation.positions.keys()), 7)
        self.assertLessEqual(fixation.deviation_max, deviation_max)
        self.assertGreaterEqual(fixation.duration, 6 * min_time)
        self.assertLessEqual(fixation.duration, 6 * max_time)
        self.assertLessEqual(fixation.finished, True)

        # Check second fixation
        ts, fixation = ts_fixations.pop_first()

        self.assertEqual(len(fixation.positions.keys()), 5)
        self.assertLessEqual(fixation.deviation_max, deviation_max)
        self.assertGreaterEqual(fixation.duration, 4 * min_time)
        self.assertLessEqual(fixation.duration, 4 * max_time)
        self.assertLessEqual(fixation.finished, True)

    def test_fixation_overlapping(self):
        """Test Fixation overlap function."""

        size = 10
        center_A = (0, 0)
        center_B = (50, 50)
        center_C = (55, 55)
        deviation_max = 15
        min_time = 0.01
        max_time = 0.1
        
        ts_gaze_positions_A = build_gaze_fixation(size, center_A, deviation_max, min_time, max_time)
        ts_gaze_positions_B = build_gaze_fixation(size, center_B, deviation_max, min_time, max_time, start_ts=ts_gaze_positions_A.last[0])
        ts_gaze_positions_C = build_gaze_fixation(size, center_C, deviation_max, min_time, max_time, start_ts=ts_gaze_positions_B.last[0])

        fixation_A = DispersionThresholdIdentification.Fixation(ts_gaze_positions_A)
        fixation_B = DispersionThresholdIdentification.Fixation(ts_gaze_positions_B)
        fixation_C = DispersionThresholdIdentification.Fixation(ts_gaze_positions_C)
        
        # Check that fixation doesn't overlap
        self.assertFalse(fixation_A.overlap(fixation_B))
        self.assertFalse(fixation_B.overlap(fixation_A))

        # Check that fixation overlaps
        self.assertTrue(fixation_B.overlap(fixation_C))
        self.assertTrue(fixation_C.overlap(fixation_B))

    @unittest.skip("Fixation overlapping is not supported anymore.")
    def test_fixation_overlapping_identification(self):
        """Test DispersionThresholdIdentification identification when fixations overlap."""

        size = 50
        center_A = (-5, 0)
        center_B = (5, 0)
        deviation_max = 15
        min_time = 0.01
        max_time = 0.1
        
        ts_gaze_positions_A = build_gaze_fixation(size, center_A, deviation_max, min_time, max_time)
        ts_gaze_positions_B = build_gaze_fixation(size, center_B, deviation_max, min_time, max_time, start_ts=ts_gaze_positions_A.last[0])

        ts_gaze_positions = ts_gaze_positions_A.append(ts_gaze_positions_B)

        gaze_movement_identifier = DispersionThresholdIdentification.GazeMovementIdentifier(deviation_max_threshold=deviation_max, duration_min_threshold=max_time*2)
        ts_fixations, ts_saccades, ts_status = gaze_movement_identifier.browse(ts_gaze_positions)

        # Check result size
        self.assertEqual(len(ts_fixations), 1)
        self.assertEqual(len(ts_saccades), 0)
        self.assertEqual(len(ts_status), size*2)

        # Check unique fixation
        ts, fixation = ts_fixations.pop_first()

        self.assertEqual(len(fixation.positions.keys()), size * 2)
        #self.assertGreaterEqual(fixation.deviation_max, deviation_max)
        self.assertGreaterEqual(fixation.duration, (2 * size - 1) * min_time)
        self.assertLessEqual(fixation.duration, (2 * size - 1) * max_time)
        self.assertLessEqual(fixation.finished, True)

    def test_identification_browsing(self):
        """Test DispersionThresholdIdentification identification browsing."""

        size = 10
        center_A = (0, 0)
        center_B = (50, 50)
        deviation_max = 10
        min_time = 0.01
        max_time = 0.1
        
        ts_gaze_positions_A = build_gaze_fixation(size, center_A, deviation_max, min_time, max_time)
        ts_gaze_positions_B = build_gaze_fixation(size, center_B, deviation_max, min_time, max_time, start_ts=ts_gaze_positions_A.last[0])

        ts_gaze_positions = ts_gaze_positions_A.append(ts_gaze_positions_B)

        gaze_movement_identifier = DispersionThresholdIdentification.GazeMovementIdentifier(deviation_max_threshold=deviation_max, duration_min_threshold=max_time*2)
           
        # Get last ts to terminate identification on last gaze position
        last_ts, _ = ts_gaze_positions.last

        # Iterate on gaze positions
        for ts, gaze_position in ts_gaze_positions.items():

            finished_gaze_movement = gaze_movement_identifier.identify(ts, gaze_position, terminate=(ts == last_ts))

            if GazeFeatures.is_fixation(finished_gaze_movement):

                self.assertEqual(len(finished_gaze_movement.positions.keys()), size)
                self.assertLessEqual(finished_gaze_movement.deviation_max, deviation_max)
                self.assertGreaterEqual(finished_gaze_movement.duration, (size-1) * min_time)
                self.assertLessEqual(finished_gaze_movement.duration, (size-1) * max_time)
                self.assertLessEqual(finished_gaze_movement.finished, True)

            elif GazeFeatures.is_saccade(finished_gaze_movement):

                self.assertEqual(len(finished_gaze_movement.positions.keys()), 2)
                self.assertGreaterEqual(finished_gaze_movement.duration, min_time)
                self.assertLessEqual(finished_gaze_movement.duration, max_time)
                self.assertLessEqual(finished_gaze_movement.finished, True)

            # Check that last gaze position date is not equal to given gaze position date
            if finished_gaze_movement.valid:

                last_ts, _ = finished_gaze_movement.positions.last

                self.assertNotEqual(last_ts, ts)

            # Check that last gaze position date of current fixation is equal to given gaze position date
            # NOTE: This is not true for saccade as, for I-DT, there is a minimal time window while the gaze movement is unknown
            current_gaze_movement = gaze_movement_identifier.current_gaze_movement
            if current_gaze_movement.valid:

                if GazeFeatures.is_fixation(current_gaze_movement):

                    last_ts, _ = current_gaze_movement.positions.last

                    self.assertEqual(last_ts, ts)

    def test_identification_generator(self):
        """Test DispersionThresholdIdentification identification using generator."""

        size = 10
        center_A = (0, 0)
        center_B = (50, 50)
        deviation_max = 10
        min_time = 0.01
        max_time = 0.1
        
        ts_gaze_positions_A = build_gaze_fixation(size, center_A, deviation_max, min_time, max_time)
        ts_gaze_positions_B = build_gaze_fixation(size, center_B, deviation_max, min_time, max_time, start_ts=ts_gaze_positions_A.last[0])

        ts_gaze_positions = ts_gaze_positions_A.append(ts_gaze_positions_B)

        gaze_movement_identifier = DispersionThresholdIdentification.GazeMovementIdentifier(deviation_max_threshold=deviation_max, duration_min_threshold=max_time*2)

        for ts, finished_gaze_movement in gaze_movement_identifier(ts_gaze_positions):

            if GazeFeatures.is_fixation(finished_gaze_movement):

                self.assertEqual(len(finished_gaze_movement.positions.keys()), size)
                self.assertLessEqual(finished_gaze_movement.deviation_max, deviation_max)
                self.assertGreaterEqual(finished_gaze_movement.duration, size * min_time)
                self.assertLessEqual(finished_gaze_movement.duration, size * max_time)
                self.assertLessEqual(finished_gaze_movement.finished, True)

            elif GazeFeatures.is_saccade(finished_gaze_movement):

                self.assertEqual(len(finished_gaze_movement.positions.keys()), 2)
                self.assertGreaterEqual(finished_gaze_movement.duration, 2 * min_time)
                self.assertLessEqual(finished_gaze_movement.duration, 2 * max_time)
                self.assertLessEqual(finished_gaze_movement.finished, True)

if __name__ == '__main__':

    unittest.main()