diff options
23 files changed, 216 insertions, 78 deletions
diff --git a/src/argaze/ArFeatures.py b/src/argaze/ArFeatures.py index 1b3e504..8c9b3c8 100644 --- a/src/argaze/ArFeatures.py +++ b/src/argaze/ArFeatures.py @@ -8,9 +8,11 @@ __copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)" __license__ = "BSD" from typing import TypeVar, Tuple, Any, Iterator, Union +from types import ModuleType from dataclasses import dataclass, field import json import os +import sys import importlib from inspect import getmembers import threading @@ -108,6 +110,7 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): aoi_scan_path: AOI scan path object aoi_scan_path_analyzers: dictionary of AOI scan path analyzers draw_parameters: default parameters passed to draw method + logging_module: path to logging module file in working directory """ name: str @@ -116,6 +119,7 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): aoi_scan_path: GazeFeatures.AOIScanPath = field(default_factory=GazeFeatures.AOIScanPath) aoi_scan_path_analyzers: dict = field(default_factory=dict) draw_parameters: dict = field(default_factory=DEFAULT_ARLAYER_DRAW_PARAMETERS) + logging_module: ModuleType = field(default=None) def __post_init__(self): @@ -152,6 +156,11 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): working_directory: folder path where to load files when a dictionary value is a relative filepath. """ + # Append working directory to the Python path + if working_directory is not None: + + sys.path.append(working_directory) + # Load name try: @@ -297,6 +306,26 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): new_layer_draw_parameters = DEFAULT_ARLAYER_DRAW_PARAMETERS + # Load logging module + try: + + new_logging_module_value = layer_data.pop('logging_module') + + # str: relative path to file + if type(new_logging_module_value) == str: + + logging_module_name = new_logging_module_value.split('.')[0] + + # Import logging module + self.logging_module = importlib.import_module(logging_module_name) + + # Register loggers as pipeline step observers + self.observers = self.logging_module.__loggers__ + + except KeyError: + + pass + # Create layer return ArLayer(new_layer_name, \ new_aoi_scene, \ @@ -388,7 +417,7 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): # Update looked aoi thanks to aoi matcher # Note: don't filter valid/unvalid and finished/unfinished fixation/saccade as we don't know how the aoi matcher works internally - self.__looked_aoi_name, _ = self.aoi_matcher.match(self.aoi_scene, gaze_movement) + self.__looked_aoi_name, _ = self.aoi_matcher.match(timestamp, self.aoi_scene, gaze_movement) # Valid and finished gaze movement has been identified if gaze_movement.valid and gaze_movement.finished: @@ -406,7 +435,7 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): # Analyze aoi scan path for aoi_scan_path_analyzer_module_path, aoi_scan_path_analyzer in self.aoi_scan_path_analyzers.items(): - aoi_scan_path_analyzer.analyze(self.aoi_scan_path) + aoi_scan_path_analyzer.analyze(timestamp, self.aoi_scan_path) # Update aoi scan path analyzed state self.__aoi_scan_path_analyzed = True @@ -486,6 +515,7 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): background: picture to draw behind layers: dictionary of AOI layers image_parameters: default parameters passed to image method + logging_module: path to logging module file in working directory """ name: str @@ -499,6 +529,7 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): background: numpy.array = field(default_factory=lambda : numpy.array([])) layers: dict = field(default_factory=dict) image_parameters: dict = field(default_factory=DEFAULT_ARFRAME_IMAGE_PARAMETERS) + logging_module: ModuleType = field(default=None) def __post_init__(self): @@ -531,6 +562,11 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): working_directory: folder path where to load files when a dictionary value is a relative filepath. """ + # Append working directory to the Python path + if working_directory is not None: + + sys.path.append(working_directory) + # Load name try: @@ -718,6 +754,26 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): new_frame_image_parameters = DEFAULT_ARFRAME_IMAGE_PARAMETERS + # Load logging module + try: + + new_logging_module_value = frame_data.pop('logging_module') + + # str: relative path to file + if type(new_logging_module_value) == str: + + logging_module_name = new_logging_module_value.split('.')[0] + + # Import logging module + self.logging_module = importlib.import_module(logging_module_name) + + # Register loggers as pipeline step observers + self.observers = self.logging_module.__loggers__ + + except KeyError: + + pass + # Create frame return ArFrame(new_frame_name, \ new_frame_size, \ @@ -852,7 +908,7 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): # Analyze aoi scan path for scan_path_analyzer_module_path, scan_path_analyzer in self.scan_path_analyzers.items(): - scan_path_analyzer.analyze(self.scan_path) + scan_path_analyzer.analyze(timestamp, self.scan_path) # Update scan path analyzed state self.__scan_path_analyzed = True @@ -869,7 +925,7 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): scale = numpy.array([self.heatmap.size[0] / self.size[0], self.heatmap.size[1] / self.size[1]]) # Update heatmap image - self.heatmap.update(self.__calibrated_gaze_position.value * scale) + self.heatmap.update(timestamp, self.__calibrated_gaze_position.value * scale) # Look layers with valid identified gaze movement # Note: don't filter valid/unvalid finished/unfished gaze movement to allow layers to reset internally diff --git a/src/argaze/ArUcoMarkers/ArUcoDetector.py b/src/argaze/ArUcoMarkers/ArUcoDetector.py index 6d12f3d..c562467 100644 --- a/src/argaze/ArUcoMarkers/ArUcoDetector.py +++ b/src/argaze/ArUcoMarkers/ArUcoDetector.py @@ -257,7 +257,7 @@ class ArUcoDetector(DataFeatures.PipelineStepObject): return output @DataFeatures.PipelineStepMethod - def detect_markers(self, image: numpy.array) -> float: + def detect_markers(self, timestamp: int|float, image: numpy.array) -> float: """Detect all ArUco markers into an image. !!! danger "DON'T MIRROR IMAGE" diff --git a/src/argaze/AreaOfInterest/AOIFeatures.py b/src/argaze/AreaOfInterest/AOIFeatures.py index 6457e8f..002b251 100644 --- a/src/argaze/AreaOfInterest/AOIFeatures.py +++ b/src/argaze/AreaOfInterest/AOIFeatures.py @@ -600,7 +600,7 @@ class Heatmap(DataFeatures.PipelineStepObject): self.__point_spread_buffer_size = self.buffer @DataFeatures.PipelineStepMethod - def update(self, point: tuple): + def update(self, timestamp: int|float, point: tuple): """Update heatmap image.""" point_spread = self.point_spread(point) diff --git a/src/argaze/DataFeatures.py b/src/argaze/DataFeatures.py index 4dfd911..ac4a176 100644 --- a/src/argaze/DataFeatures.py +++ b/src/argaze/DataFeatures.py @@ -418,15 +418,23 @@ class SharedObject(): return timestamped class PipelineStepObject(): - """Abstract class to assess pipeline step methods execution time.""" + """Abstract class to assess pipeline step methods execution time. + + Parameters: + execution_times: dictionary with each PipelineStepMethod execution time in ms. + """ execution_times: dict = {} - """Execution time for each mehtod in ms.""" + observers: dict = {} def PipelineStepMethod(method): - """Define a decorator use into PipelineStepObject class to declare pipeline method.""" + """Define a decorator use into PipelineStepObject class to declare pipeline method. + + !!! danger + PipelineStepMethod must have a timestamp as first argument. + """ - def wrapper(self, *args, **kw): + def wrapper(self, timestamp, *args, **kw): """Wrap pipeline step method to measure execution time.""" # Initialize execution time assessment @@ -434,13 +442,38 @@ def PipelineStepMethod(method): try: - result = method(self, *args, **kw) + # Execute wrapped method + result = method(self, timestamp, *args, **kw) finally: # Measure execution time self.execution_times[method.__name__] = (time.perf_counter() - start) * 1e3 + # Notify observers that method has been called + subscription_name = f'on_{method.__name__}' + + for observer_name, observer in self.observers.items(): + + # Does the observer cares about this method? + try: + + subscription = getattr(observer, subscription_name) + + # Call subscription + subscription(timestamp, self) + + except AttributeError: + + pass + return result return wrapper + +class PipelineStepObserver(): + """Define abstract class to observe pipeline step object use. + + !!! note + To subscribe to a method call, the inherited class simply needs to define 'on_<method_name>' functions with timestamp and object argument. + """ diff --git a/src/argaze/GazeAnalysis/Basic.py b/src/argaze/GazeAnalysis/Basic.py index b75932a..55c0737 100644 --- a/src/argaze/GazeAnalysis/Basic.py +++ b/src/argaze/GazeAnalysis/Basic.py @@ -27,7 +27,7 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer): self.__step_fixation_durations_average = 0 @DataFeatures.PipelineStepMethod - def analyze(self, scan_path: GazeFeatures.ScanPathType): + def analyze(self, timestamp: int|float, scan_path: GazeFeatures.ScanPathType): self.__path_duration = scan_path.duration @@ -72,7 +72,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): self.__step_fixation_durations_average = 0 @DataFeatures.PipelineStepMethod - def analyze(self, aoi_scan_path: GazeFeatures.ScanPathType): + def analyze(self, timestamp: int|float, aoi_scan_path: GazeFeatures.ScanPathType): self.__path_duration = aoi_scan_path.duration diff --git a/src/argaze/GazeAnalysis/DeviationCircleCoverage.py b/src/argaze/GazeAnalysis/DeviationCircleCoverage.py index f890701..e80b198 100644 --- a/src/argaze/GazeAnalysis/DeviationCircleCoverage.py +++ b/src/argaze/GazeAnalysis/DeviationCircleCoverage.py @@ -38,7 +38,7 @@ class AOIMatcher(GazeFeatures.AOIMatcher): self.__matched_region = None @DataFeatures.PipelineStepMethod - def match(self, aoi_scene, gaze_movement) -> Tuple[str, AOIFeatures.AreaOfInterest]: + def match(self, timestamp: int|float, aoi_scene, gaze_movement) -> Tuple[str, AOIFeatures.AreaOfInterest]: """Returns AOI with the maximal fixation's deviation circle coverage if above coverage threshold.""" if GazeFeatures.is_fixation(gaze_movement): diff --git a/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py b/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py index 011c272..39895fd 100644 --- a/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py +++ b/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py @@ -143,7 +143,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): self.__saccade_positions = GazeFeatures.TimeStampedGazePositions() @DataFeatures.PipelineStepMethod - def identify(self, ts, gaze_position, terminate=False) -> GazeMovementType: + def identify(self, ts: int|float, gaze_position, terminate=False) -> GazeMovementType: # Ignore non valid gaze position if not gaze_position.valid: diff --git a/src/argaze/GazeAnalysis/Entropy.py b/src/argaze/GazeAnalysis/Entropy.py index a62dfe6..eeccfa7 100644 --- a/src/argaze/GazeAnalysis/Entropy.py +++ b/src/argaze/GazeAnalysis/Entropy.py @@ -38,7 +38,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): self.__transition_entropy = -1 @DataFeatures.PipelineStepMethod - def analyze(self, aoi_scan_path: GazeFeatures.AOIScanPathType): + def analyze(self, timestamp: int|float, aoi_scan_path: GazeFeatures.AOIScanPathType): assert(len(aoi_scan_path) > 1) diff --git a/src/argaze/GazeAnalysis/ExploreExploitRatio.py b/src/argaze/GazeAnalysis/ExploreExploitRatio.py index 5516349..1f7fad0 100644 --- a/src/argaze/GazeAnalysis/ExploreExploitRatio.py +++ b/src/argaze/GazeAnalysis/ExploreExploitRatio.py @@ -34,7 +34,7 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer): self.__explore_exploit_ratio = 0. @DataFeatures.PipelineStepMethod - def analyze(self, scan_path: GazeFeatures.ScanPathType): + def analyze(self, timestamp: int|float, scan_path: GazeFeatures.ScanPathType): assert(len(scan_path) > 1) diff --git a/src/argaze/GazeAnalysis/FocusPointInside.py b/src/argaze/GazeAnalysis/FocusPointInside.py index d559ac2..bf97d0d 100644 --- a/src/argaze/GazeAnalysis/FocusPointInside.py +++ b/src/argaze/GazeAnalysis/FocusPointInside.py @@ -31,7 +31,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, timestamp: int|float, 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 41338a3..40e3ddd 100644 --- a/src/argaze/GazeAnalysis/KCoefficient.py +++ b/src/argaze/GazeAnalysis/KCoefficient.py @@ -31,7 +31,7 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer): self.__K = 0 @DataFeatures.PipelineStepMethod - def analyze(self, scan_path: GazeFeatures.ScanPathType): + def analyze(self, timestamp: int|float, scan_path: GazeFeatures.ScanPathType): assert(len(scan_path) > 1) @@ -88,7 +88,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): self.__K = 0 @DataFeatures.PipelineStepMethod - def analyze(self, aoi_scan_path: GazeFeatures.AOIScanPathType) -> float: + def analyze(self, timestamp: int|float, aoi_scan_path: GazeFeatures.AOIScanPathType) -> float: assert(len(aoi_scan_path) > 1) diff --git a/src/argaze/GazeAnalysis/LempelZivComplexity.py b/src/argaze/GazeAnalysis/LempelZivComplexity.py index f6a49ab..53d4285 100644 --- a/src/argaze/GazeAnalysis/LempelZivComplexity.py +++ b/src/argaze/GazeAnalysis/LempelZivComplexity.py @@ -32,7 +32,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): self.__lempel_ziv_complexity = 0 @DataFeatures.PipelineStepMethod - def analyze(self, aoi_scan_path: GazeFeatures.AOIScanPathType): + def analyze(self, timestamp: int|float, aoi_scan_path: GazeFeatures.AOIScanPathType): assert(len(aoi_scan_path) > 1) diff --git a/src/argaze/GazeAnalysis/NGram.py b/src/argaze/GazeAnalysis/NGram.py index 2526123..049da7d 100644 --- a/src/argaze/GazeAnalysis/NGram.py +++ b/src/argaze/GazeAnalysis/NGram.py @@ -36,7 +36,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): self.__ngrams_count = {} @DataFeatures.PipelineStepMethod - def analyze(self, aoi_scan_path: GazeFeatures.AOIScanPathType): + def analyze(self, timestamp: int|float, aoi_scan_path: GazeFeatures.AOIScanPathType): assert(len(aoi_scan_path) > 1) diff --git a/src/argaze/GazeAnalysis/NearestNeighborIndex.py b/src/argaze/GazeAnalysis/NearestNeighborIndex.py index 72df516..e42dea2 100644 --- a/src/argaze/GazeAnalysis/NearestNeighborIndex.py +++ b/src/argaze/GazeAnalysis/NearestNeighborIndex.py @@ -36,7 +36,7 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer): self.__nearest_neighbor_index = 0 @DataFeatures.PipelineStepMethod - def analyze(self, scan_path: GazeFeatures.ScanPathType): + def analyze(self, timestamp: int|float, scan_path: GazeFeatures.ScanPathType): assert(len(scan_path) > 1) diff --git a/src/argaze/GazeAnalysis/TransitionMatrix.py b/src/argaze/GazeAnalysis/TransitionMatrix.py index d001947..5248480 100644 --- a/src/argaze/GazeAnalysis/TransitionMatrix.py +++ b/src/argaze/GazeAnalysis/TransitionMatrix.py @@ -34,7 +34,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): self.__transition_matrix_density = 0. @DataFeatures.PipelineStepMethod - def analyze(self, aoi_scan_path: GazeFeatures.AOIScanPathType): + def analyze(self, timestamp: int|float, aoi_scan_path: GazeFeatures.AOIScanPathType): assert(len(aoi_scan_path) > 1) diff --git a/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py b/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py index 2c3ecd1..62f8bdd 100644 --- a/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py +++ b/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py @@ -143,7 +143,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): self.__saccade_positions = GazeFeatures.TimeStampedGazePositions() @DataFeatures.PipelineStepMethod - def identify(self, ts, gaze_position, terminate=False) -> GazeMovementType: + def identify(self, ts: int|float, gaze_position, terminate=False) -> GazeMovementType: # Ignore non valid gaze position if not gaze_position.valid: diff --git a/src/argaze/GazeFeatures.py b/src/argaze/GazeFeatures.py index d70cdc6..db4c1f4 100644 --- a/src/argaze/GazeFeatures.py +++ b/src/argaze/GazeFeatures.py @@ -837,7 +837,7 @@ class ScanPathAnalyzer(DataFeatures.PipelineStepObject): return DataFeatures.DataDictionary(analysis) @DataFeatures.PipelineStepMethod - def analyze(self, scan_path: ScanPathType): + def analyze(self, timestamp: int|float, scan_path: ScanPathType): """Analyze scan path.""" raise NotImplementedError('analyze() method not implemented') @@ -1175,7 +1175,7 @@ class AOIScanPathAnalyzer(DataFeatures.PipelineStepObject): return DataFeatures.DataDictionary(analysis) @DataFeatures.PipelineStepMethod - def analyze(self, aoi_scan_path: AOIScanPathType): + def analyze(self, timestamp: int|float, aoi_scan_path: AOIScanPathType): """Analyze aoi scan path.""" raise NotImplementedError('analyze() method not implemented') diff --git a/src/argaze/PupillAnalysis/WorkloadIndex.py b/src/argaze/PupillAnalysis/WorkloadIndex.py index 99f143b..1f3c586 100644 --- a/src/argaze/PupillAnalysis/WorkloadIndex.py +++ b/src/argaze/PupillAnalysis/WorkloadIndex.py @@ -34,7 +34,7 @@ class PupillDiameterAnalyzer(PupillFeatures.PupillDiameterAnalyzer): self.__last_ts = 0 @DataFeatures.PipelineStepMethod - def analyze(self, ts, pupill_diameter) -> float: + def analyze(self, ts: int|float, pupill_diameter) -> float: """Analyze workload index from successive timestamped pupill diameters.""" # Ignore non valid pupill diameter diff --git a/src/argaze/PupillFeatures.py b/src/argaze/PupillFeatures.py index 8aa7827..d8f9331 100644 --- a/src/argaze/PupillFeatures.py +++ b/src/argaze/PupillFeatures.py @@ -83,7 +83,7 @@ class PupillDiameterAnalyzer(DataFeatures.PipelineStepObject): """Abstract class to define what should provide a pupill diameter analyser.""" @DataFeatures.PipelineStepMethod - def analyze(self, ts, pupill_diameter) -> float: + def analyze(self, timestamp: int|float, pupill_diameter, float) -> float: """Analyze pupill diameter from successive timestamped pupill diameters.""" raise NotImplementedError('analyze() method not implemented') 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 f921662..5f3e3c0 100644 --- a/src/argaze/utils/demo_data/demo_gaze_analysis_setup.json +++ b/src/argaze/utils/demo_data/demo_gaze_analysis_setup.json @@ -46,7 +46,8 @@ "n_max": 3 }, "Entropy":{} - } + }, + "logging_module": "main_layer_logger.py" } }, "image_parameters": { @@ -107,5 +108,6 @@ "color": [0, 255, 255], "size": 2 } - } + }, + "logging_module": "frame_logger.py" }
\ No newline at end of file diff --git a/src/argaze/utils/demo_data/frame_logger.py b/src/argaze/utils/demo_data/frame_logger.py new file mode 100644 index 0000000..f90e63f --- /dev/null +++ b/src/argaze/utils/demo_data/frame_logger.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python + +""" """ + +__author__ = "Théo de la Hogue" +__credits__ = [] +__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)" +__license__ = "BSD" + +from argaze import ArFeatures, GazeFeatures, DataFeatures +from argaze.utils import UtilsFeatures + +class FixationLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter): + + def on_look(self, timestamp, frame): + """Log fixations""" + + #print(timestamp, "FixationLogger.on_look:", type(self), type(frame)) + + # Log fixations + if GazeFeatures.is_fixation(frame.gaze_movement) and frame.gaze_movement.finished: + + log = ( + timestamp, + frame.gaze_movement.focus, + frame.gaze_movement.duration, + frame.layers['main_layer'].looked_aoi_name + ) + + self.write(log) + +class ScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter): + + def on_look(self, timestamp, frame): + """Log scan path metrics""" + + #print(timestamp, "ScanPathAnalysisLogger.on_look:", type(self), type(frame)) + + if frame.scan_path_analyzed: + + 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) + +# Export loggers instances to register them as pipeline step object observer +__loggers__ = { + "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") + }
\ No newline at end of file diff --git a/src/argaze/utils/demo_data/main_layer_logger.py b/src/argaze/utils/demo_data/main_layer_logger.py new file mode 100644 index 0000000..e85dfcd --- /dev/null +++ b/src/argaze/utils/demo_data/main_layer_logger.py @@ -0,0 +1,38 @@ +#!/usr/bin/env python + +""" """ + +__author__ = "Théo de la Hogue" +__credits__ = [] +__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)" +__license__ = "BSD" + +from argaze import ArFeatures, GazeFeatures +from argaze.utils import UtilsFeatures + +from argaze import ArFeatures, GazeFeatures, DataFeatures +from argaze.utils import UtilsFeatures + +class AOIScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter): + + def on_look(self, timestamp, layer): + """Log aoi scan path metrics""" + + #print(timestamp, "AOIScanPathAnalysisLogger.on_look:", type(self), type(layer)) + + if layer.aoi_scan_path_analyzed: + + 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 observer +__loggers__ = { + "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 9640d18..2dcc00e 100644 --- a/src/argaze/utils/demo_gaze_analysis_run.py +++ b/src/argaze/utils/demo_gaze_analysis_run.py @@ -37,14 +37,6 @@ def main(): # Load ArFrame ar_frame = ArFeatures.ArFrame.from_json(args.frame) - # Bind to a frame layer - main_layer = ar_frame.layers['main_layer'] - - # Create FileWriter loggers - fixation_logger = UtilsFeatures.FileWriter(path="_export/logs/fixations.csv", header="Timestamp (ms), Focus (px), Duration (ms), AOI") - scan_path_logger = UtilsFeatures.FileWriter(path="_export/logs/scan_path_metrics.csv", header="Timestamp (ms), Duration (ms), Step, K, NNI, XXR") - aoi_scan_path_logger = UtilsFeatures.FileWriter(path="_export/logs/aoi_scan_path_metrics.csv", header="Timestamp (ms), Duration (ms), Step, K, LZC") - # Create a window to display ArCamera cv2.namedWindow(ar_frame.name, cv2.WINDOW_AUTOSIZE) @@ -70,45 +62,6 @@ def main(): print('Gaze projection error:', e) - # Log fixations - if GazeFeatures.is_fixation(ar_frame.gaze_movement) and ar_frame.gaze_movement.finished: - - log = ( - timestamp, - ar_frame.gaze_movement.focus, - ar_frame.gaze_movement.duration, - ar_frame.layers['main_layer'].looked_aoi_name - ) - - fixation_logger.write(log) - - # Log scan path metrics - if ar_frame.scan_path_analyzed: - - log = ( - timestamp, - ar_frame.scan_path_analyzers['argaze.GazeAnalysis.Basic'].path_duration, - ar_frame.scan_path_analyzers['argaze.GazeAnalysis.Basic'].steps_number, - ar_frame.scan_path_analyzers['argaze.GazeAnalysis.KCoefficient'].K, - ar_frame.scan_path_analyzers['argaze.GazeAnalysis.NearestNeighborIndex'].nearest_neighbor_index, - ar_frame.scan_path_analyzers['argaze.GazeAnalysis.ExploreExploitRatio'].explore_exploit_ratio - ) - - scan_path_logger.write(log) - - # Log aoi scan path metrics - if main_layer.aoi_scan_path_analyzed: - - log = ( - timestamp, - main_layer.aoi_scan_path_analyzers['argaze.GazeAnalysis.Basic'].path_duration, - main_layer.aoi_scan_path_analyzers['argaze.GazeAnalysis.Basic'].steps_number, - main_layer.aoi_scan_path_analyzers['argaze.GazeAnalysis.KCoefficient'].K, - main_layer.aoi_scan_path_analyzers['argaze.GazeAnalysis.LempelZivComplexity'].lempel_ziv_complexity - ) - - aoi_scan_path_logger.write(log) - # Attach mouse callback to window cv2.setMouseCallback(ar_frame.name, on_mouse_event) |