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
|
#!/usr/bin/env python
import argparse
import os
from argaze import GazeFeatures
from argaze.GazeAnalysis import DispersionBasedGazeMovementIdentifier
from argaze.TobiiGlassesPro2 import TobiiEntities, TobiiVideo, TobiiSpecifications
from argaze.utils import MiscFeatures
import cv2 as cv
import numpy
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='path to a tobii segment folder')
parser.add_argument('-r', '--time_range', metavar=('START_TIME', 'END_TIME'), nargs=2, type=float, default=(0., None), help='start and end time (in second)')
parser.add_argument('-d', '--dispersion_threshold', metavar='DISPERSION_THRESHOLD', type=int, default=50, help='dispersion threshold in pixel')
parser.add_argument('-t', '--duration_threshold', metavar='DURATION_THRESHOLD', type=int, default=100, help='duration threshold in millisecond')
parser.add_argument('-o', '--output', metavar='OUT', type=str, default=None, help='destination folder path (segment folder by default)')
parser.add_argument('-w', '--window', metavar='DISPLAY', type=bool, default=True, help='enable window display', action=argparse.BooleanOptionalAction)
args = parser.parse_args()
if args.segment_path != None:
# Manage destination path
destination_path = '.'
if args.output != None:
if not os.path.exists(os.path.dirname(args.output)):
os.makedirs(os.path.dirname(args.output))
print(f'{os.path.dirname(args.output)} folder created')
destination_path = args.output
else:
destination_path = args.segment_path
# Export into a dedicated time range folder
if args.time_range[1] != None:
timerange_path = f'[{int(args.time_range[0])}s - {int(args.time_range[1])}s]'
else:
timerange_path = f'[all]'
destination_path = f'{destination_path}/{timerange_path}'
if not os.path.exists(destination_path):
os.makedirs(destination_path)
print(f'{destination_path} folder created')
fixations_filepath = f'{destination_path}/gaze_fixations.csv'
saccades_filepath = f'{destination_path}/gaze_saccades.csv'
gaze_status_filepath = f'{destination_path}/gaze_status.csv'
gaze_status_video_filepath = f'{destination_path}/gaze_status.mp4'
# Load a tobii segment
tobii_segment = TobiiEntities.TobiiSegment(args.segment_path, int(args.time_range[0] * 1e6), int(args.time_range[1] * 1e6) if args.time_range[1] != None else None)
# Load a tobii segment video
tobii_segment_video = tobii_segment.load_video()
print(f'Video properties:\n\tduration: {tobii_segment_video.duration/1e6} s\n\twidth: {tobii_segment_video.width} px\n\theight: {tobii_segment_video.height} px')
# Load a tobii segment data
tobii_segment_data = tobii_segment.load_data()
print(f'Loaded data count:')
for name in tobii_segment_data.keys():
print(f'\t{name}: {len(tobii_segment_data[name])} data')
# Access to timestamped gaze position data buffer
tobii_ts_gaze_positions = tobii_segment_data['GazePosition']
# Access to timestamped gaze 3D positions data buffer
tobii_ts_gaze_positions_3d = tobii_segment_data['GazePosition3D']
# Format tobii gaze position and accuracy in pixel
ts_gaze_positions = GazeFeatures.TimeStampedGazePositions()
for ts, tobii_gaze_position in tobii_ts_gaze_positions.items():
# Test gaze position validity
if tobii_gaze_position.validity == 0:
gaze_position_px = (int(tobii_gaze_position.value[0] * tobii_segment_video.width), int(tobii_gaze_position.value[1] * tobii_segment_video.height))
# Get gaze position 3D at same gaze position timestamp
tobii_gaze_position_3d = tobii_ts_gaze_positions_3d.pop(ts)
# Test gaze position 3d validity
if tobii_gaze_position_3d.validity == 0:
gaze_accuracy_mm = numpy.sin(numpy.deg2rad(TobiiSpecifications.ACCURACY)) * tobii_gaze_position_3d.value[2]
tobii_camera_hfov_mm = numpy.sin(numpy.deg2rad(TobiiSpecifications.CAMERA_HFOV)) * tobii_gaze_position_3d.value[2]
gaze_accuracy_px = round(tobii_segment_video.width * float(gaze_accuracy_mm) / float(tobii_camera_hfov_mm))
# Store gaze position
ts_gaze_positions[ts] = GazeFeatures.GazePosition(gaze_position_px, accuracy=gaze_accuracy_px)
continue
# Store unvalid gaze position for further movement processing
ts_gaze_positions[ts] = GazeFeatures.UnvalidGazePosition()
print(f'GazeMovement identifier parameters:')
print(f'\tDispersion threshold = {args.dispersion_threshold}')
print(f'\tDuration threshold = {args.duration_threshold}')
# Start movement identification
movement_identifier = DispersionBasedGazeMovementIdentifier.GazeMovementIdentifier(args.dispersion_threshold, args.duration_threshold*1e3)
ts_fixations = GazeFeatures.TimeStampedGazeMovements()
ts_saccades = GazeFeatures.TimeStampedGazeMovements()
ts_status = GazeFeatures.TimeStampedGazeStatus()
# Initialise progress bar
MiscFeatures.printProgressBar(0, tobii_segment_video.duration, prefix = 'GazeMovements identification:', suffix = 'Complete', length = 100)
for gaze_movement in movement_identifier(ts_gaze_positions):
if isinstance(gaze_movement, DispersionBasedGazeMovementIdentifier.Fixation):
start_ts, start_position = gaze_movement.positions.first
ts_fixations[start_ts] = gaze_movement
for ts, position in gaze_movement.positions.items():
ts_status[ts] = GazeFeatures.GazeStatus.from_position(position, 'Fixation', len(ts_fixations))
elif isinstance(gaze_movement, DispersionBasedGazeMovementIdentifier.Saccade):
start_ts, start_position = gaze_movement.positions.first
end_ts, end_position = gaze_movement.positions.last
ts_saccades[start_ts] = gaze_movement
ts_status[start_ts] = GazeFeatures.GazeStatus.from_position(start_position, 'Saccade', len(ts_saccades))
ts_status[end_ts] = GazeFeatures.GazeStatus.from_position(end_position, 'Saccade', len(ts_saccades))
else:
continue
# Update Progress Bar
progress = ts - int(args.time_range[0] * 1e6)
MiscFeatures.printProgressBar(progress, tobii_segment_video.duration, prefix = 'GazeMovements identification:', suffix = 'Complete', length = 100)
print(f'\n{len(ts_fixations)} fixations and {len(ts_saccades)} saccades found')
# Export fixations analysis
ts_fixations.as_dataframe().to_csv(fixations_filepath, index=True)
print(f'Fixations saved into {fixations_filepath}')
# Export saccades analysis
ts_saccades.as_dataframe().to_csv(saccades_filepath, index=True)
print(f'Saccades saved into {saccades_filepath}')
# Export gaze status analysis
ts_status.as_dataframe().to_csv(gaze_status_filepath, index=True)
print(f'Gaze status saved into {gaze_status_filepath}')
# Prepare video exportation at the same format than segment video
output_video = TobiiVideo.TobiiVideoOutput(gaze_status_video_filepath, tobii_segment_video.stream)
# Video and data loop
try:
# Initialise progress bar
MiscFeatures.printProgressBar(0, tobii_segment_video.duration, prefix = 'Video with movements processing:', suffix = 'Complete', length = 100)
current_fixation_ts, current_fixation = ts_fixations.pop_first()
current_fixation_time_counter = 0
current_saccade_ts, current_saccade = ts_saccades.pop_first()
# Iterate on video frames
for video_ts, video_frame in tobii_segment_video.frames():
# While current time belongs to the current fixation
if video_ts >= current_fixation_ts and video_ts < current_fixation_ts + current_fixation.duration:
current_fixation_time_counter += 1
# Draw current fixation
cv.circle(video_frame.matrix, (int(current_fixation.centroid[0]), int(current_fixation.centroid[1])), int(current_fixation.dispersion), (0, 255, 0), current_fixation_time_counter)
# Check next fixation
elif video_ts >= current_fixation_ts + current_fixation.duration and len(ts_fixations) > 0:
current_fixation_ts, current_fixation = ts_fixations.pop_first()
current_fixation_time_counter = 0
# While current time belongs to the current saccade
if video_ts >= current_saccade_ts and current_fixation_time_counter == 0:
start_ts, start_position = current_saccade.positions.first
end_ts, end_position = current_saccade.positions.last
# Draw saccade
cv.line(video_frame.matrix, start_position, end_position, (0, 0, 255), 2)
# Check next saccade
elif video_ts >= current_saccade_ts + current_saccade.duration and len(ts_saccades) > 0:
current_saccade_ts, current_saccade = ts_saccades.pop_first()
# Check next gaze
try:
# Get closest gaze position before video timestamp and remove all gaze positions before
_, nearest_gaze_position = ts_gaze_positions.pop_first_until(video_ts)
# Draw gaze
nearest_gaze_position.draw(video_frame.matrix)
# Wait for gaze position
except KeyError:
pass
# Write segment timing
cv.rectangle(video_frame.matrix, (0, 0), (550, 50), (63, 63, 63), -1)
cv.putText(video_frame.matrix, f'Segment time: {int(video_ts/1e3)} ms', (20, 40), cv.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 1, cv.LINE_AA)
# Write movement identification parameters
cv.rectangle(video_frame.matrix, (0, 90), (550, 150), (63, 63, 63), -1)
cv.putText(video_frame.matrix, f'Dispersion threshold: {args.dispersion_threshold} px', (20, 100), cv.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 1, cv.LINE_AA)
cv.putText(video_frame.matrix, f'Duration threshold: {args.duration_threshold} ms', (20, 140), cv.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 1, cv.LINE_AA)
if args.window:
# Close window using 'Esc' key
if cv.waitKey(1) == 27:
break
# Display video
cv.imshow(f'Segment {tobii_segment.id} movements', video_frame.matrix)
# Write video
output_video.write(video_frame.matrix)
# Update Progress Bar
progress = video_ts - int(args.time_range[0] * 1e6)
MiscFeatures.printProgressBar(progress, tobii_segment_video.duration, prefix = 'Video with movements processing:', suffix = 'Complete', length = 100)
# Exit on 'ctrl+C' interruption
except KeyboardInterrupt:
pass
# End output video file
output_video.close()
print(f'\nVideo with movements saved into {gaze_status_video_filepath}')
if __name__ == '__main__':
main()
|