Core

EcgBatch

class EcgBatch(index, preloaded=None, unique_labels=None)[source]

Bases: cardio.dataset.dataset.batch.Batch

Batch class for ECG signals storing.

Contains ECG signals and additional metadata along with various processing methods.

Parameters:
  • index (DatasetIndex) – Unique identifiers of ECGs in the batch.
  • preloaded (tuple, optional) – Data to put in the batch if given. Defaults to None.
  • unique_labels (1-D ndarray, optional) – Array with unique labels in a dataset.
index

DatasetIndex – Unique identifiers of ECGs in the batch.

signal

1-D ndarray – Array of 2-D ndarrays with ECG signals in channels first format.

annotation

1-D ndarray – Array of dicts with different types of annotations.

meta

1-D ndarray – Array of dicts with metadata about signals.

target

1-D ndarray – Array with signals’ labels.

unique_labels

1-D ndarray – Array with unique labels in a dataset.

label_binarizer

LabelBinarizer – Object for label one-hot encoding.

Note

Some batch methods take index as their first argument after self. You should not specify it in your code, it will be passed automatically by inbatch_parallel decorator. For example, resample_signals method with index and fs arguments should be called as batch.resample_signals(fs).

Methods

Input/output methods

EcgBatch.load(src=None, fmt=None, components=None, ann_ext=None, *args, **kwargs)[source]

Load given batch components from source.

Most of the EcgBatch actions work under the assumption that both signal and meta components are loaded. In case this assumption is not fulfilled, normal operation of the actions is not guaranteed.

This method supports loading of signals from wfdb, DICOM, EDF, wav and blosc formats.

Parameters:
  • src (misc, optional) – Source to load components from.
  • fmt (str, optional) – Source format.
  • components (str or array-like, optional) – Components to load.
  • ann_ext (str, optional) – Extension of the annotation file.
Returns:

batch (EcgBatch) – Batch with loaded components. Changes batch data inplace.
EcgBatch.dump(dst=None, fmt=None, components=None, *args, **kwargs)

Save data to another array or a file.

Parameters:
  • dst – a destination (e.g. an array or a file name)
  • fmt (str) – a destination format, one of None, ‘blosc’, ‘csv’, ‘hdf5’, ‘feather’
  • components (None or str or tuple of str) – components to load
  • *args – other parameters are passed to format-specific writers
  • *kwargs – other parameters are passed to format-specific writers
EcgBatch.show_ecg(index=None, start=0, end=None, annot=None, subplot_size=(10, 4))[source]

Plot an ECG signal.

Optionally highlight QRS complexes along with P and T waves. Each channel is displayed on a separate subplot.

Parameters:
  • index (element of self.indices, optional) – Index of a signal to plot. If undefined, the first ECG in the batch is used.
  • start (int, optional) – The start point of the displayed part of the signal (in seconds).
  • end (int, optional) – The end point of the displayed part of the signal (in seconds).
  • annot (str, optional) – If not None, specifies attribute that stores annotation obtained from cardio.models.HMModel.
  • subplot_size (tuple) – Width and height of each subplot in inches.
Raises:

ValueError – If the chosen signal is not two-dimensional.

Batch processing

EcgBatch.deepcopy()

Return a deep copy of the batch.

Constructs a new Batch instance and then recursively copies all the objects found in the original batch, except the pipeline, which remains unchanged.

Returns:Batch
classmethod EcgBatch.merge(batches, batch_size=None)[source]

Concatenate a list of EcgBatch instances and split the result into two batches of sizes batch_size and sum(lens of batches) - batch_size respectively.

Parameters:
  • batches (list) – List of EcgBatch instances.
  • batch_size (positive int, optional) – Length of the first resulting batch. If None, equals the length of the concatenated batch.
Returns:

  • new_batch (EcgBatch) – Batch of no more than batch_size first items from the concatenation of input batches. Contains a deep copy of input batches’ data.
  • rest_batch (EcgBatch) – Batch of the remaining items. Contains a deep copy of input batches’ data.
