From 0423353fb03adea80a7d25bbc97338f973a643f7 Mon Sep 17 00:00:00 2001
From: Théo de la Hogue
Date: Thu, 21 Mar 2024 21:07:08 +0100
Subject: More work on new serialization system. Demo still not working.
---
src/argaze/ArFeatures.py | 58 ++---
src/argaze/ArUcoMarkers/ArUcoCamera.py | 87 +++----
src/argaze/ArUcoMarkers/ArUcoMarkersGroup.py | 66 +++--
src/argaze/ArUcoMarkers/ArUcoScene.py | 62 ++++-
src/argaze/DataFeatures.py | 86 +++----
src/argaze/GazeAnalysis/Basic.py | 4 +-
src/argaze/GazeAnalysis/DeviationCircleCoverage.py | 2 +-
.../DispersionThresholdIdentification.py | 2 +-
src/argaze/GazeAnalysis/Entropy.py | 2 +-
src/argaze/GazeAnalysis/ExploreExploitRatio.py | 2 +-
src/argaze/GazeAnalysis/FocusPointInside.py | 4 +-
src/argaze/GazeAnalysis/KCoefficient.py | 4 +-
src/argaze/GazeAnalysis/LempelZivComplexity.py | 2 +-
src/argaze/GazeAnalysis/LinearRegression.py | 2 +-
src/argaze/GazeAnalysis/NGram.py | 2 +-
src/argaze/GazeAnalysis/NearestNeighborIndex.py | 2 +-
src/argaze/GazeAnalysis/TransitionMatrix.py | 2 +-
.../VelocityThresholdIdentification.py | 12 +-
src/argaze/GazeFeatures.py | 10 +-
.../utils/demo_data/demo_aruco_markers_setup.json | 26 +-
src/argaze/utils/demo_data/demo_frame_logger.py | 70 ------
.../utils/demo_data/demo_gaze_analysis_setup.json | 58 +++--
src/argaze/utils/demo_data/demo_layer_logger.py | 42 ----
src/argaze/utils/demo_gaze_analysis_run.py | 277 +++++++++------------
24 files changed, 389 insertions(+), 495 deletions(-)
delete mode 100644 src/argaze/utils/demo_data/demo_frame_logger.py
delete mode 100644 src/argaze/utils/demo_data/demo_layer_logger.py
(limited to 'src')
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 .
-"""
-
-__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 .
-"""
-
-__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__':
--
cgit v1.1