-
Notifications
You must be signed in to change notification settings - Fork 1
/
helpers.py
176 lines (158 loc) · 6.93 KB
/
helpers.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
import os
from scipy.signal import butter, sosfilt, sosfreqz
import math
import pywt
import pywt.data
import numpy as np
import config
from typing import NamedTuple
import re
import cv2
from audio import load_audio
from video import load_video, resample_video, convert_video_to_audio_moviepy
from PIL import Image
import wfdb
import matplotlib.pyplot as plt
dataframe_cols = ['filename', 'og_filename', 'label', 'record_duration', 'num_channels', 'samples_ecg', 'samples_pcg', 'qrs_count', 'seg_num', 'avg_hr']
def check_datatype_and_filetype(datatype, filetype):
if filetype not in config.file_types_ecg.union(config.file_types_pcg):
raise ValueError(f"Error: 'filetype' is not one of {config.file_types_ecg.union(config.file_types_pcg)}")
if datatype not in config.data_types_ecg.union(config.data_types_pcg):
raise ValueError(f"Error: 'datatype' is not one of {config.data_types_ecg.union(config.data_types_pcg)}")
if filetype == 'npz' and datatype not in {'signal', 'spec', 'video_noaudio'}:
raise ValueError(f"Error: filetype '{filetype}' is not valid for datatype '{datatype}': must be one of {{'signal', 'spec', 'video_noaudio'}}")
elif filetype == 'png' and datatype not in {'spec'}:
raise ValueError(f"Error: filetype '{filetype}' is not valid for datatype '{datatype}': must be one of {{'spec'}}")
elif filetype == 'wfdb' and datatype not in {'signal'}:
raise ValueError(f"Error: filetype '{filetype}' is not valid for datatype '{datatype}': must be one of {{'signal'}}")
elif filetype == 'mp4' and datatype not in {'video', 'video_noaudio'}:
raise ValueError(f"Error: filetype '{filetype}' is not valid for datatype '{datatype}': must be one of {{'video', 'video_noaudio'}}")
elif filetype == 'wav' and datatype not in {'signal'}:
raise ValueError(f"Error: filetype '{filetype}' is not valid for datatype '{datatype}': must be one of {{'signal'}}")
else:
return True
def read_file(filepath, datatype, filetype, both_in_wfdb=False):
check_datatype_and_filetype(datatype, filetype)
return_data = None
return_data_pcg = None
if filetype == "mp4":
video_specs, fps, size = load_video(filepath)
if not config.global_opts.fps == fps:
print(f"Warning: specified fps ({config.global_opts.fps}) is different from video fps ({fps}); resampling to {config.global_opts.fps}fps")
resample_video(filepath, config.global_opts.fps)
video_specs, fps, size = load_video(filepath)
return_data = video_specs
if datatype == 'video':
audio, sr = convert_video_to_audio_moviepy(filepath)
return_data_pcg = audio
elif filetype == "npz":
if datatype == "signal":
df = np.load(filepath,allow_pickle=True)
return_data = df
elif datatype == "spec":
df = np.load(filepath,allow_pickle=True)
return_data = df
elif filetype == "wfdb":
if both_in_wfdb:
record = wfdb.rdrecord(filepath, channels=[0,1])
signal0 = record.p_signal[:,0]
signal1 = record.p_signal[:,1]
return_data = signal0
return_data_pcg = signal1
else:
record = wfdb.rdrecord(filepath, channels=[0])
signal = record.p_signal[:,0]
return_data = signal
elif filetype == "png":
img = cv2.imread(filepath)
return_data = img
elif filetype == "wav":
wav, sr = load_audio(filepath)
return_data = wav
else:
raise ValueError(f"Error: 'filetype' is not one of {config.file_types_ecg.union(config.file_types_pcg)}")
if return_data_pcg is None:
return return_data
else:
return return_data, return_data_pcg
def read_signal(filepath):
return np.load(filepath)
def linear_regression_objective(x, a, b):
return np.array(x).astype(float) * float(a) + float(b)
def get_filtered_df(df, column, value):
df = df[df[column] == value]
# Returns a df where all values of a certain column are a certain value
return df
def create_new_folder(path):
if not os.path.exists(path):
access = 0o755
try:
os.makedirs(path, access)
except Exception as e:
raise ValueError(e)
return False
return True
else:
# Only create the folder if it is not already there
return False
# Get index '00123' from a directoryname e.g. 'ECGPCG00123' => 123
def get_index_from_directory(directoryname):
matches = re.findall(r'^\D*(\d+)', directoryname)
str_ = matches[0].lstrip('0')
if len(str_) == 0:
str_ = '0'
index = int(str_)
return index
def check_filter_bounds(low, high):
if high <= low:
raise ValueError(f"Error: Upper bound supplied to filter ({high}) must be higher than the lower bound ({low})")
if low < 0:
raise ValueError(f"Error: Lower bound supplied to filter ({low}) must be >= 0")
if high <= 0:
raise ValueError(f"Error: Upper bound supplied to filter ({high}) must be > 0")
def butterworth_bandpass(lowcut, highcut, sample_rate, order=4):
nyq = sample_rate/2
low = lowcut / nyq
high = highcut / nyq
sos = butter(order, [low, high], analog=False, btype='band', output='sos')
return sos
def butterworth_bandpass_filter(data, lowcut, highcut, fs, order=4):
sos = butterworth_bandpass(lowcut, highcut, fs, order=order)
y = sosfilt(sos, data)
return y
# scipysignal.ricker Wavelet Function
def ricker(points, width):
M = []
for i, p in enumerate(np.linspace(-width/2, width/2, num=points)):
A = 2/(math.sqrt(3*width)*(math.pi**(1/3)))
M.append(A*(1 - ((p**2)/(width**2)))*math.exp(-(p**2)/(width**2)))
return M
bior_family = pywt.wavelist(family='bior', kind='continuous')
wavelet = pywt.Wavelet(name="bior2.6")
phi_d, psi_d, phi_r, psi_r, x = wavelet.wavefun(level=6)
# [Constrained transformer network for ECG signal processing and arrhythmia classification]
#
# This paper uses the Biorthogonal 2.6 (bior2.6) Wavelet Function
def bior2_6(points, width):
#ps = []
#families = pywt.families(short=False)
mother_wavelet = lambda z : np.exp(-z*z/4)*(2-z*z)/(4*math.sqrt(math.pi))
def mexican_hat_like(n,scale):
x = np.linspace(-n/2,n/2,n)
return mother_wavelet(x/scale)
print(f"XXX: {phi_d}") #psi_d
#plt.plot(psi_d, label = 'ricker')
#plt.legend(loc = 'best')
#plt.show()
return psi_d
def get_segment_num(sample_rate, samples, segment_length, factor=1):
segments = []
samples_goal = int(np.floor(sample_rate*segment_length))
if samples_goal < 1:
raise ValueError("Error: sample_rate*segment_length results in 0; segment_length is too low")
no_segs = int(np.floor((samples//samples_goal)*factor))
return no_segs
class ImageShape(NamedTuple):
height: int
width: int
channels: int