Utils¶

utils.autocorrelate(data, unbias=2, normalize=2)¶

Compute the autocorrelation coefficients for time series data. Here we use scipy.signal.correlate, but the results are the same as in Yang, et al., 2012 for unbias=1:

“The autocorrelation coefficient refers to the correlation of a time series with its own past or future values. iGAIT uses unbiased autocorrelation coefficients of acceleration data to scale the regularity and symmetry of gait. The autocorrelation coefficients are divided by \(fc(0)\), so that the autocorrelation coefficient is equal to \(1\) when \(t=0\):

\[NFC(t) = \frac{fc(t)}{fc(0)}\]

Here \(NFC(t)\) is the normalised autocorrelation coefficient, and \(fc(t)\) are autocorrelation coefficients.”

Parameters:

data (numpy array) – time series data
unbias (integer or None) – autocorrelation, divide by range (1) or by weighted range (2)
normalize (integer or None) – divide by 1st coefficient (1) or by maximum abs. value (2)

Return coefficients:

autocorrelation coefficients [normalized, unbiased]

Rtype coefficients:

numpy array

Return N:

number of coefficients

Rtype N:

integer

Examples:

>>> import numpy as np
>>> from mhealthx.signals import autocorrelate
>>> data = np.random.random(100)
>>> unbias = 2
>>> normalize = 2
>>> plot_test = True
>>> coefficients, N = autocorrelate(data, unbias, normalize, plot_test)

utils.autocorrelation(signal)¶

The correlation of a signal with a delayed copy of itself.

Parameters:: signal (array) – A 1-dimensional array or list (the signal).
Returns:: The autocorrelated signal.
Return type:: numpy.ndarray

utils.butter_lowpass_filter(data, sample_rate, cutoff=10, order=4, plot=False)¶

Low-pass filter data by the [order]th order zero lag Butterworth filter whose cut frequency is set to [cutoff] Hz.

Parameters:

data (numpy array of floats) – time-series data,
cutoff (float) – filter cutoff
order (integer) – order

Param:

sample_rate: data sample rate

Return y:

low-pass-filtered data

Rtype y:

numpy array of floats

Examples:

>>> from mhealthx.signals import butter_lowpass_filter
>>> data = np.random.random(100)
>>> sample_rate = 10
>>> cutoff = 5
>>> order = 4
>>> y = butter_lowpass_filter(data, sample_rate, cutoff, order)

utils.centroid_sort(centroids)¶

Sort centroids. This is required so that the same cluster centroid is always the 0th one. It should also be the most negative. Order defined by the Euclidean distance between the centroid and an arbitrary “small” point [-100, -100] (in each dimension) to account for possible negatives. Cluster 0 is the closest to that point, etc.

Set up

>>> from numpy.testing import assert_array_equal

Single centroids just return themselves.

>>> centroid_sort(array([[1.1, 2.2]]))

array([[ 1.1, 2.2]])

>>> centroid_sort(array([[1.1, 2.2, 3.3]]))

array([[ 1.1, 2.2, 3.3]])

Positive 2d centroids are ordered.

>>> centroids = array([
...     [5.34443858, 0.63266844],  # 3
...     [2.69156877, 0.76448578],  # 1
...     [4.74784197, 1.0815235 ],  # 2
...     [1.02330015, 0.16788118],  # 0
... ])
>>> expected_sorted_centroids = array([
...     [1.02330015, 0.16788118],  # 0
...     [2.69156877, 0.76448578],  # 1
...     [4.74784197, 1.0815235 ],  # 2
...     [5.34443858, 0.63266844],  # 3
... ])
>>> result = centroid_sort(centroids)
>>> assert_array_equal(result, expected_sorted_centroids)

3d centroids spanning the origin are ordered.

>>> centroids = array([
...     [ 3,   3,  4  ],  # 3
...     [ 1.5, 2,  3  ],  # 2
...     [-1,  -1, -1 ],   # 0
...     [ 0,   1,  0.5],  # 1
... ])
>>> expected_sorted_centroids = array([
...     [-1,  -1, -1 ],   # 0
...     [ 0,   1,  0.5],  # 1
...     [ 1.5, 2,  3  ],  # 2
...     [ 3,   3,  4  ],  # 3
... ])
>>> result = centroid_sort(centroids)
>>> assert_array_equal(result, expected_sorted_centroids)

Parameters:: centroids (numpy array) – array centroids
Return centroids:: array centroids
Rtype centroids:: numpy array

utils.compute_interpeak(data, sample_rate)¶

Compute number of samples between signal peaks using the real part of FFT.

Parameters:

data (array) – 1-dimensional time series data.
sample_rate (float) – Sample rate of accelerometer reading (Hz)

Return interpeak:

Number of samples between peaks

Rtype interpeak:

int

Examples:

