aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/argaze/DataAnalysis/DictObject.py20
-rw-r--r--src/argaze/DataAnalysis/GazeAnalysis.py (renamed from src/argaze/DataAnalysis/FixationsAndSaccades.py)103
-rw-r--r--src/argaze/DataAnalysis/GenericData.py19
-rw-r--r--src/argaze/DataAnalysis/TimeStampedData.py23
-rw-r--r--src/argaze/DataAnalysis/TimeStampedDataBuffer.py34
-rw-r--r--src/argaze/DataAnalysis/__init__.py2
-rw-r--r--src/argaze/TobiiGlassesPro2/TobiiEntities.py22
-rw-r--r--src/argaze/utils/analyse_tobii_segment_fixations.py39
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__':