1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
|
"""Dispersion threshold identification (I-DT) module."""
"""
This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by the Free Software
Foundation, either version 3 of the License, or (at your option) any later
version.
This program is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
You should have received a copy of the GNU General Public License along with
this program. If not, see <https://www.gnu.org/licenses/>.
"""
__author__ = "Théo de la Hogue"
__credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
import math
from argaze import GazeFeatures, DataFeatures
import cv2
import numpy
class Fixation(GazeFeatures.Fixation):
"""Define dispersion based fixation."""
def __init__(self, focus: tuple = (), deviation_max: float = math.nan, **kwargs):
super().__init__(**kwargs)
self._focus = focus
self.__deviation_max = deviation_max
@property
def deviation_max(self) -> float:
"""Fixation's maximal deviation."""
return self.__deviation_max
def is_overlapping(self, fixation: GazeFeatures.Fixation) -> bool:
"""Does a gaze position from another fixation having a deviation to this fixation centroïd smaller than maximal deviation?"""
positions_array = numpy.asarray(fixation.values())
centroid = numpy.mean(self.focus, axis=0)
deviations_array = numpy.sqrt(numpy.sum((positions_array - centroid)**2, axis=1))
return min(deviations_array) <= self.deviation_max
def draw(self, image: numpy.array, deviation_circle_color: tuple = None, duration_border_color: tuple = None,
duration_factor: float = 1., draw_positions: dict = None):
"""Draw fixation into image.
Parameters:
image: where to draw
deviation_circle_color: color of circle representing fixation's deviation
duration_border_color: color of border representing fixation's duration
duration_factor: how many pixels per duration unit
draw_positions:
"""
# Draw duration border if required
if duration_border_color is not None:
cv2.circle(image, (int(self.focus[0]), int(self.focus[1])), int(self.deviation_max), duration_border_color, int(self.duration * duration_factor))
# Draw deviation circle if required
if deviation_circle_color is not None:
cv2.circle(image, (int(self.focus[0]), int(self.focus[1])), int(self.deviation_max), deviation_circle_color, -1)
# Draw positions if required
if draw_positions is not None:
self.draw_positions(image, **draw_positions)
class Saccade(GazeFeatures.Saccade):
"""Define dispersion based saccade."""
def __init__(self, positions: GazeFeatures.TimeStampedGazePositions = None, **kwargs):
super().__init__(positions, **kwargs)
def draw(self, image: numpy.array, line_color: tuple = None, draw_positions: dict = None):
"""Draw saccade into image.
Parameters:
image: where to draw
line_color: color of line from first position to last position
"""
# Draw line if required
if line_color is not None:
start_position = self[0]
last_position = self[-1]
cv2.line(image, (int(start_position[0]), int(start_position[1])), (int(last_position[0]), int(last_position[1])), line_color, 2)
# Draw positions if required
if draw_positions is not None:
self.draw_positions(image, **draw_positions)
class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier):
"""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.*
Proceedings of the 2000 symposium on Eye tracking research & applications (ETRA'00, 71-78).
[https://doi.org/10.1145/355017.355028](https://doi.org/10.1145/355017.355028)
"""
@DataFeatures.PipelineStepInit
def __init__(self, **kwargs):
# Init GazeMovementIdentifier class
super().__init__()
self.__deviation_max_threshold = 0
self.__duration_min_threshold = 0
self.__valid_positions = GazeFeatures.TimeStampedGazePositions()
self.__centroid = ()
self.__deviations = []
self.__fixation = Fixation()
self.__saccade = Saccade()
@property
def deviation_max_threshold(self) -> int|float:
"""Maximal distance allowed to consider a gaze movement as a fixation."""
return self.__deviation_max_threshold
@deviation_max_threshold.setter
def deviation_max_threshold(self, deviation_max_threshold: int|float):
self.__deviation_max_threshold = deviation_max_threshold
@property
def duration_min_threshold(self) -> int|float:
"""Minimal duration allowed to consider a gaze movement as a fixation.
It is also used as maximal duration allowed to wait valid gaze positions."""
return self.__duration_min_threshold
@duration_min_threshold.setter
def duration_min_threshold(self, duration_min_threshold: int|float):
self.__duration_min_threshold = duration_min_threshold
@DataFeatures.PipelineStepMethod
def identify(self, gaze_position, terminate=False) -> GazeFeatures.GazeMovement:
# When too much time elapsed since the last valid gaze position
if self.__valid_positions:
elapsed_time = gaze_position.timestamp - self.__valid_positions[-1].timestamp
if elapsed_time > self.__duration_min_threshold:
try:
# Finish and return current gaze movement
return self.current_gaze_movement().finish()
finally:
# Reset valid gaze positions
self.__valid_positions = GazeFeatures.TimeStampedGazePositions()
# Reset centroid and deviations
self.__centroid = ()
self.__deviations = []
# Clear gaze movements
self.__fixation = Fixation()
self.__saccade = Saccade()
# Consider only valid gaze position
if gaze_position:
# Update centroid and deviations
if not self.__valid_positions:
self.__centroid = gaze_position
self.__deviations = []
else:
self.__centroid = self.__centroid + (gaze_position - self.__centroid) / (len(self.__valid_positions) + 1)
self.__deviations.append(gaze_position.distance(self.__centroid))
# Store valid gaze position
self.__valid_positions.append(gaze_position)
# Once the minimal duration is reached
if self.__valid_positions.duration >= self.__duration_min_threshold:
deviation_max = max(self.__deviations)
# Maximal deviation small enough
if deviation_max <= self.__deviation_max_threshold:
# Make valid gaze positions as current fixation
self.__fixation = Fixation(positions=self.__valid_positions, focus=self.__centroid, deviation_max=deviation_max)
# Is there a current saccade?
if self.__saccade:
try:
# Share first fixation position with current saccade
self.__saccade.append(self.__fixation[0])
# Finish and return the current saccade
return self.__saccade.finish()
# Clear saccade after the return
finally:
self.__saccade = Saccade()
# Maximal deviation too wide
else:
# Is there a current fixation?
if self.__fixation:
try:
# Share last fixation position with current saccade
self.__saccade.append(self.__fixation[-1])
# Clear valid positions
self.__valid_positions = GazeFeatures.TimeStampedGazePositions()
# Finish and return the current fixation
return self.__fixation.finish()
# Clear fixation after the return
finally:
self.__fixation = Fixation()
# No fixation case:
# Remove oldest valid position
old_gaze_position = self.__valid_positions.pop(0)
# Move oldest valid position into current saccade
self.__saccade.append(old_gaze_position)
# Update centroid and deviations
self.__centroid = self.__centroid - (old_gaze_position - self.__centroid) / (len(self.__valid_positions) + 1)
self.__deviations.pop(0)
# Return current gaze movement
return self.current_gaze_movement() if not terminate else self.current_gaze_movement().finish()
def current_gaze_movement(self) -> GazeFeatures.GazeMovement:
if self.__fixation:
return self.__fixation
if len(self.__saccade) > 1:
return self.__saccade
# Always return empty gaze movement at least
return GazeFeatures.GazeMovement()
def current_fixation(self) -> GazeFeatures.GazeMovement:
if self.__fixation:
return self.__fixation
# Always return empty gaze movement at least
return GazeFeatures.GazeMovement()
def current_saccade(self) -> GazeFeatures.GazeMovement:
if len(self.__saccade) > 1:
return self.__saccade
# Always return empty gaze movement at least
return GazeFeatures.GazeMovement()
|