Raises:

ValueError – If batch_size is non-positive or non-integer.

Versatile components processing

EcgBatch.apply_transform(func, *args, src='signal', dst='signal', **kwargs)[source]

Apply a function to each item in the batch.

Parameters:
  • func (callable) – A function to apply. Must accept an item of src as its first argument if src is not None.
  • src (str, array-like or None, optional) – The source to get the data from. If src is str, it is treated as the batch attribute or component name. Defaults to signal component.
  • dst (str, writeable array-like or None, optional) – The source to put the result in. If dst is str, it is treated as the batch attribute or component name. Defaults to signal component.
  • args (misc) – Any additional positional arguments to func.
  • kwargs (misc) – Any additional named arguments to func.
Returns:

batch (EcgBatch) – Transformed batch. If dst is str, the corresponding attribute or component is changed inplace.
EcgBatch.apply_to_each_channel(index, func, *args, src='signal', dst='signal', **kwargs)[source]

Apply a function to each slice of a signal over the axis 0 (typically the channel axis).

Parameters:
  • func (callable) – A function to apply. Must accept a signal as its first argument.
  • src (str, optional) – Batch attribute or component name to get the data from. Defaults to signal component.
  • dst (str, optional) – Batch attribute or component name to put the result in. Defaults to signal component.
  • args (misc) – Any additional positional arguments to func.
  • kwargs (misc) – Any additional named arguments to func.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.
EcgBatch.standardize(index, axis=None, eps=1e-10, *, src='signal', dst='signal')[source]

Standardize data along specified axes by removing the mean and scaling to unit variance.

Parameters:
  • axis (None or int or tuple of ints, optional) – Axis or axes along which standardization is performed. The default is to compute for the flattened array.
  • eps (float) – Small addition to avoid division by zero.
  • src (str, optional) – Batch attribute or component name to get the data from.
  • dst (str, optional) – Batch attribute or component name to put the result in.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.

Labels processing

EcgBatch.drop_labels(drop_list)[source]

Drop elements whose labels are in drop_list.

This method creates a new batch and updates only components and unique_labels attribute. The information stored in other attributes will be lost.

Parameters:drop_list (list) – Labels to be dropped from a batch.
Returns:batch (EcgBatch) – Filtered batch. Creates a new EcgBatch instance.
Raises:SkipBatchException – If all batch data was dropped. If the batch is created by a pipeline, its processing will be stopped and the pipeline will create the next batch.
EcgBatch.keep_labels(keep_list)[source]

Drop elements whose labels are not in keep_list.

This method creates a new batch and updates only components and unique_labels attribute. The information stored in other attributes will be lost.

Parameters:keep_list (list) – Labels to be kept in a batch.
Returns:batch (EcgBatch) – Filtered batch. Creates a new EcgBatch instance.
Raises:SkipBatchException – If all batch data was dropped. If the batch is created by a pipeline, its processing will be stopped and the pipeline will create the next batch.
EcgBatch.rename_labels(rename_dict)[source]

Rename labels with corresponding values from rename_dict.

Parameters:rename_dict (dict) – Dictionary containing (old label : new label) pairs.
Returns:batch (EcgBatch) – Batch with renamed labels. Changes self.target inplace.
EcgBatch.binarize_labels()[source]

Binarize labels in a batch in a one-vs-all fashion.

Returns:batch (EcgBatch) – Batch with binarized labels. Changes self.target inplace.

Channels processing

EcgBatch.drop_channels(names=None, indices=None)[source]

Drop channels whose names are in names or whose indices are in indices.

Parameters:
  • names (str or list or tuple, optional) – Names of channels to be dropped from a batch.
  • indices (int or list or tuple, optional) – Indices of channels to be dropped from a batch.
Returns:

