path: root/src/argaze/GazeFeatures.py

#!/usr/bin/env python

from dataclasses import dataclass, field
import math

from argaze import DataStructures
from argaze.AreaOfInterest import AOIFeatures

import numpy
import pandas

GazePosition = tuple
"""Define gaze position as a tuple of coordinates."""

class TimeStampedGazePositions(DataStructures.TimeStampedBuffer):
    """Define timestamped buffer to store gaze positions."""

    def __setitem__(self, key, value: GazePosition):
        super().__setitem__(key, value)

@dataclass
class Movement():
    """Define abstract movement class."""

    positions: TimeStampedGazePositions
    duration: float = field(init=False)

    def __post_init__(self):

        start_position_ts, start_position = self.positions.get_first()
        end_position_ts, end_position = self.positions.get_last()

        self.duration = round(end_position_ts - start_position_ts)

Fixation = Movement
"""Define abstract fixation as movement."""

Saccade = Movement
"""Define abstract saccade as movement."""

class TimeStampedMovements(DataStructures.TimeStampedBuffer):
    """Define timestamped buffer to store movements."""

    def __setitem__(self, key, value: Movement):
        super().__setitem__(key, value)

@dataclass
class GazeStatus():
    """Define gaze status as a position belonging to an identified and indexed movement."""

    position: GazePosition
    movement_type: str
    movement_index: int

class TimeStampedGazeStatus(DataStructures.TimeStampedBuffer):
    """Define timestamped buffer to store gaze status."""

    def __setitem__(self, key, value: GazeStatus):
        super().__setitem__(key, value)

class MovementIdentifier():
    """Abstract class to define what should provide a movement identifier."""

    def __init__(self, ts_gaze_positions: TimeStampedGazePositions):

        if type(ts_gaze_positions) != TimeStampedGazePositions:
            raise ValueError('argument must be a TimeStampedGazePositions')

    def __iter__(self):
        raise NotImplementedError('__iter__() method not implemented')

    def __next__(self):
        raise NotImplementedError('__next__() method not implemented')

class DispersionBasedMovementIdentifier(MovementIdentifier):
    """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. In Proceedings of the 2000 symposium
            on Eye tracking research & applications (ETRA '00). ACM, New York, NY, USA,
            71-78. DOI=http://dx.doi.org/10.1145/355017.355028
    """

    @dataclass
    class DispersionBasedFixation(Fixation):
        """Define dispersion based fixation as an algorithm specific fixation."""

        dispersion: float = field(init=False)
        euclidian: bool = field(default=True)

        centroid: GazePosition = field(init=False)

        def __post_init__(self):

            super().__post_init__()

            x_list = [gp[0] for (ts, gp) in list(self.positions.items())]
            y_list = [gp[1] for (ts, gp) in list(self.positions.items())]

            cx = round(numpy.mean(x_list))
            cy = round(numpy.mean(y_list))

            # select dispersion algorithm
            if self.euclidian:

                c = [cx, cy]
                points = numpy.column_stack([x_list, y_list])

                dist = (points - c)**2
                dist = numpy.sum(dist, axis=1)
                dist = numpy.sqrt(dist)

                self.dispersion = round(max(dist))

            else:

                self.dispersion = (max(x_list) - min(x_list)) + (max(y_list) - min(y_list))

            self.centroid = (cx, cy)

    @dataclass
    class DispersionBasedSaccade(Saccade):
        """Define dispersion based saccade as an algorithm specific saccade."""

        def __post_init__(self):
            super().__post_init__()

    def __init__(self, ts_gaze_positions, dispersion_threshold = 10, duration_threshold = 100):

        super().__init__(ts_gaze_positions)

        self.__dispersion_threshold = dispersion_threshold
        self.__duration_threshold = duration_threshold

        # process identification on a copy
        self.__ts_gaze_positions = ts_gaze_positions.copy()

        self.__last_fixation = None

    def __iter__(self):
        """Movement identification generator."""

        # while there are 2 gaze positions at least
        while len(self.__ts_gaze_positions) >= 2:

            # copy remaining timestamped gaze positions
            remaining_ts_gaze_positions = self.__ts_gaze_positions.copy()

            # select timestamped gaze position until a duration threshold
            (ts_start, gaze_position_start) = remaining_ts_gaze_positions.pop_first()
            (ts_current, gaze_position_current) = remaining_ts_gaze_positions.pop_first()

            ts_gaze_positions = TimeStampedGazePositions()
            ts_gaze_positions[ts_start] = gaze_position_start

            while (ts_current - ts_start) < self.__duration_threshold:

                ts_gaze_positions[ts_current] = gaze_position_current

                if len(remaining_ts_gaze_positions) > 0:
                    (ts_current, gaze_position_current) = remaining_ts_gaze_positions.pop_first()
                else:
                    break

            # is it a new fixation ?
            new_fixation = DispersionBasedMovementIdentifier.DispersionBasedFixation(ts_gaze_positions)

            # dispersion is small
            if new_fixation.dispersion <= self.__dispersion_threshold:

                # remove selected gaze positions
                for gp in ts_gaze_positions:
                    self.__ts_gaze_positions.pop_first()

                # are next gaze positions not too dispersed ?
                while len(remaining_ts_gaze_positions) > 0:

                    # select next gaze position
                    ts_next, position_next = remaining_ts_gaze_positions.pop_first()
                    ts_gaze_positions[ts_next] = position_next

                    # how much gaze is dispersed ?
                    updated_fixation = DispersionBasedMovementIdentifier.DispersionBasedFixation(ts_gaze_positions)

                    # dispersion is becomes too wide : ignore updated fixation
                    if updated_fixation.dispersion > self.__dispersion_threshold:
                        break

                    # update new fixation
                    new_fixation = updated_fixation

                    # remove selected gaze position
                    self.__ts_gaze_positions.pop_first()

                # is the new fixation have a duration ?
                if new_fixation.duration > 0:

                    if self.__last_fixation != None:

                        # store start and end positions in a timestamped buffer
                        ts_saccade_positions = TimeStampedGazePositions()

                        start_position_ts, start_position = self.__last_fixation.positions.pop_last()
                        ts_saccade_positions[start_position_ts] = start_position

                        end_position_ts, end_position = new_fixation.positions.pop_first()
                        ts_saccade_positions[end_position_ts] = end_position

                        if end_position_ts > start_position_ts:

                            new_saccade = DispersionBasedMovementIdentifier.DispersionBasedSaccade(ts_saccade_positions)
                        
                            yield new_saccade

                    self.__last_fixation = new_fixation

                    yield new_fixation

            # dispersion too wide : consider next gaze position
            else:
                self.__ts_gaze_positions.pop_first()

