diff options
| -rw-r--r-- | src/argaze/DataStructures.py | 18 | ||||
| -rw-r--r-- | src/argaze/TobiiGlassesPro2/TobiiEntities.py | 12 | ||||
| -rw-r--r-- | src/argaze/utils/export_tobii_segment_fixations.py | 76 |
3 files changed, 96 insertions, 10 deletions
diff --git a/src/argaze/DataStructures.py b/src/argaze/DataStructures.py index 5648c0e..abaea62 100644 --- a/src/argaze/DataStructures.py +++ b/src/argaze/DataStructures.py @@ -6,7 +6,7 @@ import json class DictObject(): """Convert dictionnary into object""" - def __init__(self, object_type, **dictionnary): + def __init__(self, object_type = str, **dictionnary): self.__dict__.update(dictionnary) self.__type = object_type @@ -15,9 +15,9 @@ class DictObject(): return self.__dict__[key] def __str__(self): - return json.dumps({key: self.__dict__[key] for key in self.__dict__.keys()}, default=vars) + return json.dumps({key: self.__dict__[key] for key in self.__dict__.keys()}, default = vars) - def type(self): + def get_type(self): return self.__type def keys(self): @@ -51,8 +51,16 @@ class TimeStampedBuffer(collections.OrderedDict): super().__setitem__(key, value) def __str__(self): - return json.dumps(self, default=vars) + return json.dumps(self, default = vars) def pop_first(self): """Easing FIFO access mode""" - return self.popitem(last=False)
\ No newline at end of file + return self.popitem(last=False) + + def export_as_json(self, filepath): + """Write buffer content into a json file""" + try: + with open(filepath, 'w', encoding='utf-8') as jsonfile: + json.dump(self, jsonfile, ensure_ascii = False, default=vars) + except: + raise RuntimeError(f'Can\' write {filepath}')
\ No newline at end of file diff --git a/src/argaze/TobiiGlassesPro2/TobiiEntities.py b/src/argaze/TobiiGlassesPro2/TobiiEntities.py index 6df599e..e3b0d9b 100644 --- a/src/argaze/TobiiGlassesPro2/TobiiEntities.py +++ b/src/argaze/TobiiGlassesPro2/TobiiEntities.py @@ -61,11 +61,11 @@ class TobiiSegmentData(DataStructures.DictObject): data_object = DataStructures.DictObject(data_object_type, **json_item) # append a dedicated timestamped buffer for each data object type - if data_object.type() not in ts_data_buffer_dict.keys(): - ts_data_buffer_dict[data_object.type()] = DataStructures.TimeStampedBuffer() + 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.type()][ts] = data_object + ts_data_buffer_dict[data_object.get_type()][ts] = data_object # start loading with gzip.open(self.__segment_data_path) as f: @@ -106,7 +106,7 @@ class TobiiSegmentVideo(): return self.__segment_video_path def get_duration(self): - return self.__stream.duration + return float(self.__stream.duration * self.__stream.time_base) def get_frame_number(self): return self.__stream.frames @@ -130,7 +130,9 @@ class TobiiSegmentVideo(): def __next__(self): frame = self.__container.decode(self.__stream).__next__() - return frame.time * 1000000, TobiiVideoFrame(frame.to_ndarray(format='bgr24'), frame.width, frame.height, frame.pts) # use micro second + + # return micro second timestamp and frame data + return frame.time * 1000000, TobiiVideoFrame(frame.to_ndarray(format='bgr24'), frame.width, frame.height, frame.pts) class TobiiSegment: """Handle Tobii Glasses Pro 2 segment info.""" diff --git a/src/argaze/utils/export_tobii_segment_fixations.py b/src/argaze/utils/export_tobii_segment_fixations.py new file mode 100644 index 0000000..56d147c --- /dev/null +++ b/src/argaze/utils/export_tobii_segment_fixations.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python + +import argparse +import bisect +import os + +from argaze import GazeFeatures +from argaze.TobiiGlassesPro2 import TobiiEntities + + +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 millisecond') + parser.add_argument('-o', '--output', metavar='OUT', type=str, default=None, help='destination path (segment folder by default)') + args = parser.parse_args() + + if args.segment_path != None: + + # manage destination path + 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') + + # Load a tobii segment + tobii_segment = TobiiEntities.TobiiSegment(args.segment_path) + + # Load a tobii segment video + tobii_segment_video = tobii_segment.load_video() + print(f'Video duration: {tobii_segment_video.get_duration()}, frame number: {tobii_segment_video.get_frame_number()}, width: {tobii_segment_video.get_width()}, height: {tobii_segment_video.get_height()}') + + # Load a tobii segment data + tobii_segment_data = tobii_segment.load_data() + print(f'Data keys: {tobii_segment_data.keys()}') + + # Access to timestamped gaze position data buffer + tobii_ts_gaze_positions = tobii_segment_data.gidx_l_gp + + print(f'{len(tobii_ts_gaze_positions)} gaze positions loaded') + + # Access to video timestamp index + tobii_vts = tobii_segment_data.vts + print(f'{len(tobii_vts)} video timestamps loaded') + + # Format tobii gaze data into generic gaze data and store them using millisecond unit timestamp + generic_ts_gaze_positions = GazeFeatures.TimeStampedGazePositions() + + for ts, tobii_data in tobii_ts_gaze_positions.items(): + generic_data = GazeFeatures.GazePosition(tobii_data.gp[0] * tobii_segment_video.get_width(), tobii_data.gp[1] * tobii_segment_video.get_height()) + generic_ts_gaze_positions[ts/1000] = generic_data + + print(f'Dispersion threshold: {args.dispersion_threshold}') + print(f'Duration threshold: {args.duration_threshold}') + + fixation_analyser = GazeFeatures.DispersionBasedFixationIdentifier(generic_ts_gaze_positions, args.dispersion_threshold, args.duration_threshold) + + print(f'{len(fixation_analyser.fixations)} fixations found') + + # Export fixations analysis results + if args.output != None: + fixations_filepath = args.output + else: + fixations_filepath = f'{args.segment_path}/fixations.json' + + fixation_analyser.fixations.export_as_json(fixations_filepath) + print(f'Fixations saved into {fixations_filepath}') + +if __name__ == '__main__': + + main()
\ No newline at end of file |