Imports data from the Totem Bobbi into a Pandas dataframe

basic_peak_detection.py

import heart_beat as hb 

fs = 500 # sampling frequency 500hz

#import data sample
dataset = hb.get_data("data/8-sdataHR_sample.df")
hb.process_basic_peak(dataset, 0.75, fs)

#We have imported our Python module as an object called 'hb'
#This object contains the dictionary 'measures' with all values in it
#Now we can also retrieve the BPM value (and later other values) like this:
bpm = hb.measures['bpm']
print(bpm)

heart_beat.py

import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
import math

measures = {}

def get_data(filename):
    dataset = pd.read_pickle(filename)
    return dataset

def rolmean(dataset, hrw, fs):
    mov_avg = dataset.hart.rolling(center=False, window=int(hrw*fs)).mean() #Calculate moving average
    avg_hr = (np.mean(dataset.hart)) 
    mov_avg = [avg_hr if math.isnan(x) else x for x in mov_avg]
    mov_avg = [x*1.1 for x in mov_avg]
    dataset['hart_rollingmean'] = mov_avg

def detect_peaks(dataset):
    window = []
    peaklist = []
    listpos = 0 
    for datapoint in dataset.hart:
        rollingmean = dataset.hart_rollingmean[listpos]
        if (datapoint < rollingmean) and (len(window) < 1):
            listpos += 1
        elif (datapoint > rollingmean):
            window.append(datapoint)
            listpos += 1
        else:
            if(window):
                maximum = max(window)
                beatposition = listpos - len(window) + (window.index(maximum))
                peaklist.append(beatposition)
                window = []
                listpos += 1
    measures['peaklist'] = peaklist
    measures['ybeat'] = [dataset.hart[x] for x in peaklist]

    
def calc_bpm():
    RR_list = measures['RR_list']
    measures['bpm'] = 60000 /( np.mean(RR_list))

def plotter(dataset, title):
    peaklist = measures['peaklist']
    ybeat = measures['ybeat']
    plt.title(title)
    plt.plot(dataset.hart, alpha=0.5, color='blue', label="raw signal")
    plt.plot(dataset.hart_rollingmean, color ='green', label="moving average")
    plt.scatter(peaklist, ybeat, color='red', label="average: %.1f BPM" %measures['bpm'])
    plt.legend(loc=4, framealpha=0.6)
    plt.show() 
    
    
def calc_RR(dataset, fs):
    peaklist = measures['peaklist']
    RR_list = []
    cnt = 0
    while (cnt < (len(peaklist)-1)):
        RR_interval = (peaklist[cnt+1] - peaklist[cnt])
        ms_dist = ((RR_interval / fs) * 1000.0) 
        RR_list.append(ms_dist)
        cnt += 1
    RR_diff = []
    RR_sqdiff = []
    cnt = 0 
    while (cnt < (len(RR_list)-1)): 
        RR_diff.append(abs(RR_list[cnt] - RR_list[cnt+1])) 
        RR_sqdiff.append(math.pow(RR_list[cnt] - RR_list[cnt+1], 2))
        cnt += 1
    measures['RR_list'] = RR_list
    measures['RR_diff'] = RR_diff
    measures['RR_sqdiff'] = RR_sqdiff
        

def calc_ts_measures():
    RR_list = measures['RR_list']
    RR_diff = measures['RR_diff']
    RR_sqdiff = measures['RR_sqdiff']
    measures['bpm'] = 60000 / np.mean(RR_list)
    measures['ibi'] = np.mean(RR_list)
    measures['sdnn'] = np.std(RR_list)
    measures['sdsd'] = np.std(RR_diff)
    measures['rmssd'] = np.sqrt(np.mean(RR_sqdiff))
    NN20 = [x for x in RR_diff if (x>20)]
    NN50 = [x for x in RR_diff if (x>50)]
    measures['nn20'] = NN20
    measures['nn50'] = NN50
    measures['pnn20'] = float(len(NN20)) / float(len(RR_diff))
    measures['pnn50'] = float(len(NN50)) / float(len(RR_diff))

def process_advanced(dataset, hrw, fs):
    rolmean(dataset, hrw, fs)
    detect_peaks(dataset)
    calc_RR(dataset, fs)
    calc_ts_measures()
    
def process_basic_peak(dataset, hrw, fs): 
    rolmean(dataset, hrw, fs)
    detect_peaks(dataset)
    calc_RR(dataset, fs)
    calc_bpm()
    plotter(dataset, "My Heartbeat Plot")

