aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/argaze/GazeAnalysis/DispersionBasedGazeMovementIdentifier.py143
1 files changed, 90 insertions, 53 deletions
diff --git a/src/argaze/GazeAnalysis/DispersionBasedGazeMovementIdentifier.py b/src/argaze/GazeAnalysis/DispersionBasedGazeMovementIdentifier.py
index fc565c6..b60291c 100644
--- a/src/argaze/GazeAnalysis/DispersionBasedGazeMovementIdentifier.py
+++ b/src/argaze/GazeAnalysis/DispersionBasedGazeMovementIdentifier.py
@@ -73,6 +73,13 @@ class Saccade(GazeFeatures.Saccade):
def __post_init__(self):
super().__post_init__()
+@dataclass(frozen=True)
+class UnknownGazeMovement(GazeFeatures.UnknownGazeMovement):
+ """Define dispersion based unknown movement."""
+
+ def __post_init__(self):
+ super().__post_init__()
+
@dataclass
class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier):
"""Implementation of the I-DT algorithm as described in:
@@ -93,124 +100,146 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier):
"""GazeMovement identification generator."""
self.__last_fixation = None
+ moving_gaze_positions = GazeFeatures.TimeStampedGazePositions()
- # while there are 2 gaze positions at least
+ # While there are 2 gaze positions at least
while len(self.__ts_gaze_positions) >= 2:
- # copy remaining timestamped gaze positions
- remaining_ts_gaze_positions = self.__ts_gaze_positions.copy()
+ # Copy remaining gaze positions
+ remaining_gaze_positions = self.__ts_gaze_positions.copy()
- # select timestamped gaze position until a duration threshold
- ts_start, gaze_position_start = remaining_ts_gaze_positions.pop_first()
+ # Select gaze position until a duration threshold
+ ts_start, gaze_position_start = remaining_gaze_positions.pop_first()
- # Ignore non valid start position
- if not gaze_position_start.valid:
+ # Ignore and remove non valid start positions
+ while not gaze_position_start.valid:
+ ts_start, gaze_position_start = remaining_gaze_positions.pop_first()
self.__ts_gaze_positions.pop_first()
- continue
- ts_gaze_positions = GazeFeatures.TimeStampedGazePositions()
- ts_gaze_positions[ts_start] = gaze_position_start
+ valid_gaze_positions = GazeFeatures.TimeStampedGazePositions()
+ valid_gaze_positions[ts_start] = gaze_position_start
+
+ unvalid_gaze_positions = GazeFeatures.TimeStampedGazePositions()
# Select next position
- ts_next, gaze_position_next = remaining_ts_gaze_positions.first
+ ts_next, gaze_position_next = remaining_gaze_positions.first
while (ts_next - ts_start) < self.duration_threshold:
- # Ignore non valid position
- # Should we consider invalid position to not break fixation ?
+ # Store valid position
if gaze_position_next.valid:
- # Store selected position
- ts, gaze_position = remaining_ts_gaze_positions.pop_first()
- ts_gaze_positions[ts] = gaze_position
+ ts, valid_gaze_position = remaining_gaze_positions.pop_first()
+ valid_gaze_positions[ts] = valid_gaze_position
+ # Store non valid position
else:
- remaining_ts_gaze_positions.pop_first()
+ ts, unvalid_gaze_position = remaining_gaze_positions.pop_first()
+ unvalid_gaze_positions[ts] = unvalid_gaze_position
try:
# Read next position
- ts_next, gaze_position_next = remaining_ts_gaze_positions.first
+ ts_next, gaze_position_next = remaining_gaze_positions.first
except:
break
- # is it a new fixation ?
- new_fixation = Fixation(ts_gaze_positions)
+ # Consider the last valid gaze positions as a new fixation
+ new_fixation = Fixation(valid_gaze_positions)
- # dispersion is small : extending fixation
+ # Dispersion is small : extending fixation
if new_fixation.dispersion <= self.dispersion_threshold:
- # remove selected gaze positions
- for gp in ts_gaze_positions:
+ # Remove valid gaze positions
+ for ts, gp in valid_gaze_positions.items():
+ self.__ts_gaze_positions.pop(ts)
- self.__ts_gaze_positions.pop_first()
+ # Remove unvalid gaze positions
+ for ts, gp in unvalid_gaze_positions.items():
+ self.__ts_gaze_positions.pop(ts)
- # extend fixation position from a copy
- ts_gaze_positions_extension = ts_gaze_positions.copy()
+ # Copy new fixation positions before to try to extend them
+ extended_gaze_positions = new_fixation.positions.copy()
- # are next gaze positions not too dispersed ?
- while len(remaining_ts_gaze_positions) > 0:
+ # Are next gaze positions not too dispersed ?
+ while len(remaining_gaze_positions) > 0:
# Select next gaze position
- ts_next, gaze_position_next = remaining_ts_gaze_positions.first
+ ts_next, gaze_position_next = remaining_gaze_positions.first
- # Ignore non valid position
- # Should we consider invalid position to not break fixation ?
+ # Ignore and remove non valid next positions
if not gaze_position_next.valid:
- remaining_ts_gaze_positions.pop_first()
+ ts_next, gaze_position_next = remaining_gaze_positions.pop_first()
self.__ts_gaze_positions.pop_first()
continue
- ts_gaze_positions_extension[ts_next] = gaze_position_next
+ extended_gaze_positions[ts_next] = gaze_position_next
# how much gaze is dispersed ?
- extended_fixation = Fixation(ts_gaze_positions_extension)
+ extended_fixation = Fixation(extended_gaze_positions)
# dispersion becomes too wide : ignore extended fixation
if extended_fixation.dispersion > self.dispersion_threshold:
break
# update new fixation
- new_fixation = Fixation(ts_gaze_positions_extension.copy())
+ new_fixation = Fixation(extended_gaze_positions.copy())
# remove selected gaze position
- remaining_ts_gaze_positions.pop_first()
+ remaining_gaze_positions.pop_first()
self.__ts_gaze_positions.pop_first()
- # is the new fixation have a duration ?
+ # Is the new fixation have a duration ?
if new_fixation.duration > 0:
- # does a former fixation have been identified ?
+ # Does a former fixation have been identified ?
if self.__last_fixation != None:
- # merge new fixation if it overlaps last fixation
+ # Merge new fixation if it overlaps last fixation
if self.__last_fixation.overlap(new_fixation):
self.__last_fixation.merge(new_fixation)
new_fixation = None
- # else output last fixation and create a saccade
+ # Else output last fixation and consider moving gaze positions buffer
else:
yield self.__last_fixation
-
- # store start and end positions in a timestamped buffer
- ts_saccade_positions = GazeFeatures.TimeStampedGazePositions()
- start_position_ts, start_position = self.__last_fixation.positions.last
- ts_saccade_positions[start_position_ts] = start_position
+ # Is there a new movement ?
+ if len(moving_gaze_positions) > 0:
- end_position_ts, end_position = new_fixation.positions.first
- ts_saccade_positions[end_position_ts] = end_position
+ # Check first and last gaze position of the moving positions buffer
+ start_position_ts, start_position = moving_gaze_positions.first
+ end_position_ts, end_position = moving_gaze_positions.last
- if end_position_ts > start_position_ts:
+ if not moving_gaze_positions.first[0] > self.__last_fixation.positions.last[0]:
+ print('first moving_gaze_positions not after last self.__last_fixation.positions')
+ print(moving_gaze_positions.first[0], self.__last_fixation.positions.last[0])
- new_saccade = Saccade(ts_saccade_positions)
+ if not moving_gaze_positions.last[0] < new_fixation.positions.first[0]:
+ print('last moving_gaze_positions not before first new_fixation.positions')
+ print(moving_gaze_positions.last[0], new_fixation.positions.first[0])
- yield new_saccade
+ # Saccade shouldn't be longer than fixation
+ # TODO : add a saccade duration threshold attribute ?
+ if end_position_ts - start_position_ts < self.duration_threshold:
+
+ yield Saccade(moving_gaze_positions)
+
+ # This movement is unknown
+ else:
+
+ # This unknown movement should be dispersed
+ assert(Fixation(moving_gaze_positions).dispersion > self.dispersion_threshold)
+
+ yield UnknownGazeMovement(moving_gaze_positions)
+
+ # Forget former moving gaze positions
+ moving_gaze_positions = GazeFeatures.TimeStampedGazePositions()
self.__last_fixation = new_fixation
@@ -219,10 +248,18 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier):
self.__last_fixation = new_fixation
yield self.__last_fixation
- # dispersion too wide : consider next gaze position
+ # Forget former moving gaze positions
+ moving_gaze_positions = GazeFeatures.TimeStampedGazePositions()
+
+ # Dispersion too wide :
+ # - This gaze position is part of a movement but not a fixation
+ # - Consider next gaze position
else:
- self.__ts_gaze_positions.pop_first()
- # output last fixation
+ ts_moving, moving_gaze_position = self.__ts_gaze_positions.pop_first()
+ moving_gaze_positions[ts_moving] = moving_gaze_position
+
+ # Output last fixation
if self.__last_fixation != None:
+
yield self.__last_fixation