diff options
Diffstat (limited to 'src/argaze/ArFeatures.py')
| -rw-r--r-- | src/argaze/ArFeatures.py | 64 |
1 files changed, 60 insertions, 4 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 |