Merge pull request #22 from childmindresearch/alperkent/issue21

feature/issue-21/implement-velocity-features

Author: alperkent

.pre-commit-config.yaml

@@ -20,7 +20,7 @@ repos:
     - --autofix
     - --indent=2
   - id: pretty-format-toml
-    exclude: ^poetry.lock$
+    exclude: ^uv.lock$
     args:
     - --autofix
     - --indent=2

src/graphomotor/core/config.py

@@ -0,0 +1,33 @@
+"""Configuration module for Graphomotor."""
+
+import logging
+
+import numpy as np
+
+
+class _SpiralConfig:
+    """Configuration for the reference spiral generation."""
+
+    SPIRAL_CENTER_X = 50
+    SPIRAL_CENTER_Y = 50
+    SPIRAL_START_RADIUS = 0
+    SPIRAL_GROWTH_RATE = 1.075
+    SPIRAL_START_ANGLE = 0
+    SPIRAL_END_ANGLE = 8 * np.pi
+    SPIRAL_NUM_POINTS = 10000
+
+
+def get_logger() -> logging.Logger:
+    """Get the Graphomotor logger."""
+    logger = logging.getLogger("graphomotor")
+    if logger.handlers:
+        return logger
+    logger.setLevel(logging.INFO)
+    formatter = logging.Formatter(
+        "%(asctime)s - %(name)s - %(levelname)s - "
+        "%(filename)s:%(lineno)s - %(funcName)s - %(message)s",
+    )
+    handler = logging.StreamHandler()
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    return logger

src/graphomotor/features/time.py

@@ -3,7 +3,7 @@
 from graphomotor.core import models
 
 
-def get_task_duration(spiral: models.Spiral) -> dict:
+def get_task_duration(spiral: models.Spiral) -> dict[str, float]:
     """Calculate the total duration of a spiral drawing task.
 
     Args:

src/graphomotor/features/velocity.py

@@ -0,0 +1,81 @@
+"""Feature extraction module for velocity-based metrics in spiral drawing data."""
+
+import numpy as np
+from scipy import stats
+
+from graphomotor.core import models
+from graphomotor.utils import center_spiral
+
+
+def _calculate_statistics(values: np.ndarray, name: str) -> dict[str, float]:
+    """Helper function to calculate statistics for a given array.
+
+    Args:
+        values: 1-D Numpy array of numerical values.
+        name: Name prefix for the statistics (e.g., "linear_velocity").
+
+    Returns:
+        Dictionary containing calculated metrics (sum, median, variation, skewness,
+        kurtosis) with keys prefixed by the provided name.
+    """
+    return {
+        f"{name}_sum": np.sum(np.abs(values)),
+        f"{name}_median": np.median(np.abs(values)),
+        f"{name}_variation": stats.variation(values),
+        f"{name}_skewness": stats.skew(values),
+        f"{name}_kurtosis": stats.kurtosis(values),
+    }
+
+
+def calculate_velocity_metrics(spiral: models.Spiral) -> dict[str, float]:
+    """Calculate velocity-based metrics from spiral drawing data.
+
+    This function computes three types of velocity metrics by calculating the difference
+    between consecutive points in the spiral drawing data. The three types of velocity
+    are:
+        1. Linear velocity: The magnitude of change of Euclidean distance in pixels
+           per second. This is calculated as the square root of the sum of squares of
+           the differences in x and y coordinates divided by the difference in time.
+        2. Radial velocity: The magnitude of change of distance from center (radius) in
+           pixels per second. Radius is calculated as the square root of the sum of
+           squares of x and y coordinates.
+        3. Angular velocity: The magnitude of change of angle in radians per second.
+           Angle is calculated using the arctangent of y coordinates divided by x
+           coordinates, and then unwrapped to maintain continuity across the -π to π
+           boundary.
+
+    For each velocity type, the following metrics are calculated:
+        - Sum: Sum of absolute velocity values
+        - Median: Median of absolute velocity values
+        - Variation: Coefficient of variation
+        - Skewness: Asymmetry of the velocity distribution
+        - Kurtosis: Tailedness of the velocity distribution
+
+    Args:
+        spiral: Spiral object containing drawing data.
+
+    Returns:
+        Dictionary containing calculated velocity metrics.
+    """
+    spiral = center_spiral.center_spiral(spiral)
+    x = spiral.data["x"].values
+    y = spiral.data["y"].values
+    time = spiral.data["seconds"].values
+    radius = np.sqrt(x**2 + y**2)
+    theta = np.unwrap(np.arctan2(y, x))
+
+    dx = np.diff(x)
+    dy = np.diff(y)
+    dt = np.diff(time)
+    dr = np.diff(radius)
+    dtheta = np.diff(theta)
+
+    linear_velocity = np.sqrt(dx**2 + dy**2) / dt
+    radial_velocity = dr / dt
+    angular_velocity = dtheta / dt
+
+    return {
+        **_calculate_statistics(linear_velocity, "linear_velocity"),
+        **_calculate_statistics(radial_velocity, "radial_velocity"),
+        **_calculate_statistics(angular_velocity, "angular_velocity"),
+    }

src/graphomotor/utils/center_spiral.py

@@ -0,0 +1,18 @@
+"""Utility functions for centering a spiral."""
+
+from graphomotor.core import config, models
+
+
+def center_spiral(spiral: models.Spiral) -> models.Spiral:
+    """Center a spiral by translating it to the origin.
+
+    Args:
+        spiral: Spiral object containing spiral data.
+
+    Returns:
+        Spiral object with centered spiral data.
+    """
+    spiral.data["x"] -= config._SpiralConfig.SPIRAL_CENTER_X
+    spiral.data["y"] -= config._SpiralConfig.SPIRAL_CENTER_Y
+
+    return spiral

src/graphomotor/utils/reference_spiral.py …motor/utils/generate_reference_spiral.pysrc/graphomotor/utils/reference_spiral.py renamed to src/graphomotor/utils/generate_reference_spiral.py src/graphomotor/utils/reference_spiral.py renamed to src/graphomotor/utils/generate_reference_spiral.py

@@ -1,15 +1,9 @@
-"""Generate a reference spiral with equidistant points along its arc length."""
+"""Utility functions for generating an equidistant reference spiral."""
 
 import numpy as np
 from scipy import integrate, optimize
 
-_SPIRAL_CENTER_X = 50
-_SPIRAL_CENTER_Y = 50
-_SPIRAL_START_RADIUS = 0
-_SPIRAL_GROWTH_RATE = 1.075
-_SPIRAL_START_ANGLE = 0
-_SPIRAL_END_ANGLE = 8 * np.pi
-_SPIRAL_NUM_POINTS = 10000
+from graphomotor.core.config import _SpiralConfig
 
 
 def _arc_length_integrand(t: float) -> float:
@@ -21,8 +15,8 @@ def _arc_length_integrand(t: float) -> float:
     Returns:
         Differential arc length value.
     """
