diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/argaze/TobiiGlassesPro2/TobiiData.py | 114 |
1 files changed, 87 insertions, 27 deletions
diff --git a/src/argaze/TobiiGlassesPro2/TobiiData.py b/src/argaze/TobiiGlassesPro2/TobiiData.py index 95fa947..d8ddc60 100644 --- a/src/argaze/TobiiGlassesPro2/TobiiData.py +++ b/src/argaze/TobiiGlassesPro2/TobiiData.py @@ -13,6 +13,8 @@ from argaze.TobiiGlassesPro2 import TobiiNetworkInterface from argaze.utils import MiscFeatures +import numpy + @dataclass class DirSig(): """Define dir sig data (dir sig).""" @@ -50,13 +52,13 @@ class Event(): class Accelerometer(): """Define accelerometer data (ac).""" - value: tuple((float, float, float)) + value: numpy.array @dataclass class Gyroscope(): """Define gyroscope data (gy).""" - value: tuple((float, float, float)) + value: numpy.array @dataclass class PupilCenter(): @@ -286,43 +288,50 @@ class TobiiDataSegment(): def get_path(self): return self.__segment_data_path -class TobiiDataStream(threading.Thread): +class TobiiDataStream(): """Capture Tobii Glasses Pro 2 data stream in separate thread.""" reading_callback = None def __init__(self, network_interface: TobiiNetworkInterface.TobiiNetworkInterface): - """Initialise thread super class as a deamon dedicated to data reception.""" + """Initialise network connection and prepare data reception.""" - threading.Thread.__init__(self) - threading.Thread.daemon = True - + # Enable network connection self.__network = network_interface self.__data_socket = self.__network.make_socket() - self.__data_queue = queue.Queue(50) # TODO : set queue size according technical reason - - self.__stop_event = threading.Event() - self.__read_lock = threading.Lock() - - # prepare keep alive message + # Keep connection alive 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 + self.__keep_alive_thread.start() + # Data reception + self.__data_thread = None + self.__data_queue = queue.Queue(50) # TODO : set queue size according technical reason self.__json_data_parser = TobiiJsonDataParser() + # Data capture + self.__data_stream_selector = '' + self.__data_ts_buffer = DataStructures.TimeStampedBuffer() + self.__data_ts_buffer_size = 0 + def __del__(self): - """Stop data reception before destruction.""" + """Stop data reception and network connection before destruction.""" - if self.is_alive(): - - self.close() + if self.__data_thread != None: + + threading.Thread.join(self.__data_thread) + self.__data_thread = None + + threading.Thread.join(self.__keep_alive_thread) + + self.__data_socket.close() def __keep_alive(self): - """Maintain connection.""" + """Maintain network connection.""" - while not self.__stop_event.isSet(): + while True: self.__network.send_keep_alive_msg(self.__data_socket, self.__keep_alive_msg) @@ -331,21 +340,29 @@ class TobiiDataStream(threading.Thread): def open(self): """Start data reception.""" - self.__first_ts = 0 - self.__keep_alive_thread.start() - threading.Thread.start(self) + if self.__data_thread == None: + + self.__data_thread = threading.Thread(target = self.__run) + self.__data_thread.daemon = True + + self.__stop_event = threading.Event() + self.__read_lock = threading.Lock() + + self.__first_ts = 0 + + self.__data_thread.start() def close(self): """Stop data reception definitively.""" - self.__stop_event.set() + if self.__data_thread != None: - threading.Thread.join(self.__keep_alive_thread) - threading.Thread.join(self) + self.__stop_event.set() - self.__data_socket.close() + threading.Thread.join(self.__data_thread) + self.__data_thread = None - def run(self): + def __run(self): """Managed received data for sync and async reading case. - Sync: send data to callback function. - Async: store data into a locked queue for further reading.""" @@ -465,3 +482,46 @@ class TobiiDataStream(threading.Thread): data_object_type = type(data_object).__name__ return data_ts, data_object, data_object_type + + def __capture_callback(self, data_ts, data_object, data_object_type): + """Store incoming data into timestamped buffer""" + + if data_object_type == self.__data_stream_selector: + + if len(self.__data_ts_buffer.keys()) < self.__data_ts_buffer_size: + + self.__data_ts_buffer[data_ts / 1e3] = data_object + + def capture(self, data_object_type = '', buffer_size = 500): + """Capture a data stream into buffer.""" + + # Prepare for data acquisition + self.__data_stream_selector = data_object_type + self.__data_ts_buffer = DataStructures.TimeStampedBuffer() + self.__data_ts_buffer_size = buffer_size + + # Subscribe to data stream + memo_reading_callback = self.reading_callback + self.reading_callback = self.__capture_callback + + # Start data reception + self.open() + + # Share data acquisition progress + data_size = 0 + while data_size < buffer_size: + + time.sleep(0.1) + + data_size = len(self.__data_ts_buffer.keys()) + + yield data_size, self.__data_ts_buffer + + # Stop data reception + self.close() + + # Unsubscribe to tobii data stream + self.reading_callback = memo_reading_callback + + # Return captured data + return data_size, self.__data_ts_buffer |