batch (EcgBatch) – Batch with dropped channels. Changes self.signal and self.meta inplace.
Raises:
  • ValueError – If both names and indices are empty.
  • ValueError – If all channels should be dropped.
EcgBatch.keep_channels(names=None, indices=None)[source]

Drop channels whose names are not in names and whose indices are not in indices.

Parameters:
  • names (str or list or tuple, optional) – Names of channels to be kept in a batch.
  • indices (int or list or tuple, optional) – Indices of channels to be kept in a batch.
Returns:

batch (EcgBatch) – Batch with dropped channels. Changes self.signal and self.meta inplace.
Raises:
  • ValueError – If both names and indices are empty.
  • ValueError – If all channels should be dropped.
EcgBatch.rename_channels(index, rename_dict)[source]

Rename channels with corresponding values from rename_dict.

Parameters:rename_dict (dict) – Dictionary containing (old channel name : new channel name) pairs.
Returns:batch (EcgBatch) – Batch with renamed channels. Changes self.meta inplace.

Signal processing

EcgBatch.convolve_signals(kernel, padding_mode='edge', axis=-1, **kwargs)[source]

Convolve signals with given kernel.

Parameters:
  • kernel (1-D array_like) – Convolution kernel.
  • padding_mode (str or function, optional) – np.pad padding mode.
  • axis (int, optional) – Axis along which signals are sliced. Default value is -1.
  • kwargs (misc) – Any additional named arguments to np.pad.
Returns:

batch (EcgBatch) – Convolved batch. Changes self.signal inplace.
Raises:

ValueError – If kernel is not one-dimensional or has non-numeric dtype.
EcgBatch.band_pass_signals(index, low=None, high=None, axis=-1)[source]

Reject frequencies outside a given range.

Parameters:
  • low (positive float, optional) – High-pass filter cutoff frequency (in Hz).
  • high (positive float, optional) – Low-pass filter cutoff frequency (in Hz).
  • axis (int, optional) – Axis along which signals are sliced. Default value is -1.
Returns:

batch (EcgBatch) – Filtered batch. Changes self.signal inplace.
EcgBatch.drop_short_signals(min_length, axis=-1)[source]

Drop short signals from a batch.

Parameters:
  • min_length (positive int) – Minimal signal length.
  • axis (int, optional) – Axis along which length is calculated. Default value is -1.
Returns:

batch (EcgBatch) – Filtered batch. Creates a new EcgBatch instance.
EcgBatch.flip_signals(index, window_size=None, threshold=0)[source]

Flip 2-D signals whose R-peaks are directed downwards.

Each element of self.signal must be a 2-D ndarray. Signals are flipped along axis 1 (signal axis). For each subarray of window_size length skewness is calculated and compared with threshold to decide whether this subarray should be flipped or not. Then the mode of the result is calculated to make the final decision.

Parameters:
  • window_size (int, optional) – Signal is split into K subarrays of window_size length. If it is not possible, data in the end of the signal is removed. If window_size is not given, the whole array is checked without splitting.
  • threshold (float, optional) – If skewness of a subarray is less than the threshold, it “votes” for flipping the signal. Default value is 0.
Returns:

batch (EcgBatch) – Batch with flipped signals. Changes self.signal inplace.
Raises:

ValueError – If given signal is not two-dimensional.
EcgBatch.slice_signals(index, selection_object)[source]

Perform indexing or slicing of signals in a batch. Allows basic NumPy indexing and slicing along with advanced indexing.

Parameters:selection_object (slice or int or a tuple of slices and ints) – An object that is used to slice signals.
Returns:batch (EcgBatch) – Batch with sliced signals. Changes self.signal inplace.
EcgBatch.split_signals(index, length, step, pad_value=0)[source]

Split 2-D signals along axis 1 (signal axis) with given length and step.

If signal length along axis 1 is less than length, it is padded to the left with pad_value.

Notice, that each resulting signal will be a 3-D ndarray of shape [n_segments, n_channels, length]. If you would like to get a number of 2-D signals of shape [n_channels, length] as a result, you need to apply unstack_signals method then.