@dataclass
class VisualScanStep():
    """Define a visual scan step as a start timestamp, duration, the name of the area of interest and where gaze looked at in each frame during the step."""

    timestamp: int
    duration: float
    area: str
    look_at: DataStructures.TimeStampedBuffer

class VisualScanGenerator():
    """Abstract class to define when an aoi starts to be looked and when it stops."""

    visual_scan_steps: list

    def __init__(self, ts_aoi_scenes: AOIFeatures.TimeStampedAOIScenes):

        if type(ts_aoi_scenes) != AOIFeatures.TimeStampedAOIScenes:
            raise ValueError('argument must be a TimeStampedAOIScenes')

        self.visual_scan_steps = []

        for step in self:

            if step == None:
                continue

            self.visual_scan_steps.append(step)

    def __iter__(self):
        raise NotImplementedError('__iter__() method not implemented')

    def steps(self):
        return self.visual_scan_steps

    def as_dataframe(self):
        """Convert buffer as pandas dataframe."""

        df = pandas.DataFrame.from_dict(self.visual_scan_steps)
        df.set_index('timestamp', inplace=True)
        df.sort_values(by=['timestamp'], inplace=True)

        return df

    def export_as_csv(self, filepath):
        """Write buffer content into a csv file."""
        try:

            self.as_dataframe().to_csv(filepath, index=True)

        except:
            raise RuntimeError(f'Can\' write {filepath}')

class PointerBasedVisualScan(VisualScanGenerator):
    """Build visual scan on the basis of which AOI are looked."""

    def __init__(self, ts_aoi_scenes: AOIFeatures.TimeStampedAOIScenes, ts_gaze_positions: TimeStampedGazePositions):

        # process identification on a copy
        self.__ts_aoi_scenes = ts_aoi_scenes.copy()
        self.__ts_gaze_positions = ts_gaze_positions.copy()

        # a dictionary to store when an aoi starts to be looked
        self.__step_dict = {}

        # build visual scan
        super().__init__(ts_aoi_scenes)

    def __iter__(self):
        """Visual scan generator function."""

        # while there is aoi scene to process
        while len(self.__ts_aoi_scenes) > 0:

            (ts_current, aoi_scene_current) = self.__ts_aoi_scenes.pop_first()

            try:

                gaze_position = self.__ts_gaze_positions[ts_current]

                for name, aoi in aoi_scene_current.items():

                    looked = aoi.looked(gaze_position)

                    if looked:

                        if not name in self.__step_dict.keys():

                            # aoi starts to be looked
                            self.__step_dict[name] = {
                                'start': ts_current,
                                'look_at': DataStructures.TimeStampedBuffer()
                            }

                        # store where the aoi is looked for 4 corners aoi
                        if len(aoi) == 4:
                            self.__step_dict[name]['look_at'][round(ts_current)] = aoi.look_at(gaze_position)

                    elif name in self.__step_dict.keys():

                        ts_start = self.__step_dict[name]['start']

                        # aoi stops to be looked
                        yield VisualScanStep(round(ts_start), round(ts_current - ts_start), name, self.__step_dict[name]['look_at'])

                        # forget the aoi
                        del self.__step_dict[name]

            # ignore missing gaze position
            except KeyError:
                pass

        # close started steps
        for name, step in self.__step_dict.items():

            ts_start = step['start']

            # aoi stops to be looked
            yield VisualScanStep(round(ts_start), round(ts_current - ts_start), name, step['look_at'])

class FixationBasedVisualScan(VisualScanGenerator):
    """Build visual scan on the basis of timestamped fixations."""

    def __init__(self, ts_aoi_scenes: AOIFeatures.TimeStampedAOIScenes, ts_fixations: TimeStampedMovements):

        super().__init__(ts_aoi_scenes)

        if type(ts_fixations) != TimeStampedMovements:
            raise ValueError('second argument must be a GazeFeatures.TimeStampedMovements')

        # process identification on a copy
        self.__ts_aoi_scenes = ts_aoi_scenes.copy()
        self.__ts_fixations = ts_fixations.copy()

    def __iter__(self):
        """Visual scan generator function."""

        yield -1, None