Homec4science
Diffusion EB_ECG_Smart_Resampler (master) Browse

EB_ECG_Smart_Resampler/src/build_template.py
7d3c7a24a03bmaster

Last Change
Raw File

build_template.py
View Options

import numpy as np
import pickle
import matplotlib.pyplot as plt
import os
data_beats_dir = "../data/beats/"
FREQ = 128
MIN_TO_AVG = 5
def min_max_normalization(vector):
mi_v = min(vector)
ma_v = max(vector)
norm = (np.array(vector)-mi_v)/(ma_v-mi_v)
return list(norm)
def open_file(file_name):
file_name_full = os.path.join(data_beats_dir,os.path.basename(file_name))
data = {}
with open(file_name_full,"rb") as f:
data = pickle.load(f)
for k in data.keys():
data[k] = data[k][0]
return data
def get_significant_peak(signal):
val = -1000
peak_max = None
is_peak = lambda w,x,y,z: True if (w<x and x>y) or (w<x and x==y and y>z) else False
for i in range(1,len(signal)-2):
if is_peak(signal[i-1],signal[i],signal[i+1],signal[i+2]) and signal[i]> val:
peak_max = i
val = signal[i]
return peak_max
def get_5_min_beats(data):
beats = {}
annotations = list(data.keys())
for annot in annotations[1:]:
if annot > int(MIN_TO_AVG*60*FREQ):
break
beats[annot] = data[annot]
return beats
def get_r_peaks_idx(beats):
r_times_idx = []
time_to_look = int(0.15*FREQ) #150 ms
for annot in beats.keys():
idx = 0
signal_r_peaks = []
for i,t in enumerate(beats[annot]['t']):
if t == annot - time_to_look:
idx = i
if t >= annot - time_to_look and t<= annot + time_to_look:
signal_r_peaks.append(beats[annot]['v'][i])
peak_idx = get_significant_peak(signal_r_peaks)+idx
r_times_idx.append(peak_idx)
return r_times_idx
def allign_beats(data, allignment_idxs, normalize = True):
len_bef = len_aft = 1000
data_alligned = {}
for annot, allignment_idx in zip(data.keys(),allignment_idxs):
time = data[annot]['t']
this_len_bef = len(time[:allignment_idx])
this_len_aft = len(time[allignment_idx+1:])
if this_len_bef < len_bef:
len_bef = this_len_bef
if this_len_aft < len_aft:
len_aft = this_len_aft
for annot, allignment_idx in zip(data.keys(),allignment_idxs):
new_t = data[annot]['t'][allignment_idx-len_bef:allignment_idx+len_aft+1]
new_v = data[annot]['v'][allignment_idx-len_bef:allignment_idx+len_aft+1]
if normalize:
new_v = min_max_normalization(new_v)
data_alligned[annot] = {'t':new_t,'v':new_v}
return data_alligned
def build_template(file_analyzed, normalize = True):
template = []
beats_alligned = []
data = open_file(file_analyzed)
beats = get_5_min_beats(data)
allignment_idxs = get_r_peaks_idx(beats)
alligned_beats = allign_beats(beats, allignment_idxs, normalize=normalize)
for annot in alligned_beats.keys():
beats_alligned.append(alligned_beats[annot]['v'])
template = list(np.average(beats_alligned,axis=0))
template_std = list(np.std(beats_alligned,axis=0))
return template, template_std, alligned_beats
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--file", help="Force to analyze one specific file instead of default one (first found)")
parser.add_argument("--not_norm", help="Force to NOT normalize each beats", action="store_true")
args = parser.parse_args()
files = os.listdir(data_beats_dir)
if args.file is not None:
analyzed = list(filter(lambda string: True if args.file in string else False, files))[0]
else:
analyzed = files[0]
if args.not_norm:
normalize = False
else:
normalize = True
template,std,alligned = build_template(analyzed, normalize = normalize)
plt.plot(template, color = 'C0')
plt.plot(np.array(template)+np.array(std), color = "C1")
plt.plot(np.array(template)-np.array(std), color = "C1")
for beat in alligned.keys():
plt.plot(alligned[beat]['v'],color = 'C2', alpha = 0.01)
plt.show()
c4science · Help