-    r_t = _SPIRAL_START_RADIUS + _SPIRAL_GROWTH_RATE * t
-    return np.sqrt(r_t**2 + _SPIRAL_GROWTH_RATE**2)
+    r_t = _SpiralConfig.SPIRAL_START_RADIUS + _SpiralConfig.SPIRAL_GROWTH_RATE * t
+    return np.sqrt(r_t**2 + _SpiralConfig.SPIRAL_GROWTH_RATE**2)
 
 
 def _calculate_arc_length(theta: float) -> float:
@@ -35,7 +29,7 @@ def _calculate_arc_length(theta: float) -> float:
         The arc length of the spiral from _SPIRAL_START_ANGLE to theta.
     """
     return integrate.quad(
-        lambda t: _arc_length_integrand(t), _SPIRAL_START_ANGLE, theta
+        lambda t: _arc_length_integrand(t), _SpiralConfig.SPIRAL_START_ANGLE, theta
     )[0]
 
 
@@ -50,7 +44,7 @@ def _find_theta_for_arc_length(target_arc_length: float) -> float:
     """
     solution = optimize.root_scalar(
         lambda theta: _calculate_arc_length(theta) - target_arc_length,
-        bracket=[_SPIRAL_START_ANGLE, _SPIRAL_END_ANGLE],
+        bracket=[_SpiralConfig.SPIRAL_START_ANGLE, _SpiralConfig.SPIRAL_END_ANGLE],
     )
     return solution.root
 
