Refactor/issue 53/refactor spiral specific to generic (#54)

* Initial changes

* Update pyproject.toml

* More fixes

* Typos

* Update test_models.py

* Change to task specific orchestrator

* Update reader.py

Adding all possible columns

* Refactor reader for any drawing columns

* Lots of changes

* Update reader.py

mypy fix

* Create test_io_utils.py

Added tests for missing columns

* Renaming for deidentification

* Renaming and fixing tests

* Addressing PR comments

* Update spiral_orchestrator.py

Author: Asanto32

pyproject.toml

@@ -1,9 +1,11 @@
 [project]
 name = "graphomotor"
-version = "0.2.0"
+version = "0.2.1"
 description = "A Python toolkit for analysis of graphomotor data collected via Curious"
 authors = [
   {name = "Alp Erkent", email = "alp.erkent@childmind.org"},
+  {name = "Celia Maiorano", email = "celia.maiorano@childmind.org"},
+  {name = "Iktae Kim", email = "iktae.kim@childmind.org"},
   {name = "Adam Santorelli", email = "adam.santorelli@childmind.org"}
 ]
 license = "LGPL-2.1"

src/graphomotor/core/cli.py

@@ -6,7 +6,7 @@
 
 import typer
 
-from graphomotor.core import config, orchestrator
+from graphomotor.core import config, spiral_orchestrator
 from graphomotor.plot import feature_plots, spiral_plots
 
 logger = config.get_logger()
@@ -85,7 +85,7 @@ def main(
 
 
 @app.command(
-    name="extract",
+    name="extract-spiral",
     help=(
         "Extract features from spiral drawing data. "
         "Supports both single-file and batch (directory) processing."
@@ -95,7 +95,7 @@ def main(
         "https://github.com/childmindresearch/graphomotor?tab=readme-ov-file#feature-extraction."
     ),
 )
-def extract(
+def extract_spiral(
     input_path: typing.Annotated[
         pathlib.Path,
         typer.Argument(
@@ -196,7 +196,7 @@ def extract(
     """Extract features from spiral drawing data."""
     logger.debug(f"Running Graphomotor pipeline with these arguments: {locals()}")
 
-    config_params: dict[orchestrator.ConfigParams, float | int] = {
+    config_params: dict[spiral_orchestrator.ConfigParams, float | int] = {
         "center_x": center_x,
         "center_y": center_y,
         "start_radius": start_radius,
@@ -207,11 +207,11 @@ def extract(
     }
 
     try:
-        orchestrator.run_pipeline(
+        spiral_orchestrator.run_pipeline(
             input_path=input_path,
             output_path=output_path,
             feature_categories=typing.cast(
-                list[orchestrator.FeatureCategories], features
+                list[spiral_orchestrator.FeatureCategories], features
             ),
             config_params=config_params,
         )

src/graphomotor/core/models.py

@@ -8,12 +8,13 @@
 import pydantic
 
 
-class Spiral(pydantic.BaseModel):
-    """Class representing a spiral drawing, encapsulating both raw data and metadata.
+class Drawing(pydantic.BaseModel):
+    """Class representing a drawing task, encapsulating both raw data and metadata.
 
     Attributes:
         data: DataFrame containing drawing data with required columns (line_number, x,
             y, UTC_Timestamp, seconds).
+        task_name: Name of the drawing task (e.g., 'spiral', 'trails', etc.).
         metadata: Dictionary containing metadata about the spiral:
             - id: Unique identifier for the participant,
             - hand: Hand used ('Dom' for dominant, 'NonDom' for non-dominant),
@@ -25,6 +26,7 @@ class Spiral(pydantic.BaseModel):
     model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
 
     data: pd.DataFrame
+    task_name: str
     metadata: dict[str, str | datetime.datetime]
 
     @pydantic.field_validator("data")
@@ -67,22 +69,10 @@ def validate_metadata(cls, v: dict) -> dict:
         if len(v["id"]) != 7:
             raise ValueError("'id' must be 7 digits long")
 
-        if v["hand"] not in ["Dom", "NonDom"]:
-            raise ValueError("'hand' must be either 'Dom' or 'NonDom'")
-
-        valid_tasks = ["spiral_trace", "spiral_recall"]
-        valid_tasks_trials = [
-            f"{prefix}{i}" for prefix in valid_tasks for i in range(1, 6)
-        ]
-        if v["task"] not in valid_tasks_trials:
-            raise ValueError(
-                "'task' must be either 'spiral_trace' or 'spiral_recall', numbered 1-5"
-            )
-
         return v
 
 
-class FeatureCategories:
+class SpiralFeatureCategories:
     """Class to hold valid feature categories for Graphomotor."""
 
     DURATION = "duration"
@@ -102,7 +92,7 @@ def all(cls) -> set[str]:
 
     @classmethod
     def get_extractors(
-        cls, spiral: Spiral, reference_spiral: np.ndarray
+        cls, spiral: Drawing, reference_spiral: np.ndarray
     ) -> dict[str, typing.Callable[[], dict[str, float]]]:
         """Get all feature extractors with appropriate inputs.
 

src/graphomotor/core/orchestrator.py …/graphomotor/core/spiral_orchestrator.pysrc/graphomotor/core/orchestrator.py renamed to src/graphomotor/core/spiral_orchestrator.py src/graphomotor/core/orchestrator.py renamed to src/graphomotor/core/spiral_orchestrator.py

@@ -1,8 +1,9 @@
-"""Runner for graphomotor."""
+"""Orchestrator for the Graphomotor spiral feature extraction pipeline."""
 
 import dataclasses
 import datetime
 import pathlib
+import re
 import time
 import typing
 
@@ -28,6 +29,64 @@
 ]
 
 
+def _parse_spiral_metadata(filename: str) -> dict[str, str]:
+    """Extract metadata from spiral drawing filename.
+
+    The function parses filenames of Curious exports of drawing data that are
+    typically formatted as '[5123456]curious-ID-spiral_trace2_NonDom'. It extracts
+    the participant ID (the value within the brackets), task name ('spiral_trace' or
+    'spiral_recall', followed by the trial number from 1 to 5), and hand used (dominant
+    or non-dominant). Regular expressions are used to match the expected pattern
+    and extract the relevant components.
+
+
+    Args:
+        filename: Filename of the spiral drawing CSV file from Curious export.
+
+    Returns:
+        Dictionary containing extracted metadata:
+            - id: Participant ID (e.g., '5123456')
+            - hand: Hand used for drawing ('Dom' or 'NonDom')
+            - task: Task name and trial number (e.g., 'spiral_trace2')
+
+    Raises:
+        ValueError: If filename does not match expected pattern.
+    """
+    pattern = r"\[(\d+)\].*-([^_]+)_([^_]+)_(\w+)$"
+    match = re.match(pattern, filename)
+
+    if match:
+        id, task_name, task_detail, hand = match.groups()
+        metadata = {
+            "id": id,
+            "hand": hand,
+            "task": f"{task_name}_{task_detail}",
+        }
+        return metadata
+
+    raise ValueError(f"Filename does not match expected pattern: {filename}")
+
+
+def _validate_spiral_metadata(metadata: dict[str, str | datetime.datetime]) -> None:
+    """Validate metadata extracted from spiral drawing filename.
+
+    Args:
+        metadata: Dictionary containing extracted metadata.
+
+    Raises:
+        ValueError: If metadata is invalid.
+    """
+    if metadata["hand"] not in ["Dom", "NonDom"]:
+        raise ValueError("'hand' must be either 'Dom' or 'NonDom'")
+
+    valid_tasks = ["spiral_trace", "spiral_recall"]
+    valid_tasks_trials = [f"{prefix}{i}" for prefix in valid_tasks for i in range(1, 6)]
+    if metadata["task"] not in valid_tasks_trials:
+        raise ValueError(
+            "'task' must be either 'spiral_trace' or 'spiral_recall', numbered 1-5"
+        )
+
+
 def _validate_feature_categories(
     feature_categories: list[FeatureCategories],
 ) -> set[str]:
@@ -43,7 +102,7 @@ def _validate_feature_categories(
         ValueError: If no valid feature categories are provided.
     """
     feature_categories_set: set[str] = set(feature_categories)
-    supported_categories_set = models.FeatureCategories.all()
+    supported_categories_set = models.SpiralFeatureCategories.all()
     unknown_categories = feature_categories_set - supported_categories_set
     valid_requested_categories = feature_categories_set & supported_categories_set
 
@@ -65,7 +124,7 @@ def _validate_feature_categories(
 
 
 def extract_features(
-    spiral: models.Spiral,
+    spiral: models.Drawing,
     feature_categories: list[str],
     reference_spiral: np.ndarray,
 ) -> dict[str, str]:
@@ -83,7 +142,7 @@ def extract_features(
     Returns:
         Dictionary containing the extracted features with metadata.
     """
-    feature_extractors = models.FeatureCategories.get_extractors(
+    feature_extractors = models.SpiralFeatureCategories.get_extractors(
         spiral, reference_spiral
     )
 
@@ -168,7 +227,9 @@ def _run_file(
     Returns:
         Dictionary containing the extracted features with metadata.
     """
-    spiral = reader.load_spiral(input_path)
+    spiral = reader.load_drawing_data(input_path)
+    spiral.metadata.update(_parse_spiral_metadata(input_path.stem))
+    _validate_spiral_metadata(spiral.metadata)
     centered_spiral = center_spiral.center_spiral(spiral)
     reference_spiral = generate_reference_spiral.generate_reference_spiral(
         spiral_config
@@ -326,7 +387,7 @@ def run_pipeline(
         valid_categories = sorted(_validate_feature_categories(feature_categories))
         logger.debug(f"Requested feature categories: {valid_categories}")
     else:
-        valid_categories = [*models.FeatureCategories.all()]
+        valid_categories = [*models.SpiralFeatureCategories.all()]
         logger.debug(f"Using default feature categories: {valid_categories}")
 
     if config_params and config_params != dataclasses.asdict(config.SpiralConfig()):

src/graphomotor/features/distance.py

@@ -30,7 +30,7 @@ def _segment_data(data: np.ndarray, start_prop: float, end_prop: float) -> np.nd
 
 
 def calculate_hausdorff_metrics(
-    spiral: models.Spiral, reference_spiral: np.ndarray
+    spiral: models.Drawing, reference_spiral: np.ndarray
 ) -> dict[str, float]:
     """Calculate Hausdorff distance metrics for a spiral object.
 

src/graphomotor/features/drawing_error.py

@@ -7,7 +7,7 @@
 
 
 def calculate_area_under_curve(
-    drawn_spiral: models.Spiral, reference_spiral: np.ndarray
+    drawn_spiral: models.Drawing, reference_spiral: np.ndarray
 ) -> dict[str, float]:
     """Calculate the area between drawn and reference spirals.
 

src/graphomotor/features/time.py

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

src/graphomotor/features/velocity.py

@@ -26,7 +26,7 @@ def _calculate_statistics(values: np.ndarray, name: str) -> dict[str, float]:
     }
 
 
-def calculate_velocity_metrics(spiral: models.Spiral) -> dict[str, float]:
+def calculate_velocity_metrics(spiral: models.Drawing) -> dict[str, float]:
     """Calculate velocity-based metrics from spiral drawing data.
 
     This function computes three types of velocity metrics by calculating the difference

src/graphomotor/io/io_utils.py

@@ -0,0 +1,37 @@
+"""This module contains utility functions for reading data files."""
+
+import pandas as pd
+
+DTYPE_MAP = {
+    "line_number": "int",
+    "x": "float",
+    "y": "float",
+    "UTC_Timestamp": "float",
+    "seconds": "float",
+    "epoch_time_in_seconds_start": "float",
+    "error": "str",
+    "correct_path": "str",
+    "actual_path": "str",
+    "total_time": "float",
+    "total_number_of_errors": "int",
+}
+
+
+def _check_missing_columns(data: pd.DataFrame, task_name: str) -> None:
+    """Check for missing columns in the DataFrame.
+
+    Args:
+        data: DataFrame containing spiral drawing data.
+        task_name: Name of the drawing task.
+
+    Raises:
+        KeyError: If any required columns are missing.
+    """
+    if "trail" in task_name.lower():
+        required_columns = list(DTYPE_MAP.keys())
+    else:
+        required_columns = list(DTYPE_MAP.keys())[:6]
+
+    missing_columns = set(required_columns) - set(data.columns)
+    if missing_columns:
+        raise KeyError(f"Missing required columns: {', '.join(missing_columns)}")