diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/argaze/DataAnalysis/DictObject.py | 20 | ||||
| -rw-r--r-- | src/argaze/DataAnalysis/GazeAnalysis.py (renamed from src/argaze/DataAnalysis/FixationsAndSaccades.py) | 103 | ||||
| -rw-r--r-- | src/argaze/DataAnalysis/GenericData.py | 19 | ||||
| -rw-r--r-- | src/argaze/DataAnalysis/TimeStampedData.py | 23 | ||||
| -rw-r--r-- | src/argaze/DataAnalysis/TimeStampedDataBuffer.py | 34 | ||||
| -rw-r--r-- | src/argaze/DataAnalysis/__init__.py | 2 | ||||
| -rw-r--r-- | src/argaze/TobiiGlassesPro2/TobiiEntities.py | 22 | ||||
| -rw-r--r-- | src/argaze/utils/analyse_tobii_segment_fixations.py | 39 |
8 files changed, 124 insertions, 138 deletions
diff --git a/src/argaze/DataAnalysis/DictObject.py b/src/argaze/DataAnalysis/DictObject.py new file mode 100644 index 0000000..2d6aa66 --- /dev/null +++ b/src/argaze/DataAnalysis/DictObject.py @@ -0,0 +1,20 @@ +#!/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/FixationsAndSaccades.py b/src/argaze/DataAnalysis/GazeAnalysis.py index 43bbfc0..2968b2a 100644 --- a/src/argaze/DataAnalysis/FixationsAndSaccades.py +++ b/src/argaze/DataAnalysis/GazeAnalysis.py @@ -2,52 +2,38 @@ import math -from argaze.DataAnalysis import TimeStampedData +from argaze.DataAnalysis import TimeStampedDataBuffer, DictObject import numpy FIXATION_MAX_DURATION = 1000 -class Fixation(): +class GazePosition(DictObject.DictObject): + """Define gaze position data""" - def __init__(self, start_ts, duration, dispersion, cx, cy): + def __init__(self, x, y): - self.__start_ts = start_ts - self.__duration = duration - self.__dispersion = dispersion - self.__cx = cx - self.__cy = cy + super().__init__(type(self).__name__, **{'x': x, 'y': y}) - def get_start_time(self): - return self.__start_ts +class Fixation(DictObject.DictObject): + """Define fixation data""" - def get_duration(self): - return self.__duration + def __init__(self, duration, dispersion, cx, cy): - def get_dispersion(self): - return self.__dispersion - - def get_centroid(self): - return (self.__cx, self.__cy) + super().__init__(type(self).__name__, **{'duration': duration, 'dispersion': dispersion, 'centroid': [cx, cy]}) class FixationAnalyser(): - _fixations = [] - _saccades = [] + fixations = TimeStampedDataBuffer.TimeStampedDataBuffer() + saccades = TimeStampedDataBuffer.TimeStampedDataBuffer() - def analyse(self, ts_gaze_position): + def analyse(self, ts_gaze_position_buffer): raise NotImplementedError('analyse() method not implemented') - def __init__(self, ts_gaze_position): + def __init__(self, ts_gaze_position_buffer): # do analysis on a copy - self.analyse(ts_gaze_position.copy()) - - def get_fixations(self): - return self._fixations - - def get_saccades(self): - return self._saccades + self.analyse(ts_gaze_position_buffer.copy()) class DispersionBasedFixationAnalyser(FixationAnalyser): """Implementation of the I-DT algorithm as described in: @@ -58,18 +44,18 @@ class DispersionBasedFixationAnalyser(FixationAnalyser): 71-78. DOI=http://dx.doi.org/10.1145/355017.355028 """ - def __init__(self, ts_gaze_position, dispersion_threshold = 10, duration_threshold = 100): + 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) + super().__init__(ts_gaze_position_buffer) # euclidian dispersion - def __getEuclideanDispersion(self, ts_gaze_position_list): + def __getEuclideanDispersion(self, ts_gaze_position_buffer_list): - x_list = [gp.x for (ts, gp) in ts_gaze_position_list] - y_list = [gp.y for (ts, gp) in ts_gaze_position_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) @@ -84,59 +70,60 @@ class DispersionBasedFixationAnalyser(FixationAnalyser): return max(dist), cx, cy # basic dispersion - def __getDispersion(self, ts_gaze_position_list): + def __getDispersion(self, ts_gaze_position_buffer_list): - x_list = [gp.x for (ts, gp) in ts_gaze_position_list] - y_list = [gp.y for (ts, gp) in ts_gaze_position_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): + def analyse(self, ts_gaze_position_buffer): # while there are 2 gaze positions at least - while len(ts_gaze_position) >= 2: + while len(ts_gaze_position_buffer) >= 2: # copy remaining timestamped gaze positions - remaining_ts_gaze_position = ts_gaze_position.copy() + 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.popitem() - (ts_current, gaze_position_current) = remaining_ts_gaze_position.popitem() + (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_list = [(ts_start, gaze_position_start)] + ts_gaze_position_buffer_list = [(ts_start, gaze_position_start)] while (ts_current - ts_start) < self.__duration_threshold: - ts_gaze_position_list = [(ts_current, gaze_position_current)] + ts_gaze_position_buffer_list.append( (ts_current, gaze_position_current) ) - if len(remaining_ts_gaze_position) > 0: - (ts_current, gaze_position_current) = remaining_ts_gaze_position.popitem() + 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_list) + 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_list: - ts_gaze_position.popitem() + 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) > 0: + while len(remaining_ts_gaze_position_buffer) > 0: # select next gaze position - ts_gaze_position_list.append(remaining_ts_gaze_position.popitem()) + ts_gaze_position_buffer_list.append(remaining_ts_gaze_position_buffer.pop_first()) - new_dispersion, new_cx, new_cy = self.__getEuclideanDispersion(ts_gaze_position_list) + 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_list.pop(-1) + ts_gaze_position_buffer_list.pop(-1) break # store new dispersion data @@ -145,10 +132,10 @@ class DispersionBasedFixationAnalyser(FixationAnalyser): cy = new_cy # remove selected gaze position - ts_gaze_position.popitem() + ts_gaze_position_buffer.pop_first() # we have a new fixation - ts_list = [ts for (ts, gp) in ts_gaze_position_list] + ts_list = [ts for (ts, gp) in ts_gaze_position_buffer_list] duration = ts_list[-1] - ts_list[0] if duration > FIXATION_MAX_DURATION: @@ -156,11 +143,9 @@ class DispersionBasedFixationAnalyser(FixationAnalyser): if duration > 0: - fixation = Fixation(ts_list[0], duration, dispersion, cx, cy) - - # append fixation - self._fixations.append(fixation) + # 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.popitem() + ts_gaze_position_buffer.pop_first() diff --git a/src/argaze/DataAnalysis/GenericData.py b/src/argaze/DataAnalysis/GenericData.py deleted file mode 100644 index 6c72470..0000000 --- a/src/argaze/DataAnalysis/GenericData.py +++ /dev/null @@ -1,19 +0,0 @@ -#!/usr/bin/env python - -class GenericGazePosition(): - - def get_x(self): - raise NotImplementedError('get_x() method not implemented') - - def get_y(self): - raise NotImplementedError('get_y() method not implemented') - - def __getattr__(self, key): - - if key == 'x': - return self.get_x() - - if key == 'y': - return self.get_y() - - raise NameError(f'{key} is not a valid attribute of {self.__class__}')
\ No newline at end of file diff --git a/src/argaze/DataAnalysis/TimeStampedData.py b/src/argaze/DataAnalysis/TimeStampedData.py deleted file mode 100644 index 64f717e..0000000 --- a/src/argaze/DataAnalysis/TimeStampedData.py +++ /dev/null @@ -1,23 +0,0 @@ -#!/usr/bin/env python - -import collections - -class TimeStampedData(collections.OrderedDict): - """Ordered dictionary to handle timestamped data. - ``` - { - timestamp1: data1, - timestamp2: data2, - ... - } - ``` - """ - - def __new__(cls): - return super(TimeStampedData, cls).__new__(cls) - - def __init__(self): - pass - - def __del__(self): - pass
\ No newline at end of file diff --git a/src/argaze/DataAnalysis/TimeStampedDataBuffer.py b/src/argaze/DataAnalysis/TimeStampedDataBuffer.py new file mode 100644 index 0000000..23bcf06 --- /dev/null +++ b/src/argaze/DataAnalysis/TimeStampedDataBuffer.py @@ -0,0 +1,34 @@ +#!/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 index 1181cab..ff8fb6f 100644 --- a/src/argaze/DataAnalysis/__init__.py +++ b/src/argaze/DataAnalysis/__init__.py @@ -2,4 +2,4 @@ .. include:: README.md """ __docformat__ = "restructuredtext" -__all__ = ['TimeStampedData','GenericData','FixationsAndSaccades']
\ No newline at end of file +__all__ = ['TimeStampedDataBuffer','DictObject','GazeAnalysis']
\ No newline at end of file diff --git a/src/argaze/TobiiGlassesPro2/TobiiEntities.py b/src/argaze/TobiiGlassesPro2/TobiiEntities.py index a16700d..e41bbc9 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 TimeStampedData +from argaze.DataAnalysis import TimeStampedDataBuffer, DictObject import cv2 as cv @@ -39,7 +39,7 @@ class TobiiSegmentData: def load(self): - ts_data_dict = {} + ts_data_buffer_dict = {} # define a decoder function def decode(json_item): @@ -60,22 +60,24 @@ class TobiiSegmentData: if ts < 0: return - # concatenate keys to get data signature - data_signature = '-'.join(json_item.keys()) + # convert json data into data object + data_object_type = '-'.join(json_item.keys()) + data_object = DictObject.DictObject(data_object_type, **json_item) - # append a timestamped data buffer to store all data with same signature - if data_signature not in ts_data_dict.keys(): - ts_data_dict[data_signature] = TimeStampedData.TimeStampedData() + # 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() - # store data into the dedicated timestamped data buffer - ts_data_dict[data_signature][ts] = json_item + # store data object into the timestamped buffer dedicated to its type + ts_data_buffer_dict[data_object.type()][ts] = data_object # start loading with gzip.open(self.__segment_data_path) as f: + for item in f: json.loads(item.decode('utf-8'), object_hook=decode) - return ts_data_dict + return ts_data_buffer_dict class TobiiSegmentVideo: """Handle Tobii Glasses Pro 2 segment video file.""" diff --git a/src/argaze/utils/analyse_tobii_segment_fixations.py b/src/argaze/utils/analyse_tobii_segment_fixations.py index ee9b0b9..dcd38ea 100644 --- a/src/argaze/utils/analyse_tobii_segment_fixations.py +++ b/src/argaze/utils/analyse_tobii_segment_fixations.py @@ -25,41 +25,28 @@ def main(): # Load a tobii segment video tobii_segment_video = tobii_segment.get_video() print(f'Video width: {tobii_segment_video.get_width()}, height: {tobii_segment_video.get_height()}, fps: {tobii_segment_video.get_fps()}') + + # Load a tobii segment timestamped gaze position data buffer + tobii_ts_gaze_position_buffer = tobii_segment.get_data().load()['gidx-l-gp'] - # Load a tobii segment timestamped gaze position data - tobii_ts_gaze_position = tobii_segment.get_data().load()['gidx-l-gp'] + print(f'{len(tobii_ts_gaze_position_buffer)} gaze positions loaded') - print(f'{len(tobii_ts_gaze_position)} gaze positions loaded') + # format tobii gaze data into generic gaze data + generic_ts_gaze_position_buffer = TimeStampedDataBuffer.TimeStampedDataBuffer() - # Define Tobii gaze data to Generic gaze data translater - class TobiiGazePositionTranslater(GenericData.GenericGazePosition): - - def __init__(self, tobii_gaze_position, tobii_segment_video): - - self.__tobii_gaze_position = tobii_gaze_position - self.__tobii_segment_video = tobii_segment_video - - def get_x(self): - return self.__tobii_gaze_position['gp'][0] * self.__tobii_segment_video.get_width() - - def get_y(self): - return self.__tobii_gaze_position['gp'][1] * self.__tobii_segment_video.get_height() - - generic_ts_gaze_position = TimeStampedData.TimeStampedData() - - while len(tobii_ts_gaze_position): - ts, item = tobii_ts_gaze_position.popitem() - generic_ts_gaze_position[ts] = TobiiGazePositionTranslater(item, tobii_segment_video) + 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_ts_gaze_position_buffer[ts] = generic_data print(f'dispersion_threshold = {args.dispersion_threshold}') print(f'duration_threshold = {args.duration_threshold}') - fixation_analyser = FixationsAndSaccades.DispersionBasedFixationAnalyser(generic_ts_gaze_position, args.dispersion_threshold, args.duration_threshold) + fixation_analyser = GazeAnalysis.DispersionBasedFixationAnalyser(generic_ts_gaze_position_buffer, args.dispersion_threshold, args.duration_threshold) - print(f'{len(fixation_analyser.get_fixations())} fixations found') + print(f'{len(fixation_analyser.fixations)} fixations found') - for f in fixation_analyser.get_fixations(): - print(f'start time = {f.get_start_time()}, duration = {f.get_duration()}, dispertion = {f.get_dispersion()}, centroid = {f.get_centroid()}') + for ts, f in fixation_analyser.fixations.items(): + print(f'start time = {ts}, duration = {f.duration}, dispertion = {f.dispersion}, centroid = {f.centroid}') if __name__ == '__main__': |