From cb1d236be929fd9d6bbc75debacb2de827acc852 Mon Sep 17 00:00:00 2001 From: Théo de la Hogue Date: Sat, 26 Mar 2022 01:01:32 +0100 Subject: reorganizing library elements --- src/argaze/DataAnalysis/DictObject.py | 20 --- src/argaze/DataAnalysis/GazeAnalysis.py | 174 -------------------- src/argaze/DataAnalysis/README.md | 1 - src/argaze/DataAnalysis/TimeStampedDataBuffer.py | 34 ---- src/argaze/DataAnalysis/__init__.py | 5 - src/argaze/DataStructures.py | 51 ++++++ src/argaze/GazeFeatures.py | 175 +++++++++++++++++++++ src/argaze/TobiiGlassesPro2/TobiiEntities.py | 6 +- src/argaze/__init__.py | 3 +- .../utils/analyse_tobii_segment_fixations.py | 9 +- 10 files changed, 236 insertions(+), 242 deletions(-) delete mode 100644 src/argaze/DataAnalysis/DictObject.py delete mode 100644 src/argaze/DataAnalysis/GazeAnalysis.py delete mode 100644 src/argaze/DataAnalysis/README.md delete mode 100644 src/argaze/DataAnalysis/TimeStampedDataBuffer.py delete mode 100644 src/argaze/DataAnalysis/__init__.py create mode 100644 src/argaze/DataStructures.py create mode 100644 src/argaze/GazeFeatures.py diff --git a/src/argaze/DataAnalysis/DictObject.py b/src/argaze/DataAnalysis/DictObject.py deleted file mode 100644 index 2d6aa66..0000000 --- a/src/argaze/DataAnalysis/DictObject.py +++ /dev/null @@ -1,20 +0,0 @@ -#!/usr/bin/env python - -class DictObject(): - """Convert dictionnary into object""" - - def __init__(self, object_type, **dictionnary): - - self.__dict__.update(dictionnary) - self.__type = object_type - - def __getitem__(self, key): - return self.__dict__[key] - - def type(self): - return self.__type - - def keys(self): - return list(self.__dict__.keys())[:-1] - - diff --git a/src/argaze/DataAnalysis/GazeAnalysis.py b/src/argaze/DataAnalysis/GazeAnalysis.py deleted file mode 100644 index d0e4b9e..0000000 --- a/src/argaze/DataAnalysis/GazeAnalysis.py +++ /dev/null @@ -1,174 +0,0 @@ -#!/usr/bin/env python - -import math - -from argaze.DataAnalysis import TimeStampedDataBuffer, DictObject - -import numpy - -FIXATION_MAX_DURATION = 1000 - -class GazePosition(DictObject.DictObject): - """Define gaze position data""" - - def __init__(self, x, y): - - super().__init__(type(self).__name__, **{'x': x, 'y': y}) - -class TimeStampedGazePositionBuffer(TimeStampedDataBuffer.TimeStampedDataBuffer): - """Define timestamped data buffer for gaze position only""" - - def __setitem__(self, key, value: GazePosition): - """Force value to be a GazePosition""" - if type(value) != GazePosition: - raise ValueError('value must be a GazePosition') - - super().__setitem__(key, value) - -class Fixation(DictObject.DictObject): - """Define fixation data""" - - def __init__(self, duration, dispersion, cx, cy): - - super().__init__(type(self).__name__, **{'duration': duration, 'dispersion': dispersion, 'centroid': [cx, cy]}) - -class TimeStampedFixationBuffer(TimeStampedDataBuffer.TimeStampedDataBuffer): - """Define timestamped data buffer for fixation only""" - - def __setitem__(self, key, value: Fixation): - """Force value to be a Fixation""" - if type(value) != Fixation: - raise ValueError('value must be a Fixation') - - super().__setitem__(key, value) - -class FixationAnalyser(): - - fixations = TimeStampedFixationBuffer() - saccades = TimeStampedDataBuffer.TimeStampedDataBuffer() - - def analyse(self, ts_gaze_position_buffer): - raise NotImplementedError('analyse() method not implemented') - - def __init__(self, ts_gaze_position_buffer: TimeStampedGazePositionBuffer): - - if type(ts_gaze_position_buffer) != TimeStampedGazePositionBuffer: - raise ValueError('argument must be a TimeStampedGazePositionBuffer') - - # do analysis on a copy - self.analyse(ts_gaze_position_buffer.copy()) - -class DispersionBasedFixationAnalyser(FixationAnalyser): - """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 - """ - - def __init__(self, ts_gaze_position_buffer, dispersion_threshold = 10, duration_threshold = 100): - - self.__dispersion_threshold = dispersion_threshold - self.__duration_threshold = duration_threshold - - super().__init__(ts_gaze_position_buffer) - - # euclidian dispersion - def __getEuclideanDispersion(self, ts_gaze_position_buffer_list): - - x_list = [gp.x for (ts, gp) in ts_gaze_position_buffer_list] - y_list = [gp.y for (ts, gp) in ts_gaze_position_buffer_list] - - cx = numpy.mean(x_list) - cy = numpy.mean(y_list) - 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) - - return max(dist), cx, cy - - # basic dispersion - def __getDispersion(self, ts_gaze_position_buffer_list): - - x_list = [gp.x for (ts, gp) in ts_gaze_position_buffer_list] - y_list = [gp.y for (ts, gp) in ts_gaze_position_buffer_list] - - return (max(x_list) - min(x_list)) + (max(y_list) - min(y_list)) - - def analyse(self, ts_gaze_position_buffer): - - # while there are 2 gaze positions at least - while len(ts_gaze_position_buffer) >= 2: - - # copy remaining timestamped gaze positions - remaining_ts_gaze_position_buffer = ts_gaze_position_buffer.copy() - - # select timestamped gaze position until a duration threshold - (ts_start, gaze_position_start) = remaining_ts_gaze_position_buffer.pop_first() - (ts_current, gaze_position_current) = remaining_ts_gaze_position_buffer.pop_first() - - ts_gaze_position_buffer_list = [(ts_start, gaze_position_start)] - - while (ts_current - ts_start) < self.__duration_threshold: - - ts_gaze_position_buffer_list.append( (ts_current, gaze_position_current) ) - - if len(remaining_ts_gaze_position_buffer) > 0: - (ts_current, gaze_position_current) = remaining_ts_gaze_position_buffer.pop_first() - else: - break - - # how much gaze is dispersed ? - dispersion, cx, cy = self.__getEuclideanDispersion(ts_gaze_position_buffer_list) - - - # little dispersion - if dispersion <= self.__dispersion_threshold: - - # remove selected gaze positions - for gp in ts_gaze_position_buffer_list: - ts_gaze_position_buffer.pop_first() - - # are next gaze positions not too dispersed ? - while len(remaining_ts_gaze_position_buffer) > 0: - - # select next gaze position - ts_gaze_position_buffer_list.append(remaining_ts_gaze_position_buffer.pop_first()) - - new_dispersion, new_cx, new_cy = self.__getEuclideanDispersion(ts_gaze_position_buffer_list) - - # dispersion too wide - if new_dispersion > self.__dispersion_threshold: - - # remove last gaze position - ts_gaze_position_buffer_list.pop(-1) - break - - # store new dispersion data - dispersion = new_dispersion - cx = new_cx - cy = new_cy - - # remove selected gaze position - ts_gaze_position_buffer.pop_first() - - # we have a new fixation - ts_list = [ts for (ts, gp) in ts_gaze_position_buffer_list] - duration = ts_list[-1] - ts_list[0] - - if duration > FIXATION_MAX_DURATION: - duration = FIXATION_MAX_DURATION - - if duration > 0: - - # append fixation to timestamped buffer - self.fixations[ts_list[0]] = Fixation(duration, dispersion, cx, cy) - - # dispersion too wide : consider next gaze position - else: - ts_gaze_position_buffer.pop_first() diff --git a/src/argaze/DataAnalysis/README.md b/src/argaze/DataAnalysis/README.md deleted file mode 100644 index b7431a9..0000000 --- a/src/argaze/DataAnalysis/README.md +++ /dev/null @@ -1 +0,0 @@ -Class interface to manage [data analysis](https://en.wikipedia.org/wiki/Eye_tracking). diff --git a/src/argaze/DataAnalysis/TimeStampedDataBuffer.py b/src/argaze/DataAnalysis/TimeStampedDataBuffer.py deleted file mode 100644 index 23bcf06..0000000 --- a/src/argaze/DataAnalysis/TimeStampedDataBuffer.py +++ /dev/null @@ -1,34 +0,0 @@ -#!/usr/bin/env python - -import collections - -class TimeStampedDataBuffer(collections.OrderedDict): - """Ordered dictionary to handle timestamped data. - ``` - { - timestamp1: data1, - timestamp2: data2, - ... - } - ``` - """ - - def __new__(cls): - return super(TimeStampedDataBuffer, cls).__new__(cls) - - def __init__(self): - pass - - def __del__(self): - pass - - def __setitem__(self, key: float, value): - """Force key to be a float""" - if type(key) != float: - raise KeyError('key must be a float') - - super().__setitem__(key, value) - - def pop_first(self): - """Easing FIFO access mode""" - return self.popitem(last=False) \ No newline at end of file diff --git a/src/argaze/DataAnalysis/__init__.py b/src/argaze/DataAnalysis/__init__.py deleted file mode 100644 index ff8fb6f..0000000 --- a/src/argaze/DataAnalysis/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -""" -.. include:: README.md -""" -__docformat__ = "restructuredtext" -__all__ = ['TimeStampedDataBuffer','DictObject','GazeAnalysis'] \ No newline at end of file diff --git a/src/argaze/DataStructures.py b/src/argaze/DataStructures.py new file mode 100644 index 0000000..9c0414c --- /dev/null +++ b/src/argaze/DataStructures.py @@ -0,0 +1,51 @@ +#!/usr/bin/env python + +import collections + +class DictObject(): + """Convert dictionnary into object""" + + def __init__(self, object_type, **dictionnary): + + self.__dict__.update(dictionnary) + self.__type = object_type + + def __getitem__(self, key): + return self.__dict__[key] + + def type(self): + return self.__type + + def keys(self): + return list(self.__dict__.keys())[:-1] + +class TimeStampedBuffer(collections.OrderedDict): + """Ordered dictionary to handle timestamped data. + ``` + { + timestamp1: data1, + timestamp2: data2, + ... + } + ``` + """ + + def __new__(cls): + return super(TimeStampedBuffer, cls).__new__(cls) + + def __init__(self): + pass + + def __del__(self): + pass + + def __setitem__(self, key: float, value): + """Force key to be a float""" + if type(key) != float: + raise KeyError('key must be a float') + + super().__setitem__(key, value) + + def pop_first(self): + """Easing FIFO access mode""" + return self.popitem(last=False) \ No newline at end of file diff --git a/src/argaze/GazeFeatures.py b/src/argaze/GazeFeatures.py new file mode 100644 index 0000000..f223029 --- /dev/null +++ b/src/argaze/GazeFeatures.py @@ -0,0 +1,175 @@ +#!/usr/bin/env python + +import math + +from argaze import DataStructures + +import numpy + +FIXATION_MAX_DURATION = 1000 + +class GazePosition(DataStructures.DictObject): + """Define gaze position""" + + def __init__(self, x, y): + + super().__init__(type(self).__name__, **{'x': x, 'y': y}) + +class TimeStampedGazePositions(DataStructures.TimeStampedBuffer): + """Define timestamped buffer to store gaze positions""" + + def __setitem__(self, key, value: GazePosition): + """Force value to be a GazePosition""" + if type(value) != GazePosition: + raise ValueError('value must be a GazePosition') + + super().__setitem__(key, value) + +class Fixation(DataStructures.DictObject): + """Define fixation""" + + def __init__(self, duration, dispersion, cx, cy): + + super().__init__(type(self).__name__, **{'duration': duration, 'dispersion': dispersion, 'centroid': [cx, cy]}) + +class TimeStampedFixations(DataStructures.TimeStampedBuffer): + """Define timestamped buffer to store fixations""" + + def __setitem__(self, key, value: Fixation): + """Force value to be a Fixation""" + if type(value) != Fixation: + raise ValueError('value must be a Fixation') + + super().__setitem__(key, value) + +class FixationIdentifier(): + """Abstract class to define what should provide a fixation identifier""" + + fixations = TimeStampedFixations() + saccades = DataStructures.TimeStampedBuffer() + + def identify(self, ts_gaze_positions): + raise NotImplementedError('identify() method not implemented') + + def __init__(self, ts_gaze_positions: TimeStampedGazePositions): + + if type(ts_gaze_positions) != TimeStampedGazePositions: + raise ValueError('argument must be a TimeStampedGazePositions') + + # do identification on a copy + self.identify(ts_gaze_positions.copy()) + +class DispersionBasedFixationIdentifier(FixationIdentifier): + """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 + """ + + def __init__(self, ts_gaze_positions, dispersion_threshold = 10, duration_threshold = 100): + + self.__dispersion_threshold = dispersion_threshold + self.__duration_threshold = duration_threshold + + super().__init__(ts_gaze_positions) + + # euclidian dispersion + def __getEuclideanDispersion(self, ts_gaze_positions_list): + + x_list = [gp.x for (ts, gp) in ts_gaze_positions_list] + y_list = [gp.y for (ts, gp) in ts_gaze_positions_list] + + cx = numpy.mean(x_list) + cy = numpy.mean(y_list) + 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) + + return max(dist), cx, cy + + # basic dispersion + def __getDispersion(self, ts_gaze_positions_list): + + x_list = [gp.x for (ts, gp) in ts_gaze_positions_list] + y_list = [gp.y for (ts, gp) in ts_gaze_positions_list] + + return (max(x_list) - min(x_list)) + (max(y_list) - min(y_list)) + + def identify(self, ts_gaze_positions): + + # while there are 2 gaze positions at least + while len(ts_gaze_positions) >= 2: + + # copy remaining timestamped gaze positions + remaining_ts_gaze_positions = 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_list = [(ts_start, gaze_position_start)] + + while (ts_current - ts_start) < self.__duration_threshold: + + ts_gaze_positions_list.append( (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 + + # how much gaze is dispersed ? + dispersion, cx, cy = self.__getEuclideanDispersion(ts_gaze_positions_list) + + + # little dispersion + if dispersion <= self.__dispersion_threshold: + + # remove selected gaze positions + for gp in ts_gaze_positions_list: + ts_gaze_positions.pop_first() + + # are next gaze positions not too dispersed ? + while len(remaining_ts_gaze_positions) > 0: + + # select next gaze position + ts_gaze_positions_list.append(remaining_ts_gaze_positions.pop_first()) + + new_dispersion, new_cx, new_cy = self.__getEuclideanDispersion(ts_gaze_positions_list) + + # dispersion too wide + if new_dispersion > self.__dispersion_threshold: + + # remove last gaze position + ts_gaze_positions_list.pop(-1) + break + + # store new dispersion data + dispersion = new_dispersion + cx = new_cx + cy = new_cy + + # remove selected gaze position + ts_gaze_positions.pop_first() + + # we have a new fixation + ts_list = [ts for (ts, gp) in ts_gaze_positions_list] + duration = ts_list[-1] - ts_list[0] + + if duration > FIXATION_MAX_DURATION: + duration = FIXATION_MAX_DURATION + + if duration > 0: + + # append fixation to timestamped buffer + self.fixations[ts_list[0]] = Fixation(duration, dispersion, cx, cy) + + # dispersion too wide : consider next gaze position + else: + ts_gaze_positions.pop_first() diff --git a/src/argaze/TobiiGlassesPro2/TobiiEntities.py b/src/argaze/TobiiGlassesPro2/TobiiEntities.py index e41bbc9..3cfbf91 100644 --- a/src/argaze/TobiiGlassesPro2/TobiiEntities.py +++ b/src/argaze/TobiiGlassesPro2/TobiiEntities.py @@ -5,7 +5,7 @@ import json import gzip import os -from argaze.DataAnalysis import TimeStampedDataBuffer, DictObject +from argaze import * import cv2 as cv @@ -62,11 +62,11 @@ class TobiiSegmentData: # convert json data into data object data_object_type = '-'.join(json_item.keys()) - data_object = DictObject.DictObject(data_object_type, **json_item) + data_object = DataStructures.DictObject(data_object_type, **json_item) # append a dedicated timestamped buffer for each data object type if data_object.type() not in ts_data_buffer_dict.keys(): - ts_data_buffer_dict[data_object.type()] = TimeStampedDataBuffer.TimeStampedDataBuffer() + ts_data_buffer_dict[data_object.type()] = DataStructures.TimeStampedBuffer() # store data object into the timestamped buffer dedicated to its type ts_data_buffer_dict[data_object.type()][ts] = data_object diff --git a/src/argaze/__init__.py b/src/argaze/__init__.py index 0252f36..e475324 100644 --- a/src/argaze/__init__.py +++ b/src/argaze/__init__.py @@ -1,4 +1,5 @@ """ .. include:: ../../README.md """ -__docformat__ = "restructuredtext" \ No newline at end of file +__docformat__ = "restructuredtext" +__all__ = ['DataStructures', 'GazeFeatures'] \ No newline at end of file diff --git a/src/argaze/utils/analyse_tobii_segment_fixations.py b/src/argaze/utils/analyse_tobii_segment_fixations.py index a07a515..b34979e 100644 --- a/src/argaze/utils/analyse_tobii_segment_fixations.py +++ b/src/argaze/utils/analyse_tobii_segment_fixations.py @@ -2,8 +2,9 @@ import argparse +from argaze import * from argaze.TobiiGlassesPro2 import TobiiEntities -from argaze.DataAnalysis import * + def main(): """ @@ -32,16 +33,16 @@ def main(): print(f'{len(tobii_ts_gaze_position_buffer)} gaze positions loaded') # format tobii gaze data into generic gaze data - generic_ts_gaze_position_buffer = GazeAnalysis.TimeStampedGazePositionBuffer() + generic_ts_gaze_position_buffer = GazeFeatures.TimeStampedGazePositions() for ts, tobii_data in tobii_ts_gaze_position_buffer.items(): - generic_data = GazeAnalysis.GazePosition(tobii_data.gp[0] * tobii_segment_video.get_width(), tobii_data.gp[1] * tobii_segment_video.get_height()) + generic_data = GazeFeatures.GazePosition(tobii_data.gp[0] * tobii_segment_video.get_width(), tobii_data.gp[1] * tobii_segment_video.get_height()) generic_ts_gaze_position_buffer[ts] = generic_data print(f'dispersion_threshold = {args.dispersion_threshold}') print(f'duration_threshold = {args.duration_threshold}') - fixation_analyser = GazeAnalysis.DispersionBasedFixationAnalyser(generic_ts_gaze_position_buffer, args.dispersion_threshold, args.duration_threshold) + fixation_analyser = GazeFeatures.DispersionBasedFixationIdentifier(generic_ts_gaze_position_buffer, args.dispersion_threshold, args.duration_threshold) print(f'{len(fixation_analyser.fixations)} fixations found') -- cgit v1.1