speed

gianluca-capelo · gianluca-capelo · commit e9e3414c0225 · 2025-12-21T14:52:29.000-03:00
diff --git a/neurotask/neurotask/test/test_segmentation.py b/neurotask/neurotask/test/test_segmentation.py
@@ -1,11 +1,9 @@
 import pytest
 
-from neurotask.tmt.metrics.speed_metrics import InvalidSpeedError, NonMonotonicTimeError
 from neurotask.tmt.segmentation.segmentation import classify_cursor_positions_with_hesitation
 from neurotask.tmt.model.tmt_model import (
     Coordinate,
     CursorInfo,
-    TMTSubject,
     TMTTarget,
     TMTTrial,
     TrialType,
@@ -53,48 +51,6 @@ def _build_trial(cursor_trail: list[CursorInfo], targets: list[TMTTarget] = None
 class TestClassifyCursorPositionsWithHesitation:
     """Tests for classify_cursor_positions_with_hesitation function."""
 
-    def test_raises_invalid_speed_error_when_raise_on_error_true(self):
-        """
-        Con raise_on_error=True, lanza InvalidSpeedError si velocidad > 8.0 px/ms.
-        """
-        # Movimiento de 100px en 1ms = 100 px/ms (inválido, > 8.0)
-        cursor_trail = _build_cursor_trail([
-            (0.0, 0.0, 0.0),
-            (100.0, 0.0, 1.0),  # speed = 100 px/ms (invalid)
-            (102.0, 0.0, 2.0),  # speed = 2 px/ms (valid)
-        ])
-        trial = _build_trial(cursor_trail)
-
-        with pytest.raises(InvalidSpeedError, match="exceeds INVALID_SPEED_THRESHOLD"):
-            classify_cursor_positions_with_hesitation(
-                tmt_trial=trial,
-                target_radius=10.0,
-                speed_threshold=2.0,
-                consecutive_points=2,
-                raise_on_error=True
-            )
-
-    def test_raises_non_monotonic_error_when_raise_on_error_true(self):
-        """
-        Con raise_on_error=True, lanza NonMonotonicTimeError si tiempo retrocede.
-        """
-        # Time: 0 -> 2 -> 1 (retrocede)
-        cursor_trail = _build_cursor_trail([
-            (0.0, 0.0, 0.0),
-            (2.0, 0.0, 2.0),
-            (4.0, 0.0, 1.0),  # time goes backwards
-        ])
-        trial = _build_trial(cursor_trail)
-
-        with pytest.raises(NonMonotonicTimeError, match="current_cursor.time must be greater than previous_cursor.time"):
-            classify_cursor_positions_with_hesitation(
-                tmt_trial=trial,
-                target_radius=10.0,
-                speed_threshold=2.0,
-                consecutive_points=2,
-                raise_on_error=True
-            )
-
     def test_returns_correct_states_with_valid_data(self):
         """
         Verifica que la clasificación de estados sea correcta con datos válidos.
@@ -113,8 +69,7 @@ def test_returns_correct_states_with_valid_data(self):
             tmt_trial=trial,
             target_radius=10.0,
             speed_threshold=1.5,
-            consecutive_points=2,
-            raise_on_error=True
+            consecutive_points=2
         )
 
         # Verifica estructura correcta
@@ -125,14 +80,10 @@ def test_returns_correct_states_with_valid_data(self):
             assert state in ['Search', 'Travel', 'Hesitation']
             assert isinstance(cursor_info, CursorInfo)
 
-    @pytest.mark.xfail(reason="Bug conocido: con raise_on_error=False y datos inválidos, IndexError por len(speeds) < len(cursor_trail) - 1")
-    def test_with_invalid_data_and_raise_on_error_false(self):
+    def test_handles_invalid_speed_gracefully(self):
         """
-        Con raise_on_error=False, no lanza excepción y devuelve clasificación.
-        
-        NOTA: Este test documenta un bug conocido. Cuando hay velocidades inválidas
-        y raise_on_error=False, la lista speeds tiene menos elementos que cursor_trail - 1,
-        causando IndexError en speed_increases_over_consecutive_points().
+        Con velocidades inválidas, no lanza excepción y devuelve clasificación.
+        Las velocidades inválidas son marcadas como is_valid=False en SpeedResult.
         """
         # Trial con velocidad inválida (100 px/ms > 8.0)
         cursor_trail = _build_cursor_trail([
@@ -148,8 +99,7 @@ def test_with_invalid_data_and_raise_on_error_false(self):
             tmt_trial=trial,
             target_radius=10.0,
             speed_threshold=1.5,
-            consecutive_points=2,
-            raise_on_error=False
+            consecutive_points=2
         )
 
         # Verifica estructura correcta
@@ -160,3 +110,54 @@ def test_with_invalid_data_and_raise_on_error_false(self):
             assert state in ['Search', 'Travel', 'Hesitation']
             assert isinstance(cursor_info, CursorInfo)
 
+    def test_handles_non_monotonic_time_gracefully(self):
+        """
+        Con tiempos no monótonos, no lanza excepción y devuelve clasificación.
+        Los puntos con tiempo no monótono son marcados como is_valid=False en SpeedResult.
+        """
+        # Time: 0 -> 2 -> 1 (retrocede)
+        cursor_trail = _build_cursor_trail([
+            (0.0, 0.0, 0.0),
+            (2.0, 0.0, 2.0),
+            (4.0, 0.0, 1.0),  # time goes backwards
+            (6.0, 0.0, 3.0),  # valid again
+        ])
+        trial = _build_trial(cursor_trail)
+
+        # No debe lanzar excepción
+        result = classify_cursor_positions_with_hesitation(
+            tmt_trial=trial,
+            target_radius=10.0,
+            speed_threshold=1.5,
+            consecutive_points=2
+        )
+
+        # Verifica estructura correcta
+        assert len(result) == len(cursor_trail)
+        
+        # Cada elemento es (estado, CursorInfo)
+        for state, cursor_info in result:
+            assert state in ['Search', 'Travel', 'Hesitation']
+            assert isinstance(cursor_info, CursorInfo)
+
+    def test_first_point_is_search_on_target(self):
+        """
+        Verifica que el primer punto sea Search cuando está sobre el target.
+        """
+        # Cursor empieza sobre el target (0,0)
+        cursor_trail = _build_cursor_trail([
+            (0.0, 0.0, 0.0),   # sobre target 1
+            (2.0, 0.0, 1.0),
+            (4.0, 0.0, 2.0),
+        ])
+        trial = _build_trial(cursor_trail)
+
+        result = classify_cursor_positions_with_hesitation(
+            tmt_trial=trial,
+            target_radius=10.0,
+            speed_threshold=1.5,
+            consecutive_points=2
+        )
+
+        # El primer punto debe ser Search
+        assert result[0][0] == 'Search'
diff --git a/neurotask/neurotask/tmt/metrics/speed_metrics.py b/neurotask/neurotask/tmt/metrics/speed_metrics.py
@@ -1,7 +1,13 @@
 import logging
-from typing import Dict, Any, Tuple, List
+from typing import Dict, Any, Tuple, List, NamedTuple
 
 import numpy as np
+
+
+class SpeedResult(NamedTuple):
+    """Result of a speed calculation with validity flag."""
+    is_valid: bool
+    value: float
 from neurotask.tmt.metrics.base_metric import BaseMetricCalculator
 from neurotask.tmt.metrics.distance_calculation import calculate_distance
 from neurotask.tmt.model.tmt_model import TMTTrial, CursorInfo, TMTSubject, TMTTarget
@@ -104,6 +110,45 @@ def calculate_speeds_between_cursor_positions(trial: TMTTrial, raise_on_error: b
     return calculate_speeds(cursor_trail_from_first_click, raise_on_error)
 
 
+def calculate_speeds_with_validity(cursor_trail: List[CursorInfo]) -> List[SpeedResult]:
+    """
+    Calculate speeds between consecutive cursor positions with validity flags.
+
+    Args:
+        cursor_trail: List of cursor positions.
+
+    Returns:
+        List of SpeedResult. Always len(result) == len(cursor_trail) - 1.
+        If is_valid=False, value=0.0.
+    """
+    if len(cursor_trail) < 2:
+        raise ValueError("At least two points are required to calculate velocity")
+
+    results = []
+    for i in range(1, len(cursor_trail)):
+        try:
+            speed = calculate_speed(cursor_trail[i], cursor_trail[i - 1])
+            results.append(SpeedResult(is_valid=True, value=speed))
+        except (InvalidSpeedError, NonMonotonicTimeError):
+            results.append(SpeedResult(is_valid=False, value=0.0))
+
+    return results
+
+
+def calculate_speeds_between_cursor_positions_with_validity(trial: TMTTrial) -> List[SpeedResult]:
+    """
+    Calculate speeds between cursor positions with validity flags for a trial.
+
+    Args:
+        trial: The TMT trial.
+
+    Returns:
+        List of SpeedResult. Always len(result) == len(cursor_trail) - 1.
+    """
+    cursor_trail = trial.get_cursor_trail_from_start()
+    return calculate_speeds_with_validity(cursor_trail)
+
+
 def calculate_speeds(cursor_trail: List[CursorInfo], raise_on_error: bool = False) -> List[float]:
     if len(cursor_trail) < 2:
         raise ValueError("At least two points are required to calculate velocity")
diff --git a/neurotask/neurotask/tmt/segmentation/segmentation.py b/neurotask/neurotask/tmt/segmentation/segmentation.py
@@ -6,17 +6,19 @@
 
 from neurotask.tmt.metrics.speed_metrics import (
     calculate_speeds_between_cursor_positions,
+    calculate_speeds_between_cursor_positions_with_validity,
     calculate_speeds,
     calculate_speed,
     InvalidSpeedError,
-    NonMonotonicTimeError
+    NonMonotonicTimeError,
+    SpeedResult
 )
 from ..metrics.distance_calculation import calculate_distance
 from ..model.tmt_model import CursorInfo, TMTTrial, TMTExperiment, Coordinate, TMTSubject, TrialType
 
 
 def speed_increases_over_consecutive_points(
-        speeds: List[float],
+        speeds: List[SpeedResult],
         cursor_index: int,
         speed_threshold: float,
         consecutive_points: int
@@ -25,7 +27,7 @@ def speed_increases_over_consecutive_points(
     Determines if the speed has increased over a specified number of consecutive points beyond a given speed threshold.
 
     Parameters:
-    - speeds: List of speed values between cursor positions.
+    - speeds: List of SpeedResult values between cursor positions.
     - cursor_index: The current index in the cursor trail (starting from 0).
     - speed_threshold: The minimum increase in speed between consecutive points to consider.
     - consecutive_points: Number of consecutive points over which the speed must increase.
@@ -42,14 +44,16 @@ def speed_increases_over_consecutive_points(
     for i in range(speed_index - consecutive_points + 1, speed_index + 1):
         if i <= 0:
             return False  # Not enough data
-        current_speed = speeds[i]
-        if current_speed <= speed_threshold:
+        speed_result = speeds[i]
+        if not speed_result.is_valid:
+            return False  # Invalid speed, cannot evaluate
+        if speed_result.value <= speed_threshold:
             return False  # Speed did not increase sufficiently
     return True
 
 
 def speed_decreases_over_consecutive_points(
-        speeds: List[float],
+        speeds: List[SpeedResult],
         cursor_index: int,
         speed_threshold: float,
         consecutive_points: int
@@ -58,7 +62,7 @@ def speed_decreases_over_consecutive_points(
     Determines if the speed has decreased over a specified number of consecutive points beyond a given speed threshold.
 
     Parameters:
-    - speeds: List of speed values between cursor positions.
+    - speeds: List of SpeedResult values between cursor positions.
     - cursor_index: The current index in the cursor trail (starting from 0).
     - speed_threshold: The minimum decrease in speed between consecutive points to consider.
     - consecutive_points: Number of consecutive points over which the speed must decrease.
@@ -75,8 +79,10 @@ def speed_decreases_over_consecutive_points(
     for i in range(speed_index - consecutive_points + 1, speed_index + 1):
         if i <= 0:
             return False  # Not enough data
-        current_speed = speeds[i]
-        if current_speed > speed_threshold:
+        speed_result = speeds[i]
+        if not speed_result.is_valid:
+            return False  # Invalid speed, cannot evaluate
+        if speed_result.value > speed_threshold:
             return False  # Speed did not decrease sufficiently
     return True
 
@@ -85,12 +91,11 @@ def classify_cursor_positions_with_hesitation(
         tmt_trial: TMTTrial,
         target_radius: float,
         speed_threshold,
-        consecutive_points=5,
-        raise_on_error: bool = False
+        consecutive_points=5
 ) -> List[Tuple[str, CursorInfo]]:
     classified_positions = []
     cursor_trail = tmt_trial.get_cursor_trail_from_start()
-    speeds = calculate_speeds_between_cursor_positions(tmt_trial, raise_on_error)
+    speeds = calculate_speeds_between_cursor_positions_with_validity(tmt_trial)
     over_target_flags = calculate_over_targets(cursor_trail, target_radius, tmt_trial.stimuli)
 
     current_state = 'Search'
