0
# Signal Processing
1

2
Comprehensive signal processing tools for physiological data including resampling, filtering, peak detection, QRS detection, heart rate analysis, and signal evaluation. This module provides algorithms specifically designed for biomedical signal analysis and cardiovascular data processing.
3

4
## Capabilities
5

6
### Basic Signal Operations
7

8
Fundamental signal processing operations including resampling, normalization, and filtering utilities.
9

10
```python { .api }
11
def resample_sig(x: np.ndarray, fs: float, fs_target: float) -> Tuple[np.ndarray, np.ndarray]:
12
    """
13
    Resample signal to different sampling frequency.
14
    
15
    Parameters:
16
    - x: ndarray, input signal
17
    - fs: float, original sampling frequency in Hz
18
    - fs_target: float, target sampling frequency in Hz
19
    
20
    Returns:
21
    - resampled_x: ndarray, resampled signal
22
    - resampled_t: ndarray, resampled time points
23
    """
24

25
def resample_ann(ann_sample: Union[List[int], np.ndarray], fs: float, 
26
                 fs_target: float) -> np.ndarray:
27
    """
28
    Resample annotation locations to match new sampling frequency.
29
    
30
    Parameters:
31
    - ann_sample: array-like, original annotation sample locations
32
    - fs: float, original sampling frequency
33
    - fs_target: float, target sampling frequency
34
    
35
    Returns:
36
    ndarray of resampled annotation locations
37
    """
38

39
def resample_singlechan(x: np.ndarray, ann: Annotation, fs: float, 
40
                       fs_target: float) -> Tuple[np.ndarray, Annotation]:
41
    """
42
    Resample single channel signal with annotations.
43
    
44
    Parameters:
45
    - x: ndarray, single channel signal
46
    - ann: Annotation, annotation object
47
    - fs: float, original sampling frequency
48
    - fs_target: float, target sampling frequency
49
    
50
    Returns:
51
    - resampled_x: ndarray, resampled signal
52
    - resampled_ann: Annotation, resampled annotations
53
    """
54

55
def resample_multichan(xs: np.ndarray, ann: Annotation, fs: float, 
56
                      fs_target: float, resamp_ann_chan: int = 0) -> Tuple[np.ndarray, Annotation]:
57
    """
58
    Resample multiple channel signals with annotations.
59
    
60
    Parameters:
61
    - xs: ndarray, multi-channel signals (MxN array)
62
    - ann: Annotation, annotation object
63
    - fs: float, original sampling frequency
64
    - fs_target: float, target sampling frequency
65
    - resamp_ann_chan: int, channel to use for annotation resampling
66
    
67
    Returns:
68
    - resampled_xs: ndarray, resampled signals
69
    - resampled_ann: Annotation, resampled annotations
70
    """
71

72
def normalize_bound(sig: np.ndarray, lb: float = 0, ub: float = 1) -> np.ndarray:
73
    """
74
    Normalize signal values between specified bounds.
75
    
76
    Parameters:
77
    - sig: ndarray, input signal
78
    - lb: float, lower bound (default: 0)
79
    - ub: float, upper bound (default: 1)
80
    
81
    Returns:
82
    ndarray of normalized signal values
83
    """
84

85
def get_filter_gain(b: np.ndarray, a: np.ndarray, f_gain: float, fs: float) -> float:
86
    """
87
    Calculate filter gain at specific frequency.
88
    
89
    Parameters:
90
    - b: ndarray, numerator filter coefficients
91
    - a: ndarray, denominator filter coefficients  
92
    - f_gain: float, frequency to calculate gain at
93
    - fs: float, sampling frequency
94
    
95
    Returns:
96
    Filter gain at specified frequency
97
    """
98
```
99

100
### Peak Detection
101

102
Algorithms for detecting peaks in physiological signals with various detection strategies.
103

104
```python { .api }
105
def find_peaks(sig: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
106
    """
107
    Find hard and soft peaks in signal using derivative-based method.
108
    
109
    Parameters:
110
    - sig: ndarray, input signal
111
    
112
    Returns:
113
    - hard_peaks: ndarray, indices of hard peaks (strong local maxima)
114
    - soft_peaks: ndarray, indices of soft peaks (weak local maxima)
115
    """
116

117
def find_local_peaks(sig: np.ndarray, radius: int) -> np.ndarray:
118
    """
119
    Find local peaks within specified radius.
120
    
121
    Parameters:
122
    - sig: ndarray, input signal
123
    - radius: int, search radius for local maxima
124
    
125
    Returns:
126
    ndarray of peak indices
127
    """
128

129
def correct_peaks(sig: np.ndarray, peak_inds: np.ndarray, search_radius: int,
130
                 smooth_window_size: int, peak_dir: str = "compare") -> np.ndarray:
131
    """
132
    Adjust peak locations to local maxima or minima.
133
    
134
    Parameters:
135
    - sig: ndarray, input signal
136
    - peak_inds: ndarray, initial peak indices
137
    - search_radius: int, radius to search for corrections
138
    - smooth_window_size: int, window size for smoothing
139
    - peak_dir: str, direction to search ("up", "down", "compare")
140
    
141
    Returns:
142
    ndarray of corrected peak indices
143
    """
144
```
145

