Skip to content

cds2.signals

Signal processing on scipy.signal and numpy.fft.

Signal processing — convenience re-export of scipy.signal/numpy.fft with uniform result types; see cds2.wavelets/cds2.spectral for CDS-native.

Spectrum dataclass

Power spectral density estimate.

Source code in src/cds2/signals.py 
38
39
40
41
42
43
@dataclass(frozen=True)
class Spectrum:
    """Power spectral density estimate."""

    frequencies: np.ndarray
    power: np.ndarray

SpectrogramResult dataclass

Short-time Fourier transform magnitude spectrogram.

Source code in src/cds2/signals.py 
46
47
48
49
50
51
52
@dataclass(frozen=True)
class SpectrogramResult:
    """Short-time Fourier transform magnitude spectrogram."""

    times: np.ndarray
    frequencies: np.ndarray
    power: np.ndarray

PeakResult dataclass

Indices and amplitudes of detected peaks.

Source code in src/cds2/signals.py 
55
56
57
58
59
60
@dataclass(frozen=True)
class PeakResult:
    """Indices and amplitudes of detected peaks."""

    indices: np.ndarray
    heights: np.ndarray

fft(x)

Full complex discrete Fourier transform.

Source code in src/cds2/signals.py 
63
64
65
def fft(x: object) -> np.ndarray:
    """Full complex discrete Fourier transform."""
    return np.asarray(np.fft.fft(np.asarray(x)))

ifft(x)

Inverse discrete Fourier transform.

Source code in src/cds2/signals.py 
68
69
70
def ifft(x: object) -> np.ndarray:
    """Inverse discrete Fourier transform."""
    return np.asarray(np.fft.ifft(np.asarray(x)))

rfft(x)

Real-input FFT returning the non-negative frequency half.

Source code in src/cds2/signals.py 
73
74
75
def rfft(x: object) -> np.ndarray:
    """Real-input FFT returning the non-negative frequency half."""
    return np.asarray(np.fft.rfft(np.asarray(x)))

irfft(x, n=None)

Inverse real FFT back to a time-domain signal.

Source code in src/cds2/signals.py 
78
79
80
def irfft(x: object, n: int | None = None) -> np.ndarray:
    """Inverse real FFT back to a time-domain signal."""
    return np.asarray(np.fft.irfft(np.asarray(x), n=n))

fftfreq(n, dt=1.0)

FFT sample frequencies for a window of n samples with spacing dt.

Source code in src/cds2/signals.py 
83
84
85
def fftfreq(n: int, dt: float = 1.0) -> np.ndarray:
    """FFT sample frequencies for a window of ``n`` samples with spacing ``dt``."""
    return np.asarray(np.fft.fftfreq(n, d=dt))

power_spectrum(x, fs)

Periodogram-based power spectral density of a uniformly sampled signal.

Source code in src/cds2/signals.py 
88
89
90
91
def power_spectrum(x: object, fs: float) -> Spectrum:
    """Periodogram-based power spectral density of a uniformly sampled signal."""
    freqs, psd = sps.periodogram(np.asarray(x, dtype=float), fs=fs)
    return Spectrum(frequencies=np.asarray(freqs), power=np.asarray(psd))

welch_spectrum(x, fs, nperseg=256)

Welch-averaged power spectral density (lower variance than periodogram).

Source code in src/cds2/signals.py 
94
95
96
97
98
def welch_spectrum(x: object, fs: float, nperseg: int = 256) -> Spectrum:
    """Welch-averaged power spectral density (lower variance than periodogram)."""
    values = np.asarray(x, dtype=float)
    freqs, psd = sps.welch(values, fs=fs, nperseg=min(nperseg, values.size))
    return Spectrum(frequencies=np.asarray(freqs), power=np.asarray(psd))

spectrogram(x, fs, nperseg=256)

STFT magnitude squared over sliding windows.

Source code in src/cds2/signals.py 
101
102
103
104
105
106
107
def spectrogram(x: object, fs: float, nperseg: int = 256) -> SpectrogramResult:
    """STFT magnitude squared over sliding windows."""
    values = np.asarray(x, dtype=float)
    freqs, times, sxx = sps.spectrogram(values, fs=fs, nperseg=min(nperseg, values.size))
    return SpectrogramResult(
        times=np.asarray(times), frequencies=np.asarray(freqs), power=np.asarray(sxx)
    )

butter_lowpass(cutoff, fs, order=5)

Second-order-section coefficients of a Butterworth low-pass filter.

Source code in src/cds2/signals.py 
110
111
112
113
114
def butter_lowpass(cutoff: float, fs: float, order: int = 5) -> np.ndarray:
    """Second-order-section coefficients of a Butterworth low-pass filter."""
    nyquist = fs / 2.0
    sos = sps.butter(order, cutoff / nyquist, btype="low", output="sos")
    return np.asarray(sos)

butter_highpass(cutoff, fs, order=5)

Second-order-section coefficients of a Butterworth high-pass filter.

Source code in src/cds2/signals.py 
117
118
119
120
121
def butter_highpass(cutoff: float, fs: float, order: int = 5) -> np.ndarray:
    """Second-order-section coefficients of a Butterworth high-pass filter."""
    nyquist = fs / 2.0
    sos = sps.butter(order, cutoff / nyquist, btype="high", output="sos")
    return np.asarray(sos)

