signal-processing

name	signal-processing
description	Signal processing fundamentals including Fourier analysis, filter design, sampling theory, spectral estimation, and adaptive filtering
license	MIT
compatibility	opencode
metadata	{"audience":"engineers","category":"engineering"}

What I do

• Analyze signals in time and frequency domains
• Design digital and analog filters
• Perform spectral analysis and estimation
• Implement adaptive filtering algorithms
• Process audio, image, and sensor data
• Design sampling and reconstruction systems
• Implement signal compression and encoding
• Develop detection and estimation algorithms

When to use me

When analyzing signals, designing filters, implementing spectral analysis, or processing sensor data for embedded systems and applications.

Core Concepts

• Fourier analysis (DFT, FFT, DTFT)
• Sampling theory and aliasing
• Filter design (FIR, IIR, analog)
• Z-transform and system representation
• Spectral estimation (periodogram, Welch, parametric)
• Adaptive filtering (LMS, RLS, Kalman)
• Window functions and spectral leakage
• Signal conditioning and preprocessing
• Detection and estimation theory
• Multirate signal processing

Code Examples

Fourier Analysis

import numpy as np
from dataclasses import dataclass
from typing import Tuple
import matplotlib.pyplot as plt

def dft(x: np.ndarray) -> np.ndarray:
    """Compute discrete Fourier transform."""
    N = len(x)
    X = np.zeros(N, dtype=complex)
    for k in range(N):
        for n in range(N):
            X[k] += x[n] * np.exp(-2j * np.pi * k * n / N)
    return X

