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
|
#!/usr/bin/env python
import threading
import uuid
import gzip
import json
import time
import queue
from argaze import DataStructures
from argaze.TobiiGlassesPro2 import TobiiNetworkInterface
class TobiiDataSegment(DataStructures.DictObject):
"""Handle Tobii Glasses Pro 2 segment data file."""
def __init__(self, segment_data_path, start_timestamp = 0, end_timestamp = None):
"""Load segment data from segment directory then parse and register each recorded dataflow as a TimeStampedBuffer member of the TobiiSegmentData instance."""
self.__segment_data_path = segment_data_path
self.__first_ts = 0
ts_data_buffer_dict = {}
# define a decoder function
def decode(json_data):
# accept only valid data (e.g. with status value equal to 0)
if json_data.pop('s', -1) == 0:
# convert timestamp
ts = json_data.pop('ts')
# keep first timestamp to offset all timestamps
if self.__first_ts == 0:
self.__first_ts = ts
ts -= self.__first_ts
# ignore timestamps out of the given time range
if ts < start_timestamp or ts >= end_timestamp:
return
# convert json data into data object
data_object_type = '_'.join(json_data.keys())
data_object = DataStructures.DictObject(data_object_type, **json_data)
# append a dedicated timestamped buffer for each data object type
if data_object.get_type() not in ts_data_buffer_dict.keys():
ts_data_buffer_dict[data_object.get_type()] = DataStructures.TimeStampedBuffer()
# store data object into the timestamped buffer dedicated to its type
ts_data_buffer_dict[data_object.get_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)
super().__init__(type(self).__name__, **ts_data_buffer_dict)
def keys(self):
"""Get all registered data keys"""
return list(self.__dict__.keys())[2:-1]
def get_path(self):
return self.__segment_data_path
class TobiiDataStream(threading.Thread):
"""Capture Tobii Glasses Pro 2 data stream in separate thread."""
def __init__(self, network_interface: TobiiNetworkInterface.TobiiNetworkInterface):
"""Initialise thread super class as a deamon dedicated to data reception."""
threading.Thread.__init__(self)
threading.Thread.daemon = True
self.__network = network_interface
self.__data_socket = self.__network.make_socket()
self.__data_queue = queue.Queue()
self.__stop_event = threading.Event()
self.__read_lock = threading.Lock()
# prepare keep alive message
self.__keep_alive_msg = "{\"type\": \"live.data.unicast\", \"key\": \""+ str(uuid.uuid4()) +"\", \"op\": \"start\"}"
self.__keep_alive_thread = threading.Thread(target = self.__keep_alive)
self.__keep_alive_thread.daemon = True
def __del__(self):
"""Stop data reception before destruction."""
self.close()
def __keep_alive(self):
"""Maintain connection."""
while not self.__stop_event.isSet():
self.__network.send_keep_alive_msg(self.__data_socket, self.__keep_alive_msg)
time.sleep(1)
def open(self):
"""Start data reception."""
self.__first_ts = 0
self.__keep_alive_thread.start()
threading.Thread.start(self)
def close(self):
"""Stop data reception definitively."""
self.__stop_event.set()
threading.Thread.join(self.__keep_alive_thread)
threading.Thread.join(self)
self.__data_socket.close()
def run(self):
"""Store received data into a queue for further reading."""
while not self.__stop_event.isSet():
# lock data queue access
self.__read_lock.acquire()
# write in data queue
data = self.__network.grab_data(self.__data_socket)
json_data = json.loads(data.decode('utf-8'))
self.__data_queue.put(json_data)
# unlock data queue access
self.__read_lock.release()
def read(self):
# create a dictionary of timestamped data buffers
ts_data_buffer_dict = DataStructures.DictObject('TobiiDataStream', **{})
# if the data acquisition thread is not running
if self.__stop_event.isSet():
return ts_data_buffer_dict
# lock data queue access
self.__read_lock.acquire()
# read data queue
while not self.__data_queue.empty():
json_data = self.__data_queue.get()
# accept only valid data (e.g. with status value equal to 0)
if json_data.pop('s', -1) == 0:
# convert timestamp
ts = json_data.pop('ts')
# keep first timestamp to offset all timestamps
if self.__first_ts == 0:
self.__first_ts = ts
ts -= self.__first_ts
# ignore negative timestamp
if ts < 0:
break
# convert json data into data object
data_object_type = '_'.join(json_data.keys())
data_object = DataStructures.DictObject(data_object_type, **json_data)
# append a dedicated timestamped buffer for each data object type
if data_object.get_type() not in ts_data_buffer_dict.keys():
ts_data_buffer_dict.append(data_object.get_type(), DataStructures.TimeStampedBuffer())
# store data object into the timestamped buffer dedicated to its type
ts_data_buffer_dict[data_object.get_type()][ts] = data_object
# unlock data queue access
self.__read_lock.release()
return ts_data_buffer_dict
|
