Major rewrite of GazePosition and timestampGazePositions class

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."""