processing¶
The processing subpackage contains signal-processing tools.
Basic Utility¶
Basic signal processing functions
-
wfdb.processing.
resample_ann
(resampled_t, ann_sample)¶ Compute the new annotation indices
- resampled_t : numpy array
- Array of signal locations as returned by scipy.signal.resample
- ann_sample : numpy array
- Array of annotation locations
- resampled_ann_sample : numpy array
- Array of resampled annotation locations
-
wfdb.processing.
resample_sig
(x, fs, fs_target)¶ Resample a signal to a different frequency.
- x : numpy array
- Array containing the signal
- fs : int, or float
- The original sampling frequency
- fs_target : int, or float
- The target frequency
- resampled_x : numpy array
- Array of the resampled signal values
- resampled_t : numpy array
- Array of the resampled signal locations
-
wfdb.processing.
resample_singlechan
(x, ann, fs, fs_target)¶ Resample a single-channel signal with its annotations
- x: numpy array
- The signal array
- ann : wfdb Annotation
- The wfdb annotation object
- fs : int, or float
- The original frequency
- fs_target : int, or float
- The target frequency
- resampled_x : numpy array
- Array of the resampled signal values
- resampled_ann : wfdb Annotation
- Annotation containing resampled annotation locations
-
wfdb.processing.
resample_multichan
(xs, ann, fs, fs_target, resamp_ann_chan=0)¶ Resample multiple channels with their annotations
- xs: numpy array
- The signal array
- ann : wfdb Annotation
- The wfdb annotation object
- fs : int, or float
- The original frequency
- fs_target : int, or float
- The target frequency
- resample_ann_channel : int, optional
- The signal channel used to compute new annotation indices
- resampled_xs : numpy array
- Array of the resampled signal values
- resampled_ann : wfdb Annotation
- Annotation containing resampled annotation locations
-
wfdb.processing.
normalize_bound
(sig, lb=0, ub=1)¶ Normalize a signal between the lower and upper bound
- sig : numpy array
- Original signal to be normalized
- lb : int, or float
- Lower bound
- ub : int, or float
- Upper bound
- x_normalized : numpy array
- Normalized signal
-
wfdb.processing.
get_filter_gain
(b, a, f_gain, fs)¶ Given filter coefficients, return the gain at a particular frequency.
- b : list
- List of linear filter b coefficients
- a : list
- List of linear filter a coefficients
- f_gain : int or float, optional
- The frequency at which to calculate the gain
- fs : int or float, optional
- The sampling frequency of the system
Heart Rate¶
-
wfdb.processing.
compute_hr
(sig_len, qrs_inds, fs)¶ Compute instantaneous heart rate from peak indices.
- sig_len : int
- The length of the corresponding signal
- qrs_inds : numpy array
- The qrs index locations
- fs : int, or float
- The corresponding signal’s sampling frequency.
- heart_rate : numpy array
- An array of the instantaneous heart rate, with the length of the corresponding signal. Contains numpy.nan where heart rate could not be computed.
Peaks¶
-
wfdb.processing.
find_peaks
(sig)¶ Find hard peaks and soft peaks in a signal, defined as follows:
- Hard peak: a peak that is either /or /
- Soft peak: a peak that is either /-or -/ In this case we define the middle as the peak
- sig : np array
- The 1d signal array
- hard_peaks : numpy array
- Array containing the indices of the hard peaks:
- soft_peaks : numpy array
- Array containing the indices of the soft peaks
-
wfdb.processing.
find_local_peaks
(sig, radius)¶ Find all local peaks in a signal. A sample is a local peak if it is the largest value within the <radius> samples on its left and right.
In cases where it shares the max value with nearby samples, the middle sample is classified as the local peak.
- sig : numpy array
- 1d numpy array of the signal.
- radius : int
- The radius in which to search for defining local maxima.
-
wfdb.processing.
correct_peaks
(sig, peak_inds, search_radius, smooth_window_size, peak_dir='compare')¶ Adjust a set of detected peaks to coincide with local signal maxima, and
- sig : numpy array
- The 1d signal array
- peak_inds : np array
- Array of the original peak indices
- max_gap : int
- The radius within which the original peaks may be shifted.
- smooth_window_size : int
- The window size of the moving average filter applied on the signal. Peak distance is calculated on the difference between the original and smoothed signal.
- peak_dir : str, optional
The expected peak direction: ‘up’ or ‘down’, ‘both’, or ‘compare’.
- If ‘up’, the peaks will be shifted to local maxima
- If ‘down’, the peaks will be shifted to local minima
- If ‘both’, the peaks will be shifted to local maxima of the rectified signal
- If ‘compare’, the function will try both ‘up’ and ‘down’ options, and choose the direction that gives the largest mean distance from the smoothed signal.
- corrected_peak_inds : numpy array
- Array of the corrected peak indices
QRS Detectors¶
-
class
wfdb.processing.
XQRS
(sig, fs, conf=None)¶ The qrs detector class for the xqrs algorithm.
The XQRS.Conf class is the configuration class that stores initial parameters for the detection.
The XQRS.detect method runs the detection algorithm.
The process works as follows:
- Load the signal and configuration parameters.
- Bandpass filter the signal between 5 and 20 Hz, to get the filtered signal.
- Apply moving wave integration (mwi) with a ricker (Mexican hat) wavelet onto the filtered signal, and save the square of the integrated signal.
- Conduct learning if specified, to initialize running parameters of noise and qrs amplitudes, the qrs detection threshold, and recent rr intervals. If learning is unspecified or fails, use default parameters. See the docstring for the _learn_init_params method of this class for details.
- Run the main detection. Iterate through the local maxima of
the mwi signal. For each local maxima:
- Check if it is a qrs complex. To be classified as a qrs, it must come after the refractory period, cross the qrs detection threshold, and not be classified as a t-wave if it comes close enough to the previous qrs. If successfully classified, update running detection threshold and heart rate parameters.
- If not a qrs, classify it as a noise peak and update running parameters.
- Before continuing to the next local maxima, if no qrs was detected within 1.66 times the recent rr interval, perform backsearch qrs detection. This checks previous peaks using a lower qrs detection threshold.
>>> import wfdb >>> from wfdb import processing
>>> sig, fields = wfdb.rdsamp('sample-data/100', channels=[0]) >>> xqrs = processing.XQRS(sig=sig[:,0], fs=fields['fs']) >>> xqrs.detect()
>>> wfdb.plot_items(signal=sig, ann_samp=[xqrs.qrs_inds])
-
class
Conf
(hr_init=75, hr_max=200, hr_min=25, qrs_width=0.1, qrs_thr_init=0.13, qrs_thr_min=0, ref_period=0.2, t_inspect_period=0.36)¶ Initial signal configuration object for this qrs detector
-
detect
(sampfrom=0, sampto='end', learn=True, verbose=True)¶ Detect qrs locations between two samples.
- sampfrom : int, optional
- The starting sample number to run the detection on.
- sampto : int, optional
- The final sample number to run the detection on. Set as ‘end’ to run on the entire signal.
- learn : bool, optional
- Whether to apply learning on the signal before running the main detection. If learning fails or is not conducted, the default configuration parameters will be used to initialize these variables. See the XQRS._learn_init_params docstring for details.
- verbose : bool, optional
- Whether to display the stages and outcomes of the detection process.
-
wfdb.processing.
xqrs_detect
(sig, fs, sampfrom=0, sampto='end', conf=None, learn=True, verbose=True)¶ Run the ‘xqrs’ qrs detection algorithm on a signal. See the docstring of the XQRS class for algorithm details.
- sig : numpy array
- The input ecg signal to apply the qrs detection on.
- fs : int or float
- The sampling frequency of the input signal.
- sampfrom : int, optional
- The starting sample number to run the detection on.
- sampto :
- The final sample number to run the detection on. Set as ‘end’ to run on the entire signal.
- conf : XQRS.Conf object, optional
- The configuration object specifying signal configuration parameters. See the docstring of the XQRS.Conf class.
- learn : bool, optional
- Whether to apply learning on the signal before running the main detection. If learning fails or is not conducted, the default configuration parameters will be used to initialize these variables.
- verbose : bool, optional
- Whether to display the stages and outcomes of the detection process.
- qrs_inds : numpy array
- The indices of the detected qrs complexes
>>> import wfdb >>> from wfdb import processing
>>> sig, fields = wfdb.rdsamp('sample-data/100', channels=[0]) >>> qrs_inds = processing.xqrs_detect(sig=sig[:,0], fs=fields['fs'])
-
wfdb.processing.
gqrs_detect
(sig=None, fs=None, d_sig=None, adc_gain=None, adc_zero=None, threshold=1.0, hr=75, RRdelta=0.2, RRmin=0.28, RRmax=2.4, QS=0.07, QT=0.35, RTmin=0.25, RTmax=0.33, QRSa=750, QRSamin=130)¶ Detect qrs locations in a single channel ecg. Functionally, a direct port of the gqrs algorithm from the original wfdb package. Accepts either a physical signal, or a digital signal with known adc_gain and adc_zero.
See the notes below for a summary of the program. This algorithm is not being developed/supported.
- sig : 1d numpy array, optional
- The input physical signal. The detection algorithm which replicates the original, works using digital samples, and this physical option is provided as a convenient interface. If this is the specified input signal, automatic adc is performed using 24 bit precision, to obtain the d_sig, adc_gain, and adc_zero parameters. There may be minor differences in detection results (ie. an occasional 1 sample difference) between using sig and d_sig. To replicate the exact output of the original gqrs algorithm, use the d_sig argument instead.
- fs : int, or float
- The sampling frequency of the signal.
- d_sig : 1d numpy array, optional
- The input digital signal. If this is the specified input signal rather than sig, the adc_gain and adc_zero parameters must be specified.
- adc_gain : int, or float, optional
- The analogue to digital gain of the signal (the number of adus per physical unit).
- adc_zero: int, optional
- The value produced by the ADC given a 0 volt input.
- threshold : int, or float, optional
- The relative amplitude detection threshold. Used to initialize the peak and qrs detection threshold.
- hr : int, or float, optional
- Typical heart rate, in beats per minute.
- RRdelta : int or float, optional
- Typical difference between successive RR intervals in seconds.
- RRmin : int or float, optional
- Minimum RR interval (“refractory period”), in seconds.
- RRmax : int or float, optional
- Maximum RR interval, in seconds. Thresholds will be adjusted if no peaks are detected within this interval.
- QS : int or float, optional
- Typical QRS duration, in seconds.
- QT : int or float, optional
- Typical QT interval, in seconds.
- RTmin : int or float, optional
- Minimum interval between R and T peaks, in seconds.
- RTmax : int or float, optional
- Maximum interval between R and T peaks, in seconds.
- QRSa : int or float, optional
- Typical QRS peak-to-peak amplitude, in microvolts.
- QRSamin : int or float, optional
- Minimum QRS peak-to-peak amplitude, in microvolts.
- qrs_locs : numpy array
- Detected qrs locations
This function should not be used for signals with fs <= 50Hz
The algorithm theoretically works as follows:
Load in configuration parameters. They are used to set/initialize the:
- allowed rr interval limits (fixed)
- initial recent rr interval (running)
- qrs width, used for detection filter widths (fixed)
- allowed rt interval limits (fixed)
- initial recent rt interval (running)
- initial peak amplitude detection threshold (running)
- initial qrs amplitude detection threshold (running)
- Note: this algorithm does not normalize signal amplitudes, and hence is highly dependent on configuration amplitude parameters.
Apply trapezoid low-pass filtering to the signal
Convolve a QRS matched filter with the filtered signal
Run the learning phase using a calculated signal length: detect qrs and non-qrs peaks as in the main detection phase, without saving the qrs locations. During this phase, running parameters of recent intervals and peak/qrs thresholds are adjusted.
- Run the detection::
if a sample is bigger than its immediate neighbors and larger than the peak detection threshold, it is a peak.
if it is further than RRmin from the previous qrs, and is a *primary peak.
if it is further than 2 standard deviations from the previous qrs, do a backsearch for a missed low amplitude beat
return the primary peak between the current sample and the previous qrs if any.
- if it surpasses the qrs threshold, it is a qrs complex
save the qrs location. update running rr and qrs amplitude parameters. look for the qrs complex’s t-wave and mark it if found.
- else if it is not a peak
lower the peak detection threshold if the last peak found was more than RRmax ago, and not already at its minimum.
*A peak is secondary if there is a larger peak within its neighborhood (time +- rrmin), or if it has been identified as a T-wave associated with a previous primary peak. A peak is primary if it is largest in its neighborhood, or if the only larger peaks are secondary.
The above describes how the algorithm should theoretically work, but there are bugs which make the program contradict certain parts of its supposed logic. A list of issues from the original c, code and hence this python implementation can be found here:
https://github.com/bemoody/wfdb/issues/17
gqrs will not be supported/developed in this library.
>>> import numpy as np >>> import wfdb >>> from wfdb import processing
>>> # Detect using a physical input signal >>> record = wfdb.rdrecord('sample-data/100', channels=[0]) >>> qrs_locs = processing.gqrs_detect(record.p_signal[:,0], fs=record.fs)
>>> # Detect using a digital input signal >>> record_2 = wfdb.rdrecord('sample-data/100', channels=[0], physical=False) >>> qrs_locs_2 = processing.gqrs_detect(d_sig=record_2.d_signal[:,0], fs=record_2.fs, adc_gain=record_2.adc_gain[0], adc_zero=record_2.adc_zero[0])
Annotation Evaluators¶
-
class
wfdb.processing.
Comparitor
(ref_sample, test_sample, window_width, signal=None)¶ The class to implement and hold comparisons between two sets of annotations.
See methods compare, print_summary and plot.
>>> import wfdb >>> from wfdb import processing
>>> sig, fields = wfdb.rdsamp('sample-data/100', channels=[0]) >>> ann_ref = wfdb.rdann('sample-data/100','atr') >>> xqrs = processing.XQRS(sig=sig[:,0], fs=fields['fs']) >>> xqrs.detect()
>>> comparitor = processing.Comparitor(ann_ref.sample[1:], xqrs.qrs_inds, int(0.1 * fields['fs']), sig[:,0]) >>> comparitor.compare() >>> comparitor.print_summary() >>> comparitor.plot()
-
compare
()¶ Main comparison function
-
plot
(sig_style='', title=None, figsize=None, return_fig=False)¶ Plot the comparison of two sets of annotations, possibly overlaid on their original signal.
- sig_style : str, optional
- The matplotlib style of the signal
- title : str, optional
- The title of the plot
- figsize: tuple, optional
- Tuple pair specifying the width, and height of the figure. It is the’figsize’ argument passed into matplotlib.pyplot’s figure function.
- return_fig : bool, optional
- Whether the figure is to be returned as an output argument.
-
print_summary
()¶ Print summary metrics of the annotation comparisons.
-
-
wfdb.processing.
compare_annotations
(ref_sample, test_sample, window_width, signal=None)¶ Compare a set of reference annotation locations against a set of test annotation locations.
See the Comparitor class docstring for more information.
- ref_sample : 1d numpy array
- Array of reference sample locations
- test_sample : 1d numpy array
- Array of test sample locations to compare
- window_width : int
- The maximum absolute difference in sample numbers that is permitted for matching annotations.
- signal : 1d numpy array, optional
- The original signal of the two annotations. Only used for plotting.
- comparitor : Comparitor object
- Object containing parameters about the two sets of annotations
>>> import wfdb >>> from wfdb import processing
>>> sig, fields = wfdb.rdsamp('sample-data/100', channels=[0]) >>> ann_ref = wfdb.rdann('sample-data/100','atr') >>> xqrs = processing.XQRS(sig=sig[:,0], fs=fields['fs']) >>> xqrs.detect()
>>> comparitor = processing.compare_annotations(ann_ref.sample[1:], xqrs.qrs_inds, int(0.1 * fields['fs']), sig[:,0]) >>> comparitor.print_summary() >>> comparitor.plot()
-
wfdb.processing.
benchmark_mitdb
(detector, verbose=False, print_results=False)¶ Benchmark a qrs detector against mitdb’s records.
- detector : function
- The detector function.
- verbose : bool, optional
- The verbose option of the detector function.
- print_results : bool, optional
- Whether to print the overall performance, and the results for each record.
- comparitors : dictionary
- Dictionary of Comparitor objects run on the records, keyed on the record names.
- sensitivity : float
- Aggregate sensitivity.
- positive_predictivity : float
- Aggregate positive_predictivity.
TODO: - remove non-qrs detections from reference annotations - allow kwargs
>>> import wfdb >> from wfdb.processing import benchmark_mitdb, xqrs_detect
>>> comparitors, spec, pp = benchmark_mitdb(xqrs_detect)