resurfemg.data_connector.data_classes module

Copyright 2022 Netherlands eScience Center and University of Twente Licensed under the Apache License, version 2.0. See LICENSE for details.

This file contains data classes for standardized data storage and method automation.

class EmgDataGroup(y_raw, t_data=None, fs=None, labels=None, units=None)

Bases: TimeSeriesGroup

Child-class of TimeSeriesGroup to store and handle emg data in with the additional methods filter_emg, gating, and wavelet_denoising. Enclosed TimeSeries objects can be indexed by index number or channel label. TimeSeries methods in TimeSeriesGroup._available_methods can be run on all or a subset of the TimeSeries objects through the TimeSeriesGroup.run method. When running ECG elimination methods (gating, wavelet_denoising) and no ECG channel or ecg_raw is provided, EmgDataGroup.ecg_idx is used to detect the QRS peak locations. ecg_idx is auto-detected from the labels on EmgDataGroup initialization, or can be set with the set_ecg_idx method. When no ECG channel is provided, the raw signal per TimeSeries object is used.

set_ecg_idx(ecg_idx)

Set the ECG channel index in the group.

param ecg_idx:

ECG channel index or label

type ecg_idx:

int

returns:

None

rtype:

None

class TimeSeries(y_raw, t_data=None, fs=None, label=None, units=None)

Bases: object

Data class to store, process, and plot single channel time series data. Data is stored in the _y_data dictionary, which can be accessed using the TimeSeries[signal_type]. The default signal types are: ‘raw’, ‘filt’, ‘clean’, ‘env’, ‘env_ci’, and ‘baseline’. Defined properties: - t_data: time axis data - label: channel label - y_units: channel signal units - param: dictionary of channel parameters (fs, n_samp) - peaks: dictionary of PeaksSet objects

baseline(percentile=33, window_s=None, step_s=None, base_method='default', signal_io=(None, 'baseline'), augm_percentile=25, ma_window=None, perc_window=None, **kwargs)

Derive the moving baseline of the provided signal. See postprocessing.baseline submodule. The baseline is stored in the self[‘baseline’] attribute. ———————————————————————– :returns: None :rtype: None

calculate_time_products(peak_set_name, include_aub=True, aub_window_s=None, aub_reference_signal=None, parameter_name=None, signal_io=('baseline',))

Calculate the time product, i.e. area under the curve for a PeaksSet. The results are stored as self.peaks[peak_set_name].peak_df[parameter_name]. If no parameter_name is provided, parameter_name = ‘time_product’ ———————————————————————– :param peak_set_name: PeaksSet name in self.peaks dict :type peak_set_name: str :param include_aub: Include the area under the baseline in the time product :type include_aub: bool :param signal_io: tuple of strings, the first element is the input baseline type. :type signal_io: tuple :param aub_window_s: window length in samples in which the local extreme is sought. :param aub_window_s: int :param aub_reference_signal: Optional reference signal to find the local extreme in, else the signal underlying the PeaksSet is taken. :type aub_reference_signal: ~numpy.ndarray :param parameter_name: parameter name in Dataframe self.peaks[peak_set_name].peak_df :type parameter_name: str

Returns:

None

Return type:

None

detect_emg_breaths(threshold=0, prominence_factor=0.5, min_peak_width_s=None, peak_set_name='breaths', start_idx=0, end_idx=None, overwrite=False, signal_io=(('env', 'baseline'),))

Find breath peaks in provided EMG envelope signal. See postprocessing.event_detection submodule. The peaks are stored in the self.peaks dict under the key peak_set_name. ———————————————————————– :returns: None :rtype: None

envelope(env_window=None, env_type=None, signal_io=('clean', 'env'), ci_alpha=None, **kwargs)

Derive the moving envelope of the provided signal. See preprocessing.envelope submodule. The envelope is by default stored in the self[‘env’]. If ci_alpha is not None, the confidence interval is stored in self[‘env_ci’].

Returns:

None

Return type:

None

filter_emg(signal_io=('raw', 'filt'), hp_cf=20.0, lp_cf=500.0, order=3, **kwargs)

Filter raw EMG signal to remove baseline wander and high frequency components. See preprocessing.emg_bandpass_butter submodule. The filtered signal is stored in self[signal_io[1]].

Parameters:

signal_io (tuple) – tuple of strings, the first element is the input the second element is the output signal type.

Returns:

None

Return type:

None

gating(signal_io=('filt', 'clean'), ecg_peakset_name='ecg', gate_width_samples=None, fill_method=3, **kwargs)

Eliminate ECG artifacts from the provided signal based on the peak_idx of the provided PeakSet with ecg_peakset_name. See preprocessing.ecg_removal and pipelines.ecg_removal_gating submodules. The cleaned signal is by stored in self[‘clean’].

Parameters:

signal_io (tuple) – tuple of strings, the first element is the input the second element is the output signal type.

Returns:

None

Return type:

None

get_ecg_peaks(ecg_raw=None, bp_filter=True, overwrite=False, name='ecg')