146
### QRS Detection
147

148
Specialized algorithms for detecting QRS complexes in ECG signals.
149

150
```python { .api }
151
def xqrs_detect(sig: np.ndarray, fs: float = 250, sampfrom: int = 0,
152
                sampto: Union[int, str] = 'end', conf: Dict = None,
153
                learn: bool = True, verbose: bool = True) -> np.ndarray:
154
    """
155
    Detect QRS complexes using XQRS algorithm.
156
    
157
    Parameters:
158
    - sig: ndarray, input ECG signal
159
    - fs: float, sampling frequency in Hz
160
    - sampfrom: int, starting sample for detection
161
    - sampto: int or 'end', ending sample for detection
162
    - conf: dict, configuration parameters for XQRS
163
    - learn: bool, enable learning mode for adaptation
164
    - verbose: bool, print detection progress information
165
    
166
    Returns:
167
    ndarray of QRS peak sample locations
168
    """
169

170
def gqrs_detect(sig: np.ndarray, fs: float = 250, adcgain: float = None,
171
                adczero: float = None, threshold: float = 1.0, hr: int = 75,
172
                RRdelta: float = 0.2, RRmin: float = 0.28, RRmax: float = 2.4,
173
                QS: float = 0.07, QT: float = 0.35, RTmin: float = 0.25,
174
                RTmax: float = 0.33, QRSa: float = 750, QRSamin: float = 130) -> np.ndarray:
175
    """
176
    Detect QRS complexes using GQRS algorithm.
177
    
178
    Parameters:
179
    - sig: ndarray, input ECG signal
180
    - fs: float, sampling frequency in Hz
181
    - adcgain: float, ADC gain value
182
    - adczero: float, ADC zero level
183
    - threshold: float, detection threshold
184
    - hr: int, expected heart rate in BPM
185
    - RRdelta: float, RR interval variation tolerance
186
    - RRmin: float, minimum RR interval in seconds
187
    - RRmax: float, maximum RR interval in seconds
188
    - QS: float, QRS onset to peak time in seconds
189
    - QT: float, QT interval duration in seconds
190
    - RTmin: float, minimum RT interval in seconds
191
    - RTmax: float, maximum RT interval in seconds
192
    - QRSa: float, QRS area threshold
193
    - QRSamin: float, minimum QRS area
194
    
195
    Returns:
196
    ndarray of QRS peak sample locations
197
    """
198
```
199

200
### Heart Rate Analysis
201

202
Functions for calculating heart rate metrics and RR interval analysis.
203

204
```python { .api }
205
def compute_hr(sig_len: int, qrs_inds: np.ndarray, fs: float) -> np.ndarray:
206
    """
207
    Calculate instantaneous heart rate from QRS locations.
208
    
209
    Parameters:
210
    - sig_len: int, total signal length in samples
211
    - qrs_inds: ndarray, QRS peak sample locations
212
    - fs: float, sampling frequency in Hz
213
    
214
    Returns:
215
    ndarray of instantaneous heart rate values in BPM
216
    """
217

218
def calc_rr(sample: Union[List[int], np.ndarray], fs: float = None, 
219
            ann_style: str = 'index', time_warn: bool = True) -> np.ndarray:
220
    """
221
    Calculate RR intervals from sample locations.
222
    
223
    Parameters:
224
    - sample: array-like, sample locations or time values
225
    - fs: float, sampling frequency (required if ann_style='index')
226
    - ann_style: str, 'index' for sample indices or 'time' for time values
227
    - time_warn: bool, warn about time format assumptions
228
    
229
    Returns:
230
    ndarray of RR intervals in seconds
231
    """
232

233
def calc_mean_hr(rr: np.ndarray, fs: float = None, min_rr: float = None,
234
                 max_rr: float = None, rr_units: str = "samples") -> float:
235
    """
236
    Calculate mean heart rate from RR intervals.
237
    
238
    Parameters:
239
    - rr: ndarray, RR intervals
240
    - fs: float, sampling frequency (if rr_units='samples')
241
    - min_rr: float, minimum acceptable RR interval
242
    - max_rr: float, maximum acceptable RR interval
243
    - rr_units: str, 'samples' or 'seconds'
244
    
245
    Returns:
246
    Mean heart rate in BPM
247
    """
248

249
def ann2rr(record_name: str, extension: str, pn_dir: str = None,
250
           channel: Union[str, int] = 'all', sampfrom: int = 0,
251
           sampto: Union[int, str] = 'end', shift_samps: bool = False,
252
           return_anns: bool = False, ann_style: str = 'index') -> Union[np.ndarray, Tuple[np.ndarray, Annotation]]:
253
    """
254
    Convert annotations to RR intervals.
255
    
256
    Parameters:
257
    - record_name: str, record name
258
    - extension: str, annotation extension
259
    - pn_dir: str, PhysioNet database directory
260
    - channel: str or int, channel selection ('all' or specific channel)
261
    - sampfrom: int, starting sample
262
    - sampto: int or 'end', ending sample
263
    - shift_samps: bool, shift sample numbers
264
    - return_anns: bool, also return annotation object
265
    - ann_style: str, 'index' or 'time' format
266
    
267
    Returns:
268
    RR intervals array, optionally with annotation object
269
    """
270

271
def rr2ann(rr_array: np.ndarray, record_name: str, extension: str,
272
           fs: float = 250, as_time: bool = False) -> None:
273
    """
274
    Convert RR intervals to annotation file.
275
    
276
    Parameters:
277
    - rr_array: ndarray, RR intervals in seconds
278
    - record_name: str, output record name
279
    - extension: str, output annotation extension
280
    - fs: float, sampling frequency for conversion
281
    - as_time: bool, treat RR intervals as time values
282
    """
283
```
284