Parameters:
  • length (positive int) – Length of each segment along axis 1.
  • step (positive int or dict) – Segmentation step. If step is dict, segmentation step is fetched by signal’s target key.
  • pad_value (float, optional) – Padding value. Defaults to 0.
Returns:

batch (EcgBatch) – Batch of split signals. Changes self.signal inplace.
Raises:
  • ValueError – If:
    • given signal is not two-dimensional,
    • step is not int or dict,
    • length or step for a given signal is negative or non-integer.
  • KeyError – If step dict has no signal’s target key.
EcgBatch.random_split_signals(index, length, n_segments, pad_value=0)[source]

Split 2-D signals along axis 1 (signal axis) n_segments times with random start position and given length.

If signal length along axis 1 is less than length, it is padded to the left with pad_value.

Notice, that each resulting signal will be a 3-D ndarray of shape [n_segments, n_channels, length]. If you would like to get a number of 2-D signals of shape [n_channels, length] as a result, you need to apply unstack_signals method then.

Parameters:
  • length (positive int) – Length of each segment along axis 1.
  • n_segments (positive int or dict) – Number of segments. If n_segments is dict, number of segments is fetched by signal’s target key.
  • pad_value (float, optional) – Padding value. Defaults to 0.
Returns:

batch (EcgBatch) – Batch of split signals. Changes self.signal inplace.
Raises:
  • ValueError – If:
    • given signal is not two-dimensional,
    • n_segments is not int or dict,
    • length or n_segments for a given signal is negative or non-integer.
  • KeyError – If n_segments dict has no signal’s target key.
EcgBatch.unstack_signals()[source]

Create a new batch in which each signal’s element along axis 0 is considered as a separate signal.

This method creates a new batch and updates only components and unique_labels attribute. Signal’s data from non-signal components is duplicated using a deep copy for each of the resulting signals. The information stored in other attributes will be lost.

Returns:batch (same class as self) – Batch with split signals and duplicated other components.

Examples

>>> batch.signal
array([array([[ 0,  1,  2,  3],
              [ 4,  5,  6,  7],
              [ 8,  9, 10, 11]])],
      dtype=object)

>>> batch = batch.unstack_signals()
>>> batch.signal
array([array([0, 1, 2, 3]),
       array([4, 5, 6, 7]),
       array([ 8,  9, 10, 11])],
      dtype=object)
EcgBatch.resample_signals(index, fs)[source]

Resample 2-D signals along axis 1 (signal axis) to given sampling rate.

Parameters:

fs (positive float) – New sampling rate.
Returns:

batch (EcgBatch) – Resampled batch. Changes self.signal and self.meta inplace.
Raises:
  • ValueError – If given signal is not two-dimensional.
  • ValueError – If fs is negative or non-numeric.
EcgBatch.random_resample_signals(index, distr, **kwargs)[source]

Resample 2-D signals along axis 1 (signal axis) to a new sampling rate, sampled from a given distribution.

If new sampling rate is negative, the signal is left unchanged.

Parameters:
  • distr (str or callable) – NumPy distribution name or a callable to sample from.
  • kwargs (misc) – Distribution parameters.
Returns:

batch (EcgBatch) – Resampled batch. Changes self.signal and self.meta inplace.
Raises:
  • ValueError – If given signal is not two-dimensional.
  • ValueError – If distr is not a string or a callable.

Complex ECG processing

Fourier-based transformations
EcgBatch.fft(index, func, *args, src='signal', dst='signal', **kwargs)

Compute a Discrete Fourier Transform for each slice of a signal over the axis 0 (typically the channel axis).

This method simply wraps apply_to_each_channel method by setting the func argument to numpy.fft.fft.

