aboutsummaryrefslogtreecommitdiff
path: root/src/argaze/ArFeatures.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/argaze/ArFeatures.py')
-rw-r--r--src/argaze/ArFeatures.py140
1 files changed, 69 insertions, 71 deletions
diff --git a/src/argaze/ArFeatures.py b/src/argaze/ArFeatures.py
index 7263b94..f90354a 100644
--- a/src/argaze/ArFeatures.py
+++ b/src/argaze/ArFeatures.py
@@ -116,7 +116,7 @@ class ArLayer(DataFeatures.SharedObject):
aoi_matcher: GazeFeatures.AOIMatcher = field(default_factory=GazeFeatures.AOIMatcher)
aoi_scan_path: GazeFeatures.AOIScanPath = field(default_factory=GazeFeatures.AOIScanPath)
aoi_scan_path_analyzers: dict = field(default_factory=dict)
- log: bool = field(default=False)
+ loggers: dict = field(default=dict)
draw_parameters: dict = field(default_factory=DEFAULT_ARLAYER_DRAW_PARAMETERS)
def __post_init__(self):
@@ -139,16 +139,6 @@ class ArLayer(DataFeatures.SharedObject):
self.aoi_scene = AOI3DScene.AOI3DScene(self.aoi_scene)
- # Prepare logging if needed
- self.__ts_logs = {}
-
- if self.log:
-
- # Create timestamped buffers to log each aoi scan path analysis
- for aoi_scan_path_analyzer_module_path in self.aoi_scan_path_analyzers.keys():
-
- self.__ts_logs[aoi_scan_path_analyzer_module_path] = DataFeatures.TimeStampedBuffer()
-
@classmethod
def from_dict(self, layer_data: dict, working_directory: str = None) -> ArLayerType:
"""Load attributes from dictionary.
@@ -294,14 +284,28 @@ class ArLayer(DataFeatures.SharedObject):
pass
- # Load log status
+ # Load loggers
+ new_loggers = {}
+
try:
- new_layer_log = layer_data.pop('log')
+ new_loggers_value = layer_data.pop('loggers')
+
+ for logger_module_path, logger_parameters in new_loggers_value.items():
+
+ # Prepend argaze.DataLog path when a single name is provided
+ if len(logger_module_path.split('.')) == 1:
+ logger_module_path = f'argaze.DataLog.{logger_module_path}'
+
+ logger_module = importlib.import_module(logger_module_path)
+
+ logger = logger_module.TimeStampedDataLogger(**logger_parameters)
+
+ new_loggers[logger_module_path] = logger
except KeyError:
- new_layer_log = False
+ pass
# Load image parameters
try:
@@ -318,7 +322,7 @@ class ArLayer(DataFeatures.SharedObject):
new_aoi_matcher, \
new_aoi_scan_path, \
new_aoi_scan_path_analyzers, \
- new_layer_log, \
+ new_loggers, \
new_layer_draw_parameters \
)
@@ -350,14 +354,6 @@ class ArLayer(DataFeatures.SharedObject):
self.__parent = parent
- @property
- def logs(self):
- """
- Get stored logs
- """
-
- return self.__ts_logs
-
def look(self, timestamp: int|float, gaze_movement: GazeFeatures.GazePosition = GazeFeatures.UnvalidGazePosition()) -> dict:
"""
Project timestamped gaze movement into layer.
@@ -439,11 +435,6 @@ class ArLayer(DataFeatures.SharedObject):
# Store analysis
aoi_scan_path_analysis[aoi_scan_path_analyzer_module_path] = aoi_scan_path_analyzer.analysis
- # Log analysis
- if self.log:
-
- self.__ts_logs[aoi_scan_path_analyzer_module_path][timestamp] = aoi_scan_path_analyzer.analysis
-
elif GazeFeatures.is_saccade(gaze_movement):
# Append saccade to aoi scan path
@@ -461,12 +452,20 @@ class ArLayer(DataFeatures.SharedObject):
# Assess total execution time in ms
execution_times['total'] = (time.perf_counter() - look_start) * 1e3
+
+ # Edit look data
+ look_data = looked_aoi_name, aoi_scan_path_analysis, execution_times, exception
+
+ # Log look data
+ for logger_module_path, logger in self.loggers.items():
+
+ logger.emit(timestamp, look_data)
# Unlock layer exploitation
self.release()
# Return look data
- return looked_aoi, aoi_scan_path_analysis, execution_times, exception
+ return look_data
def draw(self, image: numpy.array, draw_aoi_scene: dict = None, draw_aoi_matching: dict = None):
"""
@@ -538,7 +537,7 @@ class ArFrame(DataFeatures.SharedObject):
heatmap: heatmap object
background: picture to draw behind
layers: dictionary of AOI layers
- log: enable scan path analysis logging
+ loggers: dictionary of timestamped data loggers
image_parameters: default parameters passed to image method
"""
@@ -552,7 +551,7 @@ class ArFrame(DataFeatures.SharedObject):
heatmap: AOIFeatures.Heatmap = field(default_factory=AOIFeatures.Heatmap)
background: numpy.array = field(default_factory=lambda : numpy.array([]))
layers: dict = field(default_factory=dict)
- log: bool = field(default=False)
+ loggers: dict = field(default=dict)
image_parameters: dict = field(default_factory=DEFAULT_ARFRAME_IMAGE_PARAMETERS)
def __post_init__(self):
@@ -571,16 +570,6 @@ class ArFrame(DataFeatures.SharedObject):
# Init current gaze position
self.__gaze_position = GazeFeatures.UnvalidGazePosition()
- # Prepare logging if needed
- self.__ts_logs = {}
-
- if self.log:
-
- # Create timestamped buffers to log each aoi scan path analysis
- for scan_path_analyzer_module_path in self.scan_path_analyzers.keys():
-
- self.__ts_logs[scan_path_analyzer_module_path] = DataFeatures.TimeStampedBuffer()
-
@classmethod
def from_dict(self, frame_data: dict, working_directory: str = None) -> ArFrameType:
"""Load attributes from dictionary.
@@ -777,6 +766,29 @@ class ArFrame(DataFeatures.SharedObject):
new_frame_log = False
+ # Load loggers
+ new_loggers = {}
+
+ try:
+
+ new_loggers_value = frame_data.pop('loggers')
+
+ for logger_module_path, logger_parameters in new_loggers_value.items():
+
+ # Prepend argaze.DataLog path when a single name is provided
+ if len(logger_module_path.split('.')) == 1:
+ logger_module_path = f'argaze.DataLog.{logger_module_path}'
+
+ logger_module = importlib.import_module(logger_module_path)
+
+ logger = logger_module.TimeStampedDataLogger(**logger_parameters)
+
+ new_loggers[logger_module_path] = logger
+
+ except KeyError:
+
+ pass
+
# Load image parameters
try:
@@ -797,7 +809,7 @@ class ArFrame(DataFeatures.SharedObject):
new_heatmap, \
new_frame_background, \
new_layers, \
- new_frame_log,
+ new_loggers,
new_frame_image_parameters \
)
@@ -829,14 +841,6 @@ class ArFrame(DataFeatures.SharedObject):
self.__parent = parent
- @property
- def logs(self):
- """
- Get stored logs
- """
-
- return self.__ts_logs
-
def look(self, timestamp: int|float, gaze_position: GazeFeatures.GazePosition = GazeFeatures.UnvalidGazePosition()) -> Tuple[GazeFeatures.GazePosition, GazeFeatures.GazeMovement, dict, dict, dict, Exception]:
"""
Project gaze position into frame.
@@ -869,20 +873,19 @@ class ArFrame(DataFeatures.SharedObject):
# Init scan path analysis report
scan_step_analysis = {}
- # Init layer analysis report
- layer_analysis = {}
-
# Assess pipeline execution times
execution_times = {
'gaze_movement_identifier': None,
'scan_step_analyzers':{},
- 'heatmap': None,
- 'layers': {}
+ 'heatmap': None
}
# Catch any error
exception = None
+ # Init layers look data report
+ layers_look_data = {}
+
try:
# Apply gaze position calibration
@@ -941,11 +944,6 @@ class ArFrame(DataFeatures.SharedObject):
# Store analysis
scan_step_analysis[scan_path_analyzer_module_path] = scan_path_analyzer.analysis
- # Log analysis
- if self.log:
-
- self.__ts_logs[scan_path_analyzer_module_path][timestamp] = scan_path_analyzer.analysis
-
# No valid finished gaze movement: optionnaly stop in progress identification filtering
elif self.gaze_movement_identifier is not None and not self.filter_in_progress_identification:
@@ -970,15 +968,7 @@ class ArFrame(DataFeatures.SharedObject):
# Note: don't filter valid/unvalid finished/unfished gaze movement to allow layers to reset internally
for layer_name, layer in self.layers.items():
- looked_aoi, aoi_scan_path_analysis, layer_execution_times, layer_exception = layer.look(timestamp, identified_gaze_movement)
-
- layer_analysis[layer_name] = aoi_scan_path_analysis
-
- execution_times['layers'][layer_name] = layer_execution_times
-
- if layer_exception:
-
- raise(layer_exception)
+ layers_look_data[layer_name] = layer.look(timestamp, identified_gaze_movement)
except Exception as e:
@@ -987,17 +977,25 @@ class ArFrame(DataFeatures.SharedObject):
self.__gaze_position = GazeFeatures.UnvalidGazePosition()
identified_gaze_movement = GazeFeatures.UnvalidGazeMovement()
scan_step_analysis = {}
- layer_analysis = {}
exception = e
+ layers_look_data = {}
# Assess total execution time in ms
execution_times['total'] = (time.perf_counter() - look_start) * 1e3
+ # Edit look data
+ look_data = self.__gaze_position, identified_gaze_movement, scan_step_analysis, execution_times, exception, layers_look_data
+
+ # Log look data
+ for logger_module_path, logger in self.loggers.items():
+
+ logger.emit(timestamp, look_data)
+
# Unlock frame exploitation
self.release()
# Return look data
- return self.__gaze_position, identified_gaze_movement, scan_step_analysis, layer_analysis, execution_times, exception
+ return look_data
def __image(self, background_weight: float = None, heatmap_weight: float = None, draw_gaze_position_calibrator: dict = None, draw_scan_path: dict = None, draw_layers: dict = None, draw_gaze_positions: dict = None, draw_fixations: dict = None, draw_saccades: dict = None) -> numpy.array:
"""
generated by cgit v1.1 at 2024-09-20 13:34:43 +0000