aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/argaze/ArFeatures.py1085
-rw-r--r--src/argaze/AreaOfInterest/AOIFeatures.py53
-rw-r--r--src/argaze/DataFeatures.py314
-rw-r--r--src/argaze/GazeAnalysis/Basic.py14
-rw-r--r--src/argaze/GazeAnalysis/DeviationCircleCoverage.py25
-rw-r--r--src/argaze/GazeAnalysis/DispersionThresholdIdentification.py53
-rw-r--r--src/argaze/GazeAnalysis/Entropy.py32
-rw-r--r--src/argaze/GazeAnalysis/ExploreExploitRatio.py22
-rw-r--r--src/argaze/GazeAnalysis/FocusPointInside.py5
-rw-r--r--src/argaze/GazeAnalysis/KCoefficient.py14
-rw-r--r--src/argaze/GazeAnalysis/LempelZivComplexity.py8
-rw-r--r--src/argaze/GazeAnalysis/LinearRegression.py37
-rw-r--r--src/argaze/GazeAnalysis/NGram.py31
-rw-r--r--src/argaze/GazeAnalysis/NearestNeighborIndex.py23
-rw-r--r--src/argaze/GazeAnalysis/TransitionMatrix.py9
-rw-r--r--src/argaze/GazeAnalysis/VelocityThresholdIdentification.py41
-rw-r--r--src/argaze/GazeFeatures.py29
-rw-r--r--src/argaze/utils/Providers/__init__.py2
-rw-r--r--src/argaze/utils/__init__.py2
-rw-r--r--src/argaze/utils/demo_data/demo_aruco_markers_setup.json22
-rw-r--r--src/argaze/utils/demo_data/provider_setup.json2
21 files changed, 860 insertions, 963 deletions
diff --git a/src/argaze/ArFeatures.py b/src/argaze/ArFeatures.py
index ed152bf..0bf4a21 100644
--- a/src/argaze/ArFeatures.py
+++ b/src/argaze/ArFeatures.py
@@ -111,99 +111,179 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject):
Inherits from DataFeatures.SharedObject class to be shared by multiple threads.
"""
- def __init__(self, aoi_scene: AOIFeatures.AOIScene = None, aoi_matcher: GazeFeatures.AOIMatcher = None, aoi_scan_path: GazeFeatures.AOIScanPath = None, aoi_scan_path_analyzers: dict = None, draw_parameters: dict = None, **kwargs):
- """ Initialize ArLayer
-
- Parameters:
- aoi_scene: AOI scene description
- aoi_matcher: AOI matcher object
- 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
- """
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
+ """Initialize ArLayer."""
# Init parent classes
DataFeatures.SharedObject.__init__(self)
DataFeatures.PipelineStepObject.__init__(self, **kwargs)
# Init private attributes
- self.__aoi_scene = aoi_scene
- self.__aoi_matcher = aoi_matcher
- self.__aoi_scan_path = aoi_scan_path
- self.__aoi_scan_path_analyzers = aoi_scan_path_analyzers
- self.__draw_parameters = draw_parameters
+ self.__aoi_scene = None
+ self.__aoi_matcher = None
+ self.__aoi_scan_path = None
+ self.__aoi_scan_path_analyzers = []
+ self.__draw_parameters = DEFAULT_ARLAYER_DRAW_PARAMETERS
self.__gaze_movement = GazeFeatures.GazeMovement()
self.__looked_aoi_name = None
self.__aoi_scan_path_analyzed = False
- # Cast aoi scene to its effective dimension
- if self.__aoi_scene.dimension == 2:
+ @property
+ def aoi_scene(self) -> AOIFeatures.AOIScene:
+ """AOI scene description."""
+ return self.__aoi_scene
- self.__aoi_scene = AOI2DScene.AOI2DScene(self.__aoi_scene)
+ @aoi_scene.setter
+ def aoi_scene(self, aoi_scene_value: AOIFeatures.AOIScene|str|dict):
- elif self.__aoi_scene.dimension == 3:
+ if issubclass(type(aoi_scene_value), AOIFeatures.AOIScene):
- self.__aoi_scene = AOI3DScene.AOI3DScene(self.__aoi_scene)
+ new_aoi_scene = aoi_scene_value
- # Edit aoi_scan_path's expected aoi list by removing aoi with name equals to layer name
- if self.__aoi_scan_path is not None:
+ # str: relative path to file
+ elif type(aoi_scene_value) == str:
- expected_aoi = list(self.__aoi_scene.keys())
+ filepath = os.path.join(self.working_directory, aoi_scene_value)
+ file_format = filepath.split('.')[-1]
- if self.name in expected_aoi:
+ # JSON file format for 2D or 3D dimension
+ if file_format == 'json':
- expected_aoi.remove(self.name)
+ new_aoi_scene = AOIFeatures.AOIScene.from_json(filepath)
- self.__aoi_scan_path.expected_aoi = expected_aoi
+ # SVG file format for 2D dimension only
+ if file_format == 'svg':
- # Edit pipeline step objects parent
- if self.__aoi_scene is not None:
+ new_aoi_scene = AOI2DScene.AOI2DScene.from_svg(filepath)
- self.__aoi_scene.parent = self
+ # OBJ file format for 3D dimension only
+ elif file_format == 'obj':
- if self.__aoi_matcher is not None:
-
- self.__aoi_matcher.parent = self
+ new_aoi_scene = AOI3DScene.AOI3DScene.from_obj(filepath)
- if self.__aoi_scan_path is not None:
-
- self.__aoi_scan_path.parent = self
+ # dict:
+ elif type(aoi_scene_value) == dict:
- for name, analyzer in self.__aoi_scan_path_analyzers.items():
+ new_aoi_scene = AOIFeatures.AOIScene.from_dict(aoi_scene_value)
- analyzer.parent = self
+ # Cast aoi scene to its effective dimension
+ if new_aoi_scene.dimension == 2:
- @property
- def aoi_scene(self) -> AOIFeatures.AOIScene:
- """Get layer's aoi scene object."""
- return self.__aoi_scene
+ self.__aoi_scene = AOI2DScene.AOI2DScene(new_aoi_scene)
- @aoi_scene.setter
- def aoi_scene(self, aoi_scene: AOIFeatures.AOIScene):
- """Set layer's aoi scene object."""
- self.__aoi_scene = aoi_scene
+ elif new_aoi_scene.dimension == 3:
+
+ self.__aoi_scene = AOI3DScene.AOI3DScene(new_aoi_scene)
+
+ # Edit parent
+ if self.__aoi_scene is not None:
+
+ self.__aoi_scene.parent = self
@property
def aoi_matcher(self) -> GazeFeatures.AOIMatcher:
- """Get layer's aoi matcher object."""
+ """Select AOI matcher object."""
return self.__aoi_matcher
+
+ @aoi_matcher.setter
+ @DataFeatures.PipelineStepAttributeSetter
+ def aoi_matcher(self, aoi_matcher: GazeFeatures.AOIMatcher):
+
+ self.__aoi_matcher = aoi_matcher
+
+ # Edit parent
+ if self.__aoi_matcher is not None:
+
+ self.__aoi_matcher.parent = self
@property
def aoi_scan_path(self) -> GazeFeatures.AOIScanPath:
- """Get layer's aoi scan path object."""
+ """AOI scan path object."""
return self.__aoi_scan_path
+
+ @aoi_scan_path.setter
+ @DataFeatures.PipelineStepAttributeSetter
+ def aoi_scan_path(self, aoi_scan_path: GazeFeatures.AOIScanPath):
+
+ self.__aoi_scan_path = aoi_scan_path
+
+ # Edit aoi_scan_path's expected aoi list by removing aoi with name equals to layer name
+ expected_aoi = list(self.__aoi_scene.keys())
+
+ if self.name in expected_aoi:
+
+ expected_aoi.remove(self.name)
+
+ self.__aoi_scan_path.expected_aoi = expected_aoi
+
+ # Edit parent
+ if self.__aoi_scan_path is not None:
+
+ self.__aoi_scan_path.parent = self
@property
- def aoi_scan_path_analyzers(self) -> dict:
- """Get layer's aoi scan analyzers dictionary."""
+ def aoi_scan_path_analyzers(self) -> list:
+ """AOI scan path analyzers list."""
return self.__aoi_scan_path_analyzers
+
+ @aoi_scan_path_analyzers.setter
+ @DataFeatures.PipelineStepAttributeSetter
+ def aoi_scan_path_analyzers(self, aoi_scan_path_analyzers: list):
+
+ self.__aoi_scan_path_analyzers = aoi_scan_path_analyzers
+
+ # Connect analyzers if required
+ for analyzer in self.__aoi_scan_path_analyzers:
+
+ # DEBUG
+ print('ArLayer.aoi_scan_path_analyzers.setter type', type(analyzer))
+
+ # Check scan path analyzer parameters type
+ members = getmembers(analyzer)
+
+ for member in members:
+
+ if '__annotations__' in member:
+
+ for parameter_name, parameter_type in member[1].items():
+
+ # DEBUG
+ print('ArLayer.aoi_scan_path_analyzers.setter', parameter_name, parameter_type)
+
+ # Check if parameter is part of a package
+ if len(parameter_type.__module__.split('.')) > 1:
+
+ # Try get existing analyzer instance to append as parameter
+ try:
+
+ setattr(analyzer, parameter_name, self.__aoi_scan_path_analyzers[parameter_type.__module__])
+
+ 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:
+
+ self.scan_path = GazeFeatures.ScanPath()
+
+ # Edit parent
+ for analyzer in self.__aoi_scan_path_analyzers:
+
+ analyzer.parent = self
@property
def draw_parameters(self) -> dict:
- """Get layer's draw parameters dictionary."""
+ """Default draw method parameters dictionary."""
return self.__draw_parameters
+ @draw_parameters.setter
+ def draw_parameters(self, draw_parameters: dict):
+
+ self.__draw_parameters = draw_parameters
+
def last_looked_aoi_name(self) -> str:
"""Get last looked aoi name."""
return self.__looked_aoi_name
@@ -236,161 +316,6 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject):
"draw_parameters": self.__draw_parameters
}
- @classmethod
- def from_dict(cls, layer_data: dict, working_directory: str = None) -> ArLayerType:
- """Load ArLayer attributes from dictionary.
-
- Parameters:
- layer_data: dictionary with attributes to load
- 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 aoi scene
- try:
-
- new_aoi_scene_value = layer_data.pop('aoi_scene')
-
- # str: relative path to file
- if type(new_aoi_scene_value) == str:
-
- filepath = os.path.join(working_directory, new_aoi_scene_value)
- file_format = filepath.split('.')[-1]
-
- # JSON file format for 2D or 3D dimension
- if file_format == 'json':
-
- new_aoi_scene = AOIFeatures.AOIScene.from_json(filepath)
-
- # SVG file format for 2D dimension only
- if file_format == 'svg':
-
- new_aoi_scene = AOI2DScene.AOI2DScene.from_svg(filepath)
-
- # OBJ file format for 3D dimension only
- elif file_format == 'obj':
-
- new_aoi_scene = AOI3DScene.AOI3DScene.from_obj(filepath)
-
- # dict:
- else:
-
- new_aoi_scene = AOIFeatures.AOIScene.from_dict(new_aoi_scene_value)
-
- except KeyError:
-
- pass
-
- # Add AOI 2D Scene by default
- new_aoi_scene = AOI2DScene.AOI2DScene()
-
- # Load aoi matcher
- try:
-
- aoi_matcher_value = layer_data.pop('aoi_matcher')
-
- aoi_matcher_module_path, aoi_matcher_parameters = aoi_matcher_value.popitem()
-
- # Prepend argaze.GazeAnalysis path when a single name is provided
- if len(aoi_matcher_module_path.split('.')) == 1:
- aoi_matcher_module_path = f'argaze.GazeAnalysis.{aoi_matcher_module_path}'
-
- aoi_matcher_module = importlib.import_module(aoi_matcher_module_path)
- new_aoi_matcher = aoi_matcher_module.AOIMatcher(**aoi_matcher_parameters)
-
- except KeyError:
-
- new_aoi_matcher = None
-
- # Load AOI scan path
- try:
-
- new_aoi_scan_path_data = layer_data.pop('aoi_scan_path')
- new_aoi_scan_path = GazeFeatures.AOIScanPath(**new_aoi_scan_path_data)
-
- except KeyError:
-
- new_aoi_scan_path_data = {}
- new_aoi_scan_path = None
-
- # Load AOI scan path analyzers
- new_aoi_scan_path_analyzers = {}
-
- try:
-
- new_aoi_scan_path_analyzers_value = layer_data.pop('aoi_scan_path_analyzers')
-
- for aoi_scan_path_analyzer_module_path, aoi_scan_path_analyzer_parameters in new_aoi_scan_path_analyzers_value.items():
-
- # Prepend argaze.GazeAnalysis path when a single name is provided
- if len(aoi_scan_path_analyzer_module_path.split('.')) == 1:
- aoi_scan_path_analyzer_module_path = f'argaze.GazeAnalysis.{aoi_scan_path_analyzer_module_path}'
-
- aoi_scan_path_analyzer_module = importlib.import_module(aoi_scan_path_analyzer_module_path)
-
- # Check aoi scan path analyzer parameters type
- members = getmembers(aoi_scan_path_analyzer_module.AOIScanPathAnalyzer)
-
- for member in members:
-
- if '__annotations__' in member:
-
- for parameter, parameter_type in member[1].items():
-
- # Check if parameter is part of argaze.GazeAnalysis module
- parameter_module_path = parameter_type.__module__.split('.')
-
- # Check if parameter is part of a package
- if len(parameter_type.__module__.split('.')) > 1:
-
- # Try get existing analyzer instance to append as parameter
- try:
-
- aoi_scan_path_analyzer_parameters[parameter] = new_aoi_scan_path_analyzers[parameter_type.__module__]
-
- except KeyError:
-
- raise LoadingFailed(f'{aoi_scan_path_analyzer_module_path} aoi scan path analyzer loading fails because {parameter_type.__module__} aoi scan path analyzer is missing.')
-
- aoi_scan_path_analyzer = aoi_scan_path_analyzer_module.AOIScanPathAnalyzer(**aoi_scan_path_analyzer_parameters)
-
- new_aoi_scan_path_analyzers[aoi_scan_path_analyzer_module_path] = aoi_scan_path_analyzer
-
- # Force AOI scan path creation
- if len(new_aoi_scan_path_analyzers) > 0 and new_aoi_scan_path == None:
-
- new_aoi_scan_path = GazeFeatures.AOIScanPath(**new_aoi_scan_path_data)
-
- except KeyError:
-
- pass
-
- # Load image parameters
- try:
-
- new_layer_draw_parameters = layer_data.pop('draw_parameters')
-
- except KeyError:
-
- new_layer_draw_parameters = DEFAULT_ARLAYER_DRAW_PARAMETERS
-
- # Load temporary pipeline step object from layer_data then export it as dict
- temp_pipeline_step_object_data = DataFeatures.PipelineStepObject.from_dict(layer_data, working_directory).as_dict()
-
- # Create layer
- return ArLayer( \
- new_aoi_scene, \
- new_aoi_matcher, \
- new_aoi_scan_path, \
- new_aoi_scan_path_analyzers, \
- new_layer_draw_parameters, \
- **temp_pipeline_step_object_data \
- )
-
@DataFeatures.PipelineStepMethod
def look(self, gaze_movement: GazeFeatures.GazePosition = GazeFeatures.GazePosition()):
"""
@@ -436,7 +361,7 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject):
if aoi_scan_step is not None and len(self.__aoi_scan_path) > 1:
# Analyze aoi scan path
- for aoi_scan_path_analyzer_module_path, aoi_scan_path_analyzer in self.__aoi_scan_path_analyzers.items():
+ for aoi_scan_path_analyzer in self.__aoi_scan_path_analyzers:
aoi_scan_path_analyzer.analyze(self.__aoi_scan_path, timestamp=gaze_movement.timestamp)
@@ -506,133 +431,237 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject):
Inherits from DataFeatures.SharedObject class to be shared by multiple threads
"""
- def __init__(self, size: tuple[int] = (1, 1), provider: DataFeatures.PipelineInputProvider = None, gaze_position_calibrator: GazeFeatures.GazePositionCalibrator = None, gaze_movement_identifier: GazeFeatures.GazeMovementIdentifier = None, filter_in_progress_identification: bool = True, scan_path: GazeFeatures.ScanPath = None, scan_path_analyzers: dict = None, background: numpy.array = numpy.array([]), heatmap: AOIFeatures.Heatmap = None, layers: dict = None, image_parameters: dict = DEFAULT_ARFRAME_IMAGE_PARAMETERS, **kwargs):
- """ Initialize ArFrame
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
+ """ Initialize ArFrame."""
- Parameters:
- size: defines the dimension of the rectangular area where gaze positions are projected
- provider: provider object related to this frame
- gaze_position_calibrator: gaze position calibration algoritm
- gaze_movement_identifier: gaze movement identification algorithm
- filter_in_progress_identification: ignore in progress gaze movement identification
- scan_path: scan path object
- scan_path_analyzers: dictionary of scan path analyzers
- background: picture to draw behind
- heatmap: heatmap object
- layers: dictionary of AOI layers
- image_parameters: default parameters passed to image method
- """
+ # DEBUG
+ print('ArFrame.__init__')
# Init parent classes
DataFeatures.SharedObject.__init__(self)
DataFeatures.PipelineStepObject.__init__(self, **kwargs)
# Init private attributes
- self.__size = size
- self.__provider = provider
- self.__gaze_position_calibrator = gaze_position_calibrator
- self.__gaze_movement_identifier = gaze_movement_identifier
- self.__filter_in_progress_identification = filter_in_progress_identification
- self.__scan_path = scan_path
- self.__scan_path_analyzers = scan_path_analyzers
- self.__background = background
- self.__heatmap = heatmap
- self.__layers = layers
- self.__image_parameters = image_parameters
+ self.__size = (0, 0)
+ 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.__heatmap = None
+ self.__layers = {}
+ self.__image_parameters = DEFAULT_ARFRAME_IMAGE_PARAMETERS
self.__calibrated_gaze_position = GazeFeatures.GazePosition()
self.__identified_gaze_movement = GazeFeatures.GazeMovement()
self.__scan_path_analyzed = False
- # Edit pipeline step objects parent
- if self.__provider is not None:
-
- self.__provider.parent = self
-
- if self.__gaze_position_calibrator is not None:
-
- self.__gaze_position_calibrator.parent = self
-
- if self.__gaze_movement_identifier is not None:
-
- self.__gaze_movement_identifier.parent = self
-
- if self.__scan_path is not None:
-
- self.__scan_path.parent = self
-
- for name, analyzer in self.__scan_path_analyzers.items():
-
- analyzer.parent = self
-
- if self.__heatmap is not None:
-
- self.__heatmap.parent = self
-
- for name, layer in self.__layers.items():
-
- layer.parent = self
-
@property
def size(self) -> tuple[int]:
- """Get frame's size."""
+ """Defines the dimension of the rectangular area where gaze positions are projected."""
return self.__size
+ @size.setter
+ def size(self, size: tuple[int] = (1, 1)):
+ self.__size = size
+
@property
def provider(self) -> DataFeatures.PipelineInputProvider:
- """Get frame's provider."""
+ """Provider object related to this frame."""
return self.__provider
+
+ @provider.setter
+ @DataFeatures.PipelineStepAttributeSetter
+ def provider(self, provider: DataFeatures.PipelineInputProvider):
+
+ # DEBUG
+ print('ArFrame.provider.setter', provider)
+
+ self.__provider = provider
+
+ # Edit parent
+ if self.__provider is not None:
+
+ self.__provider.parent = self
@property
def gaze_position_calibrator(self) -> GazeFeatures.GazePositionCalibrator:
- """Get frame's gaze position calibrator object."""
+ """Select gaze position calibration algoritm."""
return self.__gaze_position_calibrator
+
+ @gaze_position_calibrator.setter
+ @DataFeatures.PipelineStepAttributeSetter
+ def gaze_position_calibrator(self, gaze_position_calibrator:GazeFeatures.GazePositionCalibrator):
+
+ self.__gaze_position_calibrator = gaze_position_calibrator
+
+ # Edit parent
+ if self.__gaze_position_calibrator is not None:
+
+ self.__gaze_position_calibrator.parent = self
@property
def gaze_movement_identifier(self) -> GazeFeatures.GazeMovementIdentifier:
- """Get frame's gaze movement identifier object."""
+ """Select gaze movement identification algorithm."""
return self.__gaze_movement_identifier
+
+ @gaze_movement_identifier.setter
+ @DataFeatures.PipelineStepAttributeSetter
+ def gaze_movement_identifier(self, gaze_movement_identifier: GazeFeatures.GazeMovementIdentifier):
+
+ self.__gaze_movement_identifier = gaze_movement_identifier
+
+ # Edit parent
+ if self.__gaze_movement_identifier is not None:
+
+ self.__gaze_movement_identifier.parent = self
@property
def filter_in_progress_identification(self) -> bool:
- """Is frame filtering in progress identification?"""
+ """Is frame ignores in progress gaze movement identification?"""
return self.__filter_in_progress_identification
+ @filter_in_progress_identification.setter
+ @DataFeatures.PipelineStepAttributeSetter
+ def filter_in_progress_identification(self, filter_in_progress_identification: bool = True):
+
+ self.__filter_in_progress_identification = filter_in_progress_identification
+
@property
def scan_path(self) -> GazeFeatures.ScanPath:
- """Get frame's scan path object."""
+ """Scan path object."""
return self.__scan_path
+ @scan_path.setter
+ @DataFeatures.PipelineStepAttributeSetter
+ def scan_path(self, scan_path: GazeFeatures.ScanPath) -> GazeFeatures.ScanPath:
+
+ self.__scan_path = scan_path
+
+ # Edit parent
+ if self.__scan_path is not None:
+
+ self.__scan_path.parent = self
+
@property
- def scan_path_analyzers(self) -> dict:
- """Get frame's scan path analyzers dictionary."""
+ def scan_path_analyzers(self) -> list:
+ """Scan path analyzers list."""
return self.__scan_path_analyzers
+ @scan_path_analyzers.setter
+ @DataFeatures.PipelineStepAttributeSetter
+ def scan_path_analyzers(self, scan_path_analyzers: list):
+
+ self.__scan_path_analyzers = scan_path_analyzers
+
+ # Connect analyzers if required
+ for analyzer in self.__scan_path_analyzers:
+
+ # DEBUG
+ print('ArFrame.scan_path_analyzers.setter type', type(analyzer))
+
+ # Check scan path analyzer parameters type
+ members = getmembers(analyzer)
+
+ for member in members:
+
+ if '__annotations__' in member:
+
+ for parameter_name, parameter_type in member[1].items():
+
+ # DEBUG
+ print('ArFrame.scan_path_analyzers.setter', parameter_name, parameter_type)
+
+ # Check if parameter is part of a package
+ if len(parameter_type.__module__.split('.')) > 1:
+
+ # Try get existing analyzer instance to append as parameter
+ try:
+
+ setattr(analyzer, parameter_name, self.__scan_path_analyzers[parameter_type.__module__])
+
+ except KeyError:
+
+ raise LoadingFailed(f'{module_path} scan path analyzer loading fails because {parameter_type.__module__} scan path analyzer is missing.')
+
+ # Force scan path creation
+ if len(self.__scan_path_analyzers) > 0 and self.scan_path == None:
+
+ self.scan_path = GazeFeatures.ScanPath()
+
+ # Edit parent
+ for analyzer in self.__scan_path_analyzers:
+
+ analyzer.parent = self
+
@property
def background(self) -> numpy.array:
- """Get frame's background matrix."""
+ """Picture to draw behind."""
return self.__background
@background.setter
- def background(self, image: numpy.array):
- """Set frame's background matrix."""
- self.__background = image
+ @DataFeatures.PipelineStepAttributeSetter
+ def background(self, background: numpy.array):
+
+ # DEBUG
+ print('ArFrame.background.setter', background)
+
+ # Resize image to frame size
+ self.__background = cv2.resize(background, dsize = self.size, interpolation = cv2.INTER_CUBIC)
@property
def heatmap(self) -> AOIFeatures.Heatmap:
- """Get frame's heatmap object."""
+ """Heatmap object."""
return self.__heatmap
+
+ @heatmap.setter
+ @DataFeatures.PipelineStepAttributeSetter
+ def heatmap(self, heatmap: AOIFeatures.Heatmap):
+
+ self.__heatmap = heatmap
+
+ # Default heatmap size equals frame size
+ if self.__heatmap.size == (1, 1):
+
+ self.__heatmap.size = self.size
+
+ # Edit parent
+ if self.__heatmap is not None:
+
+ self.__heatmap.parent = self
@property
def layers(self) -> dict:
- """Get frame's layers dictionary."""
+ """Layers dictionary."""
return self.__layers
+
+ @layers.setter
+ def layers(self, layers: dict):
+
+ self.__layers = {}
+
+ for layer_name, layer_data in layers.items():
+
+ self.__layers[layer_name] = ArLayer(working_directory = self.working_directory, name = layer_name, **layer_data)
+
+ # Edit parent
+ for name, layer in self.__layers.items():
+
+ layer.parent = self
@property
def image_parameters(self) -> dict:
- """Get frame's image parameters dictionary."""
+ """Default image method parameters dictionary."""
return self.__image_parameters
+ @image_parameters.setter
+ def image_parameters(self, image_parameters: dict = DEFAULT_ARFRAME_IMAGE_PARAMETERS):
+
+ self.__image_parameters = image_parameters
+
def last_gaze_position(self) -> object:
"""Get last calibrated gaze position"""
return self.__calibrated_gaze_position
@@ -680,256 +709,6 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject):
return d
- @classmethod
- def from_dict(cls, frame_data: dict, working_directory: str = None) -> ArFrameType:
- """Load ArFrame attributes from dictionary.
-
- Parameters:
- frame_data: dictionary with attributes to load
- 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 size
- try:
-
- new_frame_size = frame_data.pop('size')
-
- except KeyError:
-
- new_frame_size = (0, 0)
-
- # Load provider
- try:
-
- provider_value = frame_data.pop('provider')
-
- # str: relative path to file
- if type(provider_value) == str:
-
- filepath = os.path.join(working_directory, provider_value)
- file_format = filepath.split('.')[-1]
-
- # JSON file format
- if file_format == 'json':
-
- with open(filepath) as file:
-
- provider_value = json.load(file)
-
- # Create gaze position calibrator
- provider_module_path, provider_parameters = provider_value.popitem()
-
- # Prepend argaze.utils.Providers path when a single name is provided
- if len(provider_module_path.split('.')) == 1:
- provider_module_path = f'argaze.utils.Providers.{provider_module_path}'
-
- provider_module = importlib.import_module(provider_module_path)
- new_provider = provider_module.Provider(**provider_parameters)
-
- except KeyError:
-
- new_provider = None
-
- # Load gaze position calibrator
- try:
-
- gaze_position_calibrator_value = frame_data.pop('gaze_position_calibrator')
-
- # str: relative path to file
- if type(gaze_position_calibrator_value) == str:
-
- filepath = os.path.join(working_directory, gaze_position_calibrator_value)
- file_format = filepath.split('.')[-1]
-
- # JSON file format
- if file_format == 'json':
-
- with open(filepath) as file:
-
- gaze_movement_calibrator_value = json.load(file)
-
- # Create gaze position calibrator
- gaze_position_calibrator_module_path, gaze_position_calibrator_parameters = gaze_movement_calibrator_value.popitem()
-
- # Prepend argaze.GazeAnalysis path when a single name is provided
- if len(gaze_position_calibrator_module_path.split('.')) == 1:
- gaze_position_calibrator_module_path = f'argaze.GazeAnalysis.{gaze_position_calibrator_module_path}'
-
- gaze_position_calibrator_module = importlib.import_module(gaze_position_calibrator_module_path)
- new_gaze_position_calibrator = gaze_position_calibrator_module.GazePositionCalibrator(**gaze_position_calibrator_parameters)
-
- except KeyError:
-
- new_gaze_position_calibrator = None
-
- # Load gaze movement identifier
- try:
-
- gaze_movement_identifier_value = frame_data.pop('gaze_movement_identifier')
-
- gaze_movement_identifier_module_path, gaze_movement_identifier_parameters = gaze_movement_identifier_value.popitem()
-
- # Prepend argaze.GazeAnalysis path when a single name is provided
- if len(gaze_movement_identifier_module_path.split('.')) == 1:
- gaze_movement_identifier_module_path = f'argaze.GazeAnalysis.{gaze_movement_identifier_module_path}'
-
- gaze_movement_identifier_module = importlib.import_module(gaze_movement_identifier_module_path)
- new_gaze_movement_identifier = gaze_movement_identifier_module.GazeMovementIdentifier(**gaze_movement_identifier_parameters)
-
- except KeyError:
-
- new_gaze_movement_identifier = None
-
- # Current fixation matching
- try:
-
- filter_in_progress_identification = frame_data.pop('filter_in_progress_identification')
-
- except KeyError:
-
- filter_in_progress_identification = True
-
- # Load scan path
- try:
-
- new_scan_path_data = frame_data.pop('scan_path')
- new_scan_path = GazeFeatures.ScanPath(**new_scan_path_data)
-
- except KeyError:
-
- new_scan_path_data = {}
- new_scan_path = None
-
- # Load scan path analyzers
- new_scan_path_analyzers = {}
-
- try:
-
- new_scan_path_analyzers_value = frame_data.pop('scan_path_analyzers')
-
- for scan_path_analyzer_module_path, scan_path_analyzer_parameters in new_scan_path_analyzers_value.items():
-
- # Prepend argaze.GazeAnalysis path when a single name is provided
- if len(scan_path_analyzer_module_path.split('.')) == 1:
- scan_path_analyzer_module_path = f'argaze.GazeAnalysis.{scan_path_analyzer_module_path}'
-
- scan_path_analyzer_module = importlib.import_module(scan_path_analyzer_module_path)
-
- # Check scan path analyzer parameters type
- members = getmembers(scan_path_analyzer_module.ScanPathAnalyzer)
-
- for member in members:
-
- if '__annotations__' in member:
-
- for parameter, parameter_type in member[1].items():
-
- # Check if parameter is part of a package
- if len(parameter_type.__module__.split('.')) > 1:
-
- # Try get existing analyzer instance to append as parameter
- try:
-
- scan_path_analyzer_parameters[parameter] = new_scan_path_analyzers[parameter_type.__module__]
-
- except KeyError:
-
- raise LoadingFailed(f'{scan_path_analyzer_module_path} scan path analyzer loading fails because {parameter_type.__module__} scan path analyzer is missing.')
-
- scan_path_analyzer = scan_path_analyzer_module.ScanPathAnalyzer(**scan_path_analyzer_parameters)
-
- new_scan_path_analyzers[scan_path_analyzer_module_path] = scan_path_analyzer
-
- # Force scan path creation
- if len(new_scan_path_analyzers) > 0 and new_scan_path == None:
-
- new_scan_path = GazeFeatures.ScanPath(**new_scan_path_data)
-
- except KeyError:
-
- pass
-
- # Load background image
- try:
-
- new_frame_background_value = frame_data.pop('background')
- new_frame_background = cv2.imread(os.path.join(working_directory, new_frame_background_value))
- new_frame_background = cv2.resize(new_frame_background, dsize=new_frame_size, interpolation=cv2.INTER_CUBIC)
-
- except KeyError:
-
- new_frame_background = numpy.full((new_frame_size[1], new_frame_size[0], 3), 127).astype(numpy.uint8)
-
- # Load heatmap
- try:
-
- new_heatmap_data = frame_data.pop('heatmap')
-
- # Default heatmap size equals frame size
- if 'size' not in new_heatmap_data.keys():
-
- new_heatmap_data['size'] = new_frame_size
-
- new_heatmap = AOIFeatures.Heatmap(**new_heatmap_data)
-
- except KeyError:
-
- new_heatmap_data = {}
- new_heatmap = None
-
- # Load layers
- new_layers = {}
-
- try:
-
- for layer_name, layer_data in frame_data.pop('layers').items():
-
- # Append name
- layer_data['name'] = layer_name
-
- # Create layer
- new_layer = ArLayer.from_dict(layer_data, working_directory)
-
- # Append new layer
- new_layers[layer_name] = new_layer
-
- except KeyError:
-
- pass
-
- # Load image parameters
- try:
-
- new_frame_image_parameters = frame_data.pop('image_parameters')
-
- except KeyError:
-
- new_frame_image_parameters = DEFAULT_ARFRAME_IMAGE_PARAMETERS
-
- # Load temporary pipeline step object from frame_data then export it as dict
- temp_pipeline_step_object_data = DataFeatures.PipelineStepObject.from_dict(frame_data, working_directory).as_dict()
-
- # Create frame
- return ArFrame( \
- new_frame_size, \
- new_provider, \
- new_gaze_position_calibrator, \
- new_gaze_movement_identifier, \
- filter_in_progress_identification, \
- new_scan_path, \
- new_scan_path_analyzers, \
- new_frame_background, \
- new_heatmap, \
- new_layers, \
- new_frame_image_parameters, \
- **temp_pipeline_step_object_data \
- )
-
@DataFeatures.PipelineStepMethod
def look(self, timestamped_gaze_position: GazeFeatures.GazePosition = GazeFeatures.GazePosition()) -> Iterator[Union[object, type, dict]]:
"""
@@ -988,7 +767,7 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject):
if scan_step and len(self.__scan_path) > 1:
# Analyze aoi scan path
- for scan_path_analyzer_module_path, scan_path_analyzer in self.__scan_path_analyzers.items():
+ for scan_path_analyzer in self.__scan_path_analyzers:
scan_path_analyzer.analyze(self.__scan_path, timestamp=self.__identified_gaze_movement.timestamp)
@@ -1112,62 +891,98 @@ class ArScene(DataFeatures.PipelineStepObject):
Define abstract Augmented Reality scene with ArLayers and ArFrames inside.
"""
- def __init__(self, layers: dict = None, frames: dict = None, angle_tolerance: float = 0., distance_tolerance: float = 0., **kwargs):
- """ Initialize ArScene
-
- Parameters:
- layers: dictionary of ArLayers to project once the pose is estimated: see [project][argaze.ArFeatures.ArScene.project] function below.
- frames: dictionary to ArFrames to project once the pose is estimated: see [project][argaze.ArFeatures.ArScene.project] function below.
- angle_tolerance: Optional angle error tolerance to validate marker pose in degree used into [estimate_pose][argaze.ArFeatures.ArScene.estimate_pose] function.
- distance_tolerance: Optional distance error tolerance to validate marker pose in centimeter used into [estimate_pose][argaze.ArFeatures.ArScene.estimate_pose] function.
- """
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
+ """Initialize ArScene"""
# Init parent classes
super().__init__(**kwargs)
# Init private attributes
- self.__layers = layers
- self.__frames = frames
- self.__angle_tolerance = angle_tolerance
- self.__distance_tolerance = distance_tolerance
+ self.__layers = {}
+ self.__frames = {}
+ self.__angle_tolerance = 0,
+ self.__distance_tolerance = 0,
- # Edit pipeline step objects parent
- for name, layer in self.__layers.items():
+ @property
+ def layers(self) -> dict:
+ """Dictionary of ArLayers to project once the pose is estimated.
+ See [project][argaze.ArFeatures.ArScene.project] function below."""
+ return self.__layers
- layer.parent = self
+ @layers.setter
+ def layers(self, layers:dict):
- for name, frame in self.__frames.items():
+ self.__layers = {}
- frame.parent = self
+ for layer_name, layer_data in layers.items():
- @property
- def layers(self) -> dict:
- """Get scene's layers dictionary."""
- return self.__layers
+ self.__layers[layer_name] = ArLayer(working_directory = self.working_directory, name = layer_name, **layer_data)
+
+ # Edit parent
+ for name, layer in self.__layers.items():
+
+ layer.parent = self
@property
def frames(self) -> dict:
- """Get scene's frames dictionary."""
+ """Dictionary of ArFrames to project once the pose is estimated.
+ See [project][argaze.ArFeatures.ArScene.project] function below."""
return self.__frames
+
+ @frames.setter
+ def frames(self, frames: dict):
+
+ self.__frames = {}
+
+ for frame_name, frame_data in frames.items():
+
+ 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():
+
+ try:
+
+ frame_3d = scene_layer.aoi_scene[frame_name]
+
+ # Check that the frame have a layer named like this scene layer
+ aoi_2d_scene = new_frame.layers[scene_layer_name].aoi_scene
+
+ # Transform 2D frame layer AOI into 3D scene layer AOI
+ # Then, add them to scene layer
+ scene_layer.aoi_scene |= aoi_2d_scene.dimensionalize(frame_3d, new_frame.size)
+
+ except KeyError as e:
+
+ print('ArScene.from_dict: KeyError', e)
+
+ # Append new frame
+ self.__frames[frame_name] = new_frame
+
+ # Edit parent
+ for name, frame in self.__frames.items():
+
+ frame.parent = self
@property
def angle_tolerance(self) -> float:
- """Get scene's angle tolerance."""
+ """Angle error tolerance to validate marker pose in degree used into [estimate_pose][argaze.ArFeatures.ArScene.estimate_pose] function."""
return self.__angle_tolerance
@angle_tolerance.setter
def angle_tolerance(self, value: float):
- """Set scene's angle tolerance."""
+
self.__angle_tolerance = value
@property
def distance_tolerance(self) -> float:
- """Get scene's distance tolerance."""
+ """Distance error tolerance to validate marker pose in centimeter used into [estimate_pose][argaze.ArFeatures.ArScene.estimate_pose] function."""
return self.__distance_tolerance
@distance_tolerance.setter
def distance_tolerance(self, value: float):
- """Set scene's distance tolerance."""
+
self.__distance_tolerance = value
def as_dict(self) -> dict:
@@ -1181,126 +996,6 @@ class ArScene(DataFeatures.PipelineStepObject):
"distance_tolerance": self.__distance_tolerance
}
- @classmethod
- def from_dict(cls, scene_data: dict, working_directory: str = None) -> ArSceneType:
- """
- Load ArScene attributes from dictionary.
-
- Parameters:
- scene_data: dictionary
- working_directory: folder path where to load files when a dictionary value is a relative filepath.
- """
-
- # Load layers
- new_layers = {}
-
- try:
-
- for layer_name, layer_data in scene_data.pop('layers').items():
-
- # Append name
- layer_data['name'] = layer_name
-
- # Create layer
- new_layer = ArLayer.from_dict(layer_data, working_directory)
-
- # Append new layer
- new_layers[layer_name] = new_layer
-
- except KeyError:
-
- pass
-
- # Load frames
- new_frames = {}
-
- try:
-
- for frame_name, frame_data in scene_data.pop('frames').items():
-
- # str: relative path to file
- if type(frame_data) == str:
-
- filepath = os.path.join(working_directory, frame_data)
- file_format = filepath.split('.')[-1]
-
- # JSON file format for 2D or 3D dimension
- if file_format == 'json':
-
- new_frame = ArFrame.from_json(filepath)
-
- # dict:
- else:
-
- # Append name
- frame_data['name'] = frame_name
-
- new_frame = ArFrame.from_dict(frame_data, working_directory)
-
- # Look for a scene layer with an AOI named like the frame
- for scene_layer_name, scene_layer in new_layers.items():
-
- try:
-
- frame_3d = scene_layer.aoi_scene[frame_name]
-
- # Check that the frame have a layer named like this scene layer
- aoi_2d_scene = new_frame.layers[scene_layer_name].aoi_scene
-
- # Transform 2D frame layer AOI into 3D scene layer AOI
- # Then, add them to scene layer
- scene_layer.aoi_scene |= aoi_2d_scene.dimensionalize(frame_3d, new_frame.size)
-
- '''DEPRECATED: but maybe still usefull?
- # Project and reframe each layers into corresponding frame layers
- for frame_layer_name, frame_layer in new_frame.layers.items():
-
- try:
-
- layer = new_layers[frame_layer_name]
-
- layer_aoi_scene_projection = layer.aoi_scene.orthogonal_projection
- aoi_frame_projection = layer_aoi_scene_projection[frame_name]
-
- frame_layer.aoi_scene = layer_aoi_scene_projection.reframe(aoi_frame_projection, new_frame.size)
-
- if frame_layer.aoi_scan_path is not None:
-
- # Edit expected AOI list by removing AOI with name equals to frame layer name
- expected_aoi = list(layer.aoi_scene.keys())
-
- if frame_layer_name in expected_aoi:
- expected_aoi.remove(frame_layer_name)
-
- frame_layer.aoi_scan_path.expected_aoi = expected_aoi
-
- except KeyError:
-
- continue
- '''
-
- except KeyError as e:
-
- print('ArScene.from_dict: KeyError', e)
-
- # Append new frame
- new_frames[frame_name] = new_frame
-
- except KeyError:
-
- pass
-
- # Load temporary pipeline step object from scene_data then export it as dict
- temp_pipeline_step_object_data = DataFeatures.PipelineStepObject.from_dict(scene_data, working_directory).as_dict()
-
- # Create scene
- return ArScene( \
- new_layers, \
- new_frames, \
- **scene_data, \
- **temp_pipeline_step_object_data \
- )
-
@DataFeatures.PipelineStepMethod
def estimate_pose(self, detected_features: any) -> Tuple[numpy.array, numpy.array, any]:
"""Define abstract estimate scene pose method.
@@ -1370,27 +1065,17 @@ class ArCamera(ArFrame):
Define abstract Augmented Reality camera as ArFrame with ArScenes inside.
"""
- def __init__(self, scenes: dict = None, visual_hfov: float = 0., visual_vfov: float = 0., **kwargs):
- """ Initialize ArCamera
-
- Parameters:
- scenes: all scenes to project into camera frame
- visual_hfov: Optional angle in degree to clip scenes projection according visual horizontal field of view (HFOV).
- visual_vfov: Optional angle in degree to clip scenes projection according visual vertical field of view (VFOV).
- """
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
+ """Initialize ArCamera."""
# Init parent class
super().__init__(**kwargs)
# Init private attributes
- self.__scenes = scenes
- self.__visual_hfov = visual_hfov
- self.__visual_vfov = visual_vfov
-
- # Edit pipeline step objects parent
- for name, scene in self.__scenes.items():
-
- scene.parent = self
+ self.__scenes = {}
+ self.__visual_hfov = 0.
+ self.__visual_vfov = 0.
# Setup expected aoi of each layer aoi scan path with the aoi of corresponding scene layer
# Edit aoi matcher exclude attribute to ignore frame aoi
@@ -1435,12 +1120,26 @@ class ArCamera(ArFrame):
@property
def scenes(self) -> dict:
- """Get camera's scenes dictionary."""
+ """All scenes to project into camera frame."""
return self.__scenes
+ @scenes.setter
+ def scenes(self, scenes: dict):
+
+ self.__scenes = {}
+
+ for scene_name, scene_data in scenes.items():
+
+ self.__scenes[scene_name] = ArScene(working_directory = self.working_directory, name = scene_name, **scene_data)
+
+ # Edit parent
+ for name, scene in self.__scenes.items():
+
+ scene.parent = self
+
@property
def visual_hfov(self) -> float:
- """Get camera's visual horizontal field of view."""
+ """Angle in degree to clip scenes projection according visual horizontal field of view (HFOV)."""
return self.__visual_hfov
@visual_hfov.setter
@@ -1450,7 +1149,7 @@ class ArCamera(ArFrame):
@property
def visual_vfov(self) -> float:
- """Get camera's visual vertical field of view."""
+ """Angle in degree to clip scenes projection according visual vertical field of view (VFOV)."""
return self.__visual_vfov
@visual_vfov.setter
diff --git a/src/argaze/AreaOfInterest/AOIFeatures.py b/src/argaze/AreaOfInterest/AOIFeatures.py
index a396b29..31da8f4 100644
--- a/src/argaze/AreaOfInterest/AOIFeatures.py
+++ b/src/argaze/AreaOfInterest/AOIFeatures.py
@@ -557,24 +557,29 @@ class AOIScene():
HeatmapType = TypeVar('Heatmap', bound="Heatmap")
# Type definition for type annotation convenience
-@dataclass
class Heatmap(DataFeatures.PipelineStepObject):
"""Define image to draw heatmap."""
- size: tuple = field(default=(1, 1))
- """Size of heatmap image in pixels."""
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- buffer: int = field(default=0)
- """Size of heatmap buffer (0 means no buffering)."""
-
- sigma: float = field(default=0.05)
- """Point spread factor."""
+ super().__init__()
- def __post_init__(self):
+ # Init private attributes
+ self.__size = (1, 1)
+ self.__buffer = 0
+ self.__sigma = 0.05
- super().__init__()
+ @property
+ def size(self) -> tuple[int, int]:
+ """Size of heatmap image in pixels."""
+ return self.__size
- self.__rX, self.__rY = self.size
+ @size.setter
+ def size(self, size: tuple[int, int]):
+
+ self.__size = size
+ self.__rX, self.__rY = size
# Init coordinates
self.__Sx = numpy.linspace(0., self.__rX/self.__rY, self.__rX)
@@ -583,10 +588,30 @@ class Heatmap(DataFeatures.PipelineStepObject):
# Init heatmap image
self.clear()
+ @property
+ def sigma(self) -> float:
+ """Point spread factor."""
+ return self.__sigma
+
+ @sigma.setter
+ def sigma(self, sigma: float):
+
+ self.__sigma = sigma
+
+ @property
+ def buffer(self) -> int:
+ """Size of heatmap buffer (0 means no buffering)."""
+ return self.__buffer
+
+ @buffer.setter
+ def buffer(self, buffer: int):
+
+ self.__buffer = buffer
+
def point_spread(self, point: tuple):
"""Draw gaussian point spread into image."""
- div = -2 * self.sigma**2
+ div = -2 * self.__sigma**2
x = point[0] / self.__rY # we use rY not rX !!!
y = point[1] / self.__rY
@@ -603,7 +628,7 @@ class Heatmap(DataFeatures.PipelineStepObject):
self.__point_spread_sum = numpy.zeros((self.__rY, self.__rX))
self.__point_spread_buffer = []
- self.__point_spread_buffer_size = self.buffer
+ self.__point_spread_buffer_size = self.__buffer
@DataFeatures.PipelineStepMethod
def update(self, point: tuple):
@@ -615,7 +640,7 @@ class Heatmap(DataFeatures.PipelineStepObject):
self.__point_spread_sum += point_spread
# If point spread buffering enabled
- if self.buffer > 0:
+ if self.__buffer > 0:
self.__point_spread_buffer.append(point_spread)
diff --git a/src/argaze/DataFeatures.py b/src/argaze/DataFeatures.py
index 1334961..677a179 100644
--- a/src/argaze/DataFeatures.py
+++ b/src/argaze/DataFeatures.py
@@ -32,6 +32,7 @@ import time
import pandas
import numpy
+import cv2
import matplotlib.pyplot as mpyplot
import matplotlib.patches as mpatches
from colorama import Style, Fore
@@ -54,6 +55,25 @@ def module_path(obj) -> str:
"""
return obj.__class__.__module__
+def get_class(class_path: str) -> object:
+ """Get class object from 'path.to.class' string.
+
+ Parameters:
+ class_path: a 'path.to.class' string.
+
+ Returns:
+ class: a 'path.to.class' class.
+ """
+ parts = class_path.split('.')
+ module = ".".join(parts[:-1])
+
+ m = __import__(module)
+
+ for comp in parts[1:]:
+ m = getattr(m, comp)
+
+ return m
+
def properties(cls) -> list:
"""get class properties name."""
@@ -436,29 +456,149 @@ class SharedObject(TimestampedObject):
self._execution_times = {}
self._exceptions = {}
+def PipelineStepInit(method):
+ """Define a decorator use into PipelineStepObject class to declare pipeline step init method."""
+
+ def wrapper(self, **kwargs):
+ """Wrap pipeline step init method to update PipelineStepObject attributes with arguments after init call.
+
+ Parameters:
+ kwargs: Any arguments defined by PipelineStepMethodInit.
+ """
+
+ # DEBUG
+ print('@PipelineStepInit', kwargs.keys())
+
+ method(self, **kwargs)
+
+ self.update(kwargs)
+
+ return wrapper
+
+def PipelineStepAttributeSetter(method):
+ """Define a decorator use into PipelineStepObject class to declare pipeline step attribute setter."""
+
+ def wrapper(self, new_value):
+ """Wrap pipeline step attribute setter to load attribute from file.
+
+ Parameters:
+ new_value: value used to set attribute.
+ """
+
+ # Get new value type
+ new_value_type = type(new_value)
+
+ # Check setter annotations to get expected value type
+ expected_value_type = method.__annotations__.popitem()[1]
+
+ # Define function to load dict values
+ def load_dict(data: dict) -> any:
+
+ # Check if json keys are PipelineStepObject class and store them in a list
+ new_objects_list = []
+
+ for key, value in data.items():
+
+ try:
+
+ new_class = get_class(key)
+
+ except ValueError as e:
+
+ # Keys are not class name
+ if str(e) == 'Empty module name':
+
+ break
+
+ else:
+
+ raise(e)
+
+ # DEBUG
+ print('@PipelineStepAttributeSetter new_class', new_class)
+
+ new_objects_list.append( new_class(**value) )
+
+ # Only one object have been loaded: pass the object if it is a subclass of expected type
+ if len(new_objects_list) == 1 and issubclass(type(new_objects_list[0]), expected_value_type):
+
+ return new_objects_list[0]
+
+ # Pass non empty objects list
+ elif len(new_objects_list) > 0:
+
+ return new_objects_list
+
+ # Otherwise, data are parameters of the expected class
+ return expected_value_type(**data)
+
+ # DEBUG
+ print('@PipelineStepAttributeSetter', method.__name__, new_value_type, expected_value_type, type(expected_value_type))
+
+ # String not expected: load value from file
+ if new_value_type == str and new_value_type != expected_value_type:
+
+ filepath = os.path.join(self.working_directory, new_value)
+ file_format = filepath.split('.')[-1]
+
+ # Load image from JPG and PNG formats
+ if file_format == 'jpg' or file_format == 'png':
+
+ # DEBUG
+ print('@PipelineStepAttributeSetter IMAGE', filepath)
+
+ return method(self, cv2.imread(filepath))
+
+ # Load PipelineStepObject from JSON file
+ elif file_format == 'json':
+
+ # DEBUG
+ print('@PipelineStepAttributeSetter issubclass', issubclass(expected_value_type, PipelineStepObject))
+
+ #if issubclass(expected_value_type, PipelineStepObject):
+
+ # DEBUG
+ print('@PipelineStepAttributeSetter JSON', filepath)
+
+ with open(filepath) as file:
+
+ return method(self, load_dict(json.load(file)))
+
+ # DEBUG
+ print('@PipelineStepAttributeSetter unknown file format', file_format)
+
+ # Always load value from dict
+ if new_value_type == dict:
+
+ # DEBUG
+ print('@PipelineStepAttributeSetter dict', new_value)
+
+ return method(self, load_dict(new_value))
+
+ # Otherwise, pass new value to setter method
+ method(self, new_value)
+
+ return wrapper
+
class PipelineStepObject():
"""
Define class to assess pipeline step methods execution time and observe them.
"""
- def __init__(self, name: str = None, working_directory: str = None, observers: dict = None):
- """Initialize PipelineStepObject
-
- Parameters:
- name: object name
- working_directory: folder path to use for relative file path.
- observers: dictionary with observers objects.
+ @PipelineStepInit
+ def __init__(self, **kwargs):
+ """Initialize PipelineStepObject."""
- """
+ # DEBUG
+ print('PipelineStepObject.__init__')
# Init private attribute
- self.__name = name
- self.__working_directory = working_directory
- self.__observers = observers if observers is not None else {}
+ self.__name = None
+ self.__working_directory = None
+ self.__observers = {}
self.__execution_times = {}
- self.__properties = {}
- # parent attribute will be setup later by parent it self
+ # Parent attribute will be setup later by parent it self
self.__parent = None
def __enter__(self):
@@ -489,16 +629,40 @@ class PipelineStepObject():
child.__exit__(exception_type, exception_value, exception_traceback)
+ def update(self, object_data: dict):
+ """Update pipeline step object attributes with dictionary."""
+
+ for key, value in object_data.items():
+
+ setattr(self, key, value)
+
@property
def name(self) -> str:
"""Get pipeline step object's name."""
return self.__name
+ @name.setter
+ def name(self, name: str):
+ """Set pipeline step object's name."""
+ self.__name = name
+
@property
def working_directory(self) -> str:
- """Get pipeline step object's working_directory."""
+ """Get pipeline step object's working directory.
+ This path will be joined to relative file path."""
return self.__working_directory
+ @working_directory.setter
+ def working_directory(self, working_directory: str):
+ """Set pipeline step object's working directory."""
+
+ # Append working directory to the Python path
+ if working_directory is not None:
+
+ sys.path.append(working_directory)
+
+ self.__working_directory = working_directory
+
@property
def parent(self) -> object:
"""Get pipeline step object's parent object."""
@@ -514,98 +678,82 @@ class PipelineStepObject():
"""Get pipeline step object observers dictionary."""
return self.__observers
- @property
- def execution_times(self):
- """Get pipeline step object observers execution times dictionary."""
- return self.__execution_times
-
- def as_dict(self) -> dict:
- """Export PipelineStepObject attributes as dictionary.
-
- Returns:
- object_data: dictionary with pipeline step object attributes values.
- """
- return {
- "name": self.__name,
- "observers": self.__observers
- }
-
- @classmethod
- def from_dict(cls, object_data: dict, working_directory: str = None) -> object:
- """Load PipelineStepObject attributes from dictionary.
-
- Returns:
- object_data: dictionary with pipeline step object attributes values.
- working_directory: folder path where to load files when a dictionary value is a relative filepath.
+ @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
"""
- # Append working directory to the Python path
- if working_directory is not None:
-
- sys.path.append(working_directory)
-
- # Load name
- try:
-
- new_name = object_data.pop('name')
-
- except KeyError:
-
- new_name = None
-
- # Load observers
+ # Edit new observers dictionary
new_observers = {}
- try:
-
- new_observers_value = object_data.pop('observers')
+ # str: edit new observers dictionary from file
+ if type(observers_value) == str:
- # str: relative path to file
- if type(new_observers_value) == str:
+ filepath = os.path.join(self.working_directory, observers_value)
+ file_format = filepath.split('.')[-1]
- filepath = os.path.join(working_directory, new_observers_value)
- file_format = filepath.split('.')[-1]
+ # py: load __observers__ variable from Python file
+ if file_format == 'py':
- # Load module from working directory
- if file_format == 'py':
+ observer_module_path = observers_value.split('.')[0]
- observer_module_path = new_observers_value.split('.')[0]
+ observer_module = importlib.import_module(observer_module_path)
- observer_module = importlib.import_module(observer_module_path)
+ new_observers = observer_module.__observers__
- new_observers = observer_module.__observers__
+ # json: load dictionary from JSON file
+ elif file_format == 'json':
- # dict: instanciate ready-made argaze observers
- elif type(new_observers_value) == dict:
+ with open(filepath) as file:
- for observer_type, observer_data in new_observers_value.items():
+ new_observers = json.load(file)
- new_observers[observer_type] = eval(f'{observer_type}(**observer_data)')
+ # Instanciate observers from dictionary
+ for observer_type, observer_data in new_observers.items():
- except KeyError:
+ self.__observers[observer_type] = get_class(observer_type)(**observer_data)
- pass
+ @property
+ def execution_times(self):
+ """Get pipeline step object observers execution times dictionary."""
+ return self.__execution_times
+
+ def as_dict(self) -> dict:
+ """Export PipelineStepObject attributes as dictionary.
- # Create pipeline step object
- return PipelineStepObject(\
- new_name, \
- working_directory, \
- new_observers \
- )
+ Returns:
+ object_data: dictionary with pipeline step object attributes values.
+ """
+ return {
+ "name": self.__name,
+ "observers": self.__observers
+ }
@classmethod
def from_json(cls, configuration_filepath: str, patch_filepath: str = None) -> object:
"""
- Load pipeline step object from .json file.
+ Load instance from .json file.
Parameters:
configuration_filepath: path to json configuration file
patch_filepath: path to json patch file to modify any configuration entries
"""
+
+ # DEBUG
+ print('PipelineStepObject.from_json', cls)
+
+ # Load configuration from JSON file
with open(configuration_filepath) as configuration_file:
- object_data = json.load(configuration_file)
- working_directory = os.path.dirname(configuration_filepath)
+ # Edit object_data with working directory as first key
+ object_data = {
+ 'working_directory': os.path.dirname(configuration_filepath)
+ }
+
+ object_data.update(json.load(configuration_file))
# Apply patch to configuration if required
if patch_filepath is not None:
@@ -636,7 +784,11 @@ class PipelineStepObject():
object_data = update(object_data, patch_data)
- return cls.from_dict(object_data, working_directory)
+ # DEBUG
+ print('PipelineStepObject.from_json', object_data)
+
+ # Instanciate class
+ return cls(**object_data)
def to_json(self, json_filepath: str = None):
"""Save pipeline step object into .json file."""
@@ -755,7 +907,7 @@ class PipelineStepObject():
print('-', name, type(attr))
yield attr
-
+
def PipelineStepMethod(method):
"""Define a decorator use into PipelineStepObject class to declare pipeline method.
diff --git a/src/argaze/GazeAnalysis/Basic.py b/src/argaze/GazeAnalysis/Basic.py
index 724ab47..41f35ef 100644
--- a/src/argaze/GazeAnalysis/Basic.py
+++ b/src/argaze/GazeAnalysis/Basic.py
@@ -17,19 +17,17 @@ __credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
-from dataclasses import dataclass
-
from argaze import GazeFeatures, DataFeatures
import numpy
-@dataclass
class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer):
"""Basic scan path analysis."""
- def __post_init__(self):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
self.__path_duration = 0
self.__steps_number = 0
@@ -68,13 +66,13 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer):
return self.__step_fixation_durations_average
-@dataclass
class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer):
"""Basic AOI scan path analysis."""
- def __post_init__(self):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
self.__path_duration = 0
self.__steps_number = 0
diff --git a/src/argaze/GazeAnalysis/DeviationCircleCoverage.py b/src/argaze/GazeAnalysis/DeviationCircleCoverage.py
index ff23aa4..3a2d73f 100644
--- a/src/argaze/GazeAnalysis/DeviationCircleCoverage.py
+++ b/src/argaze/GazeAnalysis/DeviationCircleCoverage.py
@@ -17,7 +17,6 @@ __credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
-from typing import TypeVar, Tuple
import math
from argaze import GazeFeatures, DataFeatures
@@ -26,27 +25,27 @@ from argaze.AreaOfInterest import AOIFeatures
import numpy
import cv2
-GazeMovementType = TypeVar('GazeMovement', bound="GazeMovement")
-# Type definition for type annotation convenience
-
class AOIMatcher(GazeFeatures.AOIMatcher):
- """Matching algorithm based on fixation's deviation circle coverage over AOI.
+ """Matching algorithm based on fixation's deviation circle coverage over AOI."""
- Parameters:
- coverage_threshold: Minimal coverage ratio to consider a fixation over an AOI (1 means that whole fixation's deviation circle have to be over the AOI).
- """
- def __init__(self, coverage_threshold: float = 0, **kwargs):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
super().__init__(**kwargs)
- self.__coverage_threshold = coverage_threshold
+ self.__coverage_threshold = 0
self.__reset()
@property
- def coverage_threshold(self):
- """Get aoi matcher coverage threshold."""
+ def coverage_threshold(self) -> float:
+ """Minimal coverage ratio to consider a fixation over an AOI (1 means that whole fixation's deviation circle have to be over the AOI)."""
return self.__coverage_threshold
+
+ @coverage_threshold.setter
+ def coverage_threshold(self, coverage_threshold: float):
+
+ self.__coverage_threshold = coverage_threshold
def __reset(self):
@@ -58,7 +57,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, 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 705da57..82bb263 100644
--- a/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py
+++ b/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py
@@ -17,7 +17,6 @@ __credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
-from typing import TypeVar, Tuple
import math
from argaze import GazeFeatures, DataFeatures
@@ -25,15 +24,6 @@ from argaze import GazeFeatures, DataFeatures
import numpy
import cv2
-GazeMovementType = TypeVar('GazeMovement', bound="GazeMovement")
-# Type definition for type annotation convenience
-
-FixationType = TypeVar('Fixation', bound="Fixation")
-# Type definition for type annotation convenience
-
-SaccadeType = TypeVar('Saccade', bound="Saccade")
-# Type definition for type annotation convenience
-
class Fixation(GazeFeatures.Fixation):
"""Define dispersion based fixation."""
@@ -58,7 +48,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())
@@ -120,36 +110,43 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier):
*Identifying fixations and saccades in eye-tracking protocols.*
Proceedings of the 2000 symposium on Eye tracking research & applications (ETRA'00, 71-78).
[https://doi.org/10.1145/355017.355028](https://doi.org/10.1145/355017.355028)
-
- Parameters:
- deviation_max_threshold: Maximal distance allowed to consider a gaze movement as a fixation.
- duration_min_threshold: Minimal duration allowed to consider a gaze movement as a fixation. \
- It is also used as maximal duration allowed to wait valid gaze positions.
"""
- def __init__(self, deviation_max_threshold: int|float, duration_min_threshold: int|float):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
- self.__deviation_max_threshold = deviation_max_threshold
- self.__duration_min_threshold = duration_min_threshold
+ self.__deviation_max_threshold = 0
+ self.__duration_min_threshold = 0
self.__valid_positions = GazeFeatures.TimeStampedGazePositions()
self.__fixation_positions = GazeFeatures.TimeStampedGazePositions()
self.__saccade_positions = GazeFeatures.TimeStampedGazePositions()
@property
- def deviation_max_threshold(self):
- """Get identifier's deviation max threshold."""
+ def deviation_max_threshold(self) -> int|float:
+ """Maximal distance allowed to consider a gaze movement as a fixation."""
return self.__deviation_max_threshold
+ @deviation_max_threshold.setter
+ def deviation_max_threshold(self, deviation_max_threshold: int|float):
+
+ self.__deviation_max_threshold = deviation_max_threshold
+
@property
- def duration_min_threshold(self):
- """Get identifier duration min threshold."""
+ def duration_min_threshold(self) -> int|float:
+ """Minimal duration allowed to consider a gaze movement as a fixation.
+ It is also used as maximal duration allowed to wait valid gaze positions."""
return self.__duration_min_threshold
+
+ @duration_min_threshold.setter
+ def duration_min_threshold(self, duration_min_threshold: int|float):
+
+ 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:
@@ -241,7 +238,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 len(self.__saccade_positions) > 1))
@@ -257,7 +254,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:
@@ -266,7 +263,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/GazeAnalysis/Entropy.py b/src/argaze/GazeAnalysis/Entropy.py
index b9fa301..58617f7 100644
--- a/src/argaze/GazeAnalysis/Entropy.py
+++ b/src/argaze/GazeAnalysis/Entropy.py
@@ -17,16 +17,12 @@ __credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
-from typing import Tuple
-from dataclasses import dataclass, field
-
from argaze import GazeFeatures, DataFeatures
from argaze.GazeAnalysis import TransitionMatrix
import pandas
import numpy
-@dataclass
class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer):
"""Implementation of entropy algorithm as described in:
@@ -36,20 +32,30 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer):
[https://doi.org/10.1145/2578153.2578176](https://doi.org/10.1145/2578153.2578176)
"""
- transition_matrix_analyzer: TransitionMatrix.AOIScanPathAnalyzer = field(default_factory=TransitionMatrix.AOIScanPathAnalyzer)
- """To get its transition_matrix_probabilities result.
-
- !!! warning "Mandatory"
- TransitionMatrix analyzer have to be loaded before.
- """
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- def __post_init__(self):
-
- super().__init__()
+ super().__init__(**kwargs)
+ self.__transition_matrix_analyzer = None
self.__stationary_entropy = -1
self.__transition_entropy = -1
+ @property
+ def transition_matrix_analyzer(self) -> TransitionMatrix.AOIScanPathAnalyzer:
+ """Bind to TransitionMatrix analyzer to get its transition_matrix_probabilities.
+
+ !!! warning "Mandatory"
+ TransitionMatrix analyzer have to be loaded before.
+ """
+
+ return self.__transition_matrix_analyzer
+
+ @transition_matrix_analyzer.setter
+ def transition_matrix_analyzer(self, transition_matrix_analyzer: TransitionMatrix.AOIScanPathAnalyzer):
+
+ self.__transition_matrix_analyzer = transition_matrix_analyzer
+
@DataFeatures.PipelineStepMethod
def analyze(self, aoi_scan_path: GazeFeatures.AOIScanPathType):
diff --git a/src/argaze/GazeAnalysis/ExploreExploitRatio.py b/src/argaze/GazeAnalysis/ExploreExploitRatio.py
index 2814d75..fc67121 100644
--- a/src/argaze/GazeAnalysis/ExploreExploitRatio.py
+++ b/src/argaze/GazeAnalysis/ExploreExploitRatio.py
@@ -17,13 +17,10 @@ __credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
-from dataclasses import dataclass, field
-
from argaze import GazeFeatures, DataFeatures
import numpy
-@dataclass
class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer):
"""Implementation of explore vs exploit ratio algorithm as described in:
@@ -33,15 +30,24 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer):
[https://doi.org/10.1145/2702123.2702521](https://doi.org/10.1145/2702123.2702521)
"""
- short_fixation_duration_threshold: float = field(default=0.)
- """Time below which a fixation is considered to be short and so as exploratory."""
-
- def __post_init__(self):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
+ self.__short_fixation_duration_threshold = 0.
self.__explore_exploit_ratio = 0.
+ @property
+ def short_fixation_duration_threshold(self) -> float:
+ """Time below which a fixation is considered to be short and so as exploratory."""
+ return self.__short_fixation_duration_threshold
+
+ @short_fixation_duration_threshold.setter
+ def short_fixation_duration_threshold(self, short_fixation_duration_threshold: float):
+
+ self.__short_fixation_duration_threshold = short_fixation_duration_threshold
+
@DataFeatures.PipelineStepMethod
def analyze(self, scan_path: GazeFeatures.ScanPathType):
diff --git a/src/argaze/GazeAnalysis/FocusPointInside.py b/src/argaze/GazeAnalysis/FocusPointInside.py
index e3e5ae7..cc7b15e 100644
--- a/src/argaze/GazeAnalysis/FocusPointInside.py
+++ b/src/argaze/GazeAnalysis/FocusPointInside.py
@@ -17,7 +17,6 @@ __credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
-from typing import TypeVar, Tuple
import math
from argaze import GazeFeatures, DataFeatures
@@ -26,12 +25,10 @@ from argaze.AreaOfInterest import AOIFeatures
import numpy
import cv2
-GazeMovementType = TypeVar('GazeMovement', bound="GazeMovement")
-# Type definition for type annotation convenience
-
class AOIMatcher(GazeFeatures.AOIMatcher):
"""Matching algorithm based on fixation's focus point."""
+ @DataFeatures.PipelineStepInit
def __init__(self, **kwargs):
super().__init__(**kwargs)
diff --git a/src/argaze/GazeAnalysis/KCoefficient.py b/src/argaze/GazeAnalysis/KCoefficient.py
index 727b838..066de35 100644
--- a/src/argaze/GazeAnalysis/KCoefficient.py
+++ b/src/argaze/GazeAnalysis/KCoefficient.py
@@ -17,13 +17,10 @@ __credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
-from dataclasses import dataclass
-
from argaze import GazeFeatures, DataFeatures
import numpy
-@dataclass
class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer):
"""Implementation of the K coefficient algorithm as described in:
@@ -33,9 +30,10 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer):
[https://doi.org/10.1145/2896452](https://doi.org/10.1145/2896452)
"""
- def __post_init__(self):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
self.__K = 0
@@ -80,7 +78,6 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer):
return self.__K
-@dataclass
class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer):
"""Implementation of the K-modified coefficient algorithm as described in:
@@ -90,9 +87,10 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer):
[https://doi.org/10.1145/3379157.3391412](https://doi.org/10.1145/3379157.3391412)
"""
- def __post_init__(self):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
self.__K = 0
diff --git a/src/argaze/GazeAnalysis/LempelZivComplexity.py b/src/argaze/GazeAnalysis/LempelZivComplexity.py
index 919f25d..cf84cc0 100644
--- a/src/argaze/GazeAnalysis/LempelZivComplexity.py
+++ b/src/argaze/GazeAnalysis/LempelZivComplexity.py
@@ -17,13 +17,10 @@ __credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
-from dataclasses import dataclass
-
from argaze import GazeFeatures, DataFeatures
from lempel_ziv_complexity import lempel_ziv_complexity
-@dataclass
class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer):
"""Implementation of Lempel-Ziv complexity algorithm as described in:
@@ -34,9 +31,10 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer):
[https://doi.org/10.3929/ethz-b-000407653](https://doi.org/10.3929/ethz-b-000407653)
"""
- def __post_init__(self):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
self.__lempel_ziv_complexity = 0
diff --git a/src/argaze/GazeAnalysis/LinearRegression.py b/src/argaze/GazeAnalysis/LinearRegression.py
index 133b304..741c74b 100644
--- a/src/argaze/GazeAnalysis/LinearRegression.py
+++ b/src/argaze/GazeAnalysis/LinearRegression.py
@@ -17,17 +17,12 @@ __credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
-from typing import TypeVar, Tuple
-
from argaze import DataFeatures, GazeFeatures
from sklearn.linear_model import LinearRegression
import numpy
import cv2
-GazePositionType = TypeVar('GazePositionType', bound="GazePositionType")
-# Type definition for type annotation convenience
-
class GazePositionCalibrator(GazeFeatures.GazePositionCalibrator):
"""Implementation of linear regression algorithm as described in:
@@ -35,30 +30,36 @@ class GazePositionCalibrator(GazeFeatures.GazePositionCalibrator):
*Time- and space-efficient eye tracker calibration.*
Proceedings of the 11th ACM Symposium on Eye Tracking Research & Applications (ETRA'19, 1-8).
[https://dl.acm.org/doi/pdf/10.1145/3314111.3319818](https://dl.acm.org/doi/pdf/10.1145/3314111.3319818)
-
- Parameters:
- coefficients: linear regression coefficients.
- intercept: linear regression intercept value.
"""
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- def __init__(self, coefficients: list = [[1., 0.], [0., 1.]], intercept: list = [0., 0.]):
-
- super().__init__()
+ super().__init__(**kwargs)
self.__linear_regression = LinearRegression()
- self.__linear_regression.coef_ = numpy.array(coefficients)
- self.__linear_regression.intercept_ = numpy.array(intercept)
+ self.__linear_regression.coef_ = numpy.array([[1., 0.], [0., 1.]])
+ self.__linear_regression.intercept_ = numpy.array([0., 0.])
@property
def coefficients(self) -> list:
- """Get linear regression coefficients."""
+ """Linear regression coefficients."""
return self.__linear_regression.coef_.tolist()
+ @coefficients.setter
+ def coefficients(self, coefficients: list):
+
+ self.__linear_regression.coef_ = numpy.array(coefficients)
+
@property
- def intercept(self):
- """Get linear regression intercept value."""
+ def intercept(self) -> list:
+ """Linear regression intercept value."""
return self.__linear_regression.intercept_.tolist()
+ @intercept.setter
+ def intercept(self, intercept: list):
+
+ self.__linear_regression.intercept_ = numpy.array(intercept)
+
def is_calibrating(self) -> bool:
"""Is the calibration running?"""
return self.__linear_regression is None
@@ -85,7 +86,7 @@ class GazePositionCalibrator(GazeFeatures.GazePositionCalibrator):
# Return calibrated gaze position
return self.__linear_regression.score(self.__observed_positions, self.__expected_positions)
- def apply(self, gaze_position: GazeFeatures.GazePosition) -> GazePositionType:
+ def apply(self, gaze_position: GazeFeatures.GazePosition) -> GazeFeatures.GazePosition:
"""Apply calibration onto observed gaze position."""
if not self.is_calibrating():
diff --git a/src/argaze/GazeAnalysis/NGram.py b/src/argaze/GazeAnalysis/NGram.py
index a57a489..8cf0e1d 100644
--- a/src/argaze/GazeAnalysis/NGram.py
+++ b/src/argaze/GazeAnalysis/NGram.py
@@ -32,17 +32,34 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer):
[https://doi.org/10.1371/journal.pone.0247061](https://doi.org/10.1371/journal.pone.0247061)
"""
- n_min: int = field(default=2)
- """Minimal grams length to search."""
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- n_max: int = field(default=2)
- """Maximal grams length to search."""
+ super().__init__(**kwargs)
- def __post_init__(self):
+ self.__n_min = 2
+ self.__n_max = 2
+ self.__ngrams_count = {}
- super().__init__()
+ @property
+ def n_min(self) -> int:
+ """Minimal grams length to search."""
+ return self.__n_min
- self.__ngrams_count = {}
+ @n_min.setter
+ def n_min(self, n_min: int):
+
+ self.__n_min = n_min
+
+ @property
+ def n_max(self) -> int:
+ """Maximal grams length to search."""
+ return self.__n_max
+
+ @n_max.setter
+ def n_max(self, n_max: int):
+
+ self.__n_max = n_max
@DataFeatures.PipelineStepMethod
def analyze(self, aoi_scan_path: GazeFeatures.AOIScanPathType):
diff --git a/src/argaze/GazeAnalysis/NearestNeighborIndex.py b/src/argaze/GazeAnalysis/NearestNeighborIndex.py
index d2f129d..b0d7312 100644
--- a/src/argaze/GazeAnalysis/NearestNeighborIndex.py
+++ b/src/argaze/GazeAnalysis/NearestNeighborIndex.py
@@ -17,15 +17,11 @@ __credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
-from typing import TypeVar, Tuple, Any
-from dataclasses import dataclass, field
-
from argaze import GazeFeatures, DataFeatures
import numpy
from scipy.spatial.distance import cdist
-@dataclass
class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer):
"""Implementation of Nearest Neighbor Index algorithm as described in:
@@ -35,15 +31,24 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer):
[https://www.researchgate.net](https://www.researchgate.net/publication/239470608_Another_look_at_scanpath_distance_to_nearest_neighbour_as_a_measure_of_mental_workload)
"""
- size: tuple[float, float]
- """Frame dimension."""
-
- def __post_init__(self):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
+ self.__size = (0, 0)
self.__nearest_neighbor_index = 0
+ @property
+ def size(self) -> tuple[float, float]:
+ """Frame dimension."""
+ return self.__size
+
+ @size.setter
+ def size(self, size: tuple[float, float]):
+
+ self.__size = size
+
@DataFeatures.PipelineStepMethod
def analyze(self, scan_path: GazeFeatures.ScanPathType):
diff --git a/src/argaze/GazeAnalysis/TransitionMatrix.py b/src/argaze/GazeAnalysis/TransitionMatrix.py
index a78b8dc..25e2814 100644
--- a/src/argaze/GazeAnalysis/TransitionMatrix.py
+++ b/src/argaze/GazeAnalysis/TransitionMatrix.py
@@ -17,15 +17,11 @@ __credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
-from typing import Tuple
-from dataclasses import dataclass
-
from argaze import GazeFeatures, DataFeatures
import pandas
import numpy
-@dataclass
class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer):
"""Implementation of transition matrix probabilities and density algorithm as described in:
@@ -35,9 +31,10 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer):
[https://doi.org/10.1145/2578153.2578176](https://doi.org/10.1145/2578153.2578176)
"""
- def __post_init__(self):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
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 bd41a2e..471c688 100644
--- a/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py
+++ b/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py
@@ -16,7 +16,6 @@ __credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"
-from typing import TypeVar, Tuple
import math
from argaze import GazeFeatures, DataFeatures
@@ -24,15 +23,6 @@ from argaze import GazeFeatures, DataFeatures
import numpy
import cv2
-GazeMovementType = TypeVar('GazeMovement', bound="GazeMovement")
-# Type definition for type annotation convenience
-
-FixationType = TypeVar('Fixation', bound="Fixation")
-# Type definition for type annotation convenience
-
-SaccadeType = TypeVar('Saccade', bound="Saccade")
-# Type definition for type annotation convenience
-
class Fixation(GazeFeatures.Fixation):
"""Define dispersion based fixation."""
@@ -119,18 +109,15 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier):
saccades in eye-tracking protocols. In Proceedings of the 2000 symposium
on Eye tracking research & applications (ETRA '00). ACM, New York, NY, USA,
71-78. [http://dx.doi.org/10.1145/355017.355028](http://dx.doi.org/10.1145/355017.355028)
-
- Parameters:
- velocity_max_threshold: Maximal velocity allowed to consider a gaze movement as a fixation.
- duration_min_threshold: Minimal duration allowed to wait valid gaze positions.
"""
- def __init__(self, velocity_max_threshold: int|float, duration_min_threshold: int|float):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
- self.__velocity_max_threshold = velocity_max_threshold
- self.__duration_min_threshold = duration_min_threshold
+ self.__velocity_max_threshold = 0
+ self.__duration_min_threshold = 0
self.__last_ts = -1
self.__last_position = None
@@ -139,15 +126,25 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier):
self.__saccade_positions = GazeFeatures.TimeStampedGazePositions()
@property
- def velocity_max_threshold(self):
- """Get identifier's velocity max threshold."""
+ def velocity_max_threshold(self) -> int|float:
+ """Maximal velocity allowed to consider a gaze movement as a fixation."""
return self.__velocity_max_threshold
+ @velocity_max_threshold.setter
+ def velocity_max_threshold(self, velocity_max_threshold: int|float):
+
+ self.__velocity_max_threshold = velocity_max_threshold
+
@property
- def duration_min_threshold(self):
- """Get identifier duration min threshold."""
+ def duration_min_threshold(self) -> int|float:
+ """Minimal duration allowed to wait valid gaze positions."""
return self.__duration_min_threshold
+ @duration_min_threshold.setter
+ def duration_min_threshold(self, duration_min_threshold: int|float):
+
+ self.__duration_min_threshold = duration_min_threshold
+
@DataFeatures.PipelineStepMethod
def identify(self, gaze_position, terminate=False) -> GazeMovementType:
diff --git a/src/argaze/GazeFeatures.py b/src/argaze/GazeFeatures.py
index 82c3e50..677e9bb 100644
--- a/src/argaze/GazeFeatures.py
+++ b/src/argaze/GazeFeatures.py
@@ -300,9 +300,10 @@ GazePositionCalibratorType = TypeVar('GazePositionCalibrator', bound="GazePositi
class GazePositionCalibrator(DataFeatures.PipelineStepObject):
"""Abstract class to define what should provide a gaze position calibrator algorithm."""
- def __init__(self):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
def store(self, observed_gaze_position: GazePosition, expected_gaze_position: GazePosition):
"""Store observed and expected gaze positions.
@@ -565,9 +566,10 @@ class TimeStampedGazeStatus(DataFeatures.TimestampedObjectsList):
class GazeMovementIdentifier(DataFeatures.PipelineStepObject):
"""Abstract class to define what should provide a gaze movement identifier."""
- def __init__(self):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
@DataFeatures.PipelineStepMethod
def identify(self, timestamped_gaze_position: GazePosition, terminate:bool=False) -> GazeMovementType:
@@ -840,9 +842,10 @@ class ScanPath(list):
class ScanPathAnalyzer(DataFeatures.PipelineStepObject):
"""Abstract class to define what should provide a scan path analyzer."""
- def __init__(self):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
self.__properties = [name for (name, value) in self.__class__.__dict__.items() if isinstance(value, property)]
@@ -866,20 +869,21 @@ class ScanPathAnalyzer(DataFeatures.PipelineStepObject):
class AOIMatcher(DataFeatures.PipelineStepObject):
"""Abstract class to define what should provide an AOI matcher algorithm."""
- def __init__(self, exclude: list[str] = [], **kwargs):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
super().__init__(**kwargs)
- self.__exclude = exclude
+ self.__exclude = []
@property
def exclude(self):
- """Get list of AOI to exclude from matching."""
+ """List of AOI to exclude from matching."""
return self.__exclude
@exclude.setter
def exclude(self, exclude: list[str]):
- """Set list of AOI to exclude from matching."""
+
self.__exclude = exclude
def match(self, aoi_scene: AOIFeatures.AOIScene, gaze_movement: GazeMovement) -> Tuple[str, AOIFeatures.AreaOfInterest]:
@@ -1202,9 +1206,10 @@ class AOIScanPath(list):
class AOIScanPathAnalyzer(DataFeatures.PipelineStepObject):
"""Abstract class to define what should provide a aoi scan path analyzer."""
- def __init__(self):
+ @DataFeatures.PipelineStepInit
+ def __init__(self, **kwargs):
- super().__init__()
+ super().__init__(**kwargs)
self.__properties = [name for (name, value) in self.__class__.__dict__.items() if isinstance(value, property)]
diff --git a/src/argaze/utils/Providers/__init__.py b/src/argaze/utils/Providers/__init__.py
index f80a694..b76cd8b 100644
--- a/src/argaze/utils/Providers/__init__.py
+++ b/src/argaze/utils/Providers/__init__.py
@@ -1,4 +1,4 @@
"""
Collection of device interfaces.
"""
-__all__ = ['tobii_pro_glasses_2'] \ No newline at end of file
+__all__ = ['TobiiProGlasses2'] \ No newline at end of file
diff --git a/src/argaze/utils/__init__.py b/src/argaze/utils/__init__.py
index 4b7b4db..a2322bb 100644
--- a/src/argaze/utils/__init__.py
+++ b/src/argaze/utils/__init__.py
@@ -1,4 +1,4 @@
"""
Miscelleaneous utilities.
"""
-__all__ = ['UtilsFeatures', 'providers'] \ No newline at end of file
+__all__ = ['UtilsFeatures', 'Providers'] \ No newline at end of file
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 0a306d1..5e2d722 100644
--- a/src/argaze/utils/demo_data/demo_aruco_markers_setup.json
+++ b/src/argaze/utils/demo_data/demo_aruco_markers_setup.json
@@ -57,7 +57,7 @@
"size": [1920, 1149],
"background": "frame_background.jpg",
"gaze_movement_identifier": {
- "DispersionThresholdIdentification": {
+ "argaze.GazeAnalysis.DispersionThresholdIdentification.GazeMovementIdentifier": {
"deviation_max_threshold": 50,
"duration_min_threshold": 200
}
@@ -66,12 +66,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
}
},
@@ -79,17 +79,17 @@
"demo_layer": {
"aoi_scene": "aoi_2d_scene.json",
"aoi_matcher": {
- "FocusPointInside": {}
+ "argaze.GazeAnalysis.FocusPointInside.AOIMatcher": {}
},
"aoi_scan_path": {
"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
},
diff --git a/src/argaze/utils/demo_data/provider_setup.json b/src/argaze/utils/demo_data/provider_setup.json
index d63f914..47b6bdc 100644
--- a/src/argaze/utils/demo_data/provider_setup.json
+++ b/src/argaze/utils/demo_data/provider_setup.json
@@ -1,5 +1,5 @@
{
- "TobiiProGlasses2" : {
+ "argaze.utils.Providers.TobiiProGlasses2.Provider" : {
"address": "10.34.0.17",
"project": "MyProject",
"participant": "NewParticipant"