Waveform Analysis

Module: equser.analysis Dependencies: base (numpy only)

Functions for zero-crossing detection, cycle extraction, and waveform analysis.

find_zero_crossings(signal, time_array)

Find negative-to-positive zero crossings in a signal using linear interpolation between adjacent samples.

Args:

  • signal (ndarray): Voltage or current waveform array.
  • time_array (ndarray): Corresponding time array (same length).

Returns: (crossing_times, crossing_indices)

  • crossing_times: interpolated times of each zero crossing
  • crossing_indices: index of the sample just before each crossing
import numpy as np
from equser.data import load_cpow_scaled, SAMPLE_RATE_HZ
from equser.analysis import find_zero_crossings

result = load_cpow_scaled('cpow_data.parquet')
time = np.arange(len(result['VA'])) / SAMPLE_RATE_HZ

crossings, indices = find_zero_crossings(result['VA'], time)
print(f"Found {len(crossings)} zero crossings")

# Frequency estimation
periods = np.diff(crossings)
print(f"Mean frequency: {1.0 / np.mean(periods):.2f} Hz")
print(f"Frequency std:  {np.std(1.0 / periods):.4f} Hz")

extract_complete_cycles(signal, time_array, start_times, num_cycles=1)

Extract complete AC cycles (zero-crossing to zero-crossing) starting from specified times.

Args:

  • signal (ndarray): Waveform array.
  • time_array (ndarray): Time array.
  • start_times (list of float): Times to start extracting from.
  • num_cycles (int): Number of cycles per start time (default 1).

Returns: List of tuples (cycle_time, cycle_signal, start, end) where:

  • cycle_time: time array normalized to start at 0
  • cycle_signal: signal values for the window
  • start: actual start time (first zero crossing)
  • end: actual end time (last zero crossing)

Returns an empty list if there are not enough zero crossings for the requested number of cycles.

from equser.analysis import extract_complete_cycles

# Extract single cycles at 3 different times
cycles = extract_complete_cycles(result['VA'], time, [0.0, 0.05, 0.1])

for cycle_time, cycle_signal, start, end in cycles:
    duration_ms = (end - start) * 1000
    peak = np.max(np.abs(cycle_signal))
    print(f"Cycle at {start:.4f}s: {duration_ms:.2f} ms, peak {peak:.1f} V")

plot_extracted_cycles(signal_dict, time_array, start_times, ...)

Plot extracted cycles for one or more phases side-by-side. Requires matplotlib ([analysis] extra).

Args:

  • signal_dict: Dict mapping phase names to arrays (e.g. {'VA': va, 'VB': vb}), or a single array (treated as {'VA': array}).
  • time_array: Time array.
  • start_times: Times to extract cycles from.
  • num_cycles (int): Cycles per window (default 1).
  • epoch_start_time (datetime, optional): Absolute start time for labels.

Returns: (fig, axes, window_times)

from equser.analysis.waveform import plot_extracted_cycles

fig, axes, windows = plot_extracted_cycles(
    {'VA': result['VA'], 'VB': result['VB'], 'VC': result['VC']},
    time,
    start_times=[0.0, 0.1, 0.2],
    num_cycles=2,
    epoch_start_time=result['start_time'],
)
fig.savefig('cycles.png', dpi=150, bbox_inches='tight')