Merge branch 'main' into 96-task-write-trails-velocity-feature-functions-calculate_segment_velocity_metrics

Author: cgmaiorano

src/graphomotor/core/models.py

@@ -2,7 +2,7 @@
 
 import dataclasses
 import datetime
-import typing
+from typing import Callable, List, Optional, Tuple
 
 import numpy as np
 import pandas as pd
@@ -95,7 +95,7 @@ def all(cls) -> set[str]:
     @classmethod
     def get_extractors(
         cls, spiral: Drawing, reference_spiral: np.ndarray
-    ) -> dict[str, typing.Callable[[], dict[str, float]]]:
+    ) -> dict[str, Callable[[], dict[str, float]]]:
         """Get all feature extractors with appropriate inputs.
 
         Args:
@@ -189,6 +189,7 @@ class LineSegment:
     points: pd.DataFrame
     is_error: bool
     line_number: int
+    ink_points: np.ndarray = dataclasses.field(default_factory=lambda: np.array([]))
 
     ink_time: float = 0.0
     think_time: float = 0.0
@@ -200,8 +201,62 @@ class LineSegment:
     smoothness: float = 0.0
     hesitation_count: int = 0
     hesitation_duration: float = 0.0
-    velocities: typing.List[float] = dataclasses.field(default_factory=list)
-    accelerations: typing.List[float] = dataclasses.field(default_factory=list)
+    velocities: List[float] = dataclasses.field(default_factory=list)
+    accelerations: List[float] = dataclasses.field(default_factory=list)
+
+    def valid_ink_trajectory(
+        self,
+        start_circle: CircleTarget,
+        end_circle: CircleTarget,
+    ) -> Tuple[Optional[int], Optional[int]]:
+        """Determine whether an ink trajectory exists from a start to end circle.
+
+        An "ink trajectory" is defined as the first contiguous sequence of
+        points that:
+        1. Begins **after** the pen leaves the start circle, and
+        2. Ends when the pen first enters the end circle.
+
+        The function scans point-by-point in order. The ink start index is the
+        first point whose (x, y) location is *outside* the start circle. The
+        ink end index is the first subsequent point whose (x, y) location falls
+        *inside* the end circle. If either of these conditions never occurs,
+        the trajectory is considered invalid. If a valid trajectory is found,
+        the ink_points attribute is updated to contain only the points within
+        this trajectory.
+
+        Args:
+            points: DataFrame of points with 'x' and 'y' columns.
+            start_circle: CircleTarget representing the start circle.
+            end_circle: CircleTarget representing the end circle.
+
+        Returns:
+            Tuple of (ink_start_idx: int, ink_end_idx: int) if valid
+            trajectory exists, else (None, None).
+        """
+        ink_start_idx = None
+        ink_end_idx = None
+
+        for idx, row in self.points.iterrows():
+            if (
+                not start_circle.contains_point(row["x"], row["y"])
+                and ink_start_idx is None
+            ):
+                ink_start_idx = idx
+
+            if ink_start_idx is not None and end_circle.contains_point(
+                row["x"], row["y"]
+            ):
+                ink_end_idx = idx
+                break
+
+        if (
+            ink_start_idx is not None
+            and ink_end_idx is not None
+            and ink_end_idx > ink_start_idx
+        ):
+            self.ink_points = self.points.iloc[ink_start_idx : ink_end_idx + 1].copy()
+
+        return ink_start_idx, ink_end_idx
 
     def calculate_path_optimality(
         self,

src/graphomotor/features/trails/drawing_metrics.py

@@ -1,5 +1,8 @@
 """Feature extraction module for drawing error-based metrics in trails drawing data."""
 
+import numpy as np
+import pandas as pd
+
 from graphomotor.core import models
 
 
@@ -35,9 +38,54 @@ def percent_accurate_paths(drawing: models.Drawing) -> dict[str, float]:
         raise ValueError(
             "DataFrame must contain 'correct_path' and 'actual_path' columns."
         )
-
     return {
         "percent_accurate_paths": (
             (drawing.data["correct_path"] == drawing.data["actual_path"]).mean() * 100
         )
     }
+
+
+def calculate_smoothness(points: pd.DataFrame) -> float:
+    """Calculate path smoothness based on Root Mean Square (RMS) curvature.
+
+    Represents the curvature per unit arc length.
+    Lower values indicate smoother drawings. Penalizes sharp corners (e.g., 90° turns)
+    and noisy corrections. Normalized by arc length to reduce sampling-rate dependence.
+
+    Args:
+        points: DataFrame representing drawing points.
+
+    Returns:
+        Smoothness metric as a float.
+    """
+    if len(points) < 3:
+        return 0.0
+
+    xy = points[["x", "y"]].to_numpy()
+
+    forward_vector = xy[1:-1] - xy[:-2]
+    backward_vector = xy[2:] - xy[1:-1]
+
+    forward_norm = np.linalg.norm(forward_vector, axis=1)
+    backward_norm = np.linalg.norm(backward_vector, axis=1)
+
+    valid = (forward_norm > 0) & (backward_norm > 0)
+    if not np.any(valid):
+        return 0.0
+
+    valid_forward_vector = forward_vector[valid]
+    valid_backward_vector = backward_vector[valid]
+    valid_forward_norm = forward_norm[valid]
+    valid_backward_norm = backward_norm[valid]
+
+    cos_angle = (valid_forward_vector * valid_backward_vector).sum(axis=1) / (
+        valid_forward_norm * valid_backward_norm
+    )
+    cos_angle = np.clip(cos_angle, -1.0, 1.0)
+
+    angles = np.arccos(cos_angle)
+
+    avg_segment_length = (valid_forward_norm + valid_backward_norm) / 2.0
+    curvatures = angles / avg_segment_length
+
+    return float(np.sqrt(np.mean(curvatures**2)))

src/graphomotor/features/trails/time.py

@@ -0,0 +1,52 @@
+"""Feature extraction module for time-based metrics in trails drawing data."""
+
+from graphomotor.core import models
+
+
+def calculate_total_error_time(drawing: models.Drawing) -> dict[str, float]:
+    """Calculate the total time spent making errors.
+
+    A contiguous "error chunk" is any sequence of rows where df["error"] != "E0".
+    The start and end of each chunk is defined as the midpoint between the last
+    timestamp with a "correct" entry and the first timestamp of an "error". The total
+    error time is the sum of the durations of all error chunks.
+
+    Args:
+        drawing: Drawing object containing drawing data.
+
+    Returns:
+        Dictionary containing the total time (s) spent in error states.
+    """
+    mask = drawing.data["error"] != "E0"
+    if not mask.any():
+        return {"total_error_time": 0.0}
+
+    error_change = mask.astype(int).diff()
+    chunk_starts = error_change[error_change == 1].index.tolist()
+    chunk_ends = error_change[error_change == -1].index.tolist()
+
+    if mask.iloc[0]:
+        chunk_starts = [0] + chunk_starts
+
+    if mask.iloc[-1]:
+        chunk_ends = chunk_ends + [len(drawing.data)]
+
+    seconds = drawing.data["seconds"].to_numpy()
+    total_error_time = 0.0
+
+    for start_idx, end_idx in zip(chunk_starts, chunk_ends):
+        start_time = (
+            (seconds[start_idx - 1] + seconds[start_idx]) / 2
+            if start_idx > 0
+            else seconds[0]
+        )
+
+        end_time = (
+            (seconds[end_idx - 1] + seconds[end_idx]) / 2
+            if end_idx < len(seconds)
+            else seconds[-1]
+        )
+
+        total_error_time += end_time - start_time
+
+    return {"total_error_time": float(total_error_time)}

src/graphomotor/utils/trails_utils.py

@@ -1,54 +1,12 @@
 """Utility functions for trails management."""
 
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List
 
 import pandas as pd
 
 from graphomotor.core import models
 
 
-def valid_ink_trajectory(
-    points: pd.DataFrame,
-    start_circle: models.CircleTarget,
-    end_circle: models.CircleTarget,
-) -> Tuple[Optional[int], Optional[int]]:
-    """Determine whether an ink trajectory exists from a start circle to an end circle.
-
-    An "ink trajectory" is defined as the first contiguous sequence of points that:
-      1. Begins **after** the pen leaves the start circle, and
-      2. Ends when the pen first enters the end circle.
-
-    The function scans point-by-point in order. The ink start index is the first point
-    whose (x, y) location is *outside* the start circle. The ink end index is the first
-    subsequent point whose (x, y) location falls *inside* the end circle. If either of
-    these conditions never occurs, the trajectory is considered invalid.
-
-    Args:
-        points: DataFrame of points with 'x' and 'y' columns.
-        start_circle: CircleTarget representing the start circle.
-        end_circle: CircleTarget representing the end circle.
-
-    Returns:
-        Tuple of (ink_start_idx: int, ink_end_idx: int) if valid trajectory exists,
-        else (None, None).
-    """
-    ink_start_idx = None
-    ink_end_idx = None
-
-    for idx, row in points.iterrows():
-        if (
-            not start_circle.contains_point(row["x"], row["y"])
-            and ink_start_idx is None
-        ):
-            ink_start_idx = idx
-
-        if ink_start_idx is not None and end_circle.contains_point(row["x"], row["y"]):
-            ink_end_idx = idx
-            break
-
-    return ink_start_idx, ink_end_idx
-
-
 def segment_lines(
     trail_data: pd.DataFrame,
     trail_id: str,

tests/unit/test_models.py

@@ -1,6 +1,7 @@
 """Test cases for the Spiral model."""
 
 import datetime
+from typing import Dict, cast
 
 import pandas as pd
 import pytest
@@ -137,6 +138,108 @@ def test_path_optimality_non_positive_distance() -> None:
     assert segment.path_optimality == 0.0
 
 
+@pytest.mark.parametrize(
+    "points_data,start_params,end_params,expected_start,expected_end,test_id",
+    [
+        (
+            {"x": [0, 1, 2, 3, 4, 5], "y": [0, 1, 2, 3, 4, 5]},
+            {"order": 1, "label": "A", "center_x": 0, "center_y": 0, "radius": 0.5},
+            {"order": 2, "label": "B", "center_x": 5, "center_y": 5, "radius": 0.5},
+            1,
+            5,
+            "valid_trajectory",
+        ),
+        (
+            {"x": [0.1, 0.2, 0.3], "y": [0.1, 0.2, 0.3]},
+            {"order": 1, "label": "A", "center_x": 0, "center_y": 0, "radius": 1.0},
+            {"order": 2, "label": "B", "center_x": 10, "center_y": 10, "radius": 1.0},
+            None,
+            None,
+            "no_exit_from_start",
+        ),
+        (
+            {"x": [0, 1, 2, 3], "y": [0, 1, 2, 3]},
+            {"order": 1, "label": "A", "center_x": 0, "center_y": 0, "radius": 0.5},
+            {"order": 2, "label": "B", "center_x": 10, "center_y": 10, "radius": 0.5},
+            1,
+            None,
+            "never_reaches_end",
+        ),
+        (
+            {"x": [], "y": []},
+            {"order": 1, "label": "A", "center_x": 0, "center_y": 0, "radius": 1.0},
+            {"order": 2, "label": "B", "center_x": 5, "center_y": 5, "radius": 1.0},
+            None,
+            None,
+            "empty_dataframe",
+        ),
+        (
+            {"x": [0.1], "y": [0.1]},
+            {"order": 1, "label": "A", "center_x": 0, "center_y": 0, "radius": 1.0},
+            {"order": 2, "label": "B", "center_x": 5, "center_y": 5, "radius": 1.0},
+            None,
+            None,
+            "single_point_in_start",
+        ),
+        (
+            {"x": [2], "y": [2]},
+            {"order": 1, "label": "A", "center_x": 0, "center_y": 0, "radius": 0.5},
+            {"order": 2, "label": "B", "center_x": 5, "center_y": 5, "radius": 0.5},
+            0,
+            None,
+            "single_point_outside_start",
+        ),
+        (
+            {"x": [0, 2.5, 5], "y": [0, 2.5, 5]},
+            {"order": 1, "label": "A", "center_x": 0, "center_y": 0, "radius": 0.5},
+            {"order": 2, "label": "B", "center_x": 2.5, "center_y": 2.5, "radius": 1.0},
+            1,
+            1,
+            "immediate_transition",
+        ),
+        (
+            {"x": [3, 4, 5], "y": [3, 4, 5]},
+            {"order": 1, "label": "A", "center_x": 0, "center_y": 0, "radius": 0.5},
+            {"order": 2, "label": "B", "center_x": 5, "center_y": 5, "radius": 0.5},
+            0,
+            2,
+            "first_point_outside_start",
+        ),
+    ],
+    ids=lambda x: x if isinstance(x, str) else "",
+)
+def test_valid_ink_trajectory(
+    points_data: Dict[str, list[float]],
+    start_params: dict[str, int | str | float],
+    end_params: dict[str, int | str | float],
+    expected_start: int | None,
+    expected_end: int | None,
+    test_id: str,
+) -> None:
+    """Test valid_ink_trajectory method with various point configurations.
+
+    Tests behavior with different circle boundaries.
+    """
+    points_df = pd.DataFrame(points_data)
+
+    start_circle = models.CircleTarget(**start_params)  # type: ignore[arg-type]
+    end_circle = models.CircleTarget(**end_params)  # type: ignore[arg-type]
+
+    line_segment = models.LineSegment(
+        start_label=cast(str, start_params["label"]),
+        end_label=cast(str, end_params["label"]),
+        points=points_df,
+        is_error=False,
+        line_number=1,
+    )
+    result_start, result_end = line_segment.valid_ink_trajectory(
+        start_circle, end_circle
+    )
+
+    assert result_start == expected_start, f"Start index mismatch for {test_id}"
+    assert result_end == expected_end, f"End index mismatch for {test_id}"
+
+    
 def test_uniform_motion() -> None:
     """Test with points moving at constant velocity."""
     points = pd.DataFrame({
@@ -148,10 +251,6 @@ def test_uniform_motion() -> None:
         start_label="1",
         end_label="2",
         points=points,
-        is_error=False,
-        line_number=1,
-    )
-
     segment.calculate_velocity_metrics(points)
 
     assert segment.distance == pytest.approx(3.0)
@@ -270,3 +369,4 @@ def test_stationary_motion() -> None:
     assert all(v == pytest.approx(0.0) for v in segment.velocities)
     assert len(segment.accelerations) == 1
     assert segment.accelerations[0] == pytest.approx(0.0)
+