Detect ECG peaks in the provided signal. See preprocessing.ecg_removal submodule. ECG peaks are stored in the self.peaks dict under the key name. When no ECG channel is provided, the raw signal of the current TimeSeries object is used. When running get_ecg_peaks on a EmgDataGroup and no ECG channel or ecg_raw is provided, EmgDataGroup.ecg_idx is used to detect the QRS peak locations. ecg_idx is auto-detected from the labels on EmgDataGroup initialization, or can be set with the set_ecg_idx method. ———————————————————————– :param ecg_raw: ECG signal, if None, the raw signal is used :type ecg_raw: ~numpy.ndarray :bp_filter: Apply band-pass filter to the ECG signal :type bp_filter: bool :overwrite: Overwrite existing peaks :type overwrite: bool :param name: Name of the peak set in the self.peaks dict :type name: str

Returns:

None

Return type:

None

Find the peaks in the PeaksSet with the peak_set_name closest in time to the provided peak timings in t_reference_peaks. The results are stored in a new PeaksSet object in the self.peaks dict under the key linked_peak_set_name. If no linked_peak_set_name is provided, the key is set to peak_set_name + ‘_linked’. ———————————————————————– :param peak_set_name: PeaksSet name in self.peaks dict :type peak_set_name: str :param t_reference_peaks: Refernce peak timings in t_reference_peaks :type t_reference_peaks: ~numpy.ndarray :param linked_peak_set_name: Name of the new PeaksSet :type linked_peak_set_name: str

Returns:

None

Return type:

None

plot_aub(peak_set_name, axes, signal_io=None, valid_only=False, colors=None, **kwargs)

Plot the area under the baseline (AUB) for the peak set in the provided axes in the provided colours using the provided markers. ———————————————————————– :param peak_set_name: PeaksSet name in self.peaks dict :type peak_set_name: str :param axes: matplotlib Axes object. If none provided, a new figure is created. :type axes: matplotlib.Axes :param signal_type: the signal (‘env’, ‘clean’, ‘filt’, ‘raw’) to plot :type signal_type: str :param valid_only: when True, only valid peaks are plotted. :type valid_only: bool :param colors: 1 color or list of up to 3 colors for the peak :type colors: str or list

Returns:

None

Return type:

None

plot_curve_fits(peak_set_name, axes, valid_only=False, colors=None)

Plot the curve-fits for the peak set in the provided axes in the provided colours using the provided markers. ———————————————————————– :param peak_set_name: PeaksSet name in self.peaks dict :type peak_set_name: str :param axes: matplotlib Axes object. If none provided, a new figure is created. :type axes: matplotlib.Axes :param valid_only: when True, only valid peaks are plotted. :type valid_only: bool :param colors: 1 color or list of colors for the fitted curve peak :type colors: str or list

Returns:

None

Return type:

None

plot_full(axes=None, signal_io=(None,), colors=None, baseline_bool=True, plot_ci=False, **kwargs)

Plot the indicated signals in the provided axes. By default the most advanced signal type (envelope > clean > filt > raw) is plotted in the provided colours. ———————————————————————– :param axes: matplotlib Axis object. If none provided, a new figure is created. :type axes: matplotlib.Axis :param signal_io: tuple of strings, the first element is the input signal type. :type signal_io: tuple :param colors: list of colors to plot the 1) signal, 2) the baseline :type colors: list :param baseline_bool: plot the baseline :type baseline_bool: bool

Returns:

None

Return type:

None

plot_markers(peak_set_name, axes, valid_only=False, colors=None, markers=None)

Plot the markers for the peak set in the provided axes in the provided colours using the provided markers. ———————————————————————– :param peak_set_name: PeaksSet name in self.peaks dict :type peak_set_name: str :param axes: matplotlib Axes object. If none provided, a new figure is created. :type axes: matplotlib.Axes :param valid_only: when True, only valid peaks are plotted. :type valid_only: bool :param colors: 1 color of list of up to 3 colors for the markers, peak, start, and end markers. If 2 colors are provided, start and end have the same colors :type colors: str or list :param markers: 1 markers or list of up to 3 markers for peak, start, and end markers. If 2 markers are provided, start and end have the same marker :type markers: str or list

Returns:

None

Return type:

None

plot_peaks(peak_set_name, axes=None, signal_io=(None,), margin_s=None, valid_only=False, colors=None, baseline_bool=True, plot_ci=False, **kwargs)

Plot the indicated peaks in the provided axes. By default the most advanced signal type (envelope > clean > filt > raw) is plotted in the provided colours. ———————————————————————– :param peak_set_name: The name of the peak_set to be plotted. :type peak_set_name: str :param axes: matplotlib Axes object. If none provided, a new figure is created. :type axes: matplotlib.Axes :param signal_io: tuple of strings, the first element is the input signal type. :type signal_io: tuple :param margin_s: margins in samples plotted before the peak onset and after the peak offset :param valid_only: when True, only valid peaks are plotted. :type valid_only: bool :param colors: list of colors to plot the 1) signal, 2) the baseline :type colors: list :param baseline_bool: plot the baseline :type baseline_bool: bool

Returns:

None

Return type:

