Base

Core abstractions for the feature extraction pipeline.

This module defines the BaseFeature abstract base class and its primary specializations: StaticFeature and DynamicFeature.

BaseFeature

Bases: ABC

The root abstract class for all PyEyesWeb features.

Defines the dual-API structure for all computational components:

  1. Streaming API (__call__): Evaluates a SlidingWindow.
  2. Pure Math API (compute): Evaluates a raw NumPy array.
Source code in pyeyesweb/data_models/base.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class BaseFeature(ABC):
    """The root abstract class for all PyEyesWeb features.

    Defines the dual-API structure for all computational components:

    1. **Streaming API (`__call__`)**: Evaluates a `SlidingWindow`.
    2. **Pure Math API (`compute`)**: Evaluates a raw NumPy array.
    """

    def __init__(self):
        pass

    @abstractmethod
    def __call__(self, data: SlidingWindow) -> FeatureResult:
        """The Streaming API: Compute the feature using a SlidingWindow.

        Handles state and buffering for real-time applications by processing
        the contents of the provided window.

        Parameters
        ----------
        data : SlidingWindow
            The circular buffer containing motion samples.

        Returns
        -------
        FeatureResult
            The computed feature result.
        """
        pass

    @abstractmethod
    def compute(self, data: np.ndarray) -> FeatureResult:
        """The Pure Math API: Compute the feature from a raw NumPy array.

        Stateless and designed for composition or offline data science.

        Parameters
        ----------
        data : numpy.ndarray
            Raw motion data array.

        Returns
        -------
        FeatureResult
            The computed feature result.
        """
        pass

__call__(data) abstractmethod

The Streaming API: Compute the feature using a SlidingWindow.

Handles state and buffering for real-time applications by processing the contents of the provided window.

Parameters:
  • data (SlidingWindow) –

    The circular buffer containing motion samples.
Returns:
Source code in pyeyesweb/data_models/base.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@abstractmethod
def __call__(self, data: SlidingWindow) -> FeatureResult:
    """The Streaming API: Compute the feature using a SlidingWindow.

    Handles state and buffering for real-time applications by processing
    the contents of the provided window.

    Parameters
    ----------
    data : SlidingWindow
        The circular buffer containing motion samples.

    Returns
    -------
    FeatureResult
        The computed feature result.
    """
    pass

compute(data) abstractmethod

The Pure Math API: Compute the feature from a raw NumPy array.

Stateless and designed for composition or offline data science.

Parameters:
  • data (ndarray) –

    Raw motion data array.
Returns:
Source code in pyeyesweb/data_models/base.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@abstractmethod
def compute(self, data: np.ndarray) -> FeatureResult:
    """The Pure Math API: Compute the feature from a raw NumPy array.

    Stateless and designed for composition or offline data science.

    Parameters
    ----------
    data : numpy.ndarray
        Raw motion data array.

    Returns
    -------
    FeatureResult
        The computed feature result.
    """
    pass

DynamicFeature

Bases: BaseFeature

Base class for features that require a time-series window.

Examples include smoothness, direction change, or any feature involving temporal variations.

Source code in pyeyesweb/data_models/base.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class DynamicFeature(BaseFeature):
    """Base class for features that require a time-series window.

    Examples include smoothness, direction change, or any feature involving
    temporal variations.
    """

    @abstractmethod
    def compute(self, data: np.ndarray) -> FeatureResult:
        """The actual mathematical logic for a time-series window.

        Parameters
        ----------
        data : numpy.ndarray of shape (Time, N_nodes, N_dims)
            3D tensor of motion data within the window.

        Returns
        -------
        FeatureResult
            Must return a specialized `FeatureResult` for the feature.
        """
        pass

    def __call__(self, data: SlidingWindow) -> FeatureResult:
        """The Streaming API for dynamic features.

        Extracts the full chronological tensor from the `SlidingWindow` and
        passes it to the `compute` method.

        Parameters
        ----------
        data : SlidingWindow
            Circular buffer containing the time-series data.

        Returns
        -------
        FeatureResult
            Result for the window. Returns `FeatureResult(is_valid=False)`
            if the window is empty.
        """
        if len(data) == 0:
            return FeatureResult(is_valid=False)

        # Extract the 3D tensor
        tensor, _ = data.to_tensor()

        # Delegate to the pure math API. Exceptions will naturally bubble up!
        return self.compute(tensor)

__call__(data)

The Streaming API for dynamic features.

Extracts the full chronological tensor from the SlidingWindow and passes it to the compute method.

Parameters:
  • data (SlidingWindow) –

    Circular buffer containing the time-series data.
Returns:
  • FeatureResult –

    Result for the window. Returns FeatureResult(is_valid=False) if the window is empty.
