Skip to content

Signal Processing

Signal processing utilities for PyEyesWeb.

This module provides signal processing functions including filtering, phase extraction, and smoothing operations used throughout the library.

apply_savgol_filter(signal, rate_hz=50.0)

Apply Savitzky-Golay filter if enough data is available.

Savitzky-Golay filtering smooths data while preserving features better than moving average filters. Uses polynomial order 3 and adaptive window length.

Parameters:

Name Type Description Default
signal array - like

1D signal to filter.

 required
rate_hz float

Sampling rate in Hz (currently unused but kept for API compatibility).

 50.0

Returns:

Type Description
ndarray

Filtered signal if sufficient data (≥5 samples), otherwise original signal as array. Window length is min(n_samples, 11) and must be odd.
Notes
  • Requires at least 5 samples for filtering
  • Window length must exceed polynomial order (3)
  • Returns original signal if filtering fails
Source code in pyeyesweb/utils/signal_processing.py 
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
167
168
169
170
def apply_savgol_filter(signal, rate_hz=50.0):
    """Apply Savitzky-Golay filter if enough data is available.

    Savitzky-Golay filtering smooths data while preserving features better
    than moving average filters. Uses polynomial order 3 and adaptive
    window length.

    Parameters
    ----------
    signal : array-like
        1D signal to filter.
    rate_hz : float, optional
        Sampling rate in Hz (currently unused but kept for API compatibility).

    Returns
    -------
    ndarray
        Filtered signal if sufficient data (≥5 samples), otherwise original
        signal as array. Window length is min(n_samples, 11) and must be odd.

    Notes
    -----
    - Requires at least 5 samples for filtering
    - Window length must exceed polynomial order (3)
    - Returns original signal if filtering fails
    """
    if len(signal) < 5:
        return np.array(signal)

    N = len(signal)
    polyorder = 3
    window_length = min(N if N % 2 == 1 else N - 1, 11)
    if window_length <= polyorder:
        return np.array(signal)

    try:
        from scipy.signal import savgol_filter
        return savgol_filter(signal, window_length=window_length, polyorder=polyorder)
    except Exception:
        return np.array(signal)

bandpass_filter(data, filter_params)

Apply a band-pass filter if filter_params is set.

Uses a 4th order Butterworth band-pass filter with zero-phase filtering (filtfilt) to avoid phase distortion.

Parameters:

Name Type Description Default
data ndarray

Signal data of shape (n_samples, n_channels).

 required
filter_params tuple of (float, float, float) or None

Filter parameters as (lowcut_hz, highcut_hz, sampling_rate_hz). If None, returns data unchanged.

 required

Returns:

Type Description
ndarray

Filtered data with same shape as input. Returns original data if filter_params is None.

Examples:

>>> data = np.random.randn(1000, 2)  # 2 channels, 1000 samples
>>> filtered = bandpass_filter(data, (1.0, 10.0, 100.0))  # 1-10 Hz
Source code in pyeyesweb/utils/signal_processing.py 
55
56
57
58
59
60
61
62
63
64
65
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
def bandpass_filter(data, filter_params):
    """Apply a band-pass filter if filter_params is set.

    Uses a 4th order Butterworth band-pass filter with zero-phase
    filtering (filtfilt) to avoid phase distortion.

    Parameters
    ----------
    data : ndarray
        Signal data of shape (n_samples, n_channels).
    filter_params : tuple of (float, float, float) or None
        Filter parameters as (lowcut_hz, highcut_hz, sampling_rate_hz).
        If None, returns data unchanged.

    Returns
    -------
    ndarray
        Filtered data with same shape as input.
        Returns original data if filter_params is None.

    Examples
    --------
    >>> data = np.random.randn(1000, 2)  # 2 channels, 1000 samples
    >>> filtered = bandpass_filter(data, (1.0, 10.0, 100.0))  # 1-10 Hz
    """
    if filter_params is None:
        return data

    # Use centralized validation
    lowcut, highcut, fs = validate_filter_params(*filter_params)

    nyquist = 0.5 * fs
    low = lowcut / nyquist
    high = highcut / nyquist

    b, a = butter(4, [low, high], btype='band')

    filtered_data = np.zeros_like(data)
    for i in range(data.shape[1]):
        filtered_data[:, i] = filtfilt(b, a, data[:, i])

    return filtered_data

compute_hilbert_phases(sig)

Compute phase information from signals using Hilbert Transform.

The Hilbert transform creates an analytic signal from which instantaneous phase can be extracted. Assumes input has exactly 2 channels.

Parameters:

Name Type Description Default
sig ndarray

Signal array of shape (n_samples, 2) with two channels.

 required

Returns:

Name Type Description
phase1 ndarray

Phase values for first channel in radians [-π, π].
phase2 ndarray

Phase values for second channel in radians [-π, π].
Notes

The Hilbert transform assumes the signal is narrowband or has been appropriately filtered for meaningful phase extraction.

Source code in pyeyesweb/utils/signal_processing.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def compute_hilbert_phases(sig):
    """Compute phase information from signals using Hilbert Transform.

    The Hilbert transform creates an analytic signal from which instantaneous
    phase can be extracted. Assumes input has exactly 2 channels.

    Parameters
    ----------
    sig : ndarray
        Signal array of shape (n_samples, 2) with two channels.

    Returns
    -------
    phase1 : ndarray
        Phase values for first channel in radians [-π, π].
    phase2 : ndarray
        Phase values for second channel in radians [-π, π].

    Notes
    -----
    The Hilbert transform assumes the signal is narrowband or has been
    appropriately filtered for meaningful phase extraction.
    """
    analytic_signal1 = hilbert(sig[:, 0])
    analytic_signal2 = hilbert(sig[:, 1])

    phase1 = np.angle(analytic_signal1)
    phase2 = np.angle(analytic_signal2)

    return phase1, phase2

validate_filter_params(lowcut, highcut, fs)

Validate filter frequency parameters.

Centralized validation for filter parameters used in bandpass_filter and Synchronization class.

Parameters:

Name Type Description Default
lowcut float

Low cutoff frequency in Hz

 required
highcut float

High cutoff frequency in Hz

 required
fs float

Sampling frequency in Hz

 required

Returns:

Type Description
tuple

Validated (lowcut, highcut, fs)

Raises:

Type Description
ValueError

If parameters are invalid
Source code in pyeyesweb/utils/signal_processing.py 
11
12
13
14
15
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
def validate_filter_params(lowcut, highcut, fs):
    """Validate filter frequency parameters.

    Centralized validation for filter parameters used in bandpass_filter
    and Synchronization class.

    Parameters
    ----------
    lowcut : float
        Low cutoff frequency in Hz
    highcut : float
        High cutoff frequency in Hz
    fs : float
        Sampling frequency in Hz

    Returns
    -------
    tuple
        Validated (lowcut, highcut, fs)

    Raises
    ------
    ValueError
        If parameters are invalid
    """
    # Validate individual parameters
    if fs <= 0:
        raise ValueError(f"Sampling frequency must be positive, got {fs}")
    if lowcut <= 0:
        raise ValueError(f"Low cutoff frequency must be positive, got {lowcut}")
    if highcut <= 0:
        raise ValueError(f"High cutoff frequency must be positive, got {highcut}")

    # Validate relationships
    if lowcut >= highcut:
        raise ValueError(f"Low cutoff ({lowcut}) must be less than high cutoff ({highcut})")

    nyquist = fs / 2
    if highcut >= nyquist:
        raise ValueError(f"High cutoff ({highcut}) must be less than Nyquist frequency ({nyquist})")

    return lowcut, highcut, fs