butter_bandpass(low, high, fs, order=5)

Second-order-section coefficients of a Butterworth band-pass filter.

Source code in src/cds2/signals.py 
124
125
126
127
128
def butter_bandpass(low: float, high: float, fs: float, order: int = 5) -> np.ndarray:
    """Second-order-section coefficients of a Butterworth band-pass filter."""
    nyquist = fs / 2.0
    sos = sps.butter(order, [low / nyquist, high / nyquist], btype="band", output="sos")
    return np.asarray(sos)

filter_signal(x, sos)

Zero-phase forward-backward filtering with second-order sections.

Source code in src/cds2/signals.py 
131
132
133
134
def filter_signal(x: object, sos: object) -> np.ndarray:
    """Zero-phase forward-backward filtering with second-order sections."""
    filtered = sps.sosfiltfilt(sos, np.asarray(x, dtype=float))
    return np.asarray(filtered, dtype=float)

moving_average(x, window)

Centered moving average with edge-preserving 'same' convolution.

Source code in src/cds2/signals.py 
137
138
139
140
141
142
143
def moving_average(x: object, window: int) -> np.ndarray:
    """Centered moving average with edge-preserving 'same' convolution."""
    if window < 1:
        msg = "window must be at least 1"
        raise ValueError(msg)
    kernel = np.ones(window) / window
    return np.asarray(np.convolve(np.asarray(x, dtype=float), kernel, mode="same"))

convolve_signals(a, b, mode='same')

Linear convolution of two sequences.

Source code in src/cds2/signals.py 
146
147
148
149
150
def convolve_signals(a: object, b: object, mode: str = "same") -> np.ndarray:
    """Linear convolution of two sequences."""
    return np.asarray(
        sps.convolve(np.asarray(a, dtype=float), np.asarray(b, dtype=float), mode=mode)
    )

correlate_signals(a, b, mode='same')

Cross-correlation of two sequences.

Source code in src/cds2/signals.py 
153
154
155
156
157
def correlate_signals(a: object, b: object, mode: str = "same") -> np.ndarray:
    """Cross-correlation of two sequences."""
    return np.asarray(
        sps.correlate(np.asarray(a, dtype=float), np.asarray(b, dtype=float), mode=mode)
    )

find_peaks(x, height=None, distance=None, prominence=None)

Local maxima detection with optional height/distance/prominence filters.

Source code in src/cds2/signals.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def find_peaks(
    x: object,
    height: float | None = None,
    distance: int | None = None,
    prominence: float | None = None,
) -> PeakResult:
    """Local maxima detection with optional height/distance/prominence filters."""
    values = np.asarray(x, dtype=float)
    kwargs: dict[str, float] = {}
    if height is not None:
        kwargs["height"] = height
    if distance is not None:
        kwargs["distance"] = distance
    if prominence is not None:
        kwargs["prominence"] = prominence
    indices, _properties = sps.find_peaks(values, **kwargs)
    index_array = np.asarray(indices, dtype=np.int64)
    return PeakResult(indices=index_array, heights=np.asarray(values[index_array], dtype=float))

envelope(x)

Analytic-signal amplitude envelope via the Hilbert transform.

Source code in src/cds2/signals.py 
180
181
182
183
def envelope(x: object) -> np.ndarray:
    """Analytic-signal amplitude envelope via the Hilbert transform."""
    analytic = sps.hilbert(np.asarray(x, dtype=float))
    return np.asarray(np.abs(analytic), dtype=float)

resample_signal(x, num)

Fourier-method resampling to exactly num samples.

Source code in src/cds2/signals.py 
186
187
188
189
def resample_signal(x: object, num: int) -> np.ndarray:
    """Fourier-method resampling to exactly ``num`` samples."""
    resampled = sps.resample(np.asarray(x, dtype=float), num)
    return np.asarray(resampled, dtype=float)

detrend_signal(x)

Remove the best-fit linear trend from a signal.

Source code in src/cds2/signals.py 
192
193
194
195
def detrend_signal(x: object) -> np.ndarray:
    """Remove the best-fit linear trend from a signal."""
    detrended = sps.detrend(np.asarray(x, dtype=float))
    return np.asarray(detrended, dtype=float)

stft(x, fs, nperseg=256)

Short-time Fourier transform returning (times, frequencies, complex Zxx).

Source code in src/cds2/signals.py 
198
199
200
201
def stft(x: object, fs: float, nperseg: int = 256) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Short-time Fourier transform returning (times, frequencies, complex Zxx)."""
    freqs, times, zxx = sps.stft(np.asarray(x, dtype=float), fs=fs, nperseg=nperseg)
    return np.asarray(times), np.asarray(freqs), np.asarray(zxx)

coherence(x, y, fs, nperseg=256)

Magnitude-squared coherence between two signals.

Source code in src/cds2/signals.py 
204
205
206
207
208
209
def coherence(x: object, y: object, fs: float, nperseg: int = 256) -> Spectrum:
    """Magnitude-squared coherence between two signals."""
    signal_x = np.asarray(x, dtype=float)
    signal_y = np.asarray(y, dtype=float)
    freqs, coh = sps.coherence(signal_x, signal_y, fs=fs, nperseg=min(nperseg, signal_x.size))
    return Spectrum(frequencies=np.asarray(freqs), power=np.asarray(coh))