0%

WFDB

What’s the wfdb?

WFDB is a library of tools for reading, writing, and processing WFDB signals and annotations.

Document and Usage

See the documentation site for the public APIs.

See the demo.ipynb notebook file for more example use cases.

Install wfdb
1
pip install wfdb
show annotations classes
1
2
3
4
import wfdb

#显示所有的标注格式
wfdb.show_ann_classes()

The results are as follows:

extension description human_reviewed
atr Reference ECG annotations True
blh Human reviewed beat labels True
blm Machine beat labels False
alh Human reviewed alarms True
alm Machine alarms False
qrsc Human reviewed QRS detections True
qrs Machine QRS detections False
bph Human reviewed BP beat detections True
bpm Machine BP beat detections False
class Record

在wfdb-python包中找到record.py文件,该文件定义了record类,其声明和初始化如下:

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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
class Record(BaseRecord, _header.HeaderMixin, _signal.SignalMixin):
"""
The class representing single segment WFDB records.
Record objects can be created using the initializer, by reading a WFDB
header with `rdheader`, or a WFDB record (header and associated dat files)
with `rdrecord`.
The attributes of the Record object give information about the record as
specified by: https://www.physionet.org/physiotools/wag/header-5.htm
In addition, the d_signal and p_signal attributes store the digital and
physical signals of WFDB records with at least one channel.
Attributes
----------
p_signal : ndarray, optional
An (MxN) 2d numpy array, where M is the signal length. Gives the
physical signal values intended to be written. Either p_signal or
d_signal must be set, but not both. If p_signal is set, this method will
use it to perform analogue-digital conversion, writing the resultant
digital values to the dat file(s). If fmt is set, gain and baseline must
be set or unset together. If fmt is unset, gain and baseline must both
be unset.
d_signal : ndarray, optional
An (MxN) 2d numpy array, where M is the signal length. Gives the
digital signal values intended to be directly written to the dat
file(s). The dtype must be an integer type. Either p_signal or d_signal
must be set, but not both. In addition, if d_signal is set, fmt, gain
and baseline must also all be set.
p_signal : ndarray, optional
The expanded physical conversion of the signal. Either a 2d numpy
array or a list of 1d numpy arrays.
e_d_signal : ndarray, optional
The expanded digital conversion of the signal. Either a 2d numpy
array or a list of 1d numpy arrays.
record_name : str, optional
The name of the WFDB record to be read, without any file
extensions. If the argument contains any path delimiter
characters, the argument will be interpreted as PATH/BASE_RECORD.
Both relative and absolute paths are accepted. If the `pn_dir`
parameter is set, this parameter should contain just the base
record name, and the files fill be searched for remotely.
Otherwise, the data files will be searched for in the local path.
n_sig : int, optional
Total number of signals.
fs : float, optional
The sampling frequency of the record.
counter_freq : float, optional
The frequency used to start counting.
base_counter : float, optional
The counter used at the start of the file.
sig_len : int, optional
The total length of the signal.
base_time : str, optional
A string of the record's start time in 24h 'HH:MM:SS(.ms)' format.
base_date : str, optional
A string of the record's start date in 'DD/MM/YYYY' format.
file_name : str, optional
The name of the file used for analysis.
fmt : list, optional
A list of strings giving the WFDB format of each file used to store each
channel. Accepted formats are: '80','212",'16','24', and '32'. There are
other WFDB formats as specified by:
https://www.physionet.org/physiotools/wag/signal-5.htm
but this library will not write (though it will read) those file types.
samps_per_frame : int, optional
The total number of samples per frame.
skew : float, optional
The offset used to allign signals.
byte_offset : int, optional
The byte offset used to allign signals.
adc_gain : list, optional
A list of numbers specifying the ADC gain.
baseline : list, optional
A list of integers specifying the digital baseline.
units : list, optional
A list of strings giving the units of each signal channel.
adc_res: int, optional
The value produced by the ADC given a given Volt input.
adc_zero: int, optional
The value produced by the ADC given a 0 Volt input.
init_value : list, optional
The initial value of the signal.
checksum : list, int, optional
The checksum of the signal.
block_size : str, optional
The dimensions of the field data.
sig_name : list, optional
A list of strings giving the signal name of each signal channel.
comments : list, optional
A list of string comments to be written to the header file.
Examples
--------
>>> record = wfdb.Record(record_name='r1', fs=250, n_sig=2, sig_len=1000,
file_name=['r1.dat','r1.dat'])
"""
def __init__(self, p_signal=None, d_signal=None,
e_p_signal=None, e_d_signal=None,
record_name=None, n_sig=None,
fs=None, counter_freq=None, base_counter=None,
sig_len=None, base_time=None, base_date=None,
file_name=None, fmt=None, samps_per_frame=None,
skew=None, byte_offset=None, adc_gain=None,
baseline=None, units=None, adc_res=None,
adc_zero=None, init_value=None, checksum=None,
block_size=None, sig_name=None, comments=None):

