diff options
26 files changed, 426 insertions, 518 deletions
diff --git a/docs/user_guide/gaze_analysis_pipeline/advanced_topics/scripting.md b/docs/user_guide/gaze_analysis_pipeline/advanced_topics/scripting.md index 19d277f..927e6d7 100644 --- a/docs/user_guide/gaze_analysis_pipeline/advanced_topics/scripting.md +++ b/docs/user_guide/gaze_analysis_pipeline/advanced_topics/scripting.md @@ -95,7 +95,7 @@ Calling [ArFrame.look](../../../argaze.md/#argaze.ArFeatures.ArFrame.look) metho # Do something with scan path analysis if ar_frame.is_analysis_available(): - for scan_path_analyzer_name, scan_path_analysis in ar_frame.analysis(): + for scan_path_analyzer_name, scan_path_analysis in ar_frame.analysis: ... @@ -107,7 +107,7 @@ Calling [ArFrame.look](../../../argaze.md/#argaze.ArFeatures.ArFrame.look) metho if ar_layer.is_analysis_available(): - for aoi_scan_path_analyzer_name, aoi_scan_path_analysis in ar_layer.analysis(): + for aoi_scan_path_analyzer_name, aoi_scan_path_analysis in ar_layer.analysis: ... diff --git a/docs/user_guide/gaze_analysis_pipeline/logging.md b/docs/user_guide/gaze_analysis_pipeline/logging.md index 001c213..9cfe152 100644 --- a/docs/user_guide/gaze_analysis_pipeline/logging.md +++ b/docs/user_guide/gaze_analysis_pipeline/logging.md @@ -13,11 +13,20 @@ Here is an extract from the JSON ArFrame configuration file where logging is ena { "name": "My FullHD screen", "size": [1920, 1080], - "observers": "my_frame_logger.py", + "observers": { + "my_frame_logger.ScanPathAnalysisLogger": { + "path": "./scan_path_metrics.csv", + "header": ["Timestamp (ms)", "Duration (ms)", "Steps number"] + }, ... "layers": { "MyLayer": { - "observers": "my_layer_logger.py", + "observers": { + "my_layer_logger.AOIScanPathAnalysisLogger": { + "path": "./aoi_scan_path_metrics.csv", + "header": ["Timestamp (ms)", "NGram counts"] + } + }, ... } } @@ -42,17 +51,12 @@ class ScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.Fi log = ( timestamp, - ar_frame.scan_path_analyzers['argaze.GazeAnalysis.Basic'].path_duration, - ar_frame.scan_path_analyzers['argaze.GazeAnalysis.Basic'].steps_number + ar_frame.analysis['argaze.GazeAnalysis.Basic'].path_duration, + ar_frame.analysis['argaze.GazeAnalysis.Basic'].steps_number ) # Write to file self.write(log) - -# Export logger as observer -__observers__ = { - "Scan path analysis logger": ScanPathAnalysisLogger(path="./scan_path_metrics.csv", header=("Timestamp (ms)", "Duration (ms)", "Steps number")) - } ``` Assuming that [ArGaze.GazeAnalysis.Basic](../../argaze.md/#argaze.GazeAnalysis.Basic) scan path analysis module is enabled for 'My FullHD screen' ArFrame, a ***scan_path_metrics.csv*** file would be created: @@ -82,16 +86,11 @@ class AOIScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures log = ( timestamp, - ar_layer.aoi_scan_path_analyzers['argaze.GazeAnalysis.NGram'].ngrams_count + ar_layer.analysis['argaze.GazeAnalysis.NGram'].ngrams_count ) # Write to file self.write(log) - -# Export logger as observer -__observers__ = { - "AOI Scan path analysis logger": AOIScanPathAnalysisLogger(path="./aoi_scan_path_metrics.csv", header=("Timestamp (ms)", "NGram counts")) - } ``` Assuming that [ArGaze.GazeAnalysis.NGram](../../argaze.md/#argaze.GazeAnalysis.NGram) AOI scan path analysis module is enabled for 'MyLayer' ArLayer, a ***aoi_scan_path_metrics.csv*** file would be created: @@ -110,9 +109,28 @@ Assuming that [ArGaze.GazeAnalysis.NGram](../../argaze.md/#argaze.GazeAnalysis.N As explained in [pipeline steps visualisation chapter](visualisation.md), it is possible to get [ArFrame.image](../../argaze.md/#argaze.ArFeatures.ArFrame.image) once timestamped gaze positions have been processed by [ArFrame.look](../../argaze.md/#argaze.ArFeatures.ArFrame.look) method. +Here is the JSON ArFrame configuration file where [ArFrame](../../argaze.md/#argaze.ArFeatures.ArFrame) observers are extended with a new my_frame_logger.VideoRecorder instance: + +```json +{ + "name": "My FullHD screen", + "size": [1920, 1080], + "observers": { + ... + "my_frame_logger.VideoRecorder": { + "path": "./video.mp4", + "width": 1920, + "height": 1080, + "fps": 15 + }, + ... +} +``` + +Here is *my_frame_logger.py* file extended with a new VideoRecorder class: + ```python -from argaze import DataFeatures -from argaze.utils import UtilsFeatures +... class VideoRecorder(DataFeatures.PipelineStepObserver, UtilsFeatures.VideoWriter): @@ -121,10 +139,6 @@ class VideoRecorder(DataFeatures.PipelineStepObserver, UtilsFeatures.VideoWriter self.write(ar_frame.image()) -# Export recorder as observer -__observers__ = { - "Video recorder": VideoRecorder(path="./video.mp4", width=1920, height=1080, fps=15) - } ``` Assuming that [ArFrame.image_parameters](../../argaze.md/#argaze.ArFeatures.ArFrame.image_parameters) are provided, ***video.mp4*** file would be created.
\ No newline at end of file diff --git a/src/argaze/ArFeatures.py b/src/argaze/ArFeatures.py index 0bf4a21..c5ee535 100644 --- a/src/argaze/ArFeatures.py +++ b/src/argaze/ArFeatures.py @@ -240,13 +240,15 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): # DEBUG print('ArLayer.aoi_scan_path_analyzers.setter type', type(analyzer)) - # Check scan path analyzer parameters type - members = getmembers(analyzer) + # Check scan path analyzer properties type + for name, item in type(analyzer).__dict__.items(): - for member in members: + if isinstance(item, property): - if '__annotations__' in member: + # DEBUG + print('ArLayer.aoi_scan_path_analyzers.setter', name, type(item)) + ''' for parameter_name, parameter_type in member[1].items(): # DEBUG @@ -263,9 +265,9 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): except KeyError: raise LoadingFailed(f'{module_path} aoi scan path analyzer loading fails because {parameter_type.__module__} scan path analyzer is missing.') - + ''' # Force scan path creation - if len(self.__aoi_scan_path_analyzers) > 0 and self.scan_path == None: + if len(self.__aoi_scan_path_analyzers) > 0 and self.aoi_scan_path == None: self.scan_path = GazeFeatures.ScanPath() @@ -292,17 +294,16 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): """Are aoi scan path analysis ready?""" return self.__aoi_scan_path_analyzed - def analysis(self) -> Iterator[Union[str, dict]]: - """Iterate over aoi scan path analysis. + @property + def analysis(self) -> dict: + """Get all aoi scan path analysis into dictionary.""" + analysis = {} - Returns - iterator: analyzer module path, analysis dictionary - """ - assert(self.__aoi_scan_path_analyzed) + for analyzer in self.__aoi_scan_path_analyzers: - for aoi_scan_path_analyzer_module_path, aoi_scan_path_analyzer in self.__aoi_scan_path_analyzers.items(): + analysis[type(analyzer)] = analyzer.analysis() - yield aoi_scan_path_analyzer_module_path, aoi_scan_path_analyzer.analysis() + return analysis def as_dict(self) -> dict: """Export ArLayer properties as dictionary.""" @@ -443,14 +444,14 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): DataFeatures.PipelineStepObject.__init__(self, **kwargs) # Init private attributes - self.__size = (0, 0) + self.__size = (1, 1) self.__provider = None self.__gaze_position_calibrator = None self.__gaze_movement_identifier = None self.__filter_in_progress_identification = True self.__scan_path = None self.__scan_path_analyzers = [] - self.__background = numpy.full((new_frame_size[1], new_frame_size[0], 3), 127).astype(numpy.uint8) + self.__background = numpy.full((1, 1, 3), 127).astype(numpy.uint8) self.__heatmap = None self.__layers = {} self.__image_parameters = DEFAULT_ARFRAME_IMAGE_PARAMETERS @@ -465,7 +466,7 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): return self.__size @size.setter - def size(self, size: tuple[int] = (1, 1)): + def size(self, size: tuple[int]): self.__size = size @property @@ -658,7 +659,7 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): return self.__image_parameters @image_parameters.setter - def image_parameters(self, image_parameters: dict = DEFAULT_ARFRAME_IMAGE_PARAMETERS): + def image_parameters(self, image_parameters: dict): self.__image_parameters = image_parameters @@ -674,17 +675,16 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): """Are scan path analysis ready?""" return self.__scan_path_analyzed - def analysis(self) -> Iterator[Union[str, dict]]: - """Get scan path analysis. + @property + def analysis(self) -> dict: + """Get all scan path analysis into dictionary.""" + analysis = {} - Returns - iterator: analyzer module path, analysis dictionary - """ - assert(self.__scan_path_analyzed) + for analyzer in self.__scan_path_analyzers: - for aoi_scan_path_analyzer_module_path, aoi_scan_path_analyzer in self.__aoi_scan_path_analyzers.items(): + analysis[type(analyzer)] = analyzer.analysis() - yield aoi_scan_path_analyzer_module_path, aoi_scan_path_analyzer.analysis() + return analysis def as_dict(self) -> dict: """Export ArFrame attributes as dictionary. @@ -896,7 +896,7 @@ class ArScene(DataFeatures.PipelineStepObject): """Initialize ArScene""" # Init parent classes - super().__init__(**kwargs) + super().__init__() # Init private attributes self.__layers = {} @@ -940,7 +940,7 @@ class ArScene(DataFeatures.PipelineStepObject): new_frame = ArFrame(working_directory = self.working_directory, name = frame_name, **frame_data) # Look for a scene layer with an AOI named like the frame - for scene_layer_name, scene_layer in new_layers.items(): + for scene_layer_name, scene_layer in self.layers.items(): try: @@ -1070,7 +1070,7 @@ class ArCamera(ArFrame): """Initialize ArCamera.""" # Init parent class - super().__init__(**kwargs) + super().__init__() # Init private attributes self.__scenes = {} diff --git a/src/argaze/ArUcoMarkers/ArUcoCamera.py b/src/argaze/ArUcoMarkers/ArUcoCamera.py index ada183d..38b3e36 100644 --- a/src/argaze/ArUcoMarkers/ArUcoCamera.py +++ b/src/argaze/ArUcoMarkers/ArUcoCamera.py @@ -46,17 +46,26 @@ class ArUcoCamera(ArFeatures.ArCamera): Define an ArCamera based on ArUco marker detection. """ - def __init__(self, aruco_detector: ArUcoDetector.ArUcoDetector, **kwargs): - """ Initialize ArUcoCamera - - Parameters: - aruco_detector: ArUco marker detector - """ + @DataFeatures.PipelineStepInit + def __init__(self, **kwargs): + """Initialize ArUcoCamera""" # Init parent class super().__init__(**kwargs) # Init private attribute + self.__aruco_detector = None + self.__image_parameters = {**ArFeatures.DEFAULT_ARFRAME_IMAGE_PARAMETERS, **DEFAULT_ARUCOCAMERA_IMAGE_PARAMETERS} + + @property + def aruco_detector(self) -> ArUcoDetector.ArUcoDetector: + """ArUco marker detector.""" + return self.__aruco_detector + + @aruco_detector.setter + @DataFeatures.PipelineStepAttributeSetter + def aruco_detector(self, aruco_detector: ArUcoDetector.ArUcoDetector): + self.__aruco_detector = aruco_detector # Check optic parameters @@ -74,71 +83,37 @@ class ArUcoCamera(ArFeatures.ArCamera): # Note: The choice of 1000 for default focal length should be discussed... self.__aruco_detector.optic_parameters = ArUcoOpticCalibrator.OpticParameters(rms=-1, dimensions=self.size, K=ArUcoOpticCalibrator.K0(focal_length=(1000., 1000.), width=self.size[0], height=self.size[1])) - # Edit pipeline step objects parent + # Edit parent if self.__aruco_detector is not None: self.__aruco_detector.parent = self - @property - def aruco_detector(self) -> ArUcoDetector.ArUcoDetector: - """Get ArUco detector object.""" - return self.__aruco_detector - - @classmethod - def from_dict(cls, aruco_camera_data: dict, working_directory: str = None) -> ArUcoCameraType: - """ - Load ArUcoCamera from dictionary. - - Parameters: - aruco_camera_data: dictionary - working_directory: folder path where to load files when a dictionary value is a relative filepath. - """ - - # Load ArUco detector - new_aruco_detector = ArUcoDetector.ArUcoDetector.from_dict(aruco_camera_data.pop('aruco_detector'), working_directory) - - # Load ArUcoScenes - new_scenes = {} + @ArFeatures.ArCamera.scenes.setter + def scenes(self, scenes: dict): - try: + self.__scenes = {} - for aruco_scene_name, aruco_scene_data in aruco_camera_data.pop('scenes').items(): + for scene_name, scene_data in scenes.items(): - # Append name - aruco_scene_data['name'] = aruco_scene_name + self.__scenes[scene_name] = ArUcoScene.ArUcoScene(working_directory = self.working_directory, name = scene_name, **scene_data) - # Create new aruco scene - new_aruco_scene = ArUcoScene.ArUcoScene.from_dict(aruco_scene_data, working_directory) + # Edit parent + for name, scene in self.__scenes.items(): - # Append new scene - new_scenes[aruco_scene_name] = new_aruco_scene - - except KeyError: + scene.parent = self - pass + @ArFeatures.ArFrame.image_parameters.setter + def image_parameters(self, image_parameters: dict): - # Set image_parameters to default if there is not - if 'image_parameters' not in aruco_camera_data.keys(): - - aruco_camera_data['image_parameters'] = {**ArFeatures.DEFAULT_ARFRAME_IMAGE_PARAMETERS, **DEFAULT_ARUCOCAMERA_IMAGE_PARAMETERS} - - # Set draw_layers to default if there is not - if 'draw_layers' not in aruco_camera_data['image_parameters'].keys(): + self.__image_parameters = image_parameters - aruco_camera_data['image_parameters']['draw_layers'] = {} + if 'draw_layers' not in self.__image_parameters: - for layer_name, layer_data in aruco_camera_data['layers'].items(): - aruco_camera_data['image_parameters']['draw_layers'][layer_name] = ArFeatures.DEFAULT_ARLAYER_DRAW_PARAMETERS + self.__image_parameters['draw_layers'] = {} - # Load temporary frame from aruco_camera_data then export it as dict - temp_frame_data = ArFeatures.ArFrame.from_dict(aruco_camera_data, working_directory).as_dict() + for layer_name in self.layers.keys(): - # Create new aruco camera using temporary ar frame values - return ArUcoCamera( \ - aruco_detector = new_aruco_detector, \ - scenes = new_scenes, \ - **temp_frame_data \ - ) + self.__image_parameters['draw_layers'][layer_name] = ArFeatures.DEFAULT_ARLAYER_DRAW_PARAMETERS @DataFeatures.PipelineStepMethod def watch(self, image: numpy.array): diff --git a/src/argaze/ArUcoMarkers/ArUcoMarkersGroup.py b/src/argaze/ArUcoMarkers/ArUcoMarkersGroup.py index 49bc7df..c57497d 100644 --- a/src/argaze/ArUcoMarkers/ArUcoMarkersGroup.py +++ b/src/argaze/ArUcoMarkers/ArUcoMarkersGroup.py @@ -79,33 +79,52 @@ class Place(): corners: numpy.array marker: dict -@dataclass class ArUcoMarkersGroup(DataFeatures.PipelineStepObject): """ Handle group of ArUco markers as one unique spatial entity and estimate its pose. """ - def __init__(self, dictionary: ArUcoMarkersDictionary.ArUcoMarkersDictionary = None, places: dict = None, **kwargs): - """Initialize ArUcoMarkersGroup + @DataFeatures.PipelineStepInit + def __init__(self, **kwargs): + """Initialize ArUcoMarkersGroup""" - Parameters: - dictionary: expected dictionary of all markers in the group. - places: expected markers place. - """ + # DEBUG + print('ArUcoMarkersGroup.__init__', kwargs.keys()) # Init parent classes - DataFeatures.PipelineStepObject.__init__(self, **kwargs) + super().__init__() # Init private attributes - self.__dictionary = dictionary - self.__places = places + self.__dictionary = None + self.__places = {} self.__translation = numpy.zeros(3) self.__rotation = numpy.zeros(3) + @property + def dictionary(self) -> ArUcoMarkersDictionary.ArUcoMarkersDictionary: + """Expected dictionary of all markers in the group.""" + return self.__dictionary + + @dictionary.setter + def dictionary(self, dictionary: ArUcoMarkersDictionary.ArUcoMarkersDictionary): + + self.__dictionary = dictionary + + @property + def places(self) -> dict: + """Expected markers place.""" + return self.__places + + @places.setter + def places(self, places: dict): + + # DEBUG + print('ArUcoMarkersGroup.places.setter', places) + # Normalize places data new_places = {} - for identifier, data in self.__places.items(): + for identifier, data in places.items(): # Convert string identifier to int value if type(identifier) == str: @@ -157,15 +176,8 @@ class ArUcoMarkersGroup(DataFeatures.PipelineStepObject): self.__places = new_places - @property - def dictionary(self) -> ArUcoMarkersDictionary.ArUcoMarkersDictionary: - """Get ArUco marker group ArUco dictionary.""" - return self.__dictionary - - @property - def places(self) -> dict: - """Get ArUco marker group places dictionary.""" - return self.__places + # DEBUG + print('ArUcoMarkersGroup.places.setter *********************') @property def identifiers(self) -> list: @@ -202,20 +214,6 @@ class ArUcoMarkersGroup(DataFeatures.PipelineStepObject): } @classmethod - def from_dict(cls, aruco_markers_group_data: dict, working_directory: str = None) -> ArUcoMarkersGroupType: - """Load ArUco markers group attributes from dictionary. - - Parameters: - aruco_markers_group_data: dictionary with attributes to load - working_directory: folder path where to load files when a dictionary value is a relative filepath. - """ - - new_dictionary = aruco_markers_group_data.pop('dictionary') - new_places = aruco_markers_group_data.pop('places') - - return ArUcoMarkersGroup(new_dictionary, new_places) - - @classmethod def from_obj(self, obj_filepath: str) -> ArUcoMarkersGroupType: """Load ArUco markers group from .obj file. diff --git a/src/argaze/ArUcoMarkers/ArUcoScene.py b/src/argaze/ArUcoMarkers/ArUcoScene.py index 3d8a558..146ffc7 100644 --- a/src/argaze/ArUcoMarkers/ArUcoScene.py +++ b/src/argaze/ArUcoMarkers/ArUcoScene.py @@ -34,29 +34,65 @@ class ArUcoScene(ArFeatures.ArScene): """ Define an ArScene based on an ArUcoMarkersGroup description. """ - - def __init__(self, aruco_markers_group: ArUcoMarkersGroup.ArUcoMarkersGroup, **kwargs): - """ Initialize ArUcoScene - Parameters: - aruco_markers_group: ArUco markers 3D scene description used to estimate scene pose from detected markers: see [estimate_pose][argaze.ArFeatures.ArScene.estimate_pose] function below. - """ + @DataFeatures.PipelineStepInit + def __init__(self, **kwargs): + """Initialize ArUcoScene""" # Init parent classes - super().__init__(**kwargs) + super().__init__() # Init private attribute - self.__aruco_markers_group = aruco_markers_group + self.__aruco_markers_group = None + + @property + def aruco_markers_group(self) -> ArUcoMarkersGroup.ArUcoMarkersGroup: + """ArUco markers 3D scene description used to estimate scene pose from detected markers: see [estimate_pose][argaze.ArFeatures.ArScene.estimate_pose] function below.""" + return self.__aruco_markers_group + + @aruco_markers_group.setter + def aruco_markers_group(self, aruco_markers_group_value: ArUcoMarkersGroup.ArUcoMarkersGroup): + + # DEBUG + print('ArUcoScene.aruco_markers_group.setter', aruco_markers_group_value) + + if isinstance(aruco_markers_group_value, ArUcoMarkersGroup.ArUcoMarkersGroup): + + new_aruco_markers_group = aruco_markers_group_value + + # str: relative path to file + elif type(aruco_markers_group_value) == str: + + filepath = os.path.join(self.working_directory, aruco_markers_group_value) + file_format = filepath.split('.')[-1] - # Edit pipeline step objects parent + # OBJ file format for 3D dimension only + if file_format == 'obj': + + # DEBUG + print('ArUcoScene.aruco_markers_group.setter OBJ', filepath) + + new_aruco_markers_group = ArUcoMarkersGroup.ArUcoMarkersGroup.from_obj(filepath) + + elif file_format == 'json': + + # DEBUG + print('ArUcoScene.aruco_markers_group.setter JSON', filepath) + + with open(filepath) as file: + + new_aruco_markers_group = ArUcoMarkersGroup.ArUcoMarkersGroup(**json.load(file)) + + # Update value + self.__aruco_markers_group = new_aruco_markers_group + + # Edit parent if self.__aruco_markers_group is not None: self.__aruco_markers_group.parent = self - @property - def aruco_markers_group(self) -> ArUcoMarkersGroup.ArUcoMarkersGroup: - """Get ArUco scene markers group object.""" - return self.__aruco_markers_group + # DEBUG + print('ArUcoScene.aruco_markers_group.setter *********************') @classmethod def from_dict(cls, aruco_scene_data: dict, working_directory: str = None) -> ArUcoSceneType: diff --git a/src/argaze/DataFeatures.py b/src/argaze/DataFeatures.py index 677a179..ef6ac27 100644 --- a/src/argaze/DataFeatures.py +++ b/src/argaze/DataFeatures.py @@ -466,12 +466,9 @@ def PipelineStepInit(method): kwargs: Any arguments defined by PipelineStepMethodInit. """ - # DEBUG - print('@PipelineStepInit', kwargs.keys()) - method(self, **kwargs) - self.update(kwargs) + self.update_attributes(kwargs) return wrapper @@ -488,8 +485,20 @@ def PipelineStepAttributeSetter(method): # Get new value type new_value_type = type(new_value) + # DEBUG + print('@PipelineStepAttributeSetter new_value_type', new_value_type) + # Check setter annotations to get expected value type - expected_value_type = method.__annotations__.popitem()[1] + try: + + expected_value_type = list(method.__annotations__.values())[0] + + except KeyError: + + raise(ValueError(f'Missing annotations in {method.__name__}: {method.__annotations__}')) + + # DEBUG + print('@PipelineStepAttributeSetter expected_value_type', expected_value_type) # Define function to load dict values def load_dict(data: dict) -> any: @@ -529,6 +538,9 @@ def PipelineStepAttributeSetter(method): return new_objects_list + # DEBUG + print('@PipelineStepAttributeSetter as expected', expected_value_type, data) + # Otherwise, data are parameters of the expected class return expected_value_type(**data) @@ -595,7 +607,7 @@ class PipelineStepObject(): # Init private attribute self.__name = None self.__working_directory = None - self.__observers = {} + self.__observers = [] self.__execution_times = {} # Parent attribute will be setup later by parent it self @@ -609,8 +621,11 @@ class PipelineStepObject(): child.__enter__() + # DEBUG + print('PipelineStepObject.__enter__ observers', self.__observers) + # Start observers - for observer_name, observer in self.__observers.items(): + for observer in self.__observers: observer.__enter__() @@ -620,7 +635,7 @@ class PipelineStepObject(): """At with statement end.""" # End observers - for observer_name, observer in self.__observers.items(): + for observer in self.__observers: observer.__exit__(exception_type, exception_value, exception_traceback) @@ -629,9 +644,12 @@ class PipelineStepObject(): child.__exit__(exception_type, exception_value, exception_traceback) - def update(self, object_data: dict): + def update_attributes(self, object_data: dict): """Update pipeline step object attributes with dictionary.""" + # DEBUG + print('PipelineStepObject.update_attributes', type(self), object_data.keys()) + for key, value in object_data.items(): setattr(self, key, value) @@ -674,47 +692,16 @@ class PipelineStepObject(): self.__parent = parent @property - def observers(self) -> dict: - """Get pipeline step object observers dictionary.""" + def observers(self) -> list: + """Pipeline step object observers list.""" return self.__observers @observers.setter - def observers(self, observers_value: dict|str): - """Set pipeline step object observers dictionary. - - Parameters: - observers_value: a dictionary or a path to a file where to load dictionary - """ + @PipelineStepAttributeSetter + def observers(self, observers: list): # Edit new observers dictionary - new_observers = {} - - # str: edit new observers dictionary from file - if type(observers_value) == str: - - filepath = os.path.join(self.working_directory, observers_value) - file_format = filepath.split('.')[-1] - - # py: load __observers__ variable from Python file - if file_format == 'py': - - observer_module_path = observers_value.split('.')[0] - - observer_module = importlib.import_module(observer_module_path) - - new_observers = observer_module.__observers__ - - # json: load dictionary from JSON file - elif file_format == 'json': - - with open(filepath) as file: - - new_observers = json.load(file) - - # Instanciate observers from dictionary - for observer_type, observer_data in new_observers.items(): - - self.__observers[observer_type] = get_class(observer_type)(**observer_data) + self.__observers = observers @property def execution_times(self): @@ -825,8 +812,8 @@ class PipelineStepObject(): if len(self.__observers): output += f'{tabs}\t{Style.BRIGHT}observers{Style.RESET_ALL}:\n' - for name, observer in self.__observers.items(): - output += f'{tabs}\t - {Fore.RED}{name}{Style.RESET_ALL}: {Fore.GREEN}{Style.BRIGHT}{observer.__class__.__module__}.{observer.__class__.__name__}{Style.RESET_ALL}\n' + for observer in self.__observers: + output += f'{tabs}\t - {Fore.GREEN}{Style.BRIGHT}{observer.__class__.__module__}.{observer.__class__.__name__}{Style.RESET_ALL}\n' for name, value in self.properties: @@ -955,7 +942,7 @@ def PipelineStepMethod(method): # Notify observers that method has been called subscription_name = f'on_{method.__name__}' - for observer_name, observer in self.observers.items(): + for observer in self.observers: # Does the observer cares about this method? if subscription_name in dir(observer): @@ -1003,12 +990,13 @@ class PipelineInputProvider(PipelineStepObject): """ Define class to ... """ + @PipelineStepInit def __init__(self, **kwargs): # DEBUG print('PipelineInputProvider.__init__') - super().__init__(**kwargs) + super().__init__() def attach(self, method): diff --git a/src/argaze/GazeAnalysis/Basic.py b/src/argaze/GazeAnalysis/Basic.py index 41f35ef..7c020d7 100644 --- a/src/argaze/GazeAnalysis/Basic.py +++ b/src/argaze/GazeAnalysis/Basic.py @@ -27,7 +27,7 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__path_duration = 0 self.__steps_number = 0 @@ -72,7 +72,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__path_duration = 0 self.__steps_number = 0 diff --git a/src/argaze/GazeAnalysis/DeviationCircleCoverage.py b/src/argaze/GazeAnalysis/DeviationCircleCoverage.py index 3a2d73f..9f0121e 100644 --- a/src/argaze/GazeAnalysis/DeviationCircleCoverage.py +++ b/src/argaze/GazeAnalysis/DeviationCircleCoverage.py @@ -31,7 +31,7 @@ class AOIMatcher(GazeFeatures.AOIMatcher): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__coverage_threshold = 0 diff --git a/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py b/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py index 82bb263..3fd9613 100644 --- a/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py +++ b/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py @@ -115,7 +115,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__deviation_max_threshold = 0 self.__duration_min_threshold = 0 diff --git a/src/argaze/GazeAnalysis/Entropy.py b/src/argaze/GazeAnalysis/Entropy.py index 58617f7..e241feb 100644 --- a/src/argaze/GazeAnalysis/Entropy.py +++ b/src/argaze/GazeAnalysis/Entropy.py @@ -35,7 +35,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__transition_matrix_analyzer = None self.__stationary_entropy = -1 diff --git a/src/argaze/GazeAnalysis/ExploreExploitRatio.py b/src/argaze/GazeAnalysis/ExploreExploitRatio.py index fc67121..14b3c9d 100644 --- a/src/argaze/GazeAnalysis/ExploreExploitRatio.py +++ b/src/argaze/GazeAnalysis/ExploreExploitRatio.py @@ -33,7 +33,7 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__short_fixation_duration_threshold = 0. self.__explore_exploit_ratio = 0. diff --git a/src/argaze/GazeAnalysis/FocusPointInside.py b/src/argaze/GazeAnalysis/FocusPointInside.py index cc7b15e..48943b2 100644 --- a/src/argaze/GazeAnalysis/FocusPointInside.py +++ b/src/argaze/GazeAnalysis/FocusPointInside.py @@ -31,7 +31,7 @@ class AOIMatcher(GazeFeatures.AOIMatcher): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__reset() @@ -41,7 +41,7 @@ class AOIMatcher(GazeFeatures.AOIMatcher): self.__matched_gaze_movement = None @DataFeatures.PipelineStepMethod - def match(self, aoi_scene, gaze_movement) -> Tuple[str, AOIFeatures.AreaOfInterest]: + def match(self, aoi_scene, gaze_movement) -> tuple[str, AOIFeatures.AreaOfInterest]: """Returns AOI containing fixation focus point.""" if GazeFeatures.is_fixation(gaze_movement): diff --git a/src/argaze/GazeAnalysis/KCoefficient.py b/src/argaze/GazeAnalysis/KCoefficient.py index 066de35..9084bd9 100644 --- a/src/argaze/GazeAnalysis/KCoefficient.py +++ b/src/argaze/GazeAnalysis/KCoefficient.py @@ -33,7 +33,7 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__K = 0 @@ -90,7 +90,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__K = 0 diff --git a/src/argaze/GazeAnalysis/LempelZivComplexity.py b/src/argaze/GazeAnalysis/LempelZivComplexity.py index cf84cc0..57a82d4 100644 --- a/src/argaze/GazeAnalysis/LempelZivComplexity.py +++ b/src/argaze/GazeAnalysis/LempelZivComplexity.py @@ -34,7 +34,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__lempel_ziv_complexity = 0 diff --git a/src/argaze/GazeAnalysis/LinearRegression.py b/src/argaze/GazeAnalysis/LinearRegression.py index 741c74b..c2f532a 100644 --- a/src/argaze/GazeAnalysis/LinearRegression.py +++ b/src/argaze/GazeAnalysis/LinearRegression.py @@ -34,7 +34,7 @@ class GazePositionCalibrator(GazeFeatures.GazePositionCalibrator): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__linear_regression = LinearRegression() self.__linear_regression.coef_ = numpy.array([[1., 0.], [0., 1.]]) diff --git a/src/argaze/GazeAnalysis/NGram.py b/src/argaze/GazeAnalysis/NGram.py index 8cf0e1d..d302f07 100644 --- a/src/argaze/GazeAnalysis/NGram.py +++ b/src/argaze/GazeAnalysis/NGram.py @@ -35,7 +35,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__n_min = 2 self.__n_max = 2 diff --git a/src/argaze/GazeAnalysis/NearestNeighborIndex.py b/src/argaze/GazeAnalysis/NearestNeighborIndex.py index b0d7312..d874352 100644 --- a/src/argaze/GazeAnalysis/NearestNeighborIndex.py +++ b/src/argaze/GazeAnalysis/NearestNeighborIndex.py @@ -34,7 +34,7 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__size = (0, 0) self.__nearest_neighbor_index = 0 diff --git a/src/argaze/GazeAnalysis/TransitionMatrix.py b/src/argaze/GazeAnalysis/TransitionMatrix.py index 25e2814..b91599b 100644 --- a/src/argaze/GazeAnalysis/TransitionMatrix.py +++ b/src/argaze/GazeAnalysis/TransitionMatrix.py @@ -34,7 +34,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__transition_matrix_probabilities = pandas.DataFrame() self.__transition_matrix_density = 0. diff --git a/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py b/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py index 471c688..f881132 100644 --- a/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py +++ b/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py @@ -47,7 +47,7 @@ class Fixation(GazeFeatures.Fixation): """Get fixation's maximal deviation.""" return self.__deviation_max - def is_overlapping(self, fixation: FixationType) -> bool: + 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()) @@ -114,7 +114,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__velocity_max_threshold = 0 self.__duration_min_threshold = 0 @@ -146,7 +146,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): self.__duration_min_threshold = duration_min_threshold @DataFeatures.PipelineStepMethod - def identify(self, gaze_position, terminate=False) -> GazeMovementType: + def identify(self, gaze_position, terminate=False) -> GazeFeatures.GazeMovement: # Ignore empty gaze position if not gaze_position: @@ -234,7 +234,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): # Always return empty gaze movement at least return GazeFeatures.GazeMovement() - def current_gaze_movement(self) -> GazeMovementType: + def current_gaze_movement(self) -> GazeFeatures.GazeMovement: # It shouldn't have a current fixation and a current saccade at the same time assert(not (self.__fixation_positions and self.__saccade_positions)) @@ -250,7 +250,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): # Always return empty gaze movement at least return GazeFeatures.GazeMovement() - def current_fixation(self) -> FixationType: + def current_fixation(self) -> GazeFeatures.Fixation: if self.__fixation_positions: @@ -259,7 +259,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): # Always return empty gaze movement at least return GazeFeatures.GazeMovement() - def current_saccade(self) -> SaccadeType: + def current_saccade(self) -> GazeFeatures.Saccade: if len(self.__saccade_positions) > 1: diff --git a/src/argaze/GazeFeatures.py b/src/argaze/GazeFeatures.py index 677e9bb..a9f9c7f 100644 --- a/src/argaze/GazeFeatures.py +++ b/src/argaze/GazeFeatures.py @@ -303,7 +303,7 @@ class GazePositionCalibrator(DataFeatures.PipelineStepObject): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() def store(self, observed_gaze_position: GazePosition, expected_gaze_position: GazePosition): """Store observed and expected gaze positions. @@ -569,7 +569,7 @@ class GazeMovementIdentifier(DataFeatures.PipelineStepObject): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() @DataFeatures.PipelineStepMethod def identify(self, timestamped_gaze_position: GazePosition, terminate:bool=False) -> GazeMovementType: @@ -845,7 +845,7 @@ class ScanPathAnalyzer(DataFeatures.PipelineStepObject): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__properties = [name for (name, value) in self.__class__.__dict__.items() if isinstance(value, property)] @@ -872,7 +872,7 @@ class AOIMatcher(DataFeatures.PipelineStepObject): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__exclude = [] @@ -1209,7 +1209,7 @@ class AOIScanPathAnalyzer(DataFeatures.PipelineStepObject): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__(**kwargs) + super().__init__() self.__properties = [name for (name, value) in self.__class__.__dict__.items() if isinstance(value, property)] diff --git a/src/argaze/utils/demo_data/demo_aruco_markers_setup.json b/src/argaze/utils/demo_data/demo_aruco_markers_setup.json index 5e2d722..7c0dd2f 100644 --- a/src/argaze/utils/demo_data/demo_aruco_markers_setup.json +++ b/src/argaze/utils/demo_data/demo_aruco_markers_setup.json @@ -93,9 +93,14 @@ "n_min": 3, "n_max": 3 }, - "Entropy":{} + "argaze.GazeAnalysis.Entropy.AOIScanPathAnalyzer":{} }, - "observers": "demo_layer_logger.py" + "observers": { + "demo_loggers.AOIScanPathAnalysisLogger": { + "path": "_export/logs/aoi_scan_path_metrics.csv", + "header": ["Timestamp (ms)", "Duration (ms)", "Step", "K", "LZC"] + } + } } }, "image_parameters": { @@ -138,7 +143,22 @@ "size": 2 } }, - "observers": "demo_frame_logger.py" + "observers": { + "demo_loggers.FixationLogger": { + "path": "_export/logs/fixations.csv", + "header": ["Timestamp (ms)", "Focus (px)", "Duration (ms)", "AOI"] + }, + "demo_loggers.ScanPathAnalysisLogger": { + "path": "_export/logs/scan_path_metrics.csv", + "header": ["Timestamp (ms)", "Duration (ms)", "Step", "K", "NNI", "XXR"] + }, + "demo_loggers.VideoRecorder": { + "path": "_export/logs/video.mp4", + "width": 1920, + "height": 1080, + "fps": 15 + } + } } }, "angle_tolerance": 15.0, diff --git a/src/argaze/utils/demo_data/demo_frame_logger.py b/src/argaze/utils/demo_data/demo_frame_logger.py deleted file mode 100644 index e40fa3d..0000000 --- a/src/argaze/utils/demo_data/demo_frame_logger.py +++ /dev/null @@ -1,70 +0,0 @@ -""" - -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 <http://www.gnu.org/licenses/>. -""" - -__author__ = "Théo de la Hogue" -__credits__ = [] -__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)" -__license__ = "GPLv3" - -from argaze import DataFeatures, GazeFeatures -from argaze.utils import UtilsFeatures - -class FixationLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter): - - def on_look(self, timestamp, frame, exception): - """Log fixations.""" - - # Log fixations - if GazeFeatures.is_fixation(frame.last_gaze_movement()) and frame.last_gaze_movement().is_finished(): - - log = ( - timestamp, - frame.last_gaze_movement().focus, - frame.last_gaze_movement().duration, - frame.layers['demo_layer'].last_looked_aoi_name() - ) - - self.write(log) - -class ScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter): - - def on_look(self, timestamp, frame, exception): - """Log scan path metrics.""" - - if frame.is_analysis_available(): - - log = ( - timestamp, - frame.scan_path_analyzers['argaze.GazeAnalysis.Basic'].path_duration, - frame.scan_path_analyzers['argaze.GazeAnalysis.Basic'].steps_number, - frame.scan_path_analyzers['argaze.GazeAnalysis.KCoefficient'].K, - frame.scan_path_analyzers['argaze.GazeAnalysis.NearestNeighborIndex'].nearest_neighbor_index, - frame.scan_path_analyzers['argaze.GazeAnalysis.ExploreExploitRatio'].explore_exploit_ratio - ) - - self.write(log) - -class VideoRecorder(DataFeatures.PipelineStepObserver, UtilsFeatures.VideoWriter): - - def on_look(self, timestamp, frame, exception): - """Write frame image.""" - - self.write(frame.image()) - -# Export loggers instances to register them as pipeline step object observers -__observers__ = { - "Fixation logger": FixationLogger(path="_export/logs/fixations.csv", header=("Timestamp (ms)", "Focus (px)", "Duration (ms)", "AOI")), - "Scan path analysis logger": ScanPathAnalysisLogger(path="_export/logs/scan_path_metrics.csv", header=("Timestamp (ms)", "Duration (ms)", "Step", "K", "NNI", "XXR")), - "Video recorder": VideoRecorder(path="_export/logs/video.mp4", width=1920, height=1080, fps=15) - } - diff --git a/src/argaze/utils/demo_data/demo_gaze_analysis_setup.json b/src/argaze/utils/demo_data/demo_gaze_analysis_setup.json index 1f6a5f5..a83ecd3 100644 --- a/src/argaze/utils/demo_data/demo_gaze_analysis_setup.json +++ b/src/argaze/utils/demo_data/demo_gaze_analysis_setup.json @@ -3,7 +3,7 @@ "size": [1920, 1149], "background": "frame_background.jpg", "gaze_movement_identifier": { - "DispersionThresholdIdentification": { + "argaze.GazeAnalysis.DispersionThresholdIdentification.GazeMovementIdentifier": { "deviation_max_threshold": 50, "duration_min_threshold": 200 } @@ -13,12 +13,12 @@ "duration_max": 10000 }, "scan_path_analyzers": { - "Basic": {}, - "KCoefficient": {}, - "NearestNeighborIndex": { + "argaze.GazeAnalysis.Basic.ScanPathAnalyzer": {}, + "argaze.GazeAnalysis.KCoefficient.ScanPathAnalyzer": {}, + "argaze.GazeAnalysis.NearestNeighborIndex.ScanPathAnalyzer": { "size": [1920, 1149] }, - "ExploreExploitRatio": { + "argaze.GazeAnalysis.ExploreExploitRatio.ScanPathAnalyzer": { "short_fixation_duration_threshold": 0 } }, @@ -29,7 +29,7 @@ "demo_layer": { "aoi_scene": "aoi_2d_scene.json", "aoi_matcher": { - "DeviationCircleCoverage": { + "argaze.GazeAnalysis.DeviationCircleCoverage.AOIMatcher": { "coverage_threshold": 0.5 } }, @@ -37,17 +37,32 @@ "duration_max": 30000 }, "aoi_scan_path_analyzers": { - "Basic": {}, - "TransitionMatrix": {}, - "KCoefficient": {}, - "LempelZivComplexity": {}, - "NGram": { + "argaze.GazeAnalysis.Basic.AOIScanPathAnalyzer": {}, + "argaze.GazeAnalysis.TransitionMatrix.AOIScanPathAnalyzer": {}, + "argaze.GazeAnalysis.KCoefficient.AOIScanPathAnalyzer": {}, + "argaze.GazeAnalysis.LempelZivComplexity.AOIScanPathAnalyzer": {}, + "argaze.GazeAnalysis.NGram.AOIScanPathAnalyzer": { "n_min": 3, "n_max": 3 }, - "Entropy":{} + "argaze.GazeAnalysis.Entropy.AOIScanPathAnalyzer":{} }, - "observers": "demo_layer_logger.py" + "observers": { + "demo_loggers.FixationLogger": { + "path": "_export/logs/fixations.csv", + "header": ["Timestamp (ms)", "Focus (px)", "Duration (ms)", "AOI"] + }, + "demo_loggers.ScanPathAnalysisLogger": { + "path": "_export/logs/scan_path_metrics.csv", + "header": ["Timestamp (ms)", "Duration (ms)", "Step", "K", "NNI", "XXR"] + }, + "demo_loggers.VideoRecorder": { + "path": "_export/logs/video.mp4", + "width": 1920, + "height": 1080, + "fps": 15 + } + } } }, "image_parameters": { @@ -109,5 +124,20 @@ "size": 2 } }, - "observers": "demo_frame_logger.py" + "observers": { + "demo_loggers.FixationLogger": { + "path": "_export/logs/fixations.csv", + "header": ["Timestamp (ms)", "Focus (px)", "Duration (ms)", "AOI"] + }, + "demo_loggers.ScanPathAnalysisLogger": { + "path": "_export/logs/scan_path_metrics.csv", + "header": ["Timestamp (ms)", "Duration (ms)", "Step", "K", "NNI", "XXR"] + }, + "demo_loggers.VideoRecorder": { + "path": "_export/logs/video.mp4", + "width": 1920, + "height": 1080, + "fps": 15 + } + } }
\ No newline at end of file diff --git a/src/argaze/utils/demo_data/demo_layer_logger.py b/src/argaze/utils/demo_data/demo_layer_logger.py deleted file mode 100644 index 0173cc7..0000000 --- a/src/argaze/utils/demo_data/demo_layer_logger.py +++ /dev/null @@ -1,42 +0,0 @@ -""" - -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 <http://www.gnu.org/licenses/>. -""" - -__author__ = "Théo de la Hogue" -__credits__ = [] -__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)" -__license__ = "GPLv3" - -from argaze import DataFeatures -from argaze.utils import UtilsFeatures - -class AOIScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter): - - def on_look(self, timestamp, layer, exception): - """Log aoi scan path metrics""" - - if layer.is_analysis_available(): - - log = ( - timestamp, - layer.aoi_scan_path_analyzers['argaze.GazeAnalysis.Basic'].path_duration, - layer.aoi_scan_path_analyzers['argaze.GazeAnalysis.Basic'].steps_number, - layer.aoi_scan_path_analyzers['argaze.GazeAnalysis.KCoefficient'].K, - layer.aoi_scan_path_analyzers['argaze.GazeAnalysis.LempelZivComplexity'].lempel_ziv_complexity - ) - - self.write(log) - -# Export loggers instances to register them as pipeline step object observers -__observers__ = { - "AOI Scan path analysis logger": AOIScanPathAnalysisLogger(path="_export/logs/aoi_scan_path_metrics.csv", header=("Timestamp (ms)", "Duration (ms)", "Step", "K", "LZC")) - }
\ No newline at end of file diff --git a/src/argaze/utils/demo_gaze_analysis_run.py b/src/argaze/utils/demo_gaze_analysis_run.py index f34871a..24871f2 100644 --- a/src/argaze/utils/demo_gaze_analysis_run.py +++ b/src/argaze/utils/demo_gaze_analysis_run.py @@ -40,225 +40,184 @@ args = parser.parse_args() def main(): # Load ArFrame - ar_frame = ArFeatures.ArFrame.from_json(args.configuration) + with ArFeatures.ArFrame.from_json(args.configuration) as ar_frame: - if args.verbose: + if args.verbose: - print(ar_frame) + print(ar_frame) - # Create a window to display ArCamera - cv2.namedWindow(ar_frame.name, cv2.WINDOW_AUTOSIZE) + # Create a window to display ArCamera + cv2.namedWindow(ar_frame.name, cv2.WINDOW_AUTOSIZE) - # Heatmap buffer display option - enable_heatmap_buffer = False + # Heatmap buffer display option + enable_heatmap_buffer = False - # Init timestamp - start_time = time.time() + # Init timestamp + start_time = time.time() - # Update pointer position - def on_mouse_event(event, x, y, flags, param): + # Update pointer position + def on_mouse_event(event, x, y, flags, param): - #try: + #try: - # Project gaze position into frame with millisecond timestamp - ar_frame.look(GazeFeatures.GazePosition((x, y), timestamp=int((time.time() - start_time) * 1e3))) + # Project gaze position into frame with millisecond timestamp + ar_frame.look(GazeFeatures.GazePosition((x, y), timestamp=int((time.time() - start_time) * 1e3))) - # Catch pipeline exception - #except Exception as e: + # Catch pipeline exception + #except Exception as e: - # print('Gaze projection error:', e) + # print('Gaze projection error:', e) - # Attach mouse callback to window - cv2.setMouseCallback(ar_frame.name, on_mouse_event) + # Attach mouse callback to window + cv2.setMouseCallback(ar_frame.name, on_mouse_event) - # Waiting for 'ctrl+C' interruption - with contextlib.suppress(KeyboardInterrupt): + # Waiting for 'ctrl+C' interruption + with contextlib.suppress(KeyboardInterrupt): - # Draw frame and mouse position analysis - while True: + # Draw frame and mouse position analysis + while True: - # Get frame image - frame_image = ar_frame.image() + # Get frame image + frame_image = ar_frame.image() - # Write heatmap buffer manual - buffer_on_off = 'on' if enable_heatmap_buffer else 'off' - buffer_display_disable = 'disable' if enable_heatmap_buffer else 'enable' - cv2.putText(frame_image, f'Heatmap buffer: {buffer_on_off} (Press \'b\' key to {buffer_display_disable})', (20, 40), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255) if enable_heatmap_buffer else (255, 255, 255), 1, cv2.LINE_AA) + # Write heatmap buffer manual + buffer_on_off = 'on' if enable_heatmap_buffer else 'off' + buffer_display_disable = 'disable' if enable_heatmap_buffer else 'enable' + cv2.putText(frame_image, f'Heatmap buffer: {buffer_on_off} (Press \'b\' key to {buffer_display_disable})', (20, 40), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255) if enable_heatmap_buffer else (255, 255, 255), 1, cv2.LINE_AA) - # Write last 5 steps of aoi scan path - path = '' - for step in ar_frame.layers["demo_layer"].aoi_scan_path[-5:]: + # Write last 5 steps of aoi scan path + path = '' + for step in ar_frame.layers["demo_layer"].aoi_scan_path[-5:]: - path += f'> {step.aoi} ' - - path += f'> {ar_frame.layers["demo_layer"].aoi_scan_path.current_aoi}' - - cv2.putText(frame_image, path, (20, ar_frame.size[1]-40), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 1, cv2.LINE_AA) - - # Display Transition matrix analysis if loaded - try: - - transition_matrix_analyzer = ar_frame.layers["demo_layer"].aoi_scan_path_analyzers["argaze.GazeAnalysis.TransitionMatrix"] - - cv2.putText(frame_image, f'Transition matrix density: {transition_matrix_analyzer.transition_matrix_density:.2f}', (20, ar_frame.size[1]-160), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) + path += f'> {step.aoi} ' - # Iterate over indexes (departures) - for from_aoi, row in transition_matrix_analyzer.transition_matrix_probabilities.iterrows(): - - # Iterate over columns (destinations) - for to_aoi, probability in row.items(): - - if from_aoi != GazeFeatures.OutsideAOI and to_aoi != GazeFeatures.OutsideAOI: - - if from_aoi != to_aoi and probability > 0.0: - - from_center = ar_frame.layers["demo_layer"].aoi_scene[from_aoi].center.astype(int) - to_center = ar_frame.layers["demo_layer"].aoi_scene[to_aoi].center.astype(int) - start_line = (0.5 * from_center + 0.5 * to_center).astype(int) - - color = [int(probability*200) + 55, int(probability*200) + 55, int(probability*200) + 55] - - cv2.line(frame_image, start_line, to_center, color, int(probability*10) + 2) - cv2.line(frame_image, from_center, to_center, [55, 55, 55], 2) - - except KeyError: - pass + path += f'> {ar_frame.layers["demo_layer"].aoi_scan_path.current_aoi}' - # Display aoi scan path basic metrics analysis if loaded - try: + cv2.putText(frame_image, path, (20, ar_frame.size[1]-40), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 1, cv2.LINE_AA) - basic_analyzer = ar_frame.layers["demo_layer"].aoi_scan_path_analyzers["argaze.GazeAnalysis.Basic"] - # Write basic analysis - cv2.putText(frame_image, f'Step number: {basic_analyzer.steps_number}', (20, ar_frame.size[1]-440), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - cv2.putText(frame_image, f'Step fixation duration average: {int(basic_analyzer.step_fixation_durations_average)} ms', (20, ar_frame.size[1]-400), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) + if ar_frame.is_analysis_available() and ar_frame.layers["demo_layer"].is_analysis_available(): - except KeyError: - pass + # Display Transition matrix analysis if loaded + transition_matrix_analyzer = ar_frame.layers["demo_layer"].analysis["argaze.GazeAnalysis.TransitionMatrix"] - # Display scan path K Coefficient analysis if loaded - try: + cv2.putText(frame_image, f'Transition matrix density: {transition_matrix_analyzer.transition_matrix_density:.2f}', (20, ar_frame.size[1]-160), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) + + # Iterate over indexes (departures) + for from_aoi, row in transition_matrix_analyzer.transition_matrix_probabilities.iterrows(): - kc_analyzer = ar_frame.scan_path_analyzers["argaze.GazeAnalysis.KCoefficient"] - - # Write raw Kc analysis - if kc_analyzer.K < 0.: - - cv2.putText(frame_image, f'K coefficient: Ambient attention', (20, ar_frame.size[1]-120), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - - elif kc_analyzer.K > 0.: - - cv2.putText(frame_image, f'K coefficient: Focal attention', (20, ar_frame.size[1]-120), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 0), 1, cv2.LINE_AA) - - except KeyError: - pass + # Iterate over columns (destinations) + for to_aoi, probability in row.items(): - # Display aoi scan path K-modified coefficient analysis if loaded - try: + if from_aoi != GazeFeatures.OutsideAOI and to_aoi != GazeFeatures.OutsideAOI: - aoi_kc_analyzer = ar_frame.layers["demo_layer"].aoi_scan_path_analyzers["argaze.GazeAnalysis.KCoefficient"] + if from_aoi != to_aoi and probability > 0.0: - # Write aoi Kc analysis - if aoi_kc_analyzer.K < 0.: + from_center = ar_frame.layers["demo_layer"].aoi_scene[from_aoi].center.astype(int) + to_center = ar_frame.layers["demo_layer"].aoi_scene[to_aoi].center.astype(int) + start_line = (0.5 * from_center + 0.5 * to_center).astype(int) - cv2.putText(frame_image, f'K-modified coefficient: Ambient attention', (20, ar_frame.size[1]-80), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - - elif aoi_kc_analyzer.K > 0.: + color = [int(probability*200) + 55, int(probability*200) + 55, int(probability*200) + 55] - cv2.putText(frame_image, f'K-modified coefficient: Focal attention', (20, ar_frame.size[1]-80), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 0), 1, cv2.LINE_AA) + cv2.line(frame_image, start_line, to_center, color, int(probability*10) + 2) + cv2.line(frame_image, from_center, to_center, [55, 55, 55], 2) - except KeyError: - pass - - # Display Lempel-Ziv complexity analysis if loaded - try: + # Display aoi scan path basic metrics analysis if loaded + basic_analyzer = ar_frame.layers["demo_layer"].analysis["argaze.GazeAnalysis.Basic"] + + # Write basic analysis + cv2.putText(frame_image, f'Step number: {basic_analyzer.steps_number}', (20, ar_frame.size[1]-440), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) + cv2.putText(frame_image, f'Step fixation duration average: {int(basic_analyzer.step_fixation_durations_average)} ms', (20, ar_frame.size[1]-400), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - lzc_analyzer = ar_frame.layers["demo_layer"].aoi_scan_path_analyzers["argaze.GazeAnalysis.LempelZivComplexity"] + # Display scan path K Coefficient analysis if loaded + kc_analyzer = ar_frame.analysis["argaze.GazeAnalysis.KCoefficient"] + + # Write raw Kc analysis + if kc_analyzer.K < 0.: - cv2.putText(frame_image, f'Lempel-Ziv complexity: {lzc_analyzer.lempel_ziv_complexity}', (20, ar_frame.size[1]-200), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) + cv2.putText(frame_image, f'K coefficient: Ambient attention', (20, ar_frame.size[1]-120), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) + + elif kc_analyzer.K > 0.: - except KeyError: - pass + cv2.putText(frame_image, f'K coefficient: Focal attention', (20, ar_frame.size[1]-120), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 0), 1, cv2.LINE_AA) - # Display N-Gram analysis if loaded - try: + # Display aoi scan path K-modified coefficient analysis if loaded + aoi_kc_analyzer = ar_frame.layers["demo_layer"].analysis["argaze.GazeAnalysis.KCoefficient"] - ngram_analyzer = ar_frame.layers["demo_layer"].aoi_scan_path_analyzers["argaze.GazeAnalysis.NGram"] + # Write aoi Kc analysis + if aoi_kc_analyzer.K < 0.: - # Display only 3-gram analysis - start = ar_frame.size[1] - ((len(ngram_analyzer.ngrams_count[3]) + 1) * 40) - cv2.putText(frame_image, f'{ngram_analyzer.n_max}-Gram:', (ar_frame.size[0]-700, start-40), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) + cv2.putText(frame_image, f'K-modified coefficient: Ambient attention', (20, ar_frame.size[1]-80), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) + + elif aoi_kc_analyzer.K > 0.: - for i, (ngram, count) in enumerate(ngram_analyzer.ngrams_count[3].items()): + cv2.putText(frame_image, f'K-modified coefficient: Focal attention', (20, ar_frame.size[1]-80), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 0), 1, cv2.LINE_AA) + + # Display Lempel-Ziv complexity analysis if loaded + lzc_analyzer = ar_frame.layers["demo_layer"].analysis["argaze.GazeAnalysis.LempelZivComplexity"] - ngram_string = f'{ngram[0]}' - for g in range(1, 3): - ngram_string += f'>{ngram[g]}' + cv2.putText(frame_image, f'Lempel-Ziv complexity: {lzc_analyzer.lempel_ziv_complexity}', (20, ar_frame.size[1]-200), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - cv2.putText(frame_image, f'{ngram_string}: {count}', (ar_frame.size[0]-700, start+(i*40)), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - except KeyError: - pass + # Display N-Gram analysis if loaded + ngram_analyzer = ar_frame.layers["demo_layer"].analysis["argaze.GazeAnalysis.NGram"] - # Display Entropy analysis if loaded - try: + # Display only 3-gram analysis + start = ar_frame.size[1] - ((len(ngram_analyzer.ngrams_count[3]) + 1) * 40) + cv2.putText(frame_image, f'{ngram_analyzer.n_max}-Gram:', (ar_frame.size[0]-700, start-40), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - entropy_analyzer = ar_frame.layers["demo_layer"].aoi_scan_path_analyzers["argaze.GazeAnalysis.Entropy"] + for i, (ngram, count) in enumerate(ngram_analyzer.ngrams_count[3].items()): - cv2.putText(frame_image, f'Stationary entropy: {entropy_analyzer.stationary_entropy:.3f},', (20, ar_frame.size[1]-280), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - cv2.putText(frame_image, f'Transition entropy: {entropy_analyzer.transition_entropy:.3f},', (20, ar_frame.size[1]-240), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - - except KeyError: - pass + ngram_string = f'{ngram[0]}' + for g in range(1, 3): + ngram_string += f'>{ngram[g]}' - # Display Nearest Neighbor index analysis if loaded - try: + cv2.putText(frame_image, f'{ngram_string}: {count}', (ar_frame.size[0]-700, start+(i*40)), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - nni_analyzer = ar_frame.scan_path_analyzers["argaze.GazeAnalysis.NearestNeighborIndex"] - - cv2.putText(frame_image, f'Nearest neighbor index: {nni_analyzer.nearest_neighbor_index:.3f}', (20, ar_frame.size[1]-320), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - - except KeyError: - pass + # Display Entropy analysis if loaded + entropy_analyzer = ar_frame.layers["demo_layer"].analysis["argaze.GazeAnalysis.Entropy"] - # Display Explore/Exploit ratio analysis if loaded - try: + cv2.putText(frame_image, f'Stationary entropy: {entropy_analyzer.stationary_entropy:.3f},', (20, ar_frame.size[1]-280), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) + cv2.putText(frame_image, f'Transition entropy: {entropy_analyzer.transition_entropy:.3f},', (20, ar_frame.size[1]-240), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - xxr_analyser = ar_frame.scan_path_analyzers["argaze.GazeAnalysis.ExploreExploitRatio"] + # Display Nearest Neighbor index analysis if loaded + nni_analyzer = ar_frame.analysis["argaze.GazeAnalysis.NearestNeighborIndex"] - cv2.putText(frame_image, f'Explore/Exploit ratio: {xxr_analyser.explore_exploit_ratio:.3f}', (20, ar_frame.size[1]-360), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) + cv2.putText(frame_image, f'Nearest neighbor index: {nni_analyzer.nearest_neighbor_index:.3f}', (20, ar_frame.size[1]-320), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) + + # Display Explore/Exploit ratio analysis if loaded + xxr_analyser = ar_frame.analysis["argaze.GazeAnalysis.ExploreExploitRatio"] - except KeyError: + cv2.putText(frame_image, f'Explore/Exploit ratio: {xxr_analyser.explore_exploit_ratio:.3f}', (20, ar_frame.size[1]-360), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 1, cv2.LINE_AA) - pass - # Display frame image - cv2.imshow(ar_frame.name, frame_image) + # Display frame image + cv2.imshow(ar_frame.name, frame_image) - key_pressed = cv2.waitKey(10) + key_pressed = cv2.waitKey(10) - #if key_pressed != -1: - # print(key_pressed) + #if key_pressed != -1: + # print(key_pressed) - # Reload environment with 'h' key - if key_pressed == 114: + # Reload environment with 'h' key + if key_pressed == 114: - ar_frame = ArFeatures.ArFrame.from_json(args.frame) + ar_frame = ArFeatures.ArFrame.from_json(args.frame) - # Enable heatmap buffer with 'b' key - if key_pressed == 98: + # Enable heatmap buffer with 'b' key + if key_pressed == 98: - enable_heatmap_buffer = not enable_heatmap_buffer + enable_heatmap_buffer = not enable_heatmap_buffer - ar_frame.heatmap.buffer = 10 if enable_heatmap_buffer else 0 - ar_frame.heatmap.clear() + ar_frame.heatmap.buffer = 10 if enable_heatmap_buffer else 0 + ar_frame.heatmap.clear() - # Stop by pressing 'Esc' key - if key_pressed == 27: - break + # Stop by pressing 'Esc' key + if key_pressed == 27: + break - # Stop frame image display - cv2.destroyAllWindows() + # Stop frame image display + cv2.destroyAllWindows() if __name__ == '__main__': |