signal-processing

name	signal-processing
description	Signal processing theory and applications
license	MIT
compatibility	opencode
metadata	{"audience":"engineers, scientists, students","category":"engineering"}

What I do

• Design digital filters and signal processing algorithms
• Analyze signals in time and frequency domains
• Implement Fourier transforms and spectral analysis
• Develop signal conditioning and preprocessing
• Create data compression and feature extraction

When to use me

• When filtering noisy signals
• When analyzing frequency content
• When designing digital filters
• When extracting features from signals
• When implementing DSP algorithms

Key Concepts

Discrete Signals

import numpy as np
from scipy import signal, fft

# Signal generation
def generate_sine(freq, amplitude, duration, fs):
    """Generate sinusoidal signal"""
    t = np.arange(0, duration, 1/fs)
    return t, amplitude * np.sin(2 * np.pi * freq * t)

def generate_impulse(N):
    """Unit impulse signal"""
    impulse = np.zeros(N)
    impulse[0] = 1
    return impulse

def generate_step(N):
    """Unit step signal"""
    return np.ones(N)

# Basic operations
def convolution(x, h):
    """Linear convolution"""
    return np.convolve(x, h, mode='full')

def correlation(x, y):
    """Cross-correlation"""
    return np.correlate(x, y, mode='full')

Fourier Transform

# DFT/FFT
def compute_fft(x, fs):
    """Compute frequency spectrum"""
    N = len(x)
    X = fft.fft(x)
    freq = fft.fftfreq(N, 1/fs)
    
    # Get positive frequencies only
    positive = slice(0, N//2)
    return freq[positive], np.abs(X[positive])

def compute_stft(x, fs, nperseg=256, noverlap=None):
    """Short-Time Fourier Transform"""
    if noverlap is None:
        noverlap = nperseg // 2
    
    f, t, Sxx = signal.stft(x, fs, nperseg=nperseg, noverlap=noverlap)
    return f, t, np.abs(Sxx)

# Window functions
WINDOWS = {
    "hamming": lambda N: signal.windows.hamming(N),
    "hann": lambda N: signal.windows.hann(N),
    "blackman": lambda N: signal.windows.blackman(N),
    "kaiser": lambda N, beta=0: signal.windows.kaiser(N, beta)
}

Digital Filters

# IIR Filter design
def design_butterworth_lowpass(cutoff, fs, order=4):
    """Design Butterworth lowpass filter"""
    nyquist = fs / 2
    normalized = cutoff / nyquist
    b, a = signal.butter(order, normalized, btype='low')
    return b, a

def design_cheby1_lowpass(cutoff, fs, order=4, ripple=0.5):
    """Chebyshev Type I lowpass filter"""
    nyquist = fs / 2
    normalized = cutoff / nyquist
    b, a = signal.cheby1(order, ripple, normalized, btype='low')
    return b, a

def apply_filter(b, a, x):
    """Apply IIR filter using difference equation"""
    return signal.filtfilt(b, a, x)

# FIR Filter design
def design_fir_lowpass(cutoff, fs, num_taps=51):
    """Design FIR lowpass using window method"""
    nyquist = fs / 2
    normalized = cutoff / nyquist
    b = signal.firwin(num_taps, normalized, window='hamming')
    return b

# Filter structures
def direct_form_1(b, a, x):
    """Direct Form I IIR filter"""
    N = len(b)
    M = len(a)
    y = np.zeros(len(x))
    
    for n in range(len(x)):
         i  ((n+, N)):
            y[n] += b[i] * x[n-i]
         i  (, (n+, M)):
            y[n] -= a[i] * y[n-i]
    
     y

Filter Response Analysis

def frequency_response(b, a, fs):
    """Compute frequency response"""
    w, h = signal.freqz(b, a, worN=204    freq = w8)
 * fs / (2 * np.pi)
    return freq, 20 * np.log10(np.abs(h))

def phase_response(b, a):
    """Compute phase response"""
    w, h = signal.freqz(b, a, worN=2048)
    phase = np.unwrap(np.angle(h))
    return w, phase

def group_delay(b, a):
    """Compute group delay"""
    w, gd = signal.group_delay((b, a))
    return w, gd

Spectral Analysis

def periodogram(x, fs):
    """Compute periodogram (power spectral density)"""
    f, Pxx = signal.periodogram(x, fs)
    return f, Pxx

def welch_psd(x, fs, nperseg=256):
    """Compute PSD using Welch's method"""
    f, Pxx = signal.welch(x, fs, nperseg=nperseg)
    return f, Pxx

def spectrogram(x, fs, nperseg=256):
    """Compute spectrogram"""
    f, t, Sxx = signal.spectrogram(x, fs, nperseg=nperseg)
    return f, t, 10 * np.log10(Sxx)

Adaptive Filters

class LMSFilter:
    """Least Mean Squares adaptive filter"""
    def __init__(self, filter_length, mu):
        self.length = filter_length
        self.mu = mu  # Step size
        self.weights = np.zeros(filter_length)
    
    def filter(self, x, d):
        """Process input signal"""
        y = np.zeros_like(d)
        
        for n in range(len(d)):
            x_window = np.flip(x[max(0, n-self.length+1):n+1])
            if len(x_window) < self.length:
                x_window = np.pad(x_window, (self.length - len(x_window), 0))
            
            y[n] = np.dot(self.weights, x_window)
            error = d[n] - y[n]
            self.weights += self.mu * error * x_window
        
        return y

class RLSFilter:
    """Recursive Least Squares filter"""
    def __init__(self, filter_length, delta=0.01, lam=0.99):
        self.length = filter_length
        self.lam = lam  # Forgetting factor
        .delta = delta
        .weights = np.zeros(filter_length)
        .P = np.eye(filter_length) / delta
    
     ():
        
        y = np.zeros_like(d)
        
         n  ((d)):
            x_window = np.flip(x[(, n-.length+):n+])
             (x_window) < .length:
                x_window = np.pad(x_window, (.length - (x_window), ))
            
            y[n] = np.dot(.weights, x_window)
            error = d[n] - y[n]
            
            k = .P @ x_window / (.lam + x_window @ .P @ x_window)
            .weights += k * error
            .P = (.P - np.outer(k, x_window @ .P)) / .lam
        
         y

Feature Extraction

def extract_features(signal_data, fs):
    """Extract common time-domain features"""
    features = {
        "mean": np.mean(signal_data),
        "std": np.std(signal_data),
        "rms": np.sqrt(np.mean(signal_data**2)),
        "peak": np.max(np.abs(signal_data)),
        "crest_factor": np.max(np.abs(signal_data)) / np.sqrt(np.mean(signal_data**2)),
        "skewness": stats.skew(signal_data),
        "kurtosis": stats.kurtosis(signal_data)
    }
    
    # Zero crossing rate
    features["zcr"] = np.sum(np.diff(np.sign(signal_data)) != 0) / len(signal_data)
    
    return features