Standardize all feature functions to return dictionaries

Author: sanketsabharwal

pyeyesweb/low_level/equilibrium.py

@@ -31,28 +31,28 @@ class Equilibrium:
     >>> left = np.array([0, 0, 0])
     >>> right = np.array([400, 0, 0])
     >>> barycenter = np.array([200, 50, 0])
-    >>> value, angle = eq(left, right, barycenter)
-    >>> round(value, 2)
+    >>> result = eq(left, right, barycenter)
+    >>> round(result['value'], 2)
     0.91
-    >>> round(angle, 1)
+    >>> round(result['angle'], 1)
     0.0
 
     # Using 2D coordinates (z is optional)
     >>> left_2d = np.array([0, 0])
     >>> right_2d = np.array([400, 0])
     >>> barycenter_2d = np.array([200, 50])
-    >>> value_2d, angle_2d = eq(left_2d, right_2d, barycenter_2d)
-    >>> round(value_2d, 2)
+    >>> result_2d = eq(left_2d, right_2d, barycenter_2d)
+    >>> round(result_2d['value'], 2)
     0.91
-    >>> round(angle_2d, 1)
+    >>> round(result_2d['angle'], 1)
     0.0
     """
 
     def __init__(self, margin_mm=100, y_weight=0.5):
         self.margin = margin_mm
         self.y_weight = y_weight
 
-    def __call__(self, left_foot: np.ndarray, right_foot: np.ndarray, barycenter: np.ndarray) -> tuple[float, float]:
+    def __call__(self, left_foot: np.ndarray, right_foot: np.ndarray, barycenter: np.ndarray) -> dict:
         """
         Evaluate the equilibrium value and ellipse angle.
 