def fft(x: np.ndarray) -> np.ndarray:
    """Compute fast Fourier transform."""
    N = len(x)
    if N <= 16:
        return dft(x)
    even = fft(x[::2])
    odd = fft(x[1::2])
    factor = np.exp(-2j * np.pi * np.arange(N) / N)
    return np.concatenate([even + factor[:N//2] * odd,
                           even + factor[N//2:] * odd])

def fft_frequency_bin(
    N: int,
    fs: float
) -> np.ndarray:
    """Calculate frequency bins for FFT."""
    return np.fft.fftfreq(N, 1/fs)

def stft() -> np.ndarray:
    
    n_frames =  + ((x) - (window)) // hop
    X = np.zeros((fft_size //  + , n_frames), dtype=)
     i  (n_frames):
        x_frame = x[i * hop : i * hop + (window)] * window
        X[:, i] = fft(x_frame)[:fft_size //  + ]
     X

 () -> [np.ndarray, np.ndarray, np.ndarray]:
    
    window = np.hanning(nperseg)
    hop = nperseg - noverlap
    S = stft(x, window, hop, nperseg)
    f = fft_frequency_bin(nperseg, fs)
    t = np.arange(S.shape[]) * hop / fs
     np.(S)**, f, t


fs =   
t = np.arange(, , /fs)
x = np.sin( * np.pi *  * t) +  * np.sin( * np.pi *  * t)
X = fft(x)
freq = fft_frequency_bin((x), fs)
()

Filter Design

@dataclass
class FilterSpec:
    pass

def fir_window_design(
    cutoff: float,
    fs: float,
    window_type: str = "hamming",
    num_taps: int = None
) -> np.ndarray:
    """Design FIR filter using window method."""
    if num_taps is None:
        num_taps = int(4 / (cutoff / fs)) | 1
    
    if window_type == "rectangular":
        window = np.ones(num_taps)
    elif window_type == "hanning":
        window = np.hanning(num_taps)
    elif window_type == "hamming":
        window = np.hamming(num_taps)
    elif window_type == "blackman":
        window = np.blackman(num_taps)
    else:
        window = np.hamming(num_taps)
    
    wc = 2 * np.pi * cutoff / fs
    h = np.zeros(num_taps)
    for n in range(num_taps):
        if n == num_taps // 2:
            h[n] = wc / np.pi
        else:
            h[n] = np.sin(wc * (n - num_taps // 2)) / (np.pi * (n - num_taps // 2))
    
    return h * window

def iir_butterworth_design(
    wp: float,
    ws: float,
    gpass: float,
    gstop: ,
    fs: 
) -> [np.ndarray, np.ndarray]:
    
    wp_norm =  * wp / fs
    ws_norm =  * ws / fs
    
    n = np.ceil(np.log10(**(gpass/) - ) / np.log10(**(gstop/) - ) / 
                (np.log10(ws_norm) - np.log10(wp_norm)) / )
    wc = wp_norm / (**( * gpass) - )**(/(*n))
    
    z, p, k = butter poles(n, wc, analog=, output=)
     z, p

 () -> [np.ndarray, np.ndarray]:
    
    num = np.array([b0/a0, b1/a0, b2/a0])
    den = np.array([, a1/a0, a2/a0])
     num, den

 () -> [np.ndarray, np.ndarray]:
    
    fprewarp =  * fs * np.tan(np.pi * np.linspace(, , ) / fs)
    z = np.exp( * np.pi * np.linspace(, , ) / fs)
     z, z


cutoff =   
fs =   
h = fir_window_design(cutoff, fs, , )
()
()

Spectral Estimation

def periodogram(
    x: np.ndarray,
    fs: float,
    window: str = "hamming"
) -> Tuple[np.ndarray, np.ndarray]:
    """Compute periodogram spectral estimate."""
    N = len(x)
    w = np.hanning(N) if window == "hanning" else np.ones(N)
    xw = x * w
    X = np.fft.fft(xw)
    Pxx = (np.abs(X)**2) / (np.sum(w**2) / N)
    freqs = np.fft.fftfreq(N, 1/fs)
    return Pxx[:N//2], freqs[:N//2]

def welch_psd(
    x: np.ndarray,
    fs: float,
    nperseg: int = 256,
    noverlap: int = 128
) -> Tuple[np.ndarray, np.ndarray]:
    """Compute Welch's method PSD estimate."""
    window = np.hanning(nperseg)
    hop = nperseg - noverlap
    n_frames = 1 + (len(x) - nperseg) // hop
    
    Pxx_sum = np.zeros(nperseg // 2 + 1)
    for i in range(n_frames):
        x_frame = x[i * hop : i * hop + nperseg] * window
        X = np.fft.fft(x_frame)
        Pxx = np.abs(X[:nperseg//2 + 1])**2
        Pxx_sum += Pxx
    
    Pxx_avg = Pxx_sum / n_frames
    Pxx_avg[1:-1] *= 2  # Single-sided correction
    freqs = np.fft.fftfreq(nperseg, /fs)[:nperseg// + ]
     Pxx_avg, freqs

 () -> [np.ndarray, np.ndarray]:
    
     scipy.signal  lfilter
    n = (x)
    r = np.correlate(x, x, mode=)
    r = r[n-:n+order]
    
    
    a = np.zeros(order + )
    a[] = 
     k  (order):
        
    
    freqs = np.fft.fftfreq(nfft, /fs)[:nfft// + ]
     np.(np.fft.fft(a, nfft))**, freqs

 () -> np.ndarray:
    
    N = (x)
    R = np.correlate(x, x, mode=)
    R = R[N-:N+n_sources]
    U, S, Vh = np.linalg.svd(R)
    noise_subspace = U[:, n_sources:]
     np.arange(nfft) / nfft * fs / 

Adaptive Filtering

class LMSFilter:
    def __init__(self, order: float, mu: float):
        self.order = int(order)
        self.mu = mu
        self.w = np.zeros(order + 1)
    
    def filter(self, x: np.ndarray, d: np.ndarray) -> np.ndarray:
        """LMS adaptive filtering."""
        y = np.zeros(len(x))
        for n in range(self.order, len(x)):
            x_vec = x[n - self.order : n + 1][::-1]
            y[n] = np.dot(self.w, x_vec)
            e = d[n] - y[n]
            self.w += self.mu * e * x_vec
        return y

class RLSFilter:
    def __init__(self, order: float, delta: float, lambda_rls: float):
        self.order = int(order)
        self.lambda_rls = lambda_rls
        self.delta = delta
        self.w = np.zeros(order + 1)
        self.P = np.eye(order + 1) / delta
    
    def () -> np.ndarray:
        
        y = np.zeros((x))
         n  (.order, (x)):
            x_vec = x[n - .order : n + ][::-]
            y[n] = np.dot(.w, x_vec)
            e = d[n] - y[n]
            Px = .P @ x_vec
            k = Px / (.lambda_rls + np.dot(x_vec, Px))
            .w += k * e
            .P = (.P - np.outer(k, x_vec) @ .P) / .lambda_rls
         y

 () -> [np.ndarray, np.ndarray]:
    
    x = np.zeros((z))
    P = np.zeros((z))
    x[] = x0
    P[] = P0
    
     k  (, (z)):
        x_pred = x[k-]
        P_pred = P[k-] + Q
        K = P_pred / (P_pred + R)
        x[k] = x_pred + K * (z[k] - x_pred)
        P[k] = ( - K) * P_pred
    
     x, P


np.random.seed()
N = 
mu = 
order = 


v = np.random.randn(N)
s = np.sin( * np.pi *  * np.arange(N) / )
d = s +  * v


x = np.concatenate([[], v[:-]])

lms = LMSFilter(order, mu)
y = lms.(x, d)
()

Sampling and Reconstruction

def nyquist_rate(
    signal_bandwidth: float
) -> float:
    """Calculate Nyquist sampling rate."""
    return 2 * signal_bandwidth

def anti_aliasing_design(
    fp: float,  # passband edge Hz
    fs: float,  # stopband edge Hz
    As: float,  # stopband attenuation dB
    delta_f: float  # transition width Hz
) -> Tuple[float, int]:
    """Design anti-aliasing filter requirements."""
    delta_p = 10**(-0.05) - 1
    delta_s = 10**(-As/20)
    delta_f_norm = delta_f / fs
    
    # Kaiser window design
    if As > 50:
        beta = 0.1102 * (As - 8.7)
    elif As > 21:
        beta = 0.5842 * (As - 21)**0.4 + 0.07886 * (As - 21)
    else:
        beta = 5
    
    N = int((As - 8) / (2.285 * 2 * np.pi * delta_f_norm)) + 1
    return N, beta

def sinc_interpolation() -> np.ndarray:
    
    y = np.zeros((t))
     n, xn  (x):
        y += xn * np.sinc((t - n * T) / T)
     y

 () -> np.ndarray:
    
    N = (x)
    t_original = np.arange(N) / fs
    t_hold = np.arange(, N, T_hold * fs) / fs
     np.interp(t_original, t_hold, x)

Best Practices

• Always consider anti-aliasing filtering before sampling
• Use appropriate window functions to minimize spectral leakage
• Choose FFT size based on frequency resolution requirements
• Consider numerical precision in IIR filter implementations
• Use cascaded biquad sections for high-order IIR filters
• Implement proper initialization for adaptive filters
• Verify filter stability before deployment
• Consider quantization effects in fixed-point implementations
• Use overlap-add or overlap-save for efficient convolution
• Document filter specifications and design parameters