Parameters:
  • src (str, optional) – Batch attribute or component name to get the data from.
  • dst (str, optional) – Batch attribute or component name to put the result in.
  • args (misc) – Any additional positional arguments to numpy.fft.fft.
  • kwargs (misc) – Any additional named arguments to numpy.fft.fft.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.
EcgBatch.ifft(index, func, *args, src='signal', dst='signal', **kwargs)

Compute an inverse Discrete Fourier Transform for each slice of a signal over the axis 0 (typically the channel axis).

This method simply wraps apply_to_each_channel method by setting the func argument to numpy.fft.ifft.

Parameters:
  • src (str, optional) – Batch attribute or component name to get the data from.
  • dst (str, optional) – Batch attribute or component name to put the result in.
  • args (misc) – Any additional positional arguments to numpy.fft.ifft.
  • kwargs (misc) – Any additional named arguments to numpy.fft.ifft.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.
EcgBatch.rfft(index, func, *args, src='signal', dst='signal', **kwargs)

Compute a real-input Discrete Fourier Transform for each slice of a signal over the axis 0 (typically the channel axis).

This method simply wraps apply_to_each_channel method by setting the func argument to numpy.fft.rfft.

Parameters:
  • src (str, optional) – Batch attribute or component name to get the data from.
  • dst (str, optional) – Batch attribute or component name to put the result in.
  • args (misc) – Any additional positional arguments to numpy.fft.rfft.
  • kwargs (misc) – Any additional named arguments to numpy.fft.rfft.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.
EcgBatch.irfft(index, func, *args, src='signal', dst='signal', **kwargs)

Compute a real-input inverse Discrete Fourier Transform for each slice of a signal over the axis 0 (typically the channel axis).

This method simply wraps apply_to_each_channel method by setting the func argument to numpy.fft.irfft.

Parameters:
  • src (str, optional) – Batch attribute or component name to get the data from.
  • dst (str, optional) – Batch attribute or component name to put the result in.
  • args (misc) – Any additional positional arguments to numpy.fft.irfft.
  • kwargs (misc) – Any additional named arguments to numpy.fft.irfft.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.
EcgBatch.spectrogram(index, *args, src='signal', dst='signal', **kwargs)[source]

Compute a spectrogram for each slice of a signal over the axis 0 (typically the channel axis).

This method is a wrapper around scipy.signal.spectrogram, that accepts the same arguments, except the fs which is substituted automatically from signal’s meta. The method returns only the spectrogram itself.

Parameters:
  • src (str, optional) – Batch attribute or component name to get the data from.
  • dst (str, optional) – Batch attribute or component name to put the result in.
  • args (misc) – Any additional positional arguments to scipy.signal.spectrogram.
  • kwargs (misc) – Any additional named arguments to scipy.signal.spectrogram.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.
Wavelet-based transformations
EcgBatch.dwt(index, func, *args, src='signal', dst='signal', **kwargs)

Compute a single level Discrete Wavelet Transform for each slice of a signal over the axis 0 (typically the channel axis).

This method simply wraps apply_to_each_channel method by setting the func argument to pywt.dwt.

Parameters:
  • src (str, optional) – Batch attribute or component name to get the data from.
  • dst (str, optional) – Batch attribute or component name to put the result in.
  • args (misc) – Any additional positional arguments to pywt.dwt.
  • kwargs (misc) – Any additional named arguments to pywt.dwt.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.
EcgBatch.idwt(index, func, *args, src='signal', dst='signal', **kwargs)

Compute a single level inverse Discrete Wavelet Transform for each slice of a signal over the axis 0 (typically the channel axis).

This method simply wraps apply_to_each_channel method by setting the func argument to pywt.idwt.

Parameters:
  • src (str, optional) – Batch attribute or component name to get the data from.
  • dst (str, optional) – Batch attribute or component name to put the result in.
  • args (misc) – Any additional positional arguments to pywt.idwt.
  • kwargs (misc) – Any additional named arguments to pywt.idwt.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.
EcgBatch.wavedec(index, func, *args, src='signal', dst='signal', **kwargs)