>>> import numpy as np
>>> from mhealthx.signals import compute_interpeak
>>> data = np.random.random(10000)
>>> sample_rate = 100
>>> interpeak = compute_interpeak(data, sample_rate)

utils.crossings_nonzero_pos2neg(data)¶

Find indices of zero crossings from positive to negative values.

Parameters:: data (numpy array of floats) – numpy array of floats
Return crossings:: crossing indices to data
Rtype crossings:: numpy array of integers
Examples:

>>> import numpy as np
>>> from mhealthx.signals import crossings_nonzero_pos2neg
>>> data = np.random.random(100)
>>> crossings = crossings_nonzero_pos2neg(data)

utils.get_signal_peaks_and_prominences(data)¶

Get the signal peaks and peak prominences.

Parameters:: array (data) – One-dimensional array.
Return peaks array:: The peaks of our signal.
Return prominences array:: The prominences of the peaks.

utils.load_cloudupdrs_data(filename, convert_times=1000000000.0)¶

This method loads data in the cloudupdrs format

Usually the data will be saved in a csv file and it should look like this:

timestamp_0, x_0, y_0, z_0
timestamp_1, x_1, y_1, z_1
timestamp_2, x_2, y_2, z_2
.
.
.
timestamp_n, x_n, y_n, z_n

where x, y, z are the components of the acceleration

Parameters:

filename (string) – The path to load data from
convert_times (float) – Convert times. The default is from from nanoseconds to seconds.

utils.load_data(filename, format_file='cloudupdrs', button_left_rect=None, button_right_rect=None)¶

This is a general load data method where the format of data to load can be passed as a parameter,

Parameters:

filename (str) – The path to load data from
format_file (str) – format of the file. Default is CloudUPDRS. Set to mpower for mpower data.
button_left_rect (str) – mpower param
button_right_rect (str) – mpower param

utils.load_finger_tapping_cloudupdrs_data(filename, convert_times=1000.0)¶

This method loads data in the cloudupdrs format for the finger tapping processor

Usually the data will be saved in a csv file and it should look like this:

timestamp_0, . , action_type_0, x_0, y_0, . , . , x_target_0, y_target_0
timestamp_1, . , action_type_1, x_1, y_1, . , . , x_target_1, y_target_1
timestamp_2, . , action_type_2, x_2, y_2, . , . , x_target_2, y_target_2
.
.
.
timestamp_n, . , action_type_n, x_n, y_n, . , . , x_target_n, y_target_n

where data_frame.x, data_frame.y: components of tapping position. data_frame.x_target, data_frame.y_target their target.

Parameters:

filename (string) – The path to load data from
convert_times (float) – Convert times. The default is from from milliseconds to seconds.

utils.load_finger_tapping_mpower_data(filename, button_left_rect, button_right_rect, convert_times=1000.0)¶: This method loads data in the mpower format

utils.load_finger_tapping_opdc_data(filename, convert_times=1000000000.0)¶

This method loads data in the cloudupdrs format for the finger tapping processor

Usually the data will be saved in a csv file and it should look like this:

timestamp_0, x_0, y_0,
timestamp_1, x_1, y_1,
timestamp_2, x_2, y_2,
.
.
.
timestamp_n, x_n, y_n

where data_frame.x, data_frame.y: components of tapping position.

Parameters:

filename (string) – The path to load data from
convert_times (float) – Convert times. The default is from from milliseconds to seconds.

utils.load_mpower_data(filename, convert_times=1000000000.0)¶

This method loads data in the mpower format

The format is like:

[
   {
      "timestamp":19298.67999479167,
      "x": ... ,
      "y": ...,
      "z": ...,
   },
   {...},
   {...}
]

Parameters:

filename (string) – The path to load data from
convert_times (float) – Convert times. The default is from from nanoseconds to seconds.

utils.load_opdc_data(filename, convert_times=1000000000.0)¶

This method loads data in the OPDC format

Usually the data will be saved in a csv file and it should look like this:

timestamp_0, x_0, y_0, z_0
timestamp_1, x_1, y_1, z_1
timestamp_2, x_2, y_2, z_2
.
.
.
timestamp_n, x_n, y_n, z_n

where x, y, z are the components of the acceleration

Parameters:

filename (string) – The path to load data from
convert_times (float) – Convert times. The default is from from nanoseconds to seconds.

utils.load_reaction_opdc_data(filename, convert_times=1000000000.0)¶

This method loads data in the OPDC format for the reaction time processor

The data is expected in a csv file with the following format:

timestamp_0, x_0, y_0, buttonVisible, buttonPressed
timestamp_1, x_1, y_1, buttonVisible, buttonPressed
timestamp_2, x_2, y_2, buttonVisible, buttonPressed
.
.
.
timestamp_n, x_n, y_n, buttonVisible, buttonPressed

