Skip to content

Synchronization

Synchronization analysis module for real time signal phase locking.

This module provides tools for computing phase synchronization between paired signals using the Hilbert Transform and Phase Locking Value (PLV) analysis. It is designed for real time analysis of motion capture or sensor data streams.

The synchronization analysis follows these steps: 1. Optional band pass filtering to isolate frequencies of interest 2. Signal centering (mean removal) to eliminate DC bias 3. Hilbert Transform to extract instantaneous phase information 4. Phase Locking Value computation to quantify synchronization strength

Typical use cases include: 1. Movement coordination analysis between limbs 2. Human-human or human-robot interaction studies 3. Neural oscillation synchronization 4. Periodic signal coupling analysis

References
  1. Lachaux et al. (1999). Measuring phase synchrony in brain signals. Human Brain Mapping, 8(4), 194-208.
  2. Rosenblum et al. (1996). Phase synchronization of chaotic oscillators. Physical Review Letters, 76(11), 1804.

Synchronization

Real time phase synchronization analyzer for paired signals.

This class computes the Phase Locking Value (PLV) between two signals using the Hilbert Transform to extract instantaneous phase information. It maintains a history buffer for tracking synchronization over time and can optionally apply band-pass filtering to focus on specific frequency bands.

The PLV ranges from 0 (no synchronization) to 1 (perfect synchronization) and is computed as the absolute value of the mean complex phase difference between the two signals.

Read more in the User Guide

Parameters:

Name Type Description Default
sensitivity int

Size of the PLV history buffer. Larger values provide more temporal context but increase memory usage. Must be positive integer between 1 and 10,000 (default: 100).

 100
output_phase bool

If True, outputs phase synchronization status as "IN PHASE" or "OUT OF PHASE" based on the phase_threshold. Must be boolean (default: False).

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

Band-pass filter parameters as (lowcut_hz, highcut_hz, sampling_rate_hz). All frequencies must be positive with lowcut < highcut < sampling_rate/2. Example: (0.5, 30, 100) for 0.5-30 Hz band with 100 Hz sampling. If None, no filtering is applied (default: None).

 None
phase_threshold float

PLV threshold for phase status determination. Values above this are considered "IN PHASE". Must be between 0 and 1 inclusive (default: 0.7).

 0.7

Raises:

Type Description
TypeError

If sensitivity is not int, output_phase is not bool, phase_threshold is not numeric, or filter_params is not tuple/list.
ValueError

If sensitivity <= 0 or > 10,000, phase_threshold outside [0, 1], or filter_params contains invalid frequencies.

Attributes:

Name Type Description
plv_history deque

Rolling buffer storing recent PLV values for temporal analysis. Thread safe access is ensured via internal locking.
output_phase bool

Flag controlling phase status output.
filter_params tuple or None

Band-pass filter configuration.
phase_threshold float

Threshold for phase synchronization classification.

Examples:

>>> from pyeyesweb.analysis_primitives.synchronization import Synchronization
>>> from pyeyesweb.data_models.sliding_window import SlidingWindow
>>>
>>> # Create synchronization analyzer with filtering
>>> sync = Synchronization(
...     sensitivity=50,
...     output_phase=True,
...     filter_params=(1.0, 10.0, 100.0),  # 1-10 Hz band at 100 Hz
...     phase_threshold=0.8
... )
>>>
>>> # Create sliding window for two signals
>>> window = SlidingWindow(max_length=200, n_columns=2)
>>>
>>> # Add signal data (e.g., from two sensors)
>>> for i in range(200):
...     window.append([signal1[i], signal2[i]])
>>>
>>> # Compute synchronization
>>> result = sync(window)
>>> print(f"PLV: {result['plv']:.3f}, Status: {result['phase_status']}")
Notes
  • Requires at least a full window of data to compute meaningful results
  • The Hilbert Transform assumes narrowband or filtered signals for best results
  • Phase differences are most meaningful for signals with similar frequencies
  • For broadband signals, consider using filter_params to isolate frequency bands