heart_measures.py

import heart_beat as hb 

fs = 500 # sampling frequency 120

dataset = hb.get_data("data/8-sdataHR_sample.df")
hb.process_advanced(dataset, 0.75, fs)

#The module dict now contains all the variables computed over our signal:
print ("BPM: "+str(hb.measures['bpm']))
print ("ibi: "+str(hb.measures['ibi']))
print ("sdnn: "+str(hb.measures['sdnn']))
print ("RR_list: "+str(hb.measures['RR_list']))
print ("RR_diff: "+str(hb.measures['RR_diff']))
print ("RR_sqdiff: "+str(hb.measures['RR_sqdiff']))
print ("RR_sqdiff: "+str(hb.measures['RR_sqdiff']))
print ("sdnn: "+str(hb.measures['sdnn']))
print ("sdsd: "+str(hb.measures['sdsd']))
print ("rmssd: "+str(hb.measures['rmssd']))
print ("nn20: "+str(hb.measures['nn20']))
print ("nn50: "+str(hb.measures['nn50']))
print ("pnn20: "+str(hb.measures['pnn20']))
print ("pnn50: "+str(hb.measures['pnn50']))

import_bobbi.py

import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
import os 
from pathlib import Path

###################### Set csv filename ######################

csv_file = "data/8-sdataHR.csv"

##############################################################

# Retrieve some file info
f = Path(csv_file)                                      # path object
f_no_ext = os.path.join(str(f.parent), f.stem)          # remove extension

# Save Pickled DF - faster to save and load later
is_cached = Path(f_no_ext +".df")                       # check for cache file
if is_cached.is_file():                                 # cache exists?
    dataset = pd.read_pickle(f_no_ext +".df")           # yes - load cache
    
else:
    dataset = pd.read_csv( f_no_ext +".csv",            # no  - load csv
                index_col=False,
                sep=';',
                dtype={'ms': np.int32,
                       'heartrate': np.int16, 
                       'BPM': np.int16 } ) 

    dataset['hart'] = dataset['heartrate']              # set hart
    # dataset['hart'] = -dataset['hart']+1000           # if needed, flip
    dataset.to_pickle(f_no_ext +".df")                  # save to cache
    
# Save small subset of data as a sample
dataset = dataset[(dataset['ms'] >= 104000) 
                    & (dataset['ms'] <= 115000)]

dataset = dataset.reset_index(drop=True)                # Start index at 0
dataset.to_csv(f_no_ext + "_sample.csv"
               ,sep=';',index=False)                    # Save CSV sample
dataset.to_pickle(f_no_ext +"_sample.df")               # Save DF sample

# Lets plot our sample, see if its ok
plt.title("Heart Rate Signal")                          #The title of our plot
plt.plot(dataset.ms,dataset.hart)                       #Draw the plot object
plt.show()                                              #Display the plot 

output.sh

BPM: 84.4409695074
ibi: 710.555555556
sdnn: 19.1162783712
RR_list: [725.0, 725.0, 725.0, 683.3333333333334, 741.6666666666667, 716.6666666666666, 700.0, 666.6666666666666, 708.3333333333334, 700.0, 708.3333333333334, 700.0, 700.0, 733.3333333333333, 725.0]
RR_diff: [0.0, 0.0, 41.66666666666663, 58.33333333333337, 25.000000000000114, 16.66666666666663, 33.33333333333337, 41.66666666666674, 8.333333333333371, 8.333333333333371, 8.333333333333371, 0.0, 33.33333333333326, 8.333333333333258]
RR_sqdiff: [0.0, 0.0, 1736.111111111108, 3402.7777777777824, 625.0000000000057, 277.7777777777765, 1111.1111111111136, 1736.1111111111175, 69.44444444444508, 69.44444444444508, 69.44444444444508, 0.0, 1111.111111111106, 69.44444444444318]
RR_sqdiff: [0.0, 0.0, 1736.111111111108, 3402.7777777777824, 625.0000000000057, 277.7777777777765, 1111.1111111111136, 1736.1111111111175, 69.44444444444508, 69.44444444444508, 69.44444444444508, 0.0, 1111.111111111106, 69.44444444444318]
sdnn: 19.1162783712
sdsd: 18.0151737505
rmssd: 27.0947778018
nn20: [41.66666666666663, 58.33333333333337, 25.000000000000114, 33.33333333333337, 41.66666666666674, 33.33333333333326]
nn50: [58.33333333333337]
pnn20: 0.42857142857142855
pnn50: 0.07142857142857142