@@ -86,14 +80,19 @@ def generate_reference_spiral() -> np.ndarray:
     Returns:
         Array with shape (N, 2) containing Cartesian coordinates of the spiral points.
     """
-    total_arc_length = _calculate_arc_length(_SPIRAL_END_ANGLE)
+    total_arc_length = _calculate_arc_length(_SpiralConfig.SPIRAL_END_ANGLE)
 
-    arc_length_values = np.linspace(0, total_arc_length, _SPIRAL_NUM_POINTS)
+    arc_length_values = np.linspace(
+        0, total_arc_length, _SpiralConfig.SPIRAL_NUM_POINTS
+    )
 
     theta_values = np.array([_find_theta_for_arc_length(s) for s in arc_length_values])
 
-    r_values = _SPIRAL_START_RADIUS + _SPIRAL_GROWTH_RATE * theta_values
-    x_values = _SPIRAL_CENTER_X + r_values * np.cos(theta_values)
-    y_values = _SPIRAL_CENTER_Y + r_values * np.sin(theta_values)
+    r_values = (
+        _SpiralConfig.SPIRAL_START_RADIUS
+        + _SpiralConfig.SPIRAL_GROWTH_RATE * theta_values
+    )
+    x_values = _SpiralConfig.SPIRAL_CENTER_X + r_values * np.cos(theta_values)
+    y_values = _SpiralConfig.SPIRAL_CENTER_Y + r_values * np.sin(theta_values)
 
     return np.column_stack((x_values, y_values))

tests/conftest.py

@@ -8,7 +8,7 @@
 import pytest
 
 from graphomotor.core import models
-from graphomotor.utils import reference_spiral
+from graphomotor.utils import generate_reference_spiral
 
 
 @pytest.fixture
@@ -56,4 +56,4 @@ def valid_spiral(
 @pytest.fixture
 def ref_spiral() -> np.ndarray:
     """Create a reference spiral for testing."""
-    return reference_spiral.generate_reference_spiral()
+    return generate_reference_spiral.generate_reference_spiral()

tests/unit/test_config.py

@@ -0,0 +1,33 @@
+"""Test cases for config.py functions."""
+
+import logging
+
+import pytest
+
+from graphomotor.core import config
+
+
+def test_get_logger(caplog: pytest.LogCaptureFixture) -> None:
+    """Test the graphomotor logger with level set to INFO (20)."""
+    if logging.getLogger("graphomotor").handlers:
+        logging.getLogger("graphomotor").handlers.clear()
+    logger = config.get_logger()
+
+    logger.debug("Debug message here.")
+    logger.info("Info message here.")
+    logger.warning("Warning message here.")
+
+    assert logger.getEffectiveLevel() == logging.INFO
+    assert "Debug message here" not in caplog.text
+    assert "Info message here." in caplog.text
+    assert "Warning message here." in caplog.text
+
+
+def test_get_logger_second_call() -> None:
+    """Test get logger when a handler already exists."""
+    logger = config.get_logger()
+    second_logger = config.get_logger()
+
+    assert len(logger.handlers) == len(second_logger.handlers) == 1
+    assert logger.handlers[0] is second_logger.handlers[0]
+    assert logger is second_logger

tests/unit/test_drawing_error.py

@@ -20,4 +20,4 @@ def test_calculate_area_under_curve(valid_spiral: models.Spiral) -> None:
         valid_spiral, np.column_stack((x, y2))
     )["area_under_curve"]
 
-    assert np.isclose(calculated_area, expected_area, rtol=1e-3)
+    assert np.isclose(calculated_area, expected_area, atol=0, rtol=1e-3)

tests/unit/test_reference_spiral.py

@@ -2,33 +2,36 @@
 
 import numpy as np
 
-from graphomotor.utils import reference_spiral
+from graphomotor.core import config
+from graphomotor.utils import generate_reference_spiral
 
 
 def test_generate_reference_spiral() -> None:
     """Test the generation of a reference spiral."""
-    expected_mean_arc_length = reference_spiral._calculate_arc_length(
-        reference_spiral._SPIRAL_END_ANGLE
-    ) / (reference_spiral._SPIRAL_NUM_POINTS - 1)
+    expected_mean_arc_length = generate_reference_spiral._calculate_arc_length(
+        config._SpiralConfig.SPIRAL_END_ANGLE
+    ) / (config._SpiralConfig.SPIRAL_NUM_POINTS - 1)
 
-    spiral = reference_spiral.generate_reference_spiral()
+    spiral = generate_reference_spiral.generate_reference_spiral()
     arc_lengths = np.linalg.norm(spiral[1:] - spiral[:-1], axis=1)
     mean_arc_length = np.mean(arc_lengths)
 
     assert isinstance(spiral, np.ndarray)
-    assert spiral.shape == (reference_spiral._SPIRAL_NUM_POINTS, 2)
+    assert spiral.shape == (config._SpiralConfig.SPIRAL_NUM_POINTS, 2)
     assert np.array_equal(
         spiral[0],
-        [reference_spiral._SPIRAL_CENTER_X, reference_spiral._SPIRAL_CENTER_Y],
+        [config._SpiralConfig.SPIRAL_CENTER_X, config._SpiralConfig.SPIRAL_CENTER_Y],
     )
     assert np.allclose(
         spiral[-1],
         [
-            reference_spiral._SPIRAL_CENTER_X
-            + reference_spiral._SPIRAL_GROWTH_RATE * reference_spiral._SPIRAL_END_ANGLE,
-            reference_spiral._SPIRAL_CENTER_Y,
+            config._SpiralConfig.SPIRAL_CENTER_X
+            + config._SpiralConfig.SPIRAL_GROWTH_RATE
+            * config._SpiralConfig.SPIRAL_END_ANGLE,
+            config._SpiralConfig.SPIRAL_CENTER_Y,
         ],
-        atol=1e-8,
+        atol=0,
+        rtol=1e-8,
     )
-    assert np.allclose(arc_lengths, mean_arc_length, rtol=1e-3)
-    assert np.isclose(mean_arc_length, expected_mean_arc_length, rtol=1e-6)
+    assert np.allclose(arc_lengths, mean_arc_length, atol=0, rtol=1e-3)
+    assert np.isclose(mean_arc_length, expected_mean_arc_length, atol=0, rtol=1e-6)