Source code in pyeyesweb/analysis_primitives/synchronization.py 
 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
 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
 97
 98
 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
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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
class Synchronization:
    """Real time phase synchronization analyzer for paired signals.

    This class computes the Phase Locking Value (PLV) between two signals
    using the Hilbert Transform to extract instantaneous phase information.
    It maintains a history buffer for tracking synchronization over time and
    can optionally apply band-pass filtering to focus on specific frequency bands.

    The PLV ranges from 0 (no synchronization) to 1 (perfect synchronization)
    and is computed as the absolute value of the mean complex phase difference
    between the two signals.

    Read more in the [User Guide](/PyEyesWeb/user_guide/theoretical_framework/analysis_primitives/synchronization/)

    Parameters
    ----------
    sensitivity : int, optional
        Size of the PLV history buffer. Larger values provide more temporal
        context but increase memory usage. Must be positive integer between
        1 and 10,000 (default: 100).
    output_phase : bool, optional
        If True, outputs phase synchronization status as "IN PHASE" or
        "OUT OF PHASE" based on the phase_threshold. Must be boolean
        (default: False).
    filter_params : tuple of (float, float, float) or None, optional
        Band-pass filter parameters as (lowcut_hz, highcut_hz, sampling_rate_hz).
        All frequencies must be positive with lowcut < highcut < sampling_rate/2.
        Example: (0.5, 30, 100) for 0.5-30 Hz band with 100 Hz sampling.
        If None, no filtering is applied (default: None).
    phase_threshold : float, optional
        PLV threshold for phase status determination. Values above this are
        considered "IN PHASE". Must be between 0 and 1 inclusive (default: 0.7).

    Raises
    ------
    TypeError
        If sensitivity is not int, output_phase is not bool, phase_threshold
        is not numeric, or filter_params is not tuple/list.
    ValueError
        If sensitivity <= 0 or > 10,000, phase_threshold outside [0, 1],
        or filter_params contains invalid frequencies.

    Attributes
    ----------
    plv_history : collections.deque
        Rolling buffer storing recent PLV values for temporal analysis.
        Thread safe access is ensured via internal locking.
    output_phase : bool
        Flag controlling phase status output.
    filter_params : tuple or None
        Band-pass filter configuration.
    phase_threshold : float
        Threshold for phase synchronization classification.

    Examples
    --------
    >>> from pyeyesweb.analysis_primitives.synchronization import Synchronization
    >>> from pyeyesweb.data_models.sliding_window import SlidingWindow
    >>>
    >>> # Create synchronization analyzer with filtering
    >>> sync = Synchronization(
    ...     sensitivity=50,
    ...     output_phase=True,
    ...     filter_params=(1.0, 10.0, 100.0),  # 1-10 Hz band at 100 Hz
    ...     phase_threshold=0.8
    ... )
    >>>
    >>> # Create sliding window for two signals
    >>> window = SlidingWindow(max_length=200, n_columns=2)
    >>>
    >>> # Add signal data (e.g., from two sensors)
    >>> for i in range(200):
    ...     window.append([signal1[i], signal2[i]])
    >>>
    >>> # Compute synchronization
    >>> result = sync(window)
    >>> print(f"PLV: {result['plv']:.3f}, Status: {result['phase_status']}")

    Notes
    -----
    - Requires at least a full window of data to compute meaningful results
    - The Hilbert Transform assumes narrowband or filtered signals for best results
    - Phase differences are most meaningful for signals with similar frequencies
    - For broadband signals, consider using filter_params to isolate frequency bands
    """

    def __init__(self, sensitivity=100, output_phase=False, filter_params=None, phase_threshold=0.7):
        sensitivity = validate_integer(sensitivity, 'sensitivity', min_val=1, max_val=10000)
        self.output_phase = validate_boolean(output_phase, 'output_phase')
        self.phase_threshold = validate_numeric(phase_threshold, 'phase_threshold', min_val=0, max_val=1)

        if filter_params is not None:
            filter_params = validate_filter_params_tuple(filter_params)
            lowcut, highcut, fs = validate_filter_params(*filter_params)
            filter_params = (lowcut, highcut, fs)

        self.plv_history = deque(maxlen=sensitivity)
        self._history_lock = threading.Lock()
        self.filter_params = filter_params


    def compute_synchronization(self, signals: SlidingWindow):
        """Compute phase synchronization between two signals.

        Processes the signal pair through filtering (optional), centering,
        Hilbert Transform, and PLV computation to quantify synchronization.

        Parameters
        ----------
        signals : SlidingWindow
            Sliding window buffer containing exactly 2 columns of signal data.
            Must be full (contain max_length samples) for computation.

        Returns
        -------
        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
        -----
        The computation pipeline:
        1. Check if window has sufficient data (is_full)
        2. Apply band-pass filter if filter_params is set
        3. Center signals by removing mean (eliminates DC component)
        4. Apply Hilbert Transform to get analytic signal and phase
        5. Compute PLV from phase difference
        6. Update PLV history buffer
        7. Determine phase status if requested
        """
        # Validate input has exactly 2 columns
        if signals._n_columns != 2:
            raise ValueError(f"Synchronization requires exactly 2 signal channels, got {signals._n_columns}")

        if not signals.is_full():
            return {"plv": float("nan"), "phase_status": None}

        sig, _ = signals.to_array()

        # Apply band-pass filtering if filter parameters are provided
        sig = bandpass_filter(sig, self.filter_params)

        # Remove the mean from each signal to center the data
        sig = center_signals(sig)

        # Extract the phase information from the analytic signals
        phase1, phase2 = compute_hilbert_phases(sig)

        # Compute the Phase Locking Value (PLV)
        plv = compute_phase_locking_value(phase1, phase2)
        with self._history_lock:
            self.plv_history.append(plv)

        phase_status = None
        if self.output_phase:
            # Determine phase synchronization status based on threshold
            phase_status = "IN PHASE" if plv > self.phase_threshold else "OUT OF PHASE"

        return {"plv": plv, "phase_status": phase_status}

    def __call__(self, sliding_window: SlidingWindow):
        """Compute and optionally display synchronization metrics.

        This method allows the class to be used as a callable, providing a convenient interface for real time processing pipelines.

        Parameters
        ----------
        sliding_window : SlidingWindow
            Buffer containing two signal columns to analyze.

        Returns
        -------
        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.
        """
        result = self.compute_synchronization(sliding_window)
        plv = result["plv"]
        phase_status = result["phase_status"]

        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 result

