diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/argaze.test/GazeAnalysis/DispersionThresholdIdentification.py | 54 | ||||
| -rw-r--r-- | src/argaze/DataFeatures.py | 16 | ||||
| -rw-r--r-- | src/argaze/GazeAnalysis/DispersionThresholdIdentification.py | 106 | ||||
| -rw-r--r-- | src/argaze/GazeFeatures.py | 101 |
4 files changed, 128 insertions, 149 deletions
diff --git a/src/argaze.test/GazeAnalysis/DispersionThresholdIdentification.py b/src/argaze.test/GazeAnalysis/DispersionThresholdIdentification.py index 7e74c1d..07496c3 100644 --- a/src/argaze.test/GazeAnalysis/DispersionThresholdIdentification.py +++ b/src/argaze.test/GazeAnalysis/DispersionThresholdIdentification.py @@ -121,12 +121,12 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check fixation fixation = ts_fixations.pop(0) - self.assertEqual(len(fixation.positions.keys()), size) + self.assertEqual(len(fixation), size) self.assertLessEqual(fixation.deviation_max, deviation_max) self.assertGreaterEqual(fixation.duration, (size - 1) * min_time) self.assertLessEqual(fixation.duration, (size - 1) * max_time) self.assertLessEqual(fixation.finished, True) - + def test_fixation_and_direct_saccade_identification(self): """Test DispersionThresholdIdentification fixation and saccade identification.""" @@ -153,7 +153,7 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check first fixation fixation = ts_fixations.pop(0) - self.assertEqual(len(fixation.positions.keys()), size) + self.assertEqual(len(fixation), size) self.assertLessEqual(fixation.deviation_max, deviation_max) self.assertGreaterEqual(fixation.duration, (size - 1) * min_time) self.assertLessEqual(fixation.duration, (size - 1) * max_time) @@ -162,7 +162,7 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check first saccade saccade = ts_saccades.pop(0) - self.assertEqual(len(saccade.positions.keys()), 2) + self.assertEqual(len(saccade), 2) self.assertGreaterEqual(saccade.duration, min_time) self.assertLessEqual(saccade.duration, max_time) self.assertLessEqual(saccade.finished, True) @@ -174,7 +174,7 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check second fixation fixation = ts_fixations.pop(0) - self.assertEqual(len(fixation.positions.keys()), size) + self.assertEqual(len(fixation), size) self.assertLessEqual(fixation.deviation_max, deviation_max) self.assertGreaterEqual(fixation.duration, (size - 1) * min_time) self.assertLessEqual(fixation.duration, (size - 1) * max_time) @@ -183,7 +183,7 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check that last position of a movement is equal to first position of next movement self.assertEqual(saccade[-1].timestamp, fixation[0].timestamp) self.assertEqual(saccade[-1].value, fixation[0].value) - + def test_fixation_and_short_saccade_identification(self): """Test DispersionThresholdIdentification fixation and saccade identification.""" @@ -213,7 +213,7 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check first fixation fixation = ts_fixations.pop(0) - self.assertEqual(len(fixation.positions.keys()), size) + self.assertEqual(len(fixation), size) self.assertLessEqual(fixation.deviation_max, deviation_max) self.assertGreaterEqual(fixation.duration, (size - 1) * min_time) self.assertLessEqual(fixation.duration, (size - 1) * max_time) @@ -222,7 +222,7 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check first saccade saccade = ts_saccades.pop(0) - self.assertEqual(len(saccade.positions.keys()), move + 2) + self.assertEqual(len(saccade), move + 2) self.assertGreaterEqual(saccade.duration, (move + 1) * min_time) self.assertLessEqual(saccade.duration, (move + 1) * max_time) self.assertLessEqual(saccade.finished, True) @@ -234,7 +234,7 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check second fixation fixation = ts_fixations.pop(0) - self.assertEqual(len(fixation.positions.keys()), size) + self.assertEqual(len(fixation), size) self.assertLessEqual(fixation.deviation_max, deviation_max) self.assertGreaterEqual(fixation.duration, (size - 1) * min_time) self.assertLessEqual(fixation.duration, (size - 1) * max_time) @@ -243,9 +243,9 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check that last position of a movement is equal to first position of next movement self.assertEqual(saccade[-1].timestamp, fixation[0].timestamp) self.assertEqual(saccade[-1].value, fixation[0].value) - - def test_invalid_gaze_position(self): - """Test DispersionThresholdIdentification fixation and saccade identification with invalid gaze position.""" + + def test_empty_gaze_position(self): + """Test DispersionThresholdIdentification fixation and saccade identification with empty gaze position.""" size = 15 center = (0, 0) @@ -267,7 +267,7 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check first fixation fixation = ts_fixations.pop(0) - self.assertEqual(len(fixation.positions.keys()), 7) + self.assertEqual(len(fixation), 7) self.assertLessEqual(fixation.deviation_max, deviation_max) self.assertGreaterEqual(fixation.duration, 6 * min_time) self.assertLessEqual(fixation.duration, 6 * max_time) @@ -276,12 +276,12 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check second fixation fixation = ts_fixations.pop(0) - self.assertEqual(len(fixation.positions.keys()), 5) + self.assertEqual(len(fixation), 5) self.assertLessEqual(fixation.deviation_max, deviation_max) self.assertGreaterEqual(fixation.duration, 4 * min_time) self.assertLessEqual(fixation.duration, 4 * max_time) self.assertLessEqual(fixation.finished, True) - + def test_fixation_overlapping(self): """Test Fixation overlap function.""" @@ -336,12 +336,12 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check unique fixation fixation = ts_fixations.pop(0) - self.assertEqual(len(fixation.positions.keys()), size * 2) + self.assertEqual(len(fixation), size * 2) #self.assertGreaterEqual(fixation.deviation_max, deviation_max) self.assertGreaterEqual(fixation.duration, (2 * size - 1) * min_time) self.assertLessEqual(fixation.duration, (2 * size - 1) * max_time) self.assertLessEqual(fixation.finished, True) - + def test_identification_browsing(self): """Test DispersionThresholdIdentification identification browsing.""" @@ -355,7 +355,7 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): ts_gaze_positions_A = build_gaze_fixation(size, center_A, deviation_max, min_time, max_time) ts_gaze_positions_B = build_gaze_fixation(size, center_B, deviation_max, min_time, max_time, start_ts=ts_gaze_positions_A[-1].timestamp) - ts_gaze_positions = ts_gaze_positions_A.append(ts_gaze_positions_B) + ts_gaze_positions = ts_gaze_positions_A + ts_gaze_positions_B gaze_movement_identifier = DispersionThresholdIdentification.GazeMovementIdentifier(deviation_max_threshold=deviation_max, duration_min_threshold=max_time*2) @@ -369,7 +369,7 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): if GazeFeatures.is_fixation(finished_gaze_movement): - self.assertEqual(len(finished_gaze_movement.positions.keys()), size) + self.assertEqual(len(finished_gaze_movement), size) self.assertLessEqual(finished_gaze_movement.deviation_max, deviation_max) self.assertGreaterEqual(finished_gaze_movement.duration, (size-1) * min_time) self.assertLessEqual(finished_gaze_movement.duration, (size-1) * max_time) @@ -377,13 +377,13 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): elif GazeFeatures.is_saccade(finished_gaze_movement): - self.assertEqual(len(finished_gaze_movement.positions.keys()), 2) + self.assertEqual(len(finished_gaze_movement), 2) self.assertGreaterEqual(finished_gaze_movement.duration, min_time) self.assertLessEqual(finished_gaze_movement.duration, max_time) self.assertLessEqual(finished_gaze_movement.finished, True) # Check that last gaze position date is not equal to given gaze position date - if finished_gaze_movement.valid: + if finished_gaze_movement: last_ts = finished_gaze_movement[-1].timestamp @@ -392,14 +392,14 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): # Check that last gaze position date of current fixation is equal to given gaze position date # NOTE: This is not true for saccade as, for I-DT, there is a minimal time window while the gaze movement is unknown current_gaze_movement = gaze_movement_identifier.current_gaze_movement - if current_gaze_movement.valid: + if current_gaze_movement: if GazeFeatures.is_fixation(current_gaze_movement): last_ts = current_gaze_movement[-1].timestamp self.assertEqual(last_ts, gaze_position.timestamp) - + def test_identification_generator(self): """Test DispersionThresholdIdentification identification using generator.""" @@ -413,15 +413,15 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): ts_gaze_positions_A = build_gaze_fixation(size, center_A, deviation_max, min_time, max_time) ts_gaze_positions_B = build_gaze_fixation(size, center_B, deviation_max, min_time, max_time, start_ts=ts_gaze_positions_A[-1].timestamp) - ts_gaze_positions = ts_gaze_positions_A.append(ts_gaze_positions_B) + ts_gaze_positions = ts_gaze_positions_A + ts_gaze_positions_B gaze_movement_identifier = DispersionThresholdIdentification.GazeMovementIdentifier(deviation_max_threshold=deviation_max, duration_min_threshold=max_time*2) - for ts, finished_gaze_movement in gaze_movement_identifier(ts_gaze_positions): + for finished_gaze_movement in gaze_movement_identifier(ts_gaze_positions): if GazeFeatures.is_fixation(finished_gaze_movement): - self.assertEqual(len(finished_gaze_movement.positions.keys()), size) + self.assertEqual(len(finished_gaze_movement), size) self.assertLessEqual(finished_gaze_movement.deviation_max, deviation_max) self.assertGreaterEqual(finished_gaze_movement.duration, size * min_time) self.assertLessEqual(finished_gaze_movement.duration, size * max_time) @@ -429,7 +429,7 @@ class TestDispersionThresholdIdentificationClass(unittest.TestCase): elif GazeFeatures.is_saccade(finished_gaze_movement): - self.assertEqual(len(finished_gaze_movement.positions.keys()), 2) + self.assertEqual(len(finished_gaze_movement), 2) self.assertGreaterEqual(finished_gaze_movement.duration, 2 * min_time) self.assertLessEqual(finished_gaze_movement.duration, 2 * max_time) self.assertLessEqual(finished_gaze_movement.finished, True) diff --git a/src/argaze/DataFeatures.py b/src/argaze/DataFeatures.py index 6be946f..ff9baec 100644 --- a/src/argaze/DataFeatures.py +++ b/src/argaze/DataFeatures.py @@ -205,9 +205,17 @@ class TimestampedObjectsList(list): super().append(ts_object) + def look_for(self, timestamp: TimeStampType) -> TimestampedObjectType: + """Look for object at given timestamp.""" + for ts_object in self: + + if ts_object.timestamp == timestamp: + + return ts_object + def __add__(self, ts_objects: list = []) -> TimestampedObjectsListType: """Append timestamped objects list.""" - + for ts_object in ts_objects: self.append(ts_object) @@ -774,7 +782,7 @@ def PipelineStepMethod(method): PipelineStepMethod must have a timestamp as first argument. """ - def wrapper(self, timestamp, *args, unwrap: bool = False): + def wrapper(self, timestamp, *args, unwrap: bool = False, **kwargs): """Wrap pipeline step method to measure execution time. Parameters: @@ -784,7 +792,7 @@ def PipelineStepMethod(method): """ if unwrap: - return method(self, timestamp, *args) + return method(self, timestamp, *args, **kwargs) # Initialize execution time assessment start = time.perf_counter() @@ -794,7 +802,7 @@ def PipelineStepMethod(method): try: # Execute wrapped method - result = method(self, timestamp, *args) + result = method(self, timestamp, *args, **kwargs) except Exception as e: diff --git a/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py b/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py index a452e98..c85e576 100644 --- a/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py +++ b/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py @@ -26,49 +26,39 @@ FixationType = TypeVar('Fixation', bound="Fixation") SaccadeType = TypeVar('Saccade', bound="Saccade") # Type definition for type annotation convenience -@dataclass(frozen=True) class Fixation(GazeFeatures.Fixation): """Define dispersion based fixation.""" - deviation_max: float = field(init=False) - """Maximal gaze position distance to the centroïd.""" + def __init__(self, positions: GazeFeatures.TimeStampedGazePositions = (), finished: bool = False, message: str = None, **kwargs): - def __post_init__(self): + super().__init__(positions, finished, message, **kwargs) - super().__post_init__() + if positions: - positions_array = numpy.asarray(self.positions.values()) - centroid = numpy.mean(positions_array, axis=0) - deviations_array = numpy.sqrt(numpy.sum((positions_array - centroid)**2, axis=1)) - - # Update frozen focus attribute using centroid - object.__setattr__(self, 'focus', (centroid[0], centroid[1])) + positions_array = numpy.asarray(self.values()) + centroid = numpy.mean(positions_array, axis=0) + deviations_array = numpy.sqrt(numpy.sum((positions_array - centroid)**2, axis=1)) - # Update frozen deviation_max attribute - object.__setattr__(self, 'deviation_max', deviations_array.max()) + # Set focus as positions centroid + self.focus = (centroid[0], centroid[1]) - def point_deviation(self, gaze_position) -> float: - """Get distance of a point from the fixation's centroïd.""" + # Set deviation_max attribute + self.__deviation_max = deviations_array.max() - return numpy.sqrt((self.focus[0] - gaze_position.value[0])**2 + (self.focus[1] - gaze_position.value[1])**2) + @property + def deviation_max(self): + """Get fixation's maximal deviation.""" + return self.__deviation_max - def overlap(self, fixation) -> bool: + def overlap(self, fixation: FixationType) -> bool: """Does a gaze position from another fixation having a deviation to this fixation centroïd smaller than maximal deviation?""" - - positions_array = numpy.asarray(self.positions.values()) - centroid = numpy.array(list(self.focus)) + + positions_array = numpy.asarray(fixation.values()) + centroid = numpy.mean(self.focus, axis=0) deviations_array = numpy.sqrt(numpy.sum((positions_array - centroid)**2, axis=1)) - return deviations_array.min() <= self.deviation_max - - def merge(self, fixation) -> FixationType: - """Merge another fixation into this fixation.""" - - self.positions.append(fixation.positions) - self.__post_init__() - - return self - + return min(deviations_array) <= self.deviation_max + def draw(self, image: numpy.array, deviation_circle_color: tuple = None, duration_border_color: tuple = None, duration_factor: float = 1., draw_positions: dict = None): """Draw fixation into image. @@ -93,12 +83,12 @@ class Fixation(GazeFeatures.Fixation): self.draw_positions(image, **draw_positions) -@dataclass(frozen=True) class Saccade(GazeFeatures.Saccade): """Define dispersion based saccade.""" - def __post_init__(self): - super().__post_init__() + def __init__(self, positions: GazeFeatures.TimeStampedGazePositions = (), finished: bool = False, message: str = None, **kwargs): + + super().__init__(positions, finished, message, **kwargs) def draw(self, image: numpy.array, line_color: tuple = None): """Draw saccade into image. @@ -115,7 +105,6 @@ class Saccade(GazeFeatures.Saccade): cv2.line(image, (int(start_position[0]), int(start_position[1])), (int(last_position[0]), int(last_position[1])), line_color, 2) -@dataclass class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): """Implementation of the I-DT algorithm as described in: @@ -123,25 +112,36 @@ 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. """ - deviation_max_threshold: int|float - """Maximal distance allowed to consider a gaze movement as a fixation.""" - - duration_min_threshold: 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.""" - - def __post_init__(self): + def __init__(self, deviation_max_threshold: int|float, duration_min_threshold: int|float): super().__init__() + self.__deviation_max_threshold = deviation_max_threshold + self.__duration_min_threshold = duration_min_threshold + 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.""" + return self.__deviation_max_threshold + + @property + def duration_min_threshold(self): + """Get identifier duration min threshold.""" + return self.__duration_min_threshold + @DataFeatures.PipelineStepMethod - def identify(self, ts: int|float, gaze_position, terminate=False) -> GazeMovementType: + def identify(self, timestamp: int|float, gaze_position, terminate=False) -> GazeMovementType: # Ignore non valid gaze position if not gaze_position: @@ -151,9 +151,9 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): # Check if too much time elapsed since last valid gaze position if len(self.__valid_positions) > 0: - ts_last, _ = self.__valid_positions[-1] + ts_last = self.__valid_positions[-1].timestamp - if (ts - ts_last) > self.duration_min_threshold: + if (timestamp - ts_last) > self.__duration_min_threshold: # Get last movement last_movement = self.current_gaze_movement.finish() @@ -172,17 +172,14 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): # Store gaze positions until a minimal duration self.__valid_positions.append(gaze_position) - first_ts, _ = self.__valid_positions[0] - last_ts, _ = self.__valid_positions[-1] - # Once the minimal duration is reached - if last_ts - first_ts >= self.duration_min_threshold: + if self.__valid_positions.duration >= self.__duration_min_threshold: # Calculate the deviation of valid gaze positions deviation = Fixation(self.__valid_positions).deviation_max # Valid gaze positions deviation small enough - if deviation <= self.deviation_max_threshold: + if deviation <= self.__deviation_max_threshold: last_saccade = GazeFeatures.GazeMovement() @@ -190,8 +187,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): if len(self.__saccade_positions) > 0: # Copy oldest valid position into saccade positions - first_ts, first_position = self.__valid_positions[0] - self.__saccade_positions.append(first_position) + self.__saccade_positions.append(self.__valid_positions[0]) # Finish last saccade last_saccade = self.current_saccade.finish() @@ -214,8 +210,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): if len(self.__fixation_positions) > 0: # Copy most recent fixation position into saccade positions - last_ts, last_position = self.__fixation_positions[-1] - self.__saccade_positions.append(last_position) + self.__saccade_positions.append(self.__fixation_positions[-1]) # Finish last fixation last_fixation = self.current_fixation.finish() @@ -233,8 +228,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): return last_fixation if not terminate else self.current_saccade.finish() # Move oldest valid position into saccade positions - first_ts, first_position = self.__valid_positions.pop_first() - self.__saccade_positions.append(first_position) + self.__saccade_positions.append(self.__valid_positions.pop(0)) # Always return unvalid gaze movement at least return GazeFeatures.GazeMovement() diff --git a/src/argaze/GazeFeatures.py b/src/argaze/GazeFeatures.py index 71c643e..eac9e5c 100644 --- a/src/argaze/GazeFeatures.py +++ b/src/argaze/GazeFeatures.py @@ -520,31 +520,34 @@ class TimeStampedGazeMovements(DataFeatures.TimestampedObjectsList): GazeStatusType = TypeVar('GazeStatus', bound="GazeStatus") # Type definition for type annotation convenience -@dataclass(frozen=True) -class GazeStatus(GazePosition): - """Define gaze status as a gaze position belonging to an identified and indexed gaze movement.""" +class GazeStatus(list, DataFeatures.TimestampedObject): + """Define gaze status as a list of 1 or 2 (index, GazeMovementType) tuples. - movement_type: str = field(kw_only=True) - """GazeMovement type to which gaze position belongs.""" + Parameters: + position: the position that the status represents. + """ - movement_index: int = field(kw_only=True) - """GazeMovement index to which gaze positon belongs.""" + def __init__(self, position: GazePosition): - @classmethod - def from_position(cls, gaze_position: GazePosition, movement_type: str, movement_index: int) -> GazeStatusType: - """Initialize from a gaze position instance.""" + DataFeatures.TimestampedObject.__init__(self, timestamp=position.timestamp) + + self.__position = position + + @property + def position(self) -> GazePosition: + """Get gaze status position.""" + return self.__position + + def append(self, movement_index: int, movement_type:type): + """Append movement index and type.""" - return cls(gaze_position, precision=gaze_position.precision, movement_type=movement_type, movement_index=movement_index) + super().append((movement_index, movement_type)) TimeStampedGazeStatusType = TypeVar('TimeStampedGazeStatus', bound="TimeStampedGazeStatus") # Type definition for type annotation convenience 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. - """ + """Handle timestamped gaze status into a list.""" def __init__(self): @@ -568,7 +571,7 @@ class GazeMovementIdentifier(DataFeatures.PipelineStepObject): terminate: allows to notify identification algorithm that given gaze position will be the last one. Returns: - finished_gaze_movement: identified gaze movement once it is finished otherwise it returns unvalid gaze movement. + gaze_movement: identified gaze movement once it is finished otherwise it returns empty gaze movement. """ raise NotImplementedError('identify() method not implemented') @@ -612,57 +615,31 @@ class GazeMovementIdentifier(DataFeatures.PipelineStepObject): # Iterate on gaze positions for gaze_position in ts_gaze_positions: - finished_gaze_movement = self.identify(gaze_position.timestamp, gaze_position, terminate=(gaze_position.timestamp == last_ts)) - - if is_fixation(finished_gaze_movement): + gaze_movement = self.identify(gaze_position.timestamp, gaze_position, terminate=(gaze_position.timestamp == last_ts)) - ts_fixations.append(finished_gaze_movement) + if gaze_movement: # First gaze movement position is always shared with previous gaze movement - for movement_position in finished_gaze_movement: - - gaze_status = GazeStatus.from_position(movement_position, 'Fixation', len(ts_fixations)) - - if movement_position.timestamp != finished_gaze_movement.timestamp: - - ts_status.append([gaze_status]) - - else: - - try: - - ts_status[finished_gaze_movement.timestamp].append(gaze_status) - - except KeyError: - - ts_status[finished_gaze_movement.timestamp] = [gaze_status] - - elif is_saccade(finished_gaze_movement): - - ts_saccades.append(finished_gaze_movement) - - # First gaze movement position is always shared with previous gaze movement - for movement_position in finished_gaze_movement: - - gaze_status = GazeStatus.from_position(position, 'Saccade', len(ts_saccades)) - - if movement_position.timestamp != finished_gaze_movement.timestamp: - - ts_status.append([gaze_status]) + for movement_position in gaze_movement: - else: + # Is a status already exist for this position? + gaze_status = ts_status.look_for(movement_position.timestamp) - try: + if not gaze_status: + + gaze_status = GazeStatus(movement_position) + ts_status.append(gaze_status) - ts_status[finished_gaze_movement.timestamp].append(gaze_status) + gaze_status.append(len(ts_fixations), type(gaze_movement)) - except KeyError: + # Store gaze movment into the appropriate list + if is_fixation(gaze_movement): - ts_status[finished_gaze_movement.timestamp] = [gaze_status] + ts_fixations.append(gaze_movement) - else: + elif is_saccade(gaze_movement): - continue + ts_saccades.append(gaze_movement) return ts_fixations, ts_saccades, ts_status @@ -674,7 +651,7 @@ class GazeMovementIdentifier(DataFeatures.PipelineStepObject): Returns: timestamp: first gaze position date of identified gaze movement - finished_gaze_movement: identified gaze movement once it is finished + gaze_movement: identified gaze movement once it is finished """ assert(type(ts_gaze_positions) == TimeStampedGazePositions) @@ -685,11 +662,11 @@ class GazeMovementIdentifier(DataFeatures.PipelineStepObject): # Iterate on gaze positions for gaze_position in ts_gaze_positions: - finished_gaze_movement = self.identify(gaze_position.timestamp, gaze_position, terminate=(gaze_position.timestamp == last_ts)) + gaze_movement = self.identify(gaze_position.timestamp, gaze_position, terminate=(gaze_position.timestamp == last_ts)) - if finished_gaze_movement: + if gaze_movement: - yield finished_gaze_movement + yield gaze_movement ScanStepType = TypeVar('ScanStep', bound="ScanStep") # Type definition for type annotation convenience |