diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/argaze/GazeFeatures.py | 267 |
1 files changed, 156 insertions, 111 deletions
diff --git a/src/argaze/GazeFeatures.py b/src/argaze/GazeFeatures.py index cfd7419..1c27540 100644 --- a/src/argaze/GazeFeatures.py +++ b/src/argaze/GazeFeatures.py @@ -25,57 +25,124 @@ import cv2 GazePositionType = TypeVar('GazePosition', bound="GazePosition") # Type definition for type annotation convenience -@dataclass(frozen=True) -class GazePosition(): - """Define gaze position as a tuple of coordinates with precision.""" +class GazePosition(tuple, DataFeatures.TimestampedObject): + """Define gaze position as a tuple of coordinates with precision. - value: tuple[int | float] = field(default=(0, 0)) - """Position's value.""" + Parameters: + precision: the radius of a circle around value where other same gaze position measurements could be. + message: a string to describe why the the position is what it is. + """ - precision: float = field(default=0., kw_only=True) - """Position's precision represents the radius of a circle around \ - this gaze position value where other same gaze position measurements could be.""" + def __new__(cls, position: tuple = (), precision: int|float = None, message: str = None, **kwargs): - def __getitem__(self, axis: int) -> int | float: - """Get position value along a particular axis.""" + return tuple.__new__(cls, position) - return self.value[axis] + def __init__(self, position: tuple = (), precision: int|float = None, message: str = None, **kwargs): - def __iter__(self) -> iter: - """Iterate over each position value axis.""" + DataFeatures.TimestampedObject.__init__(self, **kwargs) + self.__precision = precision + self.__message = message - return iter(self.value) - - def __len__(self) -> int: - """Number of axis in position value.""" + @property + def value(self): + """Get position's tuple value.""" + return tuple(self) + + @property + def precision(self): + """Get position's precision.""" + return self.__precision + + @property + def message(self): + """Get position's message.""" + return self.__message + + @classmethod + def from_dict(self, position_data: dict) -> GazePositionType: + + if 'value' in position_data.keys(): + + value = position_data.pop('value') + return GazePosition(value, **position_data) + + else: - return len(self.value) + return GazePosition(**position_data) + + def __bool__(self) -> bool: + """Is the position value valid?""" + return len(self) > 0 def __repr__(self): """String representation""" - return json.dumps(self, ensure_ascii = False, default=vars) + return json.dumps(DataFeatures.as_dict(self)) - def __mul__(self, value) -> GazePositionType: - """Multiply gaze position.""" + def __add__(self, position: GazePositionType) -> GazePositionType: + """Add position. - return GazePosition(numpy.array(self.value) * value, precision= self.precision * numpy.linalg.norm(value)) + !!! note + The returned position precision is the maximal precision. + """ + if self.__precision is not None and position.precision is not None: - def __array__(self): - """Cast as numpy array.""" + return GazePosition(numpy.array(self) + numpy.array(position), precision = max(self.__precision, position.precision)) - return numpy.array(self.value) + else: - @property - def valid(self) -> bool: - """Is the precision not None?""" + return GazePosition(numpy.array(self) + numpy.array(position)) + + __radd__ = __add__ + + def __sub__(self, position: GazePositionType) -> GazePositionType: + """Substract position. + + !!! note + The returned position precision is the maximal precision. + """ + if self.__precision is not None and position.precision is not None: + + return GazePosition(numpy.array(self) - numpy.array(position), precision = max(self.__precision, position.precision)) + + else: + + return GazePosition(numpy.array(self) - numpy.array(position)) + + def __rsub__(self, position: GazePositionType) -> GazePositionType: + """Reversed substract position. + + !!! note + The returned position precision is the maximal precision. + """ + if self.__precision is not None and position.precision is not None: + + return GazePosition(numpy.array(position) - numpy.array(self), precision = max(self.__precision, position.precision)) + + else: + + return GazePosition(numpy.array(position) - numpy.array(self)) + + def __mul__(self, factor: int|float) -> GazePositionType: + """Multiply position by a factor. + + !!! note + The returned position precision is also multiplied by the factor. + """ + return GazePosition(numpy.array(self) * factor, precision = self.__precision * factor if self.__precision is not None else None) + + def __pow__(self, factor: int|float) -> GazePositionType: + """Power position by a factor. - return self.precision is not None + !!! note + The returned position precision is also powered by the factor. + """ + return GazePosition(numpy.array(self) ** factor, precision = self.__precision ** factor if self.__precision is not None else None) def distance(self, gaze_position) -> float: """Distance to another gaze positions.""" - distance = (self.value[0] - gaze_position.value[0])**2 + (self.value[1] - gaze_position.value[1])**2 + distance = (self[0] - gaze_position[0])**2 + (self[1] - gaze_position[1])**2 distance = numpy.sqrt(distance) return distance @@ -84,82 +151,50 @@ class GazePosition(): """Does this gaze position overlap another gaze position considering its precision? Set both to True to test if the other gaze position overlaps this one too.""" - distance = (self.value[0] - gaze_position.value[0])**2 + (self.value[1] - gaze_position.value[1])**2 - distance = numpy.sqrt(distance) + distance = numpy.sqrt(numpy.sum((self - gaze_position)**2)) if both: - return distance < min(self.precision, gaze_position.precision) + return distance < min(self.__precision, gaze_position.precision) else: - return distance < self.precision + return distance < self.__precision def draw(self, image: numpy.array, color: tuple = None, size: int = None, draw_precision=True): """Draw gaze position point and precision circle.""" if self.valid: - int_value = (int(self.value[0]), int(self.value[1])) + int_value = (int(self[0]), int(self[1])) # Draw point at position if required if color is not None: cv2.circle(image, int_value, size, color, -1) # Draw precision circle - if self.precision > 0 and draw_precision: - cv2.circle(image, int_value, round(self.precision), color, 1) - -class UnvalidGazePosition(GazePosition): - """Unvalid gaze position.""" - - def __init__(self, message=None): - - self.message = message - - super().__init__((None, None), precision=None) + if self.__precision > 0 and draw_precision: + cv2.circle(image, int_value, round(self.__precision), color, 1) TimeStampedGazePositionsType = TypeVar('TimeStampedGazePositions', bound="TimeStampedGazePositions") # Type definition for type annotation convenience -class TimeStampedGazePositions(DataFeatures.TimeStampedBuffer): - """Define timestamped buffer to store gaze positions.""" - - def __setitem__(self, key, value: GazePosition|dict): - """Force GazePosition storage.""" - - # Convert dict into GazePosition - if type(value) == dict: - - assert(set(['value', 'precision']).issubset(value.keys())) - - if math.isnan(value['precision']): - - if 'message' in value.keys(): - - value = UnvalidGazePosition(value['message']) - - else : - - value = UnvalidGazePosition() - - else: - - value = GazePosition(value['value'], precision=value['precision']) - - assert(type(value) == GazePosition or type(value) == UnvalidGazePosition) +class TimeStampedGazePositions(DataFeatures.TimeStampedObjectsList): + """Handle timestamped gaze positions into a list""" + + def __init__(self, gaze_positions: list = []): - super().__setitem__(key, value) + super().__init__(GazePosition, gaze_positions) @classmethod def from_json(self, json_filepath: str) -> TimeStampedGazePositionsType: """Create a TimeStampedGazePositionsType from .json file.""" - with open(json_filepath, encoding='utf-8') as ts_buffer_file: + with open(json_filepath, encoding='utf-8') as ts_positions_file: - json_buffer = json.load(ts_buffer_file) + json_positions = json.load(ts_positions_file) - return TimeStampedGazePositions({ast.literal_eval(ts_str): json_buffer[ts_str] for ts_str in json_buffer}) + return TimeStampedGazePositions({ast.literal_eval(ts_str): json_positions[ts_str] for ts_str in json_positions}) @classmethod - def from_dataframe(self, dataframe: pandas.DataFrame, timestamp: str, x: str, y: str, precision: str = None) -> TimeStampedGazePositionsType: + def from_dataframe(self, dataframe: pandas.DataFrame, timestamp: str, x: str, y: str, precision: str = None, message: str = None) -> TimeStampedGazePositionsType: """Create a TimeStampedGazePositions from [Pandas DataFrame](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html). Parameters: @@ -167,6 +202,7 @@ class TimeStampedGazePositions(DataFeatures.TimeStampedBuffer): x: specific x column label. y: specific y column label. precision: specific precision column label if exist. + message: specific message column label if exist. """ # Copy columns @@ -178,29 +214,52 @@ class TimeStampedGazePositions(DataFeatures.TimeStampedBuffer): df = dataframe.loc[:, (timestamp, x, y)] + # DEBUG + print(df) + # Merge x and y columns into one 'value' column df['value'] = tuple(zip(df[x], df[y])) - df.drop(columns= [x, y], inplace=True, axis=1) + df.drop(columns=[x, y], inplace=True, axis=1) + + # DEBUG + print(df) + + # Replace (NaN, NaN) values by () + df['value'] = df.apply(lambda row: print(row.values[1], pandas.isnull(row.values[1])), axis=True) + + # DEBUG + print(df) # Handle precision data if precision: - # Rename precision column into 'precision' column + # Rename precision column into 'precision' column df.rename(columns={precision: 'precision'}, inplace=True) else: - # Append a precision column where precision is NaN if value is a tuple of NaN else 0 - df['precision'] = df.apply(lambda row: numpy.nan if math.isnan(row.value[0]) or math.isnan(row.value[1]) else 0, axis=True) + # Append a None precision column + df['precision'] = df.apply(lambda row: None, axis=True) + + # Handle message data + if message: + + # Rename message column into 'message' column + df.rename(columns={precision: 'message'}, inplace=True) + + else: + + # Append a None message column + df['message'] = df.apply(lambda row: None, axis=True) - # Rename timestamp column into 'timestamp' column then use it as index + # Rename timestamp column into 'timestamp' column df.rename(columns={timestamp: 'timestamp'}, inplace=True) - df.set_index('timestamp', inplace=True) # Filter duplicate timestamps - df = df[df.index.duplicated() == False] + df = df[df.timestamp.duplicated() == False] - return TimeStampedGazePositions(df.to_dict('index')) + # Create timestamped gaze positions + return TimeStampedGazePositions(df.apply(lambda row: GazePosition(row.value, precision=row.precision, message=row.message, timestamp=row.timestamp), axis=True)) class GazePositionCalibrationFailed(Exception): """Exception raised by GazePositionCalibrator.""" @@ -300,7 +359,7 @@ GazeMovementType = TypeVar('GazeMovement', bound="GazeMovement") # Type definition for type annotation convenience @dataclass(frozen=True) -class GazeMovement(): +class GazeMovement(DataFeatures.TimestampedObject): """Define abstract gaze movement class as a buffer of timestamped positions.""" positions: TimeStampedGazePositions @@ -328,7 +387,7 @@ class GazeMovement(): _, start_position = self.positions.first _, end_position = self.positions.last - amplitude = numpy.linalg.norm( numpy.array(start_position.value) - numpy.array(end_position.value)) + amplitude = numpy.linalg.norm(start_position - end_position) # Update frozen amplitude attribute object.__setattr__(self, 'amplitude', amplitude) @@ -350,7 +409,7 @@ class GazeMovement(): for ts, position in self.positions.items(): - output += f'\n\t{ts}:\n\t\tvalue={position.value},\n\t\tprecision={position.precision}' + output += f'\n\t{ts}:\n\t\tvalue={position},\n\t\tprecision={position.precision}' else: @@ -453,24 +512,12 @@ def is_saccade(gaze_movement): TimeStampedGazeMovementsType = TypeVar('TimeStampedGazeMovements', bound="TimeStampedGazeMovements") # Type definition for type annotation convenience -class TimeStampedGazeMovements(DataFeatures.TimeStampedBuffer): - """Define timestamped buffer to store gaze movements.""" +class TimeStampedGazeMovements(DataFeatures.TimeStampedObjectsList): + """Handle timestamped gaze movements into a list""" - def __setitem__(self, key, value: GazeMovement): - """Force value to be or inherit from GazeMovement.""" - - assert(isinstance(value, GazeMovement) or type(value).__bases__[0] == Fixation or type(value).__bases__[0] == Saccade) - - super().__setitem__(key, value) - - def __str__(self): - - output = '' - for ts, item in self.items(): - - output += f'\n{item}' + def __init__(self): - return output + super().__init__(GazeMovement) GazeStatusType = TypeVar('GazeStatus', bound="GazeStatus") # Type definition for type annotation convenience @@ -489,23 +536,21 @@ class GazeStatus(GazePosition): def from_position(cls, gaze_position: GazePosition, movement_type: str, movement_index: int) -> GazeStatusType: """Initialize from a gaze position instance.""" - return cls(gaze_position.value, precision=gaze_position.precision, movement_type=movement_type, movement_index=movement_index) + return cls(gaze_position, precision=gaze_position.precision, movement_type=movement_type, movement_index=movement_index) TimeStampedGazeStatusType = TypeVar('TimeStampedGazeStatus', bound="TimeStampedGazeStatus") # Type definition for type annotation convenience -class TimeStampedGazeStatus(DataFeatures.TimeStampedBuffer): - """Define timestamped buffer to store list of gaze statusa. +class TimeStampedGazeStatus(DataFeatures.TimeStampedObjectsList): + """Handle timestamped gaze movements into a list !!! note List of gaze status are required as a gaze position can belongs to two consecutive gaze movements as last and first position. """ - def __setitem__(self, key, value: list): - - assert(isinstance(value, list)) + def __init__(self): - super().__setitem__(key, value) + super().__init__(GazeStatus) class GazeMovementIdentifier(DataFeatures.PipelineStepObject): """Abstract class to define what should provide a gaze movement identifier.""" |