# Note the lack of the 'n_seg' field. Single segment records cannot
# have this field. Even n_seg = 1 makes the header a multi-segment
# header.
super(Record, self).__init__(record_name, n_sig,
fs, counter_freq, base_counter, sig_len,
base_time, base_date, comments, sig_name)

self.p_signal = p_signal
self.d_signal = d_signal
self.e_p_signal = e_p_signal
self.e_d_signal = e_d_signal

self.file_name = file_name
self.fmt = fmt
self.samps_per_frame = samps_per_frame
self.skew = skew
self.byte_offset = byte_offset
self.adc_gain = adc_gain
self.baseline = baseline
self.units = units
self.adc_res = adc_res
self.adc_zero = adc_zero
self.init_value = init_value
self.checksum = checksum
self.block_size = block_size

# Equal comparison operator for objects of this type
def __eq__(self, other, verbose=False):
"""
Equal comparison operator for objects of this type.
Parameters
----------
other : object
The object that is being compared to self.
verbose : bool, optional
Whether to print details about equality (True) or not (False).
Returns
-------
bool
Determines if the objects are equal (True) or not equal (False).
"""
att1 = self.__dict__
att2 = other.__dict__

if set(att1.keys()) != set(att2.keys()):
if verbose:
print('Attributes members mismatch.')
return False

for k in att1.keys():

v1 = att1[k]
v2 = att2[k]

if type(v1) != type(v2):
if verbose:
print('Mismatch in attribute: %s' % k, v1, v2)
return False

if type(v1) == np.ndarray:
# Necessary for nans
np.testing.assert_array_equal(v1, v2)
else:
if v1 != v2:
if verbose:
print('Mismatch in attribute: %s' % k, v1, v2)
return False

return True


def wrsamp(self, expanded=False, write_dir=''):
"""
Write a WFDB header file and any associated dat files from this
object.
Parameters
----------
expanded : bool, optional
Whether to write the expanded signal (e_d_signal) instead
of the uniform signal (d_signal).
write_dir : str, optional
The directory in which to write the files.
Returns
-------
N/A
"""
# Perform field validity and cohesion checks, and write the
# header file.
self.wrheader(write_dir=write_dir)
if self.n_sig > 0:
# Perform signal validity and cohesion checks, and write the
# associated dat files.
self.wr_dats(expanded=expanded, write_dir=write_dir)


def _arrange_fields(self, channels, sampfrom=0, expanded=False):
"""
Arrange/edit object fields to reflect user channel and/or signal
range input.
Parameters
----------
channels : list
List of channel numbers specified.
sampfrom : int, optional
Starting sample number read.
expanded : bool, optional
Whether the record was read in expanded mode.
Returns
-------
N/A
"""
# Rearrange signal specification fields
for field in _header.SIGNAL_SPECS.index:
item = getattr(self, field)
setattr(self, field, [item[c] for c in channels])

# Expanded signals - multiple samples per frame.
if expanded:
# Checksum and init_value to be updated if present
# unless the whole signal length was input
if self.sig_len != int(len(self.e_d_signal[0]) / self.samps_per_frame[0]):
self.checksum = self.calc_checksum(expanded)
self.init_value = [s[0] for s in self.e_d_signal]

self.n_sig = len(channels)
self.sig_len = int(len(self.e_d_signal[0]) / self.samps_per_frame[0])

# MxN numpy array d_signal
else:
# Checksum and init_value to be updated if present
# unless the whole signal length was input
if self.sig_len != self.d_signal.shape[0]:

if self.checksum is not None:
self.checksum = self.calc_checksum()
if self.init_value is not None:
ival = list(self.d_signal[0, :])
self.init_value = [int(i) for i in ival]

# Update record specification parameters
# Important that these get updated after^^
self.n_sig = len(channels)
self.sig_len = self.d_signal.shape[0]

# Adjust date and time if necessary
self._adjust_datetime(sampfrom=sampfrom)