diff options
| -rw-r--r-- | src/argaze/DataAnalysis/FixationsAndSaccades.py | 166 | ||||
| -rw-r--r-- | src/argaze/DataAnalysis/GenericData.py | 19 | ||||
| -rw-r--r-- | src/argaze/DataAnalysis/README.md | 1 | ||||
| -rw-r--r-- | src/argaze/DataAnalysis/TimeStampedData.py | 23 | ||||
| -rw-r--r-- | src/argaze/DataAnalysis/__init__.py | 5 | ||||
| -rw-r--r-- | src/argaze/TobiiGlassesPro2/_TobiiGazeData_temp.py | 333 | ||||
| -rw-r--r-- | src/argaze/utils/README.md | 6 | ||||
| -rw-r--r-- | src/argaze/utils/analyse_tobii_segment_fixations.py | 66 |
8 files changed, 619 insertions, 0 deletions
diff --git a/src/argaze/DataAnalysis/FixationsAndSaccades.py b/src/argaze/DataAnalysis/FixationsAndSaccades.py new file mode 100644 index 0000000..43bbfc0 --- /dev/null +++ b/src/argaze/DataAnalysis/FixationsAndSaccades.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python + +import math + +from argaze.DataAnalysis import TimeStampedData + +import numpy + +FIXATION_MAX_DURATION = 1000 + +class Fixation(): + + def __init__(self, start_ts, duration, dispersion, cx, cy): + + self.__start_ts = start_ts + self.__duration = duration + self.__dispersion = dispersion + self.__cx = cx + self.__cy = cy + + def get_start_time(self): + return self.__start_ts + + def get_duration(self): + return self.__duration + + def get_dispersion(self): + return self.__dispersion + + def get_centroid(self): + return (self.__cx, self.__cy) + +class FixationAnalyser(): + + _fixations = [] + _saccades = [] + + def analyse(self, ts_gaze_position): + raise NotImplementedError('analyse() method not implemented') + + def __init__(self, ts_gaze_position): + + # 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 + +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, dispersion_threshold = 10, duration_threshold = 100): + + self.__dispersion_threshold = dispersion_threshold + self.__duration_threshold = duration_threshold + + super().__init__(ts_gaze_position) + + # euclidian dispersion + def __getEuclideanDispersion(self, ts_gaze_position_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] + + 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_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] + + return (max(x_list) - min(x_list)) + (max(y_list) - min(y_list)) + + def analyse(self, ts_gaze_position): + + # while there are 2 gaze positions at least + while len(ts_gaze_position) >= 2: + + # copy remaining timestamped gaze positions + remaining_ts_gaze_position = ts_gaze_position.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_gaze_position_list = [(ts_start, gaze_position_start)] + + while (ts_current - ts_start) < self.__duration_threshold: + + ts_gaze_position_list = [(ts_current, gaze_position_current)] + + if len(remaining_ts_gaze_position) > 0: + (ts_current, gaze_position_current) = remaining_ts_gaze_position.popitem() + else: + break + + # how much gaze is dispersed ? + dispersion, cx, cy = self.__getEuclideanDispersion(ts_gaze_position_list) + + # little dispersion + if dispersion <= self.__dispersion_threshold: + + # remove selected gaze positions + for gp in ts_gaze_position_list: + ts_gaze_position.popitem() + + # are next gaze positions not too dispersed ? + while len(remaining_ts_gaze_position) > 0: + + # select next gaze position + ts_gaze_position_list.append(remaining_ts_gaze_position.popitem()) + + new_dispersion, new_cx, new_cy = self.__getEuclideanDispersion(ts_gaze_position_list) + + # dispersion too wide + if new_dispersion > self.__dispersion_threshold: + + # remove last gaze position + ts_gaze_position_list.pop(-1) + break + + # store new dispersion data + dispersion = new_dispersion + cx = new_cx + cy = new_cy + + # remove selected gaze position + ts_gaze_position.popitem() + + # we have a new fixation + ts_list = [ts for (ts, gp) in ts_gaze_position_list] + duration = ts_list[-1] - ts_list[0] + + if duration > FIXATION_MAX_DURATION: + duration = FIXATION_MAX_DURATION + + if duration > 0: + + fixation = Fixation(ts_list[0], duration, dispersion, cx, cy) + + # append fixation + self._fixations.append(fixation) + + # dispersion too wide : consider next gaze position + else: + ts_gaze_position.popitem() diff --git a/src/argaze/DataAnalysis/GenericData.py b/src/argaze/DataAnalysis/GenericData.py new file mode 100644 index 0000000..6c72470 --- /dev/null +++ b/src/argaze/DataAnalysis/GenericData.py @@ -0,0 +1,19 @@ +#!/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/README.md b/src/argaze/DataAnalysis/README.md new file mode 100644 index 0000000..b7431a9 --- /dev/null +++ b/src/argaze/DataAnalysis/README.md @@ -0,0 +1 @@ +Class interface to manage [data analysis](https://en.wikipedia.org/wiki/Eye_tracking). diff --git a/src/argaze/DataAnalysis/TimeStampedData.py b/src/argaze/DataAnalysis/TimeStampedData.py new file mode 100644 index 0000000..64f717e --- /dev/null +++ b/src/argaze/DataAnalysis/TimeStampedData.py @@ -0,0 +1,23 @@ +#!/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/__init__.py b/src/argaze/DataAnalysis/__init__.py new file mode 100644 index 0000000..1181cab --- /dev/null +++ b/src/argaze/DataAnalysis/__init__.py @@ -0,0 +1,5 @@ +""" +.. include:: README.md +""" +__docformat__ = "restructuredtext" +__all__ = ['TimeStampedData','GenericData','FixationsAndSaccades']
\ No newline at end of file diff --git a/src/argaze/TobiiGlassesPro2/_TobiiGazeData_temp.py b/src/argaze/TobiiGlassesPro2/_TobiiGazeData_temp.py new file mode 100644 index 0000000..5579948 --- /dev/null +++ b/src/argaze/TobiiGlassesPro2/_TobiiGazeData_temp.py @@ -0,0 +1,333 @@ +#!/usr/bin/env python + +import math +import os + +import numpy as np + +from sortedcontainers import SortedList, SortedDict + +from tobiiglasses.livedata import * + +import tobiiglasses.events + +def import_json_items_from_gzipfile(source_dir, filename, decode_JSON=None): + logging.info('Importing JSON items from %s in %s' % (filename, source_dir)) + with gzip.open(os.path.join(source_dir, filename)) as f: + for item in f: + json.loads(item.decode('utf-8'), object_hook=decode_JSON) + +class GazeItem: + + def __init__(self, label, dtype): + self.__label__ = label + self.__data__ = SortedDict({}) + self.__dtype__ = dtype + + def __getitem__(self, key): + return self.__data__[key] + + def __setitem__(self, key, value): + self.__data__[key] = value + + def getData(self): + return self.__data__ + + def getLabel(self): + return self.__label__ + + def getType(self): + return self.__dtype__ + + def keys(self): + return list(self.__data__.keys()) + + def pop(self, key): + self.__data__.pop(key) + + def values(self): + return list(self.__data__.values()) + +class GazeData: + + Timestamp = "Timestamp" + Gidx = "Gaze Index" + LoggedEvents = "Logged Events" + JSONEvents = "JSON Events" + GazePositionX = "Gaze Position X" + GazePositionY = "Gaze Position Y" + Gaze3DPositionX = "Gaze 3D Position X" + Gaze3DPositionY = "Gaze 3D Position Y" + Gaze3DPositionZ = "Gaze 3D Position Z" + Depth = "Gaze Depth" + Tilt = "Gaze Tilt" + Vergence = "Gaze Vergence" + Version = "Gaze Version" + GazeDirectionX_Left = "Gaze Direction Left X" + GazeDirectionY_Left = "Gaze Direction Left Y" + GazeDirectionZ_Left = "Gaze Direction Left Z" + GazeDirectionX_Right = "Gaze Direction Right X" + GazeDirectionY_Right = "Gaze Direction Right Y" + GazeDirectionZ_Right = "Gaze Direction Right Z" + GazePixelX = "Gaze Pixel X" + GazePixelY = "Gaze Pixel Y" + PupilCenterX_Left = "Pupil Center Left X" + PupilCenterY_Left = "Pupil Center Left Y" + PupilCenterZ_Left = "Pupil Center Left Z" + PupilCenterX_Right = "Pupil Center Right X" + PupilCenterY_Right = "Pupil Center Right Y" + PupilCenterZ_Right = "Pupil Center Right Z" + PupilDiameter_Left = "Pupil Diameter Left" + PupilDiameter_Right = "Pupil Diameter Right" + + + def __init__(self, segment): + + self.__gazedata__ = {} + self.__gd_right__ = {} + self.__gd_left__ = {} + self.__experiment_vars__ = {} # keys: var_name, values: {keys: ts, values: var_value} + self.__experiment_vars_headers__ = [] + + self.__rec_id__ = None + self.__pa_name__ = None + self.__pr_name__ = None + self.__segment__ = segment + + self.__FRAME_HEIGHT__ = segment.getFrameHeight() + self.__FRAME_WIDTH__ = segment.getFrameWidth() + self.__FRAME_FPS__ = segment.getFrameFPS() + self.__vts__ = SortedDict({}) + + self.__init_datatypes__() + self.__importGazeData__(segment) + self.__importExpVars__() + + def __addItem__(self, livedatajson_item): + if livedatajson_item is None: + return + if livedatajson_item.s.getValue() != 0: # Discard invalid samples + return + ts = livedatajson_item.ts.getValue()/1000.0 + + if isinstance(livedatajson_item, VTS): + self.__vts__[ts] = livedatajson_item.vts.getValue()/1000.0 + return + else: + self.__gazedata__[GazeData.Timestamp][ts] = ts # Conversion in milliseconds + + if isinstance(livedatajson_item, APISynch): + tvalue = livedatajson_item.type.getValue() + tag = livedatajson_item.tag.getValue() + tag_value = livedatajson_item.value.getValue() + if tvalue != "JsonEvent": + if tvalue.startswith('#') and tvalue.endswith('#'): + var_name = tvalue[1:-1] + if not var_name in self.__experiment_vars__.keys(): + self.__experiment_vars__[var_name] = SortedDict({}) + self.__experiment_vars__[var_name][ts] = tag + + elif tvalue.startswith('@') and tvalue.endswith('@'): + vars_list = eval(tvalue[1:-1]) + values_list = eval(tag) + for i in range(0, len(vars_list)): + var_name = vars_list[i] + value = values_list[i] + if not var_name in self.__experiment_vars__.keys(): + self.__experiment_vars__[var_name] = SortedDict({}) + self.__experiment_vars__[var_name][ts] = value + else: + self.__gazedata__[GazeData.LoggedEvents][ts] = tvalue + + else: + #self.__gazedata__[GazeData.Timestamp].pop(ts) + self.__gazedata__[GazeData.JSONEvents][ts] = tag_value + return + else: + gidx = livedatajson_item.gidx.getValue() + self.__gazedata__[GazeData.Gidx][ts] = gidx + if isinstance(livedatajson_item, GazePosition): + self.__gazedata__[GazeData.GazePositionX][ts] = livedatajson_item.gp.getValue()[0] + self.__gazedata__[GazeData.GazePositionY][ts] = livedatajson_item.gp.getValue()[1] + self.__gazedata__[GazeData.GazePixelX][ts] = int(self.__FRAME_WIDTH__*self.__gazedata__[GazeData.GazePositionX][ts]) + self.__gazedata__[GazeData.GazePixelY][ts] = int(self.__FRAME_HEIGHT__*self.__gazedata__[GazeData.GazePositionY][ts]) + return + elif isinstance(livedatajson_item, GazePosition3d): + self.__gazedata__[GazeData.Gaze3DPositionX][ts] = livedatajson_item.gp3.getValue()[0] + self.__gazedata__[GazeData.Gaze3DPositionY][ts] = livedatajson_item.gp3.getValue()[1] + self.__gazedata__[GazeData.Gaze3DPositionZ][ts] = livedatajson_item.gp3.getValue()[2] + self.__gazedata__[GazeData.Depth][ts] = (self.__gazedata__[GazeData.Gaze3DPositionX][ts]**2 + self.__gazedata__[GazeData.Gaze3DPositionY][ts]**2 + self.__gazedata__[GazeData.Gaze3DPositionZ][ts]**2)**0.5 + self.__gazedata__[GazeData.Tilt][ts] = math.atan(self.__gazedata__[GazeData.Gaze3DPositionY][ts]/self.__gazedata__[GazeData.Gaze3DPositionZ][ts])*180./math.pi + return + elif isinstance(livedatajson_item, GazeDirection): + if livedatajson_item.eye.getValue() == "left": + self.__gazedata__[GazeData.GazeDirectionX_Left][ts] = livedatajson_item.gd.getValue()[0] + self.__gazedata__[GazeData.GazeDirectionY_Left][ts] = livedatajson_item.gd.getValue()[1] + self.__gazedata__[GazeData.GazeDirectionZ_Left][ts] = livedatajson_item.gd.getValue()[2] + self.__gazedata__[GazeData.Version][ts] = math.acos(self.__gazedata__[GazeData.GazeDirectionZ_Left][ts])*180./math.pi + self.__gd_left__[gidx] = np.array([self.__gazedata__[GazeData.GazeDirectionX_Left][ts],self.__gazedata__[GazeData.GazeDirectionY_Left][ts], self.__gazedata__[GazeData.GazeDirectionZ_Left][ts]]) + return + elif livedatajson_item.eye.getValue() == "right": + self.__gazedata__[GazeData.GazeDirectionX_Right][ts] = livedatajson_item.gd.getValue()[0] + self.__gazedata__[GazeData.GazeDirectionY_Right][ts] = livedatajson_item.gd.getValue()[1] + self.__gazedata__[GazeData.GazeDirectionZ_Right][ts] = livedatajson_item.gd.getValue()[2] + self.__gazedata__[GazeData.Version][ts] = math.acos(self.__gazedata__[GazeData.GazeDirectionZ_Right][ts])*180./math.pi + self.__gd_right__[gidx] = np.array([self.__gazedata__[GazeData.GazeDirectionX_Right][ts], self.__gazedata__[GazeData.GazeDirectionY_Right][ts], self.__gazedata__[GazeData.GazeDirectionZ_Right][ts]]) + try: + self.__gazedata__[GazeData.Vergence][ts] = math.acos( np.dot(self.__gd_left__[gidx], self.__gd_right__[gidx])/(np.linalg.norm(self.__gd_left__[gidx])*np.linalg.norm(self.__gd_right__[gidx])) )*180./math.pi + except: + pass + elif isinstance(livedatajson_item, PupilCenter): + if livedatajson_item.eye.getValue() == "left": + self.__gazedata__[GazeData.PupilCenterX_Left][ts] = livedatajson_item.pc.getValue()[0] + self.__gazedata__[GazeData.PupilCenterY_Left][ts] = livedatajson_item.pc.getValue()[1] + self.__gazedata__[GazeData.PupilCenterZ_Left][ts] = livedatajson_item.pc.getValue()[2] + return + elif livedatajson_item.eye.getValue() == "right": + self.__gazedata__[GazeData.PupilCenterX_Right][ts] = livedatajson_item.pc.getValue()[0] + self.__gazedata__[GazeData.PupilCenterY_Right][ts] = livedatajson_item.pc.getValue()[1] + self.__gazedata__[GazeData.PupilCenterZ_Right][ts] = livedatajson_item.pc.getValue()[2] + return + elif isinstance(livedatajson_item, PupilDiameter): + if livedatajson_item.eye.getValue() == "left": + self.__gazedata__[GazeData.PupilDiameter_Left][ts] = livedatajson_item.pd.getValue() + return + elif livedatajson_item.eye.getValue() == "right": + self.__gazedata__[GazeData.PupilDiameter_Right][ts] = livedatajson_item.pd.getValue() + return + + def __decodeJSON__(self, json_item): + + if TobiiJSONProperties.PupilCenter.key in json_item: + item = PupilCenter(json_item) + elif TobiiJSONProperties.PupilDiameter.key in json_item: + item = PupilDiameter(json_item) + elif TobiiJSONProperties.GazeDirection.key in json_item: + item = GazeDirection(json_item) + elif TobiiJSONProperties.GazePosition.key in json_item: + item = GazePosition(json_item) + elif TobiiJSONProperties.GazePosition3d.key in json_item: + item = GazePosition3d(json_item) + elif TobiiJSONProperties.APISynch_ETS.key in json_item: + item = APISynch(json_item) + elif TobiiJSONProperties.VTS.key in json_item: + item = VTS(json_item) + else: + item = None + + self.__addItem__(item) + + def __importExpVars__(self): + self.__experiment_vars_headers__ = [] + for k,v in self.__experiment_vars__.items(): + current_var = None + self.__gazedata__[k] = GazeItem(k, np.dtype(str)) + self.__experiment_vars_headers__.append(k) + for ts in self.__gazedata__[GazeData.Timestamp].getData().values(): + try: + current_var = v[ts] + except: + pass + self.__gazedata__[k][ts] = current_var + + def __importGazeData__(self, segment): + filepath = segment.getSegmentPath() + logging.info('Importing segment %s in %s' % (segment.getId(), filepath)) + tobiiglasses.utils.import_json_items_from_gzipfile(filepath, tobiiglasses.entities.TobiiSegment.livedata_filename, self.__decodeJSON__) + + def __init_datatypes__(self): + self.__gazedata__[GazeData.Timestamp] = GazeItem(GazeData.Timestamp, np.dtype('float')) + self.__gazedata__[GazeData.Gidx] = GazeItem(GazeData.Gidx, np.dtype('u4')) + self.__gazedata__[GazeData.LoggedEvents] = GazeItem(GazeData.LoggedEvents, np.dtype(object)) + self.__gazedata__[GazeData.JSONEvents] = GazeItem(GazeData.JSONEvents, np.dtype(object)) + self.__gazedata__[GazeData.GazePositionX] = GazeItem(GazeData.GazePositionX, np.dtype('float')) + self.__gazedata__[GazeData.GazePositionY] = GazeItem(GazeData.GazePositionY, np.dtype('float')) + self.__gazedata__[GazeData.GazePixelX] = GazeItem(GazeData.GazePixelX, np.dtype('u4')) + self.__gazedata__[GazeData.GazePixelY] = GazeItem(GazeData.GazePixelY, np.dtype('u4')) + self.__gazedata__[GazeData.Gaze3DPositionX] = GazeItem(GazeData.Gaze3DPositionX, np.dtype('float')) + self.__gazedata__[GazeData.Gaze3DPositionY] = GazeItem(GazeData.Gaze3DPositionY, np.dtype('float')) + self.__gazedata__[GazeData.Gaze3DPositionZ] = GazeItem(GazeData.Gaze3DPositionZ, np.dtype('float')) + self.__gazedata__[GazeData.Depth] = GazeItem(GazeData.Depth, np.dtype('float')) + self.__gazedata__[GazeData.Vergence] = GazeItem(GazeData.Vergence, np.dtype('float')) + self.__gazedata__[GazeData.Version] = GazeItem(GazeData.Version, np.dtype('float')) + self.__gazedata__[GazeData.Tilt] = GazeItem(GazeData.Tilt, np.dtype('float')) + self.__gazedata__[GazeData.GazeDirectionX_Left] = GazeItem(GazeData.GazeDirectionX_Left, np.dtype('float')) + self.__gazedata__[GazeData.GazeDirectionY_Left] = GazeItem(GazeData.GazeDirectionY_Left, np.dtype('float')) + self.__gazedata__[GazeData.GazeDirectionZ_Left] = GazeItem(GazeData.GazeDirectionZ_Left, np.dtype('float')) + self.__gazedata__[GazeData.GazeDirectionX_Right] = GazeItem(GazeData.GazeDirectionX_Right, np.dtype('float')) + self.__gazedata__[GazeData.GazeDirectionY_Right] = GazeItem(GazeData.GazeDirectionY_Right, np.dtype('float')) + self.__gazedata__[GazeData.GazeDirectionZ_Right] = GazeItem(GazeData.GazeDirectionZ_Right, np.dtype('float')) + self.__gazedata__[GazeData.PupilCenterX_Left] = GazeItem(GazeData.PupilCenterX_Left, np.dtype('float')) + self.__gazedata__[GazeData.PupilCenterY_Left] = GazeItem(GazeData.PupilCenterY_Left, np.dtype('float')) + self.__gazedata__[GazeData.PupilCenterZ_Left] = GazeItem(GazeData.PupilCenterZ_Left, np.dtype('float')) + self.__gazedata__[GazeData.PupilCenterX_Right] = GazeItem(GazeData.PupilCenterX_Right, np.dtype('float')) + self.__gazedata__[GazeData.PupilCenterY_Right] = GazeItem(GazeData.PupilCenterY_Right, np.dtype('float')) + self.__gazedata__[GazeData.PupilCenterZ_Right] = GazeItem(GazeData.PupilCenterZ_Right, np.dtype('float')) + self.__gazedata__[GazeData.PupilDiameter_Left] = GazeItem(GazeData.PupilDiameter_Left, np.dtype('float')) + self.__gazedata__[GazeData.PupilDiameter_Right] = GazeItem(GazeData.PupilDiameter_Right, np.dtype('float')) + + def getData(self): + return self.__gazedata__.values + + def getExpVarsHeaders(self): + return self.__experiment_vars_headers__ + + def getFrameHeight(self): + return self.__FRAME_HEIGHT__ + + def getFrameWidth(self): + return self.__FRAME_WIDTH__ + + def getFrameFPS(self): + return self.__FRAME_FPS__ + + def getLoggedEvents(self): + return self.__gazedata__[GazeData.LoggedEvents].getData() + + def getJSONEvents(self): + return self.__gazedata__[GazeData.JSONEvents].getData() + + def getTimestamps(self): + return self.__gazedata__[GazeData.Timestamp].values() + + def getVTS(self): + return self.__vts__ + + def importEvents(self, events): + self.__gazedata__[tobiiglasses.events.GazeEvents.GazeType] = GazeItem(tobiiglasses.events.GazeEvents.GazeType, np.dtype(object)) + self.__gazedata__[tobiiglasses.events.GazeEvents.Fixation_X] = GazeItem(tobiiglasses.events.GazeEvents.Fixation_X, np.dtype('u4')) + self.__gazedata__[tobiiglasses.events.GazeEvents.Fixation_Y] = GazeItem(tobiiglasses.events.GazeEvents.Fixation_Y, np.dtype('u4')) + self.__gazedata__[tobiiglasses.events.GazeEvents.EventIndex] = GazeItem(tobiiglasses.events.GazeEvents.EventIndex, np.dtype('u4')) + self.__gazedata__[tobiiglasses.events.GazeEvents.EventDuration] = GazeItem(tobiiglasses.events.GazeEvents.EventDuration, np.dtype('u4')) + self.__gazedata__[tobiiglasses.events.GazeEvents.AOI_Mapped_Fixation_X] = GazeItem(tobiiglasses.events.GazeEvents.AOI_Mapped_Fixation_X, np.dtype('u4')) + self.__gazedata__[tobiiglasses.events.GazeEvents.AOI_Mapped_Fixation_Y] = GazeItem(tobiiglasses.events.GazeEvents.AOI_Mapped_Fixation_Y, np.dtype('u4')) + self.__gazedata__[tobiiglasses.events.GazeEvents.AOI] = GazeItem(tobiiglasses.events.GazeEvents.AOI, np.dtype(object)) + self.__gazedata__[tobiiglasses.events.GazeEvents.AOI_Score] = GazeItem(tobiiglasses.events.GazeEvents.AOI_Score, np.dtype('f2')) + + events = events.getEvents() + events_ts = events[GazeData.Timestamp].getData().values() + + evt_idx = 0 + evt_ts = 0 + evt_duration = 0 + for ts in self.__gazedata__[GazeData.Timestamp].getData().values(): + + if (events_ts[evt_idx] - ts) <= 10.0: + evt_ts = events_ts[evt_idx] + evt_duration = events[tobiiglasses.events.GazeEvents.EventDuration][evt_ts] + if events_ts[evt_idx] != events_ts[-1]: + evt_idx+=1 + + if ts <= (evt_ts + evt_duration): + self.__gazedata__[tobiiglasses.events.GazeEvents.GazeType][ts] = events[tobiiglasses.events.GazeEvents.GazeType][evt_ts] + self.__gazedata__[tobiiglasses.events.GazeEvents.EventIndex][ts] = events[tobiiglasses.events.GazeEvents.EventIndex][evt_ts] + self.__gazedata__[tobiiglasses.events.GazeEvents.EventDuration][ts] = evt_duration + self.__gazedata__[tobiiglasses.events.GazeEvents.Fixation_X][ts] = events[tobiiglasses.events.GazeEvents.Fixation_X][evt_ts] + self.__gazedata__[tobiiglasses.events.GazeEvents.Fixation_Y][ts] = events[tobiiglasses.events.GazeEvents.Fixation_Y][evt_ts] + try: + self.__gazedata__[tobiiglasses.events.GazeEvents.AOI][ts] = events[tobiiglasses.events.GazeEvents.AOI][evt_ts] + self.__gazedata__[tobiiglasses.events.GazeEvents.AOI_Score][ts] = events[tobiiglasses.events.GazeEvents.AOI][evt_ts] + self.__gazedata__[tobiiglasses.events.GazeEvents.AOI_Mapped_Fixation_X][ts] = events[tobiiglasses.events.GazeEvents.AOI_Mapped_Fixation_X][evt_ts] + self.__gazedata__[tobiiglasses.events.GazeEvents.AOI_Mapped_Fixation_Y][ts] = events[tobiiglasses.events.GazeEvents.AOI_Mapped_Fixation_Y][evt_ts] + except: + pass diff --git a/src/argaze/utils/README.md b/src/argaze/utils/README.md index c215916..9f6f9b9 100644 --- a/src/argaze/utils/README.md +++ b/src/argaze/utils/README.md @@ -54,6 +54,12 @@ python ./src/argaze/utils/explore_tobii_sdcard.py -r RECORDING_PATH python ./src/argaze/utils/explore_tobii_sdcard.py -s SEGMENT_PATH ``` +- Analyse Tobii segment fixations (replace SEGMENT_PATH) + +``` +python ./src/argaze/utils/analyse_tobii_segment_fixations.py -s SEGMENT_PATH +``` + - Track any 6cm ArUco marker into calibrated Tobii camera video stream (replace IP_ADDRESS). Load an roi scene (replace ROI_SCENE) .obj file, position it virtually like the detected ArUco markers and project the scene into camera frame. Then, detect if Tobii gaze point is inside any ROI. Export all collected datas into an export folder. ``` diff --git a/src/argaze/utils/analyse_tobii_segment_fixations.py b/src/argaze/utils/analyse_tobii_segment_fixations.py new file mode 100644 index 0000000..ee9b0b9 --- /dev/null +++ b/src/argaze/utils/analyse_tobii_segment_fixations.py @@ -0,0 +1,66 @@ +#!/usr/bin/env python + +import argparse + +from argaze.TobiiGlassesPro2 import TobiiEntities +from argaze.DataAnalysis import * + +def main(): + """ + Analyse Tobii segment fixations + """ + + # manage arguments + parser = argparse.ArgumentParser(description=main.__doc__.split('-')[0]) + parser.add_argument('-s', '--segment_path', metavar='SEGMENT_PATH', type=str, default=None, help='segment path') + parser.add_argument('-d', '--dispersion_threshold', metavar='DISPERSION_THRESHOLD', type=int, default=10, help='dispersion threshold in pixel') + parser.add_argument('-t', '--duration_threshold', metavar='DURATION_THRESHOLD', type=int, default=100, help='duration threshold in ms') + args = parser.parse_args() + + if args.segment_path != None: + + # Load a tobii segment + tobii_segment = TobiiEntities.TobiiSegment(args.segment_path) + + # 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 + tobii_ts_gaze_position = tobii_segment.get_data().load()['gidx-l-gp'] + + print(f'{len(tobii_ts_gaze_position)} gaze positions loaded') + + # 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) + + 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) + + print(f'{len(fixation_analyser.get_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()}') + +if __name__ == '__main__': + + main()
\ No newline at end of file |