285
### Signal Evaluation
286

287
Tools for comparing and evaluating signal processing results.
288

289
```python { .api }
290
def compare_annotations(ref_sample: np.ndarray, test_sample: np.ndarray,
291
                       window_width: float, signal: np.ndarray = None) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
292
    """
293
    Compare reference and test annotations for accuracy evaluation.
294
    
295
    Parameters:
296
    - ref_sample: ndarray, reference annotation sample locations
297
    - test_sample: ndarray, test annotation sample locations
298
    - window_width: float, matching window width in samples
299
    - signal: ndarray, optional signal for additional analysis
300
    
301
    Returns:
302
    - true_positives: ndarray, matched test annotations 
303
    - false_positives: ndarray, unmatched test annotations
304
    - false_negatives: ndarray, unmatched reference annotations
305
    """
306

307
def benchmark_mitdb(detector: Callable, verbose: bool = False, 
308
                   print_results: bool = False) -> Dict[str, float]:
309
    """
310
    Benchmark detector performance on MIT-BIH Arrhythmia Database.
311
    
312
    Parameters:
313
    - detector: callable, detection function to benchmark
314
    - verbose: bool, print detailed progress information
315
    - print_results: bool, print benchmark results
316
    
317
    Returns:
318
    Dictionary with performance metrics (sensitivity, PPV, F1-score, etc.)
319
    """
320
```
321

322
### Signal Filtering
323

324
Signal filtering and averaging operations.
325

326
```python { .api }
327
def sigavg(sig_list: List[np.ndarray], mode: str = 'auto') -> np.ndarray:
328
    """
329
    Signal averaging/ensemble averaging.
330
    
331
    Parameters:
332
    - sig_list: list of ndarray, signals to average
333
    - mode: str, averaging mode ('auto', 'mean', 'median')
334
    
335
    Returns:
336
    ndarray of averaged signal
337
    """
338
```
339

340
## Types
341

342
```python { .api }
343
class XQRS:
344
    """Configurable QRS detector class."""
345
    
346
    def __init__(self, sig: np.ndarray, fs: float = 250):
347
        """
348
        Initialize XQRS detector.
349
        
350
        Parameters:
351
        - sig: ndarray, input ECG signal
352
        - fs: float, sampling frequency in Hz
353
        """
354
    
355
    def detect(self, sig: np.ndarray, verbose: bool = False) -> np.ndarray:
356
        """
357
        Detect QRS complexes in signal.
358
        
359
        Parameters:
360
        - sig: ndarray, input ECG signal
361
        - verbose: bool, print detection progress
362
        
363
        Returns:
364
        ndarray of QRS peak sample locations
365
        """
366

367
class Comparitor:
368
    """Class for comparing annotation sets."""
369
    
370
    def compare_annotations(self, ref_sample: np.ndarray, test_sample: np.ndarray,
371
                          window_width: float, signal: np.ndarray = None) -> Dict[str, np.ndarray]:
372
        """
373
        Compare reference and test annotations.
374
        
375
        Parameters:
376
        - ref_sample: ndarray, reference annotations
377
        - test_sample: ndarray, test annotations
378
        - window_width: float, matching window width
379
        - signal: ndarray, optional signal data
380
        
381
        Returns:
382
        Dictionary with comparison results
383
        """
384
```
385