Source code in pyeyesweb/data_models/base.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def __call__(self, data: SlidingWindow) -> FeatureResult:
    """The Streaming API for dynamic features.

    Extracts the full chronological tensor from the `SlidingWindow` and
    passes it to the `compute` method.

    Parameters
    ----------
    data : SlidingWindow
        Circular buffer containing the time-series data.

    Returns
    -------
    FeatureResult
        Result for the window. Returns `FeatureResult(is_valid=False)`
        if the window is empty.
    """
    if len(data) == 0:
        return FeatureResult(is_valid=False)

    # Extract the 3D tensor
    tensor, _ = data.to_tensor()

    # Delegate to the pure math API. Exceptions will naturally bubble up!
    return self.compute(tensor)

compute(data) abstractmethod

The actual mathematical logic for a time-series window.

Parameters:
  • data (numpy.ndarray of shape (Time, N_nodes, N_dims)) –

    3D tensor of motion data within the window.
Returns:
  • FeatureResult –

    Must return a specialized FeatureResult for the feature.
Source code in pyeyesweb/data_models/base.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@abstractmethod
def compute(self, data: np.ndarray) -> FeatureResult:
    """The actual mathematical logic for a time-series window.

    Parameters
    ----------
    data : numpy.ndarray of shape (Time, N_nodes, N_dims)
        3D tensor of motion data within the window.

    Returns
    -------
    FeatureResult
        Must return a specialized `FeatureResult` for the feature.
    """
    pass

StaticFeature

Bases: BaseFeature

Base class for features computed on a single frame.

Examples include posture analysis, spatial density, or instantaneous geometric properties.

Source code in pyeyesweb/data_models/base.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class StaticFeature(BaseFeature):
    """Base class for features computed on a single frame.

    Examples include posture analysis, spatial density, or instantaneous
    geometric properties.
    """

    @abstractmethod
    def compute(self, data: np.ndarray) -> FeatureResult:
        """The actual mathematical logic for a single frame.

        Parameters
        ----------
        data : numpy.ndarray of shape (N_nodes, N_dims)
            Snapshot of motion data for a single point in time.

        Returns
        -------
        FeatureResult
            Must return a specialized `FeatureResult` for the feature.
        """
        pass

    def __call__(self, data: SlidingWindow, **kwargs) -> FeatureResult:
        """The Streaming API for static features.

        Automatically extracts the latest frame from the `SlidingWindow` and
        delegates to the `compute` method.

        Parameters
        ----------
        data : SlidingWindow
            Circular buffer containing the data.
        **kwargs
            Additional arguments passed to `compute`.

        Returns
        -------
        FeatureResult
            Result for the newest frame. Returns `FeatureResult(is_valid=False)`
            if the window is empty.
        """
        if len(data) == 0:
            return FeatureResult(is_valid=False)

        # Extract the 3D tensor and grab the very last frame (the newest one)
        tensor, _ = data.to_tensor()
        frame_data = tensor[-1]

        # Delegate to the pure math API. Exceptions will naturally bubble up!
        return self.compute(frame_data)

__call__(data, **kwargs)

The Streaming API for static features.

Automatically extracts the latest frame from the SlidingWindow and delegates to the compute method.

Parameters:
  • data (SlidingWindow) –

    Circular buffer containing the data.

  • **kwargs –

    Additional arguments passed to compute.
Returns:
  • FeatureResult –

    Result for the newest frame. Returns FeatureResult(is_valid=False) if the window is empty.
Source code in pyeyesweb/data_models/base.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def __call__(self, data: SlidingWindow, **kwargs) -> FeatureResult:
    """The Streaming API for static features.

    Automatically extracts the latest frame from the `SlidingWindow` and
    delegates to the `compute` method.

    Parameters
    ----------
    data : SlidingWindow
        Circular buffer containing the data.
    **kwargs
        Additional arguments passed to `compute`.

    Returns
    -------
    FeatureResult
        Result for the newest frame. Returns `FeatureResult(is_valid=False)`
        if the window is empty.
    """
    if len(data) == 0:
        return FeatureResult(is_valid=False)

    # Extract the 3D tensor and grab the very last frame (the newest one)
    tensor, _ = data.to_tensor()
    frame_data = tensor[-1]

    # Delegate to the pure math API. Exceptions will naturally bubble up!
    return self.compute(frame_data)

compute(data) abstractmethod

The actual mathematical logic for a single frame.

Parameters:
  • data (numpy.ndarray of shape (N_nodes, N_dims)) –

    Snapshot of motion data for a single point in time.
Returns:
  • FeatureResult –

    Must return a specialized FeatureResult for the feature.
Source code in pyeyesweb/data_models/base.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@abstractmethod
def compute(self, data: np.ndarray) -> FeatureResult:
    """The actual mathematical logic for a single frame.

    Parameters
    ----------
    data : numpy.ndarray of shape (N_nodes, N_dims)
        Snapshot of motion data for a single point in time.

    Returns
    -------
    FeatureResult
        Must return a specialized `FeatureResult` for the feature.
    """
    pass