__call__(sliding_window)

Compute and optionally display synchronization metrics.

This method allows the class to be used as a callable, providing a convenient interface for real time processing pipelines.

Parameters:

Name Type Description Default
sliding_window SlidingWindow

Buffer containing two signal columns to analyze.

 required

Returns:

Type Description
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.

Source code in pyeyesweb/analysis_primitives/synchronization.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def __call__(self, sliding_window: SlidingWindow):
    """Compute and optionally display synchronization metrics.

    This method allows the class to be used as a callable, providing a convenient interface for real time processing pipelines.

    Parameters
    ----------
    sliding_window : SlidingWindow
        Buffer containing two signal columns to analyze.

    Returns
    -------
    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.
    """
    result = self.compute_synchronization(sliding_window)
    plv = result["plv"]
    phase_status = result["phase_status"]

    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 result

compute_synchronization(signals)

Compute phase synchronization between two signals.

Processes the signal pair through filtering (optional), centering, Hilbert Transform, and PLV computation to quantify synchronization.

Parameters:

Name Type Description Default
signals SlidingWindow

Sliding window buffer containing exactly 2 columns of signal data. Must be full (contain max_length samples) for computation.

 required

Returns:

Type Description
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

The computation pipeline: 1. Check if window has sufficient data (is_full) 2. Apply band-pass filter if filter_params is set 3. Center signals by removing mean (eliminates DC component) 4. Apply Hilbert Transform to get analytic signal and phase 5. Compute PLV from phase difference 6. Update PLV history buffer 7. Determine phase status if requested

Source code in pyeyesweb/analysis_primitives/synchronization.py 
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
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def compute_synchronization(self, signals: SlidingWindow):
    """Compute phase synchronization between two signals.

    Processes the signal pair through filtering (optional), centering,
    Hilbert Transform, and PLV computation to quantify synchronization.

    Parameters
    ----------
    signals : SlidingWindow
        Sliding window buffer containing exactly 2 columns of signal data.
        Must be full (contain max_length samples) for computation.

    Returns
    -------
    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
    -----
    The computation pipeline:
    1. Check if window has sufficient data (is_full)
    2. Apply band-pass filter if filter_params is set
    3. Center signals by removing mean (eliminates DC component)
    4. Apply Hilbert Transform to get analytic signal and phase
    5. Compute PLV from phase difference
    6. Update PLV history buffer
    7. Determine phase status if requested
    """
    # Validate input has exactly 2 columns
    if signals._n_columns != 2:
        raise ValueError(f"Synchronization requires exactly 2 signal channels, got {signals._n_columns}")

    if not signals.is_full():
        return {"plv": float("nan"), "phase_status": None}

    sig, _ = signals.to_array()

    # Apply band-pass filtering if filter parameters are provided
    sig = bandpass_filter(sig, self.filter_params)

    # Remove the mean from each signal to center the data
    sig = center_signals(sig)

    # Extract the phase information from the analytic signals
    phase1, phase2 = compute_hilbert_phases(sig)

    # Compute the Phase Locking Value (PLV)
    plv = compute_phase_locking_value(phase1, phase2)
    with self._history_lock:
        self.plv_history.append(plv)

    phase_status = None
    if self.output_phase:
        # Determine phase synchronization status based on threshold
        phase_status = "IN PHASE" if plv > self.phase_threshold else "OUT OF PHASE"

    return {"plv": plv, "phase_status": phase_status}