None

set_peaks(peak_idxs, signal, peak_set_name, overwrite=False)

Store a new PeaksSet object in the self.peaks dict under the key peak_set_name. ———————————————————————– :returns: None :rtype: None

signal_type_data(signal_type=None)

Automatically select the most advanced data type eligible for a subprocess (‘env’ {=envelope} > ‘clean’ > ‘filt’ > ‘raw’) ———————————————————————– :param signal_type: one of ‘env’, ‘clean’, ‘filt’, or ‘raw’ :type signal_type: str

Returns y_data:

data of the selected signal type

Rtype y_data:

~numpy.ndarray

test_emg_quality(peak_set_name, cutoff=None, skip_tests=None, parameter_names=None, verbose=True)

See helper_functions.data_classes_quality_assessment submodule. The results are stored in the self.peaks[peak_set_name].quality_outcomes_df and self.peaks[peak_set_name].quality_values_df DataFrames.

test_linked_peak_sets(peak_set_name, linked_timeseries, linked_peak_set_name, parameter_names=None, cutoff=None, skip_tests=None, verbose=True)

See helper_functions.data_classes_quality_assessment submodule. The results are stored in the self.peaks[peak_set_name].quality_outcomes_df and self.peaks[peak_set_name].quality_values_df DataFrames.

test_pocc_quality(peak_set_name, cutoff=None, skip_tests=None, parameter_names=None, verbose=True)

See helper_functions.data_classes_quality_assessment submodule. The results are stored in the self.peaks[peak_set_name].quality_outcomes_df and self.peaks[peak_set_name].quality_values_df DataFrames.

wavelet_denoising(signal_io=('filt', 'clean'), ecg_peakset_name='ecg', n=None, fixed_threshold=None, **kwargs)

Eliminate ECG artifacts from the provided signal. See preprocessing.wavelet_denoising submodules. The cleaned signal is by default stored in self[‘clean’].

Parameters:

signal_io (tuple) – tuple of strings, the first element is the input the second element is the output signal type.

Returns:

None

Return type:

None

class TimeSeriesGroup(y_raw, t_data=None, fs=None, labels=None, units=None)

Bases: object

Data class to store, process, and plot time series data. TimeSeriesGroup is a collection of TimeSeries objects. Enclosed TimeSeries objects can be indexed by index number or channel label. TimeSeries methods in TimeSeriesGroup._available_methods can be run on all or a subset of the TimeSeries objects through the TimeSeriesGroup.run method.

run(method, channel_idxs=None, **kwargs)

Run a TimeSeries function on the provided channels in the group. The function is run with the provided keyword arguments. ———————————————————————– :param method: method to run on the TimeSeries objects :type method: str :param channel_idxs: channel indices to run the method on :type channel_idxs: ~numpy.ndarray or int :param kwargs: keyword arguments for the method

to_numpy(channel_idxs=None, signal_io=(None,))

Convert the TimeSeriesGroup to a numpy array. The output is a 2D array with the shape (n_channels, n_samples). The signal type is determined by the signal_io parameter. If signal_io is (None,), the most advanced signal type (envelope > clean > filt > raw) is used.

Parameters:

  • channel_idxs (ndarray or int) – channel indices to convert to numpy array

  • signal_io (tuple) – tuple of strings, the first element is the input signal type.

Returns:

numpy array of the TimeSeriesGroup data

Return type:

ndarray

class VentilatorDataGroup(y_raw, t_data=None, fs=None, labels=None, units=None)

Bases: TimeSeriesGroup

Child-class of TimeSeriesGroup to store and handle ventilator data in. Default channels are ‘Paw’/ ‘Pvent’, ‘F’, and ‘Vvent’, which are auto- detected from the labels. The PEEP-level (VentilatorDataGroup.peep) is auto-detected from the pressure channel when a pressure channel is set.

find_occluded_breaths(pressure_idx=None, peep=None, overwrite=False, **kwargs)

Find end-expiratory occlusion manoeuvres in ventilator pressure timeseries data. See postprocessing.event_detection submodule. ———————————————————————– :param pressure_idx: Channel index of the ventilator pressure data :type pressure_idx: int For other arguments, see postprocessing.event_detection submodule.

Returns:

None

Return type:

None

find_peep(pressure_idx, volume_idx)

Calculate PEEP as the median value of p_vent at end-expiration.

param pressure_idx:

Channel index of the ventilator pressure data

type pressure_idx:

int

param volume_idx:

Channel index of the ventilator volume data

type volume_idx:

int

returns:

None

rtype:

None

find_tidal_volume_peaks(volume_idx=None, pressure_idx=None, overwrite=False, **kwargs)

Find tidal-volume peaks in ventilator volume signal. Peaks are stored in PeaksSet named ‘ventilator_breaths’ in ventilator pressure and volume TimeSeries. ———————————————————————– :param volume_idx: Channel index of the ventilator volume data :type volume_idx: int :param pressure_idx: Channel index of the ventilator pressure data :type pressure_idx: int For other arguments, see postprocessing.event_detection submodule.

Returns:

None

Return type:

None