where data_frame.x, data_frame.y: components of tapping position.

Parameters:

filename (string) – The path to load data from
convert_times (float) – Convert times. The default is from from milliseconds to seconds.
buttonVisible (Boolean) – True if button is visible.
buttonPressed (Boolean) – True if button is being pressed.

utils.load_segmented_data(filename)¶

Helper function to load segmented gait time series data.

Parameters:: filename (str) – The full path of the file that contais our data. This should be a comma separated value (csv file).
Returns:: The gait time series segmented data, with a x, y, z, mag_acc_sum and segmented columns.
Return type:: pandas.DataFrame

utils.non_zero_index(arr)¶

Raises: ValueError: If no-non-zero rows can be found.

Empty array raises.

>>> arr = array([])
>>> non_zero_index(arr)

Array with zero values raises.

>>> arr = array([
...     [0, 0],
...     [0, 0],
...     [0, 0, 0],
... ])
>>> non_zero_index(arr)

Array with a non-zero value will have that index returned.

>>> arr = array([
...     [0, 0],
...     [0, 0, 0],
...     [1, 0, 0],  # Still has zeros
...     [1, 1, 0],
...     [0, 1, 1],
...     [-1, 0, 0],
...     [-1, 2, 3],  # First non-zero array
...     [1, 2, 3],
... ])
>>> non_zero_index(arr)

6

Parameters:: arr (numpy array) – array
Return index:: Index of first non-zero entry in an array.
Rtype index:: int

utils.non_zero_row(arr)¶

Empty row returns False.

>>> arr = array([])
>>> non_zero_row(arr)

False

Row with a zero returns False.

>>> arr = array([1, 4, 3, 0, 5, -1, -2])
>>> non_zero_row(arr)

False

Row with no zeros returns True.

>>> arr = array([-1, -0.1, 0.001, 2])
>>> non_zero_row(arr)

True

Parameters:: arr (numpy array) – array
Return empty:: If row is completely free of zeros
Rtype empty:: bool

utils.numerical_integration(signal, sampling_frequency)¶

Numerically integrate a signal with it’s sampling frequency.

Parameters:

signal (array) – A 1-dimensional array or list (the signal).
sampling_frequency (float) – The sampling frequency for the signal.

Returns:

The integrated signal.

Return type:

numpy.ndarray

utils.peakdet(signal, delta, x=None)¶

Find the local maxima and minima (peaks) in a 1-dimensional signal. Converted from MATLAB script <http://billauer.co.il/peakdet.html>

Parameters:

signal (array) – A 1-dimensional array or list (the signal).
delta (float) – The peak threashold. A point is considered a maximum peak if it has the maximal value, and was preceded (to the left) by a value lower by delta.
x (array) – Indices in local maxima and minima are replaced with the corresponding values in x (None default).

Return maxtab:

The highest peaks.

Rtype maxtab:

numpy.ndarray

Return mintab:

The lowest peaks.

Rtype mintab:

numpy.ndarray

utils.plot_segmentation(data, peaks, segment_indexes, figsize=(10, 5))¶

Will plot the data and segmentation based on the peaks and segment indexes.

Parameters:

data (1d-array) – The orginal axis of the data that was segmented into sections.
peaks (1d-array) – Peaks of the data.
segment_indexes (1d-array) – These are the different classes, corresponding to each peak.

Will not return anything, instead it will plot the data and peaks with different colors for each class.

utils.separate_walks_turns(data, window=[1, 1, 1])¶

Will separate peaks into the clusters by following the trend in the clusters array. This is usedful because scipy’s k-mean clustering will give us a continous clusters array.

Parameters:

array (peaks) – A continous array representing different classes.
array – The peaks that we want to separate into the classes from the custers.

Return walks arrays:

An array of arrays that will have all the peaks corresponding to every individual walk.

Return turns arraays:

Array of array which has all the indices of the peaks that correspond to turning.

utils.smoothing_window(data, window=[1, 1, 1])¶

This is a smoothing functionality so we can fix misclassifications. It will run a sliding window of form [border, smoothing, border] on the signal and if the border elements are the same it will change the smooth elements to match the border. An example would be for a window of [2, 1, 2] we have the following elements [1, 1, 0, 1, 1], this will transform it into [1, 1, 1, 1, 1]. So if the border elements match it will transform the middle (smoothing) into the same as the border.

Parameters:

array (window) – One-dimensional array.
array – Used to define the [border, smoothing, border] regions.

Return data array:

The smoothed version of the original data.

utils.window_features(idx, window_size=100, overlap=10)¶

Generate indexes for a sliding window with overlap

Parameters:

idx (array) – The indexes that need to be windowed.
window_size (int) – The size of the window.
overlap (int) – How much should each window overlap.

Return array view:

The indexes for the windows with overlap.

Utils¶

PDkit

Navigation

Related Topics