Compute a multilevel 1D Discrete Wavelet Transform for each slice of a signal over the axis 0 (typically the channel axis).

This method simply wraps apply_to_each_channel method by setting the func argument to pywt.wavedec.

Parameters:
  • src (str, optional) – Batch attribute or component name to get the data from.
  • dst (str, optional) – Batch attribute or component name to put the result in.
  • args (misc) – Any additional positional arguments to pywt.wavedec.
  • kwargs (misc) – Any additional named arguments to pywt.wavedec.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.
EcgBatch.waverec(index, func, *args, src='signal', dst='signal', **kwargs)

Compute a multilevel 1D Inverse Discrete Wavelet Transform for each slice of a signal over the axis 0 (typically the channel axis).

This method simply wraps apply_to_each_channel method by setting the func argument to pywt.waverec.

Parameters:
  • src (str, optional) – Batch attribute or component name to get the data from.
  • dst (str, optional) – Batch attribute or component name to put the result in.
  • args (misc) – Any additional positional arguments to pywt.waverec.
  • kwargs (misc) – Any additional named arguments to pywt.waverec.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.
EcgBatch.pdwt(index, func, *args, src='signal', dst='signal', **kwargs)

Compute a partial Discrete Wavelet Transform data decomposition for each slice of a signal over the axis 0 (typically the channel axis).

This method simply wraps apply_to_each_channel method by setting the func argument to pywt.downcoef.

Parameters:
  • src (str, optional) – Batch attribute or component name to get the data from.
  • dst (str, optional) – Batch attribute or component name to put the result in.
  • args (misc) – Any additional positional arguments to pywt.downcoef.
  • kwargs (misc) – Any additional named arguments to pywt.downcoef.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.
EcgBatch.cwt(index, func, *args, src='signal', dst='signal', **kwargs)

Compute a Continuous Wavelet Transform for each slice of a signal over the axis 0 (typically the channel axis).

This method simply wraps apply_to_each_channel method by setting the func argument to pywt.cwt.

Parameters:
  • src (str, optional) – Batch attribute or component name to get the data from.
  • dst (str, optional) – Batch attribute or component name to put the result in.
  • args (misc) – Any additional positional arguments to pywt.cwt.
  • kwargs (misc) – Any additional named arguments to pywt.cwt.
Returns:

batch (EcgBatch) – Transformed batch. Changes dst attribute or component.
Other methods
EcgBatch.calc_ecg_parameters(index, src=None)[source]

Calculate ECG report parameters and write them to the meta component.

Calculates PQ, QT, QRS intervals along with their borders and the heart rate value based on the annotation and writes them to the meta component.

Parameters:src (str) – Batch attribute or component name to get the annotation from.
Returns:batch (EcgBatch) – Batch with report parameters stored in the meta component.
Raises:ValueError – If src is None or is not an attribute of a batch.

EcgDataset

class EcgDataset(index=None, batch_class=<class 'cardio.core.ecg_batch.EcgBatch'>, preloaded=None, index_class=<class 'cardio.dataset.dataset.dsindex.FilesIndex'>, *args, **kwargs)[source]

Bases: cardio.dataset.dataset.dataset.Dataset

Dataset that generates batches of EcgBatch class.

Contains indices of ECGs and a specific batch_class to create and process batches - small subsets of data.

Parameters:
  • index (DatasetIndex or None, optional) – Unique identifiers of ECGs in a dataset. If index is not given, it is constructed by instantiating index_class with args and kwargs.
  • batch_class (type, optional) – Class of batches, generated by dataset. Must be inherited from Batch.
  • preloaded (tuple, optional) – Data to put in created batches. Defaults to None.
  • index_class (type, optional) – Class of built index if index is not given. Must be inherited from DatasetIndex.
  • args (misc, optional) – Additional positional argments to index_class.__init__.
  • kwargs (misc, optional) – Additional named argments to index_class.__init__.