import datetime
from typing import Collection, List, Tuple
import numpy as np
import pandas as pd
from wfdb.io import _signal
from wfdb.io import util
from wfdb.io.header import HeaderSyntaxError, rx_record, rx_segment, rx_signal
"""
Notes
-----
In the original WFDB package, certain fields have default values, but
not all of them. Some attributes need to be present for core
functionality, i.e. baseline, whereas others are not essential, yet have
defaults, i.e. base_time.
This inconsistency has likely resulted in the generation of incorrect
files, and general confusion. This library aims to make explicit,
whether certain fields are present in the file, by setting their values
to None if they are not written in, unless the fields are essential, in
which case an actual default value will be set.
The read vs write default values are different for 2 reasons:
1. We want to force the user to be explicit with certain important
fields when writing WFDB records fields, without affecting
existing WFDB headers when reading.
2. Certain unimportant fields may be dependencies of other
important fields. When writing, we want to fill in defaults
so that the user doesn't need to. But when reading, it should
be clear that the fields are missing.
"""
int_types = (int, np.int64, np.int32, np.int16, np.int8)
float_types = (float, np.float64, np.float32) + int_types
_SPECIFICATION_COLUMNS = [
"allowed_types",
"delimiter",
"dependency",
"write_required",
"read_default",
"write_default",
]
RECORD_SPECS = pd.DataFrame(
index=[
"record_name",
"n_seg",
"n_sig",
"fs",
"counter_freq",
"base_counter",
"sig_len",
"base_time",
"base_date",
],
columns=_SPECIFICATION_COLUMNS,
dtype="object",
data=[
[(str,), "", None, True, None, None], # record_name
[int_types, "/", "record_name", True, None, None], # n_seg
[int_types, " ", "record_name", True, None, None], # n_sig
[float_types, " ", "n_sig", True, 250, None], # fs
[float_types, "/", "fs", False, None, None], # counter_freq
[float_types, "(", "counter_freq", False, None, None], # base_counter
[int_types, " ", "fs", True, None, None], # sig_len
[
(datetime.time,),
" ",
"sig_len",
False,
None,
"00:00:00",
], # base_time
[(datetime.date,), " ", "base_time", False, None, None], # base_date
],
)
SIGNAL_SPECS = pd.DataFrame(
index=[
"file_name",
"fmt",
"samps_per_frame",
"skew",
"byte_offset",
"adc_gain",
"baseline",
"units",
"adc_res",
"adc_zero",
"init_value",
"checksum",
"block_size",
"sig_name",
],
columns=_SPECIFICATION_COLUMNS,
dtype="object",
data=[
[(str,), "", None, True, None, None], # file_name
[(str,), " ", "file_name", True, None, None], # fmt
[int_types, "x", "fmt", False, 1, None], # samps_per_frame
[int_types, ":", "fmt", False, None, None], # skew
[int_types, "+", "fmt", False, None, None], # byte_offset
[float_types, " ", "fmt", True, 200.0, None], # adc_gain
[int_types, "(", "adc_gain", True, 0, None], # baseline
[(str,), "/", "adc_gain", True, "mV", None], # units
[int_types, " ", "adc_gain", False, None, 0], # adc_res
[int_types, " ", "adc_res", False, None, 0], # adc_zero
[int_types, " ", "adc_zero", False, None, None], # init_value
[int_types, " ", "init_value", False, None, None], # checksum
[int_types, " ", "checksum", False, None, 0], # block_size
[(str,), " ", "block_size", False, None, None], # sig_name
],
)
SEGMENT_SPECS = pd.DataFrame(
index=["seg_name", "seg_len"],
columns=_SPECIFICATION_COLUMNS,
dtype="object",
data=[
[(str), "", None, True, None, None], # seg_name
[int_types, " ", "seg_name", True, None, None], # seg_len
],
)
# Specifications of all WFDB header fields, except for comments
FIELD_SPECS = pd.concat((RECORD_SPECS, SIGNAL_SPECS, SEGMENT_SPECS))
class BaseHeaderMixin(object):
"""
Mixin class with multi-segment header methods. Inherited by Record and
MultiRecord classes.
Attributes
----------
N/A
"""
def get_write_subset(self, spec_type):
"""
Get a set of fields used to write the header; either 'record'
or 'signal' specification fields. Helper function for
`get_write_fields`. Gets the default required fields, the user
defined fields, and their dependencies.
Parameters
----------
spec_type : str
The set of specification fields desired. Either 'record' or
'signal'.
Returns
-------
write_fields : list or dict
For record fields, returns a list of all fields needed. For
signal fields, it returns a dictionary of all fields needed,
with keys = field and value = list of channels that must be
present for the field.
"""
if spec_type == "record":
write_fields = []
record_specs = RECORD_SPECS.copy()
# Remove the n_seg requirement for single segment items
if not hasattr(self, "n_seg"):
record_specs.drop("n_seg", inplace=True)
for field in record_specs.index[-1::-1]:
# Continue if the field has already been included
if field in write_fields:
continue
# If the field is required by default or has been
# defined by the user
if (
record_specs.loc[field, "write_required"]
or getattr(self, field) is not None
):
req_field = field
# Add the field and its recursive dependencies
while req_field is not None:
write_fields.append(req_field)
req_field = record_specs.loc[req_field, "dependency"]
# Add comments if any
if getattr(self, "comments") is not None:
write_fields.append("comments")
# signal spec field. Need to return a potentially different list for each channel.
elif spec_type == "signal":
# List of lists for each channel
write_fields = []
signal_specs = SIGNAL_SPECS.copy()
for ch in range(self.n_sig):
# The fields needed for this channel
write_fields_ch = []
for field in signal_specs.index[-1::-1]:
if field in write_fields_ch:
continue
item = getattr(self, field)
# If the field is required by default or has been defined by the user
if signal_specs.loc[field, "write_required"] or (
item is not None and item[ch] is not None
):
req_field = field
# Add the field and its recursive dependencies
while req_field is not None:
write_fields_ch.append(req_field)
req_field = signal_specs.loc[
req_field, "dependency"
]
write_fields.append(write_fields_ch)
# Convert the list of lists to a single dictionary.
# keys = field and value = list of channels in which the
# field is required.
dict_write_fields = {}
# For fields present in any channel:
for field in set(
[i for write_fields_ch in write_fields for i in write_fields_ch]
):
dict_write_fields[field] = []
for ch in range(self.n_sig):
if field in write_fields[ch]:
dict_write_fields[field].append(ch)
write_fields = dict_write_fields
return write_fields
class HeaderMixin(BaseHeaderMixin):
"""
Mixin class with single-segment header methods. Inherited by Record class.
Attributes
----------
N/A
"""
def set_defaults(self):
"""
Set defaults for fields needed to write the header if they have
defaults.
Parameters
----------
N/A
Returns
-------
N/A
Notes
-----
- This is NOT called by `rdheader`. It is only automatically
called by the gateway `wrsamp` for convenience.
- This is also not called by `wrheader` since it is supposed to
be an explicit function.
- This is not responsible for initializing the attributes. That
is done by the constructor.
See also `set_p_features` and `set_d_features`.
"""
rfields, sfields = self.get_write_fields()
for f in rfields:
self.set_default(f)
for f in sfields:
self.set_default(f)
def wrheader(self, write_dir="", expanded=True):
"""
Write a WFDB header file. The signals are not used. Before
writing:
- Get the fields used to write the header for this instance.
- Check each required field.
- Check that the fields are cohesive with one another.
Parameters
----------
write_dir : str, optional
The output directory in which the header is written.
expanded : bool, optional
Whether the header file should include `samps_per_frame` (this
should only be true if the signal files are written using
`expanded=True`).
Returns
-------
N/A
Notes
-----
This function does NOT call `set_defaults`. Essential fields
must be set beforehand.
"""
# Get all the fields used to write the header
# sig_write_fields is a dictionary of
# {field_name:required_channels}
rec_write_fields, sig_write_fields = self.get_write_fields()
if not expanded:
sig_write_fields.pop("samps_per_frame", None)
# Check the validity of individual fields used to write the header
# Record specification fields (and comments)
for field in rec_write_fields:
self.check_field(field)
# Signal specification fields.
for field in sig_write_fields:
self.check_field(field, required_channels=sig_write_fields[field])
# Check the cohesion of fields used to write the header
self.check_field_cohesion(rec_write_fields, list(sig_write_fields))
# Write the header file using the specified fields
self.wr_header_file(rec_write_fields, sig_write_fields, write_dir)
def get_write_fields(self):
"""
Get the list of fields used to write the header, separating
record and signal specification fields. Returns the default
required fields, the user defined fields, and their dependencies.
Does NOT include `d_signal` or `e_d_signal`.
Parameters
----------
N/A
Returns
-------
rec_write_fields : list
Record specification fields to be written. Includes
'comment' if present.
sig_write_fields : dict
Dictionary of signal specification fields to be written,
with values equal to the channels that need to be present
for each field.
"""
# Record specification fields
rec_write_fields = self.get_write_subset("record")
# Add comments if any
if self.comments != None:
rec_write_fields.append("comments")
# Get required signal fields if signals are present.
self.check_field("n_sig")
if self.n_sig > 0:
sig_write_fields = self.get_write_subset("signal")
else:
sig_write_fields = None
return rec_write_fields, sig_write_fields
def _auto_signal_file_names(self):
fmt = self.fmt or [None] * self.n_sig
spf = self.samps_per_frame or [None] * self.n_sig
num_groups = 0
group_number = []
prev_fmt = prev_spf = None
channels_in_group = 0
for ch_fmt, ch_spf in zip(fmt, spf):
if ch_fmt != prev_fmt:
num_groups += 1
channels_in_group = 0
elif ch_fmt in ("508", "516", "524"):
if channels_in_group >= 8 or ch_spf != prev_spf:
num_groups += 1
channels_in_group = 0
group_number.append(num_groups)
prev_fmt = ch_fmt
prev_spf = ch_spf
if num_groups < 2:
return [self.record_name + ".dat"] * self.n_sig
else:
digits = len(str(group_number[-1]))
return [
self.record_name + "_" + str(g).rjust(digits, "0") + ".dat"
for g in group_number
]
def set_default(self, field):
"""
Set the object's attribute to its default value if it is missing
and there is a default. Not responsible for initializing the
attribute. That is done by the constructor.
Parameters
----------
field : str
The desired attribute of the object.
Returns
-------
N/A
"""
# Record specification fields
if field in RECORD_SPECS.index:
# Return if no default to set, or if the field is already
# present.
if (
RECORD_SPECS.loc[field, "write_default"] is None
or getattr(self, field) is not None
):
return
setattr(self, field, RECORD_SPECS.loc[field, "write_default"])
# Signal specification fields
# Setting entire list default, not filling in blanks in lists.
elif field in SIGNAL_SPECS.index:
# Specific dynamic case
if field == "file_name" and self.file_name is None:
self.file_name = self._auto_signal_file_names()
return
item = getattr(self, field)
# Return if no default to set, or if the field is already
# present.
if (
SIGNAL_SPECS.loc[field, "write_default"] is None
or item is not None
):
return
# Set more specific defaults if possible
if field == "adc_res" and self.fmt is not None:
self.adc_res = _signal._fmt_res(self.fmt)
return
setattr(
self,
field,
[SIGNAL_SPECS.loc[field, "write_default"]] * self.n_sig,
)
def check_field_cohesion(self, rec_write_fields, sig_write_fields):
"""
Check the cohesion of fields used to write the header.
Parameters
----------
rec_write_fields : list
List of record specification fields to write.
sig_write_fields : dict
Dictionary of signal specification fields to write, values
being equal to a list of channels to write for each field.
Returns
-------
N/A
"""
# If there are no signal specification fields, there is nothing to check.
if self.n_sig > 0:
# The length of all signal specification fields must match n_sig
# even if some of its elements are None.
for f in sig_write_fields:
if len(getattr(self, f)) != self.n_sig:
raise ValueError(
"The length of field: " + f + " must match field n_sig."
)
# Each file_name must correspond to only one fmt, (and only one byte offset if defined).
datfmts = {}
for ch in range(self.n_sig):
if self.file_name[ch] not in datfmts:
datfmts[self.file_name[ch]] = self.fmt[ch]
else:
if datfmts[self.file_name[ch]] != self.fmt[ch]:
raise ValueError(
"Each file_name (dat file) specified must have the same fmt"
)
datoffsets = {}
if self.byte_offset is not None:
# At least one byte offset value exists
for ch in range(self.n_sig):
if self.byte_offset[ch] is None:
continue
if self.file_name[ch] not in datoffsets:
datoffsets[self.file_name[ch]] = self.byte_offset[ch]
else:
if (
datoffsets[self.file_name[ch]]
!= self.byte_offset[ch]
):
raise ValueError(
"Each file_name (dat file) specified must have the same byte offset"
)
def wr_header_file(self, rec_write_fields, sig_write_fields, write_dir):
"""
Write a header file using the specified fields. Converts Record
attributes into appropriate WFDB format strings.
Parameters
----------
rec_write_fields : list
List of record specification fields to write.
sig_write_fields : dict
Dictionary of signal specification fields to write, values
being equal to a list of channels to write for each field.
write_dir : str
The directory in which to write the header file.
Returns
-------
N/A
"""
# Create record specification line
record_line = ""
# Traverse the ordered dictionary
for field in RECORD_SPECS.index:
# If the field is being used, add it with its delimiter
if field in rec_write_fields:
string_field = str(getattr(self, field))
# Certain fields need extra processing
if field == "fs" and isinstance(self.fs, float):
if round(self.fs, 8) == float(int(self.fs)):
string_field = str(int(self.fs))
elif field == "base_time" and "." in string_field:
string_field = string_field.rstrip("0")
elif field == "base_date":
string_field = "/".join(
(string_field[8:], string_field[5:7], string_field[:4])
)
record_line += (
RECORD_SPECS.loc[field, "delimiter"] + string_field
)
# The 'base_counter' field needs to be closed with ')'
if field == "base_counter":
record_line += ")"
header_lines = [record_line]
# Create signal specification lines (if any) one channel at a time
if self.n_sig > 0:
signal_lines = self.n_sig * [""]
for ch in range(self.n_sig):
# Traverse the signal fields
for field in SIGNAL_SPECS.index:
# If the field is being used, add each of its
# elements with the delimiter to the appropriate
# line
if (
field in sig_write_fields
and ch in sig_write_fields[field]
):
signal_lines[ch] += SIGNAL_SPECS.loc[
field, "delimiter"
] + str(getattr(self, field)[ch])
# The 'baseline' field needs to be closed with ')'
if field == "baseline":
signal_lines[ch] += ")"
header_lines += signal_lines
# Create comment lines (if any)
if "comments" in rec_write_fields:
comment_lines = ["# " + comment for comment in self.comments]
header_lines += comment_lines
util.lines_to_file(self.record_name + ".hea", write_dir, header_lines)
class MultiHeaderMixin(BaseHeaderMixin):
"""
Mixin class with multi-segment header methods. Inherited by
MultiRecord class.
Attributes
----------
N/A
"""
def set_defaults(self):
"""
Set defaults for fields needed to write the header if they have
defaults. This is NOT called by rdheader. It is only called by the
gateway wrsamp for convenience. It is also not called by wrheader since
it is supposed to be an explicit function. Not responsible for
initializing the attributes. That is done by the constructor.
Parameters
----------
N/A
Returns
-------
N/A
"""
for field in self.get_write_fields():
self.set_default(field)
def wrheader(self, write_dir=""):
"""
Write a multi-segment WFDB header file. The signals or segments are
not used. Before writing:
- Get the fields used to write the header for this instance.
- Check each required field.
- Check that the fields are cohesive with one another.
Parameters
----------
write_dir : str, optional
The output directory in which the header is written.
Returns
-------
N/A
Notes
-----
This function does NOT call `set_defaults`. Essential fields
must be set beforehand.
"""
# Get all the fields used to write the header
write_fields = self.get_write_fields()
# Check the validity of individual fields used to write the header
for field in write_fields:
self.check_field(field)
# Check the cohesion of fields used to write the header
self.check_field_cohesion()
# Write the header file using the specified fields
self.wr_header_file(write_fields, write_dir)
def get_write_fields(self):
"""
Get the list of fields used to write the multi-segment header.
Parameters
----------
N/A
Returns
-------
write_fields : list
All the default required fields, the user defined fields,
and their dependencies.
"""
# Record specification fields
write_fields = self.get_write_subset("record")
# Segment specification fields are all mandatory
write_fields = write_fields + ["seg_name", "seg_len"]
# Comments
if self.comments != None:
write_fields.append("comments")
return write_fields
def set_default(self, field):
"""
Set a field to its default value if there is a default.
Parameters
----------
field : str
The desired attribute of the object.
Returns
-------
N/A
"""
# Record specification fields
if field in RECORD_SPECS:
# Return if no default to set, or if the field is already present.
if (
RECORD_SPECS[field].write_def is None
or getattr(self, field) is not None
):
return
setattr(self, field, RECORD_SPECS[field].write_def)
def check_field_cohesion(self):
"""
Check the cohesion of fields used to write the header.
Parameters
----------
N/A
Returns
-------
N/A
"""
# The length of seg_name and seg_len must match n_seg
for f in ["seg_name", "seg_len"]:
if len(getattr(self, f)) != self.n_seg:
raise ValueError(
"The length of field: " + f + " does not match field n_seg."
)
# Check the sum of the 'seg_len' fields against 'sig_len'
if np.sum(self.seg_len) != self.sig_len:
raise ValueError(
"The sum of the 'seg_len' fields do not match the 'sig_len' field"
)
def wr_header_file(self, write_fields, write_dir):
"""
Write a header file using the specified fields.
Parameters
----------
write_fields : list
All the default required fields, the user defined fields,
and their dependencies.
write_dir : str
The output directory in which the header is written.
Returns
-------
N/A
"""
# Create record specification line
record_line = ""
# Traverse the ordered dictionary
for field in RECORD_SPECS.index:
# If the field is being used, add it with its delimiter
if field in write_fields:
record_line += RECORD_SPECS.loc[field, "delimiter"] + str(
getattr(self, field)
)
header_lines = [record_line]
# Create segment specification lines
segment_lines = self.n_seg * [""]
# For both fields, add each of its elements with the delimiter
# to the appropriate line
for field in SEGMENT_SPECS.index:
for seg_num in range(self.n_seg):
segment_lines[seg_num] += SEGMENT_SPECS.loc[
field, "delimiter"
] + str(getattr(self, field)[seg_num])
header_lines = header_lines + segment_lines
# Create comment lines (if any)
if "comments" in write_fields:
comment_lines = ["# " + comment for comment in self.comments]
header_lines += comment_lines
util.lines_to_file(self.record_name + ".hea", header_lines, write_dir)
def get_sig_segments(self, sig_name=None):
"""
Get a list of the segment numbers that contain a particular signal
(or a dictionary of segment numbers for a list of signals).
Only works if information about the segments has been read in.
Parameters
----------
sig_name : str, list
The name of the signals to be segmented.
Returns
-------
sig_dict : dict
Segments for each desired signal.
sig_segs : list
Segments for the desired signal.
"""
if self.segments is None:
raise Exception(
"The MultiRecord's segments must be read in before this method is called. ie. Call rdheader() with rsegment_fieldsments=True"
)
# Default value = all signal names.
if sig_name is None:
sig_name = self.get_sig_name()
if isinstance(sig_name, list):
sig_dict = {}
for sig in sig_name:
sig_dict[sig] = self.get_sig_segments(sig)
return sig_dict
elif isinstance(sig_name, str):
sig_segs = []
for i in range(self.n_seg):
if (
self.seg_name[i] != "~"
and sig_name in self.segments[i].sig_name
):
sig_segs.append(i)
return sig_segs
else:
raise TypeError("sig_name must be a string or a list of strings")
def get_sig_name(self):
"""
Get the signal names for the entire record.
Parameters
----------
N/A
Returns
-------
sig_name : str, list
The name of the signals to be segmented.
"""
if self.segments is None:
raise Exception(
"The MultiRecord's segments must be read in before this method is called. ie. Call rdheader() with rd_segments=True"
)
if self.layout == "fixed":
for i in range(self.n_seg):
if self.seg_name[i] != "~":
sig_name = self.segments[i].sig_name
break
else:
sig_name = self.segments[0].sig_name
return sig_name
def contained_ranges(self, sig_name: str) -> List[Tuple[int, int]]:
"""
Given a signal name, return the sample ranges that contain signal values,
relative to the start of the full record. Does not account for NaNs/missing
values.
This function is mainly useful for variable layout records, but can also be
used for fixed-layout records. Only works if the headers from the individual
segment records have already been read in.
Parameters
----------
sig_name : str
The name of the signal to query.
Returns
-------
ranges : List[Tuple[int, int]]
Tuple pairs which specify thee sample ranges in which the signal is contained.
The second value of each tuple pair will be one beyond the signal index.
eg. A length 1000 signal would generate a tuple of: (0, 1000), allowing
selection using signal[0:1000].
"""
if self.segments is None:
raise Exception(
"The MultiRecord's segments must be read in before this method is called. ie. Call rdheader() with rd_segments=True"
)
ranges = []
seg_start = 0
range_start = None
# TODO: Add shortcut for fixed-layout records
# Cannot process segments only because missing segments are None
# and do not contain length information.
for seg_num in range(self.n_seg):
seg_len = self.seg_len[seg_num]
segment = self.segments[seg_num]
if seg_len == 0:
continue
# Open signal range
if (
range_start is None
and segment is not None
and sig_name in segment.sig_name
):
range_start = seg_start
# Close signal range
elif range_start is not None and (
segment is None or sig_name not in segment.sig_name
):
ranges.append((range_start, seg_start))
range_start = None
seg_start += seg_len
# Account for final segment
if range_start is not None:
ranges.append((range_start, seg_start))
return ranges
def contained_combined_ranges(
self,
sig_names: Collection[str],
) -> List[Tuple[int, int]]:
"""
Given a collection of signal name, return the sample ranges that
contain all of the specified signals, relative to the start of the
full record. Does not account for NaNs/missing values.
This function is mainly useful for variable layout records, but can also be
used for fixed-layout records. Only works if the headers from the individual
segment records have already been read in.
Parameters
----------
sig_names : List[str]
The names of the signals to query.
Returns
-------
ranges : List[Tuple[int, int]]
Tuple pairs which specify thee sample ranges in which the signal is contained.
The second value of each tuple pair will be one beyond the signal index.
eg. A length 1000 signal would generate a tuple of: (0, 1000), allowing
selection using signal[0:1000].
"""
# TODO: Add shortcut for fixed-layout records
if len(sig_names) == 0:
return []
combined_ranges = self.contained_ranges(sig_names[0])
if len(sig_names) > 1:
for name in sig_names[1:]:
combined_ranges = util.overlapping_ranges(
combined_ranges, self.contained_ranges(name)
)
return combined_ranges
def wfdb_strptime(time_string: str) -> datetime.time:
"""
Given a time string in an acceptable WFDB format, return
a datetime.time object.
Valid formats: SS, MM:SS, HH:MM:SS, all with and without microsec.
Parameters
----------
time_string : str
The time to be converted to a datetime.time object.
Returns
-------
datetime.time object
The time converted from str format.
"""
n_colons = time_string.count(":")
if n_colons == 0:
time_fmt = "%S"
elif n_colons == 1:
time_fmt = "%M:%S"
elif n_colons == 2:
time_fmt = "%H:%M:%S"
if "." in time_string:
time_fmt += ".%f"
return datetime.datetime.strptime(time_string, time_fmt).time()
def _parse_record_line(record_line: str) -> dict:
"""
Extract fields from a record line string into a dictionary.
Parameters
----------
record_line : str
The record line contained in the header file
Returns
-------
record_fields : dict
The fields for the given record line.
"""
# Dictionary for record fields
record_fields = {}
# Read string fields from record line
match = rx_record.match(record_line)
if match is None:
raise HeaderSyntaxError("invalid syntax in record line")
(
record_fields["record_name"],
record_fields["n_seg"],
record_fields["n_sig"],
record_fields["fs"],
record_fields["counter_freq"],
record_fields["base_counter"],
record_fields["sig_len"],
record_fields["base_time"],
record_fields["base_date"],
) = match.groups()
for field in RECORD_SPECS.index:
# Replace empty strings with their read defaults (which are
# mostly None)
if record_fields[field] == "":
record_fields[field] = RECORD_SPECS.loc[field, "read_default"]
# Typecast non-empty strings for non-string (numerical/datetime)
# fields
else:
if RECORD_SPECS.loc[field, "allowed_types"] == int_types:
record_fields[field] = int(record_fields[field])
elif RECORD_SPECS.loc[field, "allowed_types"] == float_types:
record_fields[field] = float(record_fields[field])
# cast fs to an int if it is close
if field == "fs":
fs = float(record_fields["fs"])
if round(fs, 8) == float(int(fs)):
fs = int(fs)
record_fields["fs"] = fs
elif field == "base_time":
record_fields["base_time"] = wfdb_strptime(
record_fields["base_time"]
)
elif field == "base_date":
record_fields["base_date"] = datetime.datetime.strptime(
record_fields["base_date"], "%d/%m/%Y"
).date()
# This is not a standard WFDB field, but is useful to set.
if record_fields["base_date"] and record_fields["base_time"]:
record_fields["base_datetime"] = datetime.datetime.combine(
record_fields["base_date"], record_fields["base_time"]
)
return record_fields
def _parse_signal_lines(signal_lines):
"""
Extract fields from a list of signal line strings into a dictionary.
Parameters
----------
signal_lines : list
The name of the signal line that will be used to extact fields.
Returns
-------
signal_fields : dict
The fields for the given signal line.
"""
n_sig = len(signal_lines)
# Dictionary for signal fields
signal_fields = {}
# Each dictionary field is a list
for field in SIGNAL_SPECS.index:
signal_fields[field] = n_sig * [None]
# Read string fields from signal line
for ch in range(n_sig):
match = rx_signal.match(signal_lines[ch])
if match is None:
raise HeaderSyntaxError("invalid syntax in signal line")
(
signal_fields["file_name"][ch],
signal_fields["fmt"][ch],
signal_fields["samps_per_frame"][ch],
signal_fields["skew"][ch],
signal_fields["byte_offset"][ch],
signal_fields["adc_gain"][ch],
signal_fields["baseline"][ch],
signal_fields["units"][ch],
signal_fields["adc_res"][ch],
signal_fields["adc_zero"][ch],
signal_fields["init_value"][ch],
signal_fields["checksum"][ch],
signal_fields["block_size"][ch],
signal_fields["sig_name"][ch],
) = match.groups()
for field in SIGNAL_SPECS.index:
# Replace empty strings with their read defaults (which are mostly None)
# Note: Never set a field to None. [None]* n_sig is accurate, indicating
# that different channels can be present or missing.
if signal_fields[field][ch] == "":
signal_fields[field][ch] = SIGNAL_SPECS.loc[
field, "read_default"
]
# Special case: missing baseline defaults to ADCzero if present
if field == "baseline" and signal_fields["adc_zero"][ch] != "":
signal_fields["baseline"][ch] = int(
signal_fields["adc_zero"][ch]
)
# Typecast non-empty strings for numerical fields
else:
if SIGNAL_SPECS.loc[field, "allowed_types"] is int_types:
signal_fields[field][ch] = int(signal_fields[field][ch])
elif SIGNAL_SPECS.loc[field, "allowed_types"] is float_types:
signal_fields[field][ch] = float(signal_fields[field][ch])
# Special case: adc_gain of 0 means 200
if (
field == "adc_gain"
and signal_fields["adc_gain"][ch] == 0
):
signal_fields["adc_gain"][ch] = 200.0
return signal_fields
def _read_segment_lines(segment_lines):
"""
Extract fields from segment line strings into a dictionary.
Parameters
----------
segment_line : list
The name of the segment line that will be used to extact fields.
Returns
-------
segment_fields : dict
The fields for the given segment line.
"""
# Dictionary for segment fields
segment_fields = {}
# Each dictionary field is a list
for field in SEGMENT_SPECS.index:
segment_fields[field] = [None] * len(segment_lines)
# Read string fields from signal line
for i in range(len(segment_lines)):
match = rx_segment.match(segment_lines[i])
if match is None:
raise HeaderSyntaxError("invalid syntax in segment line")
(
segment_fields["seg_name"][i],
segment_fields["seg_len"][i],
) = match.groups()
# Typecast strings for numerical field
if field == "seg_len":
segment_fields["seg_len"][i] = int(segment_fields["seg_len"][i])
return segment_fields