diff options
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" |