386
## Usage Examples
387

388
### QRS Detection
389

390
```python
391
import wfdb
392
import numpy as np
393

394
# Read ECG record
395
record = wfdb.rdrecord('100', pn_dir='mitdb', channels=[0])
396
ecg_signal = record.p_signal[:, 0]
397

398
# Detect QRS complexes using XQRS
399
qrs_inds = wfdb.processing.xqrs_detect(ecg_signal, fs=record.fs)
400
print(f"Detected {len(qrs_inds)} QRS complexes")
401

402
# Alternative: use GQRS algorithm  
403
qrs_inds_gqrs = wfdb.processing.gqrs_detect(ecg_signal, fs=record.fs)
404

405
# Use XQRS class for more control
406
detector = wfdb.processing.XQRS(ecg_signal, fs=record.fs)
407
qrs_inds_class = detector.detect(ecg_signal, verbose=True)
408
```
409

410
### Heart Rate Analysis
411

412
```python
413
import wfdb
414
import numpy as np
415

416
# Read record and detect QRS
417
record = wfdb.rdrecord('100', pn_dir='mitdb')
418
qrs_inds = wfdb.processing.xqrs_detect(record.p_signal[:, 0], fs=record.fs)
419

420
# Calculate instantaneous heart rate
421
hr = wfdb.processing.compute_hr(record.sig_len, qrs_inds, record.fs)
422
print(f"Mean HR: {np.mean(hr):.1f} BPM")
423
print(f"HR range: {np.min(hr):.1f} - {np.max(hr):.1f} BPM")
424

425
# Calculate RR intervals
426
rr_intervals = wfdb.processing.calc_rr(qrs_inds, fs=record.fs, ann_style='index')
427
print(f"Mean RR: {np.mean(rr_intervals):.3f} seconds")
428

429
# Mean heart rate from RR intervals
430
mean_hr = wfdb.processing.calc_mean_hr(rr_intervals, rr_units='seconds')
431
print(f"Mean HR from RR: {mean_hr:.1f} BPM")
432
```
433

434
### Signal Resampling
435

436
```python
437
import wfdb
438
import numpy as np
439

440
# Read signal at original sampling rate
441
record = wfdb.rdrecord('100', pn_dir='mitdb')  # 360 Hz
442
original_signal = record.p_signal[:, 0]
443

444
# Resample to different frequency
445
target_fs = 250  # Hz
446
resampled_sig, resampled_t = wfdb.processing.resample_sig(
447
    original_signal, record.fs, target_fs)
448

449
print(f"Original: {len(original_signal)} samples at {record.fs} Hz")
450
print(f"Resampled: {len(resampled_sig)} samples at {target_fs} Hz")
451

452
# Normalize signal
453
normalized_sig = wfdb.processing.normalize_bound(resampled_sig, lb=-1, ub=1)
454
```
455

456
### Peak Detection
457

458
```python
459
import wfdb
460
import numpy as np
461

462
# Read signal
463
record = wfdb.rdrecord('100', pn_dir='mitdb', channels=[0])
464
signal = record.p_signal[:, 0]
465

466
# Find peaks using derivative method
467
hard_peaks, soft_peaks = wfdb.processing.find_peaks(signal)
468
print(f"Hard peaks: {len(hard_peaks)}, Soft peaks: {len(soft_peaks)}")
469

470
# Find local peaks with radius
471
local_peaks = wfdb.processing.find_local_peaks(signal, radius=10)
472

473
# Correct peak locations to local maxima
474
corrected_peaks = wfdb.processing.correct_peaks(
475
    signal, hard_peaks, search_radius=5, smooth_window_size=3)
476
```
477

478
### Performance Evaluation
479

480
```python
481
import wfdb
482

483
# Read reference annotations
484
ref_ann = wfdb.rdann('100', 'atr', pn_dir='mitdb')
485
ref_qrs = ref_ann.sample
486

487
# Detect QRS using algorithm
488
record = wfdb.rdrecord('100', pn_dir='mitdb', channels=[0])
489
test_qrs = wfdb.processing.xqrs_detect(record.p_signal[:, 0], fs=record.fs)
490

491
# Compare results
492
tp, fp, fn = wfdb.processing.compare_annotations(
493
    ref_qrs, test_qrs, window_width=50)  # 50 sample tolerance
494

495
# Calculate metrics
496
sensitivity = len(tp) / (len(tp) + len(fn))
497
ppv = len(tp) / (len(tp) + len(fp))
498
print(f"Sensitivity: {sensitivity:.3f}")
499
print(f"Positive Predictive Value: {ppv:.3f}")
500
```