@@ -70,13 +70,13 @@ def __call__(self, left_foot: np.ndarray, right_foot: np.ndarray, barycenter: np
 
         Returns
         -------
-        value : float
-            Equilibrium value in [0, 1].
-            - 1 means the barycenter is perfectly at the ellipse center.
-            - 0 means the barycenter is outside the ellipse.
-        angle : float
-            Orientation of the ellipse in degrees, measured counter-clockwise
-            from the X-axis (line connecting left and right foot).
+        dict
+            Dictionary containing:
+            - 'value': Equilibrium value in [0, 1].
+                      1 means the barycenter is perfectly at the ellipse center.
+                      0 means the barycenter is outside the ellipse.
+            - 'angle': Orientation of the ellipse in degrees, measured counter-clockwise
+                      from the X-axis (line connecting left and right foot).
 
         Notes
         -----
@@ -157,4 +157,4 @@ def __call__(self, left_foot: np.ndarray, right_foot: np.ndarray, barycenter: np
             else:
                 value = 0.0
 
-        return max(0.0, value), np.degrees(angle)
+        return {"value": max(0.0, value), "angle": np.degrees(angle)}

pyeyesweb/mid_level/smoothness.py

@@ -53,8 +53,8 @@ class Smoothness:
     >>> for value in movement_data:
     ...     window.append([value])
     >>>
-    >>> sparc, jerk = smooth(window)
-    >>> print(f"SPARC: {sparc:.3f}, Jerk RMS: {jerk:.3f}")
+    >>> result = smooth(window)
+    >>> print(f"SPARC: {result['sparc']:.3f}, Jerk RMS: {result['jerk_rms']:.3f}")
 
     Notes
     -----
@@ -106,20 +106,15 @@ def __call__(self, sliding_window: SlidingWindow):
 
         Returns
         -------
-        sparc : float or None
-            Spectral Arc Length (more negative = smoother).
-            Returns None if insufficient data.
-        jerk : float or None
-            RMS of jerk (third derivative).
-            Returns None if insufficient data.
-
-        Returns
-        -------
-        tuple of (float, float)
-            (SPARC value, Jerk RMS value) or (NaN, NaN) if insufficient data.
+        dict
+            Dictionary containing:
+            - 'sparc': Spectral Arc Length (more negative = smoother).
+                      Returns NaN if insufficient data.
+            - 'jerk_rms': RMS of jerk (third derivative).
+                         Returns NaN if insufficient data.
         """
         if len(sliding_window) < 5:
-            return float("nan"), float("nan")
+            return {"sparc": float("nan"), "jerk_rms": float("nan")}
 
         signal, _ = sliding_window.to_array()
 
@@ -133,4 +128,4 @@ def __call__(self, sliding_window: SlidingWindow):
         sparc = compute_sparc(normalized, self.rate_hz)
         jerk = compute_jerk_rms(filtered, self.rate_hz)
 
-        return sparc, jerk
+        return {"sparc": sparc, "jerk_rms": jerk}

pyeyesweb/sync.py

@@ -30,6 +30,7 @@
     sys.path.append(os.getcwd())
 
 from collections import deque
+import numpy as np
 
 from pyeyesweb.data_models.sliding_window import SlidingWindow
 from pyeyesweb.utils.signal_processing import bandpass_filter, compute_hilbert_phases
@@ -110,8 +111,8 @@ class Synchronization:
     ...     window.append([signal1[i], signal2[i]])
     >>>
     >>> # Compute synchronization
-    >>> plv, status = sync(window)
-    >>> print(f"PLV: {plv:.3f}, Status: {status}")
+    >>> result = sync(window)
+    >>> print(f"PLV: {result['plv']:.3f}, Status: {result['phase_status']}")
 
     Notes
     -----
@@ -176,13 +177,13 @@ def compute_synchronization(self, signals: SlidingWindow):
 
         Returns
         -------
-        plv : float or None
-            Phase Locking Value between 0 (no sync) and 1 (perfect sync).
-            Returns None if the window is not full.
-        phase_status : str or None
-            If output_phase is True, returns "IN PHASE" when PLV > phase_threshold,
-            "OUT OF PHASE" otherwise. Returns None if output_phase is False or
-            if the window is not full.
+        dict
+            Dictionary containing synchronization metrics:
+            - 'plv': Phase Locking Value between 0 (no sync) and 1 (perfect sync).
+                    Returns NaN if the window is not full.
+            - 'phase_status': If output_phase is True, returns "IN PHASE" when PLV > phase_threshold,
+                            "OUT OF PHASE" otherwise. Returns None if output_phase is False or
+                            if the window is not full.
 
         Notes
         -----
@@ -200,7 +201,7 @@ def compute_synchronization(self, signals: SlidingWindow):
             raise ValueError(f"Synchronization requires exactly 2 signal channels, got {signals._n_columns}")
 
         if not signals.is_full():
-            return float("nan"), None
+            return {"plv": float("nan"), "phase_status": None}
 
         sig, _ = signals.to_array()
 
@@ -222,7 +223,7 @@ def compute_synchronization(self, signals: SlidingWindow):
             # Determine phase synchronization status based on threshold
             phase_status = "IN PHASE" if plv > self.phase_threshold else "OUT OF PHASE"
 
-        return plv, phase_status
+        return {"plv": plv, "phase_status": phase_status}
 
     def __call__(self, sliding_window: SlidingWindow):
         """Compute and optionally display synchronization metrics.
@@ -236,21 +237,23 @@ def __call__(self, sliding_window: SlidingWindow):
 
         Returns
         -------
-        plv : float or None
-            Phase Locking Value (0-1) or None if insufficient data.
-        phase_status : str or None
-            Phase status ("IN PHASE"/"OUT OF PHASE") or None.
+        dict
+            Dictionary containing synchronization metrics:
+            - 'plv': Phase Locking Value (0-1) or NaN if insufficient data.
+            - 'phase_status': Phase status ("IN PHASE"/"OUT OF PHASE") or None.
 
         Output
         ------------
         Prints synchronization metrics to stdout if PLV is computed successfully.
         Format depends on output_phase setting.
         """
-        plv, phase_status = self.compute_synchronization(sliding_window)
+        result = self.compute_synchronization(sliding_window)
+        plv = result["plv"]
+        phase_status = result["phase_status"]
 
-        if plv is not None:
+        if not np.isnan(plv):
             if self.output_phase:
                 print(f"Synchronization Index: {plv:.3f}, Phase Status: {phase_status}")
             else:
                 print(f"Synchronization Index: {plv:.3f}")
-        return plv, phase_status
+        return result

test_dict_output.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""Test script to verify all features return dictionaries."""
+
+import numpy as np
+from pyeyesweb.sync import Synchronization
+from pyeyesweb.low_level.equilibrium import Equilibrium
+from pyeyesweb.mid_level.smoothness import Smoothness
+from pyeyesweb.data_models.sliding_window import SlidingWindow
+
+def test_synchronization():
+    """Test Synchronization class returns dictionary."""
+    print("Testing Synchronization...")
+
+    # Create synchronization analyzer
+    sync = Synchronization(sensitivity=50, output_phase=True)
+
+    # Create sliding window with two signals
+    window = SlidingWindow(max_length=100, n_columns=2)
+
+    # Generate sample data
+    t = np.linspace(0, 2 * np.pi, 100)
+    signal1 = np.sin(t)
+    signal2 = np.sin(t + np.pi/4)  # Phase shifted signal
+
+    # Fill window
+    for i in range(100):
+        window.append([signal1[i], signal2[i]])
+
+    # Compute synchronization
+    result = sync(window)
+
+    # Check that result is a dictionary
+    assert isinstance(result, dict), f"Expected dict, got {type(result)}"
+    assert 'plv' in result, "Missing 'plv' key in result"
+    assert 'phase_status' in result, "Missing 'phase_status' key in result"
+
+    print(f"  ✓ Synchronization returns dict with keys: {list(result.keys())}")
+    print(f"    PLV: {result['plv']:.3f}")
+    print(f"    Phase Status: {result['phase_status']}")
+
+def test_equilibrium():
+    """Test Equilibrium class returns dictionary."""
+    print("\nTesting Equilibrium...")
+
+    # Create equilibrium analyzer
+    eq = Equilibrium(margin_mm=100, y_weight=0.5)
+
+    # Define foot positions and barycenter
+    left_foot = np.array([0, 0, 0])
+    right_foot = np.array([400, 0, 0])
+    barycenter = np.array([200, 50, 0])
+
+    # Compute equilibrium
+    result = eq(left_foot, right_foot, barycenter)
+
+    # Check that result is a dictionary
+    assert isinstance(result, dict), f"Expected dict, got {type(result)}"
+    assert 'value' in result, "Missing 'value' key in result"
+    assert 'angle' in result, "Missing 'angle' key in result"
+
+    print(f"  ✓ Equilibrium returns dict with keys: {list(result.keys())}")
+    print(f"    Value: {result['value']:.3f}")
+    print(f"    Angle: {result['angle']:.1f}°")
+
+def test_smoothness():
+    """Test Smoothness class returns dictionary."""
+    print("\nTesting Smoothness...")
+
+    # Create smoothness analyzer
+    smooth = Smoothness(rate_hz=100.0, use_filter=True)
+
+    # Create sliding window
+    window = SlidingWindow(max_length=200, n_columns=1)
+
+    # Generate sample movement data (smooth sinusoidal motion)
+    t = np.linspace(0, 2, 200)
+    movement = np.sin(2 * np.pi * t)
+
+    # Fill window
+    for value in movement:
+        window.append([value])
+
+    # Compute smoothness
+    result = smooth(window)
+
+    # Check that result is a dictionary
+    assert isinstance(result, dict), f"Expected dict, got {type(result)}"
+    assert 'sparc' in result, "Missing 'sparc' key in result"
+    assert 'jerk_rms' in result, "Missing 'jerk_rms' key in result"
+
+    print(f"  ✓ Smoothness returns dict with keys: {list(result.keys())}")
+    print(f"    SPARC: {result['sparc']:.3f}")
+    print(f"    Jerk RMS: {result['jerk_rms']:.3f}")
+
+def main():
+    """Run all tests."""
+    print("=" * 50)
+    print("Testing Feature Dictionary Output Standardization")
+    print("=" * 50)
+
+    try:
+        test_synchronization()
+        test_equilibrium()
+        test_smoothness()
+
+        print("\n" + "=" * 50)
+        print("✅ All tests passed! All features return dictionaries.")
+        print("=" * 50)
+
+    except AssertionError as e:
+        print(f"\n❌ Test failed: {e}")
+        return 1
+    except Exception as e:
+        print(f"\n❌ Unexpected error: {e}")
+        return 1
+
+    return 0
+
+if __name__ == "__main__":
+    exit(main())