edf.py

import datetime
import functools
import math
import os
import posixpath
import struct

import numpy as np
import pandas as pd

from wfdb.io.annotation import Annotation, format_ann_from_df, wrann
from wfdb.io.record import Record, rdrecord, SIG_UNITS
from wfdb.io import _url
from wfdb.io import download


def read_edf(
    record_name,
    pn_dir=None,
    header_only=False,
    verbose=False,
    rdedfann_flag=False,
    encoding="iso8859-1",
):
    """
    Read a EDF format file into a WFDB Record.

    Many EDF files contain signals at widely varying sampling frequencies.
    `read_edf` handles these properly, but the default behavior of most WFDB
    applications is to read such data in low-resolution mode (in which all
    signals are resampled at the lowest sampling frequency used for any signal
    in the record). This is almost certainly not what you want if, for
    example, the record contains EEG signals sampled at 200 Hz and body
    temperature sampled at 1 Hz; by default, applications such as `rdsamp`
    will resample the EEGs (and any other signals in the record) at 1 Hz. To
    avoid this behavior, you can set `smooth_frames` to False (high resolution)
    provided by `rdrecord` and a few other WFDB applications.

    Note that applications built using version 3.1.0 and later versions of
    the WFDB-Python library can read EDF files directly, so that the conversion
    performed by `read_edf` is no longer necessary. However, one can still use
    this function to produce WFDB-compatible files from EDF files if desired.

    Parameters
    ----------
    record_name : str
        The name of the input EDF record to be read.
    pn_dir : str, optional
        Option used to stream data from Physionet. The Physionet
        database directory from which to find the required record files.
        eg. For record '100' in 'http://physionet.org/content/mitdb'
        pn_dir='mitdb'.
    header_only : bool, optional
        Whether to only return the header information (True) or not (False).
        If true, this function will only return `['fs', 'sig_len', 'n_sig',
        'base_date', 'base_time', 'units', 'sig_name', 'comments']`.
    verbose : bool, optional
        Whether to print all the information read about the file (True) or
        not (False).
    rdedfann_flag : bool, optional
        Whether the function is being called by `rdedfann` or the user. If it
        is being called by the user and the file has annotations, then warn
        them that the EDF file has annotations and that they should use
        `rdedfann` instead.
    encoding : str, optional
        The encoding to use for strings in the header. Although the edf
        specification requires ascii strings, some files do not adhere to it.

    Returns
    -------
    record : dict, optional
        All of the record information needed to generate MIT formatted files.
        Only returns if 'record_only' is set to True, else generates the
        corresponding .dat and .hea files. This record file will not match the
        `rdrecord` output since it will only give us the digital signal for now.

    Notes
    -----
    The entire file is composed of (seen here: https://www.edfplus.info/specs/edf.html):

    HEADER RECORD (we suggest to also adopt the 12 simple additional EDF+ specs)
    8 ascii : version of this data format (0)
    80 ascii : local patient identification (mind item 3 of the additional EDF+ specs)
    80 ascii : local recording identification (mind item 4 of the additional EDF+ specs)
    8 ascii : startdate of recording (dd.mm.yy) (mind item 2 of the additional EDF+ specs)
    8 ascii : starttime of recording (hh.mm.ss)
    8 ascii : number of bytes in header record
    44 ascii : reserved
    8 ascii : number of data records (-1 if unknown, obey item 10 of the additional EDF+ specs)
    8 ascii : duration of a data record, in seconds
    4 ascii : number of signals (ns) in data record
    ns * 16 ascii : ns * label (e.g. EEG Fpz-Cz or Body temp) (mind item 9 of the additional EDF+ specs)
    ns * 80 ascii : ns * transducer type (e.g. AgAgCl electrode)
    ns * 8 ascii : ns * physical dimension (e.g. uV or degreeC)
    ns * 8 ascii : ns * physical minimum (e.g. -500 or 34)
    ns * 8 ascii : ns * physical maximum (e.g. 500 or 40)
    ns * 8 ascii : ns * digital minimum (e.g. -2048)
    ns * 8 ascii : ns * digital maximum (e.g. 2047)
    ns * 80 ascii : ns * prefiltering (e.g. HP:0.1Hz LP:75Hz)
    ns * 8 ascii : ns * nr of samples in each data record
    ns * 32 ascii : ns * reserved

    DATA RECORD
    nr of samples[1] * integer : first signal in the data record
    nr of samples[2] * integer : second signal
    ..
    ..
    nr of samples[ns] * integer : last signal

    Bytes   0 - 127: descriptive text
    Bytes 128 - 131: master tag (data type = matrix)
    Bytes 132 - 135: master tag (data size)
    Bytes 136 - 151: array flags (4 byte tag with data type, 4 byte
                     tag with subelement size, 8 bytes of content)
    Bytes 152 - 167: array dimension (4 byte tag with data type, 4
                     byte tag with subelement size, 8 bytes of content)
    Bytes 168 - 183: array name (4 byte tag with data type, 4 byte
                     tag with subelement size, 8 bytes of content)
    Bytes 184 - ...: array content (4 byte tag with data type, 4 byte
                     tag with subelement size, ... bytes of content)

    Examples
    --------
    >>> record = read_edf('x001_FAROS.edf',
                           pn_dir='simultaneous-measurements/raw_data')

    """
    if pn_dir is not None:
        if "." not in pn_dir:
            dir_list = pn_dir.split("/")
            pn_dir = posixpath.join(
                dir_list[0], download.get_version(dir_list[0]), *dir_list[1:]
            )

        file_url = posixpath.join(download.PN_INDEX_URL, pn_dir, record_name)
        # Currently must download file for MNE to read it though can give the
        # user the option to delete it immediately afterwards
        with _url.openurl(file_url, "rb") as f:
            open(record_name, "wb").write(f.read())

    # Open the desired file
    edf_file = open(record_name, mode="rb")

    # Version of this data format (8 bytes)
    version = struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)

    # Check to see that the input is an EDF file. (This check will detect
    # most but not all other types of files.)
    if version != "0       ":
        raise Exception(
            "Input does not appear to be EDF -- no conversion attempted"
        )
    else:
        if verbose:
            print("EDF version number: {}".format(version.strip()))

    # Local patient identification (80 bytes)
    patient_id = struct.unpack("<80s", edf_file.read(80))[0].decode(encoding)
    if verbose:
        print("Patient ID: {}".format(patient_id))

    # Local recording identification (80 bytes)
    # Bob Kemp recommends using this field to encode the start date
    # including an abbreviated month name in English and a full (4-digit)
    # year, as is done here if this information is available in the input
    # record. EDF+ requires this.
    record_id = struct.unpack("<80s", edf_file.read(80))[0].decode(encoding)
    if verbose:
        print("Recording ID: {}".format(record_id))

    # Start date of recording (dd.mm.yy) (8 bytes)
    start_date = struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)
    if verbose:
        print("Recording Date: {}".format(start_date))
    start_day, start_month, start_year = [int(i) for i in start_date.split(".")]
    # This should work for a while
    if start_year < 1970:
        start_year += 1900
    if start_year < 1970:
        start_year += 100

    # Start time of recording (hh.mm.ss) (8 bytes)
    start_time = struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)
    if verbose:
        print("Recording Time: {}".format(start_time))
    start_hour, start_minute, start_second = [
        int(i) for i in start_time.split(".")
    ]

    # Number of bytes in header (8 bytes)
    header_bytes = int(
        struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)
    )
    if verbose:
        print("Number of bytes in header record: {}".format(header_bytes))

    # Reserved (44 bytes)
    reserved_notes = (
        struct.unpack("<44s", edf_file.read(44))[0].decode(encoding).strip()
    )
    if reserved_notes[:5] == "EDF+C":
        # The file is EDF compatible and will work without issue
        # See: Bob Kemp, Jesus Olivan, European data format ‘plus’ (EDF+), an
        #      EDF alike standard format for the exchange of physiological
        #      data, Clinical Neurophysiology, Volume 114, Issue 9, 2003,
        #      Pages 1755-1761, ISSN 1388-2457
        pass
    elif reserved_notes[:5] == "EDF+D":
        raise Exception(
            "EDF+ File: interrupted data records (not currently supported)"
        )
    else:
        if verbose:
            print("Free Space: {}".format(reserved_notes))

    # Number of blocks (-1 if unknown) (8 bytes)
    num_blocks = int(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding))
    if verbose:
        print("Number of data records: {}".format(num_blocks))
    if num_blocks == -1:
        raise Exception(
            "Number of data records in unknown (not currently supported)"
        )

    # Duration of a block, in seconds (8 bytes)
    block_duration = float(
        struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)
    )
    if verbose:
        print(
            "Duration of each data record in seconds: {}".format(block_duration)
        )
    if block_duration <= 0.0:
        block_duration = 1.0

    # Number of signals (4 bytes)
    n_sig = int(struct.unpack("<4s", edf_file.read(4))[0].decode(encoding))
    if verbose:
        print("Number of signals: {}".format(n_sig))
    if n_sig < 1:
        raise Exception("Done: not any signals left to read")

    # Label (e.g., EEG FpzCz or Body temp) (16 bytes each)
    sig_name = []
    for _ in range(n_sig):
        temp_sig = (
            struct.unpack("<16s", edf_file.read(16))[0].decode(encoding).strip()
        )
        if temp_sig == "EDF Annotations" and not rdedfann_flag:
            print(
                "*** This may be an EDF+ Annotation file instead, please see "
                "the `rdedfann` function. ***"
            )
        sig_name.append(temp_sig)
    if verbose:
        print("Signal Labels: {}".format(sig_name))

    # Transducer type (e.g., AgAgCl electrode) (80 bytes each)
    transducer_types = []
    for _ in range(n_sig):
        transducer_types.append(
            struct.unpack("<80s", edf_file.read(80))[0].decode(encoding).strip()
        )
    if verbose:
        print("Transducer Types: {}".format(transducer_types))

    # Physical dimension (e.g., uV or degreeC) (8 bytes each)
    physical_dims = []
    for _ in range(n_sig):
        physical_dims.append(
            struct.unpack("<8s", edf_file.read(8))[0].decode(encoding).strip()
        )
    if verbose:
        print("Physical Dimensions: {}".format(physical_dims))

    # Physical minimum (e.g., -500 or 34) (8 bytes each)
    physical_min = np.array([])
    for _ in range(n_sig):
        physical_min = np.append(
            physical_min,
            float(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)),
        )
    if verbose:
        print("Physical Minimums: {}".format(physical_min))

    # Physical maximum (e.g., 500 or 40) (8 bytes each)
    physical_max = np.array([])
    for _ in range(n_sig):
        physical_max = np.append(
            physical_max,
            float(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)),
        )
    if verbose:
        print("Physical Maximums: {}".format(physical_max))

    # Digital minimum (e.g., -2048) (8 bytes each)
    digital_min = np.array([])
    for _ in range(n_sig):
        digital_min = np.append(
            digital_min,
            float(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)),
        )
    if verbose:
        print("Digital Minimums: {}".format(digital_min))

    # Digital maximum (e.g., 2047) (8 bytes each)
    digital_max = np.array([])
    for _ in range(n_sig):
        digital_max = np.append(
            digital_max,
            float(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)),
        )
    if verbose:
        print("Digital Maximums: {}".format(digital_max))

    # Prefiltering (e.g., HP:0.1Hz LP:75Hz) (80 bytes each)
    prefilter_info = []
    for _ in range(n_sig):
        prefilter_info.append(
            struct.unpack("<80s", edf_file.read(80))[0].decode(encoding).strip()
        )
    if verbose:
        print("Prefiltering Information: {}".format(prefilter_info))

    # Number of samples per block (8 bytes each)
    samps_per_block = []
    for _ in range(n_sig):
        samps_per_block.append(
            int(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding))
        )
    if verbose:
        print("Number of Samples per Record: {}".format(samps_per_block))

    # The last 32*nsig bytes in the header are unused
    for _ in range(n_sig):
        struct.unpack("<32s", edf_file.read(32))[0].decode(encoding)

    # Pre-process the acquired data before creating the record
    record_name_out = (
        record_name.split(os.sep)[-1].replace("-", "_").replace(".edf", "")
    )
    sample_rate = [int(i / block_duration) for i in samps_per_block]
    fs = functools.reduce(math.gcd, sample_rate)
    samps_per_frame = [int(s / min(samps_per_block)) for s in samps_per_block]
    sig_len = int(fs * num_blocks * block_duration)
    base_time = datetime.time(start_hour, start_minute, start_second)
    base_date = datetime.date(start_year, start_month, start_day)
    file_name = n_sig * [record_name_out + ".dat"]
    fmt = n_sig * ["16"]
    skew = n_sig * [None]
    byte_offset = n_sig * [None]
    adc_gain_all = (digital_max - digital_min) / (physical_max - physical_min)
    adc_gain = [float(format(a, ".12g")) for a in adc_gain_all]
    baseline = (digital_max - (physical_max * adc_gain_all) + 1).astype("int64")

    units = n_sig * [""]
    for i, f in enumerate(physical_dims):
        if f == "n/a":
            label = sig_name[i].lower().split()[0]
            if label in list(SIG_UNITS.keys()):
                units[i] = SIG_UNITS[label]
            else:
                units[i] = "n/a"
        else:
            f = f.replace("µ", "u")  # Maybe more weird symbols to check for?
            if f == "":
                units[i] = "mV"
            else:
                units[i] = f

    adc_res = [int(math.log2(f)) for f in (digital_max - digital_min)]
    adc_zero = [int(f) for f in ((digital_max + 1 + digital_min) / 2)]
    block_size = n_sig * [0]
    base_datetime = datetime.datetime(
        start_year,
        start_month,
        start_day,
        start_hour,
        start_minute,
        start_second,
    )
    base_time = datetime.time(
        base_datetime.hour, base_datetime.minute, base_datetime.second
    )
    base_date = datetime.date(
        base_datetime.year, base_datetime.month, base_datetime.day
    )

    if header_only:
        return {
            "fs": fs,
            "sig_len": sig_len,
            "n_sig": n_sig,
            "base_date": base_date,
            "base_time": base_time,
            "units": units,
            "sig_name": sig_name,
            "comments": [],
        }

    sig_data = np.empty((sig_len, n_sig))
    temp_sig_data = np.fromfile(edf_file, dtype=np.int16)
    temp_sig_data = temp_sig_data.reshape((-1, sum(samps_per_block)))
    temp_all_sigs = np.hsplit(temp_sig_data, np.cumsum(samps_per_block)[:-1])

    for i in range(n_sig):
        # Check if `samps_per_frame` has all equal values
        if samps_per_frame.count(samps_per_frame[0]) == len(samps_per_frame):
            sig_data[:, i] = (
                temp_all_sigs[i].flatten().astype(np.int64) - baseline[i]
            ) / adc_gain_all[i]
        else:
            temp_sig_data = temp_all_sigs[i].flatten()

            if samps_per_frame[i] == 1:
                sig_data[:, i] = (
                    temp_sig_data.astype(np.int64) - baseline[i]
                ) / adc_gain_all[i]
            else:
                for j in range(sig_len):
                    start_ind = j * samps_per_frame[i]
                    stop_ind = start_ind + samps_per_frame[i]
                    sig_data[j, i] = np.mean(
                        temp_sig_data[start_ind:stop_ind].astype(np.int64)
                        - baseline[i] / adc_gain_all[i]
                    )

    # This is the closest I can get to the original implementation
    # NOTE: This is done using `np.testing.assert_array_equal()`
    # Mismatched elements: 15085545 / 15400000 (98%)
    # Max absolute difference: 3.75166564e-12
    # Max relative difference: 5.41846079e-15
    #  x: array([[  -3.580728,   42.835293, -102.818048,   54.978632,  -52.354247],
    #        [  -8.340205,   43.079939, -102.106351,   56.402027,  -44.992626],
    #        [  -5.004123,   43.546991,  -99.481966,   51.64255 ,  -43.079939],...
    #  y: array([[  -3.580728,   42.835293, -102.818048,   54.978632,  -52.354247],
    #        [  -8.340205,   43.079939, -102.106351,   56.402027,  -44.992626],
    #        [  -5.004123,   43.546991,  -99.481966,   51.64255 ,  -43.079939],...

    init_value = [int(s[0, 0]) for s in temp_all_sigs]
    checksum = [
        int(np.sum(v) % 65536) for v in np.transpose(sig_data)
    ]  # not all values correct?

    edf_file.close()

    record = Record(
        record_name=record_name_out,
        n_sig=n_sig,
        fs=fs,
        samps_per_frame=samps_per_frame,
        counter_freq=None,
        base_counter=None,
        sig_len=sig_len,
        base_time=base_time,
        base_date=base_date,
        comments=[],
        sig_name=sig_name,  # Remove whitespace to make compatible later?
        p_signal=sig_data,
        d_signal=None,
        e_p_signal=None,
        e_d_signal=None,
        file_name=n_sig * [record_name_out + ".dat"],
        fmt=n_sig * ["16"],
        skew=n_sig * [None],
        byte_offset=n_sig * [None],
        adc_gain=adc_gain,
        baseline=baseline,
        units=units,
        adc_res=[int(math.log2(f)) for f in (digital_max - digital_min)],
        adc_zero=[int(f) for f in ((digital_max + 1 + digital_min) / 2)],
        init_value=init_value,
        checksum=checksum,
        block_size=n_sig * [0],
    )

    record.base_datetime = base_datetime

    return record


def wfdb_to_edf(
    record_name,
    pn_dir=None,
    sampfrom=0,
    sampto=None,
    channels=None,
    output_filename="",
    edf_plus=False,
):
    """
    These programs convert EDF (European Data Format) files into
    WFDB-compatible files (as used in PhysioNet) and vice versa. European
    Data Format (EDF) was originally designed for storage of polysomnograms.

    Note that WFDB format does not include a standard way to specify the
    transducer type or the prefiltering specification; these parameters are
    not preserved by these conversion programs. Also note that use of the
    standard signal and unit names specified for EDF is permitted but not
    enforced by `wfdb_to_edf`.

    Parameters
    ----------
    record_name : str
        The name of the input WFDB record to be read. Can also work with both
        EDF and WAV files.
    pn_dir : str, optional
        Option used to stream data from Physionet. The Physionet
        database directory from which to find the required record files.
        eg. For record '100' in 'http://physionet.org/content/mitdb'
        pn_dir='mitdb'.
    sampfrom : int, optional
        The starting sample number to read for all channels.
    sampto : int, 'end', optional
        The sample number at which to stop reading for all channels.
        Reads the entire duration by default.
    channels : list, optional
        List of integer indices specifying the channels to be read.
        Reads all channels by default.
    output_filename : str, optional
        The desired name of the output file. If this value set to the
        default value of '', then the output filename will be 'REC.edf'.
    edf_plus : bool, optional
        Whether to write the output file in EDF (False) or EDF+ (True) format.

    Returns
    -------
    N/A

    Notes
    -----
    The entire file is composed of (seen here: https://www.edfplus.info/specs/edf.html):

    HEADER RECORD (we suggest to also adopt the 12 simple additional EDF+ specs)
    8 ascii : version of this data format (0)
    80 ascii : local patient identification (mind item 3 of the additional EDF+ specs)
    80 ascii : local recording identification (mind item 4 of the additional EDF+ specs)
    8 ascii : startdate of recording (dd.mm.yy) (mind item 2 of the additional EDF+ specs)
    8 ascii : starttime of recording (hh.mm.ss)
    8 ascii : number of bytes in header record
    44 ascii : reserved
    8 ascii : number of data records (-1 if unknown, obey item 10 of the additional EDF+ specs)
    8 ascii : duration of a data record, in seconds
    4 ascii : number of signals (ns) in data record
    ns * 16 ascii : ns * label (e.g. EEG Fpz-Cz or Body temp) (mind item 9 of the additional EDF+ specs)
    ns * 80 ascii : ns * transducer type (e.g. AgAgCl electrode)
    ns * 8 ascii : ns * physical dimension (e.g. uV or degreeC)
    ns * 8 ascii : ns * physical minimum (e.g. -500 or 34)
    ns * 8 ascii : ns * physical maximum (e.g. 500 or 40)
    ns * 8 ascii : ns * digital minimum (e.g. -2048)
    ns * 8 ascii : ns * digital maximum (e.g. 2047)
    ns * 80 ascii : ns * prefiltering (e.g. HP:0.1Hz LP:75Hz)
    ns * 8 ascii : ns * nr of samples in each data record
    ns * 32 ascii : ns * reserved

    DATA RECORD
    nr of samples[1] * integer : first signal in the data record
    nr of samples[2] * integer : second signal
    ..
    ..
    nr of samples[ns] * integer : last signal

    Bytes   0 - 127: descriptive text
    Bytes 128 - 131: master tag (data type = matrix)
    Bytes 132 - 135: master tag (data size)
    Bytes 136 - 151: array flags (4 byte tag with data type, 4 byte
                     tag with subelement size, 8 bytes of content)
    Bytes 152 - 167: array dimension (4 byte tag with data type, 4
                     byte tag with subelement size, 8 bytes of content)
    Bytes 168 - 183: array name (4 byte tag with data type, 4 byte
                     tag with subelement size, 8 bytes of content)
    Bytes 184 - ...: array content (4 byte tag with data type, 4 byte
                     tag with subelement size, ... bytes of content)

    Examples
    --------
    >>> wfdb.wfdb_to_edf('100', pn_dir='pwave')

    The output file name is '100.edf'

    """
    record = rdrecord(
        record_name,
        pn_dir=pn_dir,
        sampfrom=sampfrom,
        sampto=sampto,
        smooth_frames=False,
        channels=channels,
    )
    record_name_out = record_name.split(os.sep)[-1].replace("-", "_")

    # Maximum data block length, in bytes
    edf_max_block = 61440

    # Convert to the expected month name formatting
    month_names = [
        "JAN",
        "FEB",
        "MAR",
        "APR",
        "MAY",
        "JUN",
        "JUL",
        "AUG",
        "SEP",
        "OCT",
        "NOV",
        "DEC",
    ]

    # Calculate block duration. (In the EDF spec, blocks are called "records"
    # or "data records", but this would be confusing here since "record"
    # refers to the entire recording -- so here we say "blocks".)
    samples_per_frame = sum(record.samps_per_frame)

    # i.e., The number of frames per minute, divided by 60
    frames_per_minute = record.fs * 60 + 0.5
    frames_per_second = frames_per_minute / 60

    # Ten seconds
    frames_per_block = 10 * frames_per_second + 0.5

    # EDF specifies 2 bytes per sample
    bytes_per_block = int(2 * samples_per_frame * frames_per_block)

    # Blocks would be too long -- reduce their length by a factor of 10
    while bytes_per_block > edf_max_block:
        frames_per_block /= 10
        bytes_per_block = samples_per_frame * 2 * frames_per_block

    seconds_per_block = int(frames_per_block / frames_per_second)

    if (frames_per_block < 1) and (bytes_per_block < edf_max_block / 60):
        # The number of frames/minute
        frames_per_block = frames_per_minute
        bytes_per_block = 2 * samples_per_frame * frames_per_block
        seconds_per_block = 60

    if bytes_per_block > edf_max_block:
        print(
            (
                "Can't convert record %s to EDF: EDF blocks can't be larger "
                "than {} bytes, but each input frame requires {} bytes.  Use "
                "'channels' to select a subset of the input signals or trim "
                "using 'sampfrom' and 'sampto'."
            ).format(edf_max_block, samples_per_frame * 2)
        )

    # Calculate the number of blocks to be written. The calculation rounds
    # up so that we don't lose any frames, even if the number of frames is not
    # an exact multiple of frames_per_block
    total_frames = record.sig_len
    num_blocks = int(total_frames / int(frames_per_block)) + 1

    digital_min = []
    digital_max = []
    physical_min = []
    physical_max = []
    # Calculate the physical and digital extrema
    for i in range(record.n_sig):
        # Invalid ADC resolution in input .hea file
        if record.adc_res[i] < 1:
            # Guess the ADC resolution based on format
            if record.fmt[i] == "24":
                temp_adc_res = 24
            elif record.fmt[i] == "32":
                temp_adc_res = 32
            elif record.fmt[i] == "80":
                temp_adc_res = 8
            elif record.fmt[i] == "212":
                temp_adc_res = 12
            elif (record.fmt[i] == "310") or (record.fmt[i] == "311"):
                temp_adc_res = 10
            else:
                temp_adc_res = 16
        else:
            temp_adc_res = record.adc_res[i]
        # Determine the physical and digital extrema
        digital_max.append(
            int(record.adc_zero[i] + (1 << (temp_adc_res - 1)) - 1)
        )
        digital_min.append(int(record.adc_zero[i] - (1 << (temp_adc_res - 1))))
        physical_max.append(
            (digital_max[i] - record.baseline[i]) / record.adc_gain[i]
        )
        physical_min.append(
            (digital_min[i] - record.baseline[i]) / record.adc_gain[i]
        )

    # The maximum record name length to write is 80 bytes
    if len(record_name_out) > 80:
        record_name_write = record_name_out[:79] + "\0"
    else:
        record_name_write = record_name_out

    # The maximum seconds per block length to write is 8 bytes
    if len(str(seconds_per_block)) > 8:
        seconds_per_block_write = seconds_per_block[:7] + "\0"
    else:
        seconds_per_block_write = seconds_per_block

    # The maximum signal name length to write is 16 bytes
    sig_name_write = len(record.sig_name) * []
    for s in record.sig_name:
        if len(s) > 16:
            sig_name_write.append(s[:15] + "\0")
        else:
            sig_name_write.append(s)

    # The maximum units length to write is 8 bytes
    units_write = len(record.units) * []
    for s in record.units:
        if len(s) > 8:
            units_write.append(s[:7] + "\0")
        else:
            units_write.append(s)

    # Configure the output datetime
    if hasattr("record", "base_datetime"):
        start_second = int(record.base_datetime.second)
        start_minute = int(record.base_datetime.minute)
        start_hour = int(record.base_datetime.hour)
        start_day = int(record.base_datetime.day)
        start_month = int(record.base_datetime.month)
        start_year = int(record.base_datetime.year)
    else:
        # Set date to start of EDF epoch
        start_second = 0
        start_minute = 0
        start_hour = 0
        start_day = 1
        start_month = 1
        start_year = 1985

    # Determine the number of bytes in the header
    header_bytes = 256 * (record.n_sig + 1)

    # Determine the number of samples per data record
    samps_per_record = []
    for spf in record.samps_per_frame:
        samps_per_record.append(int(frames_per_block) * spf)

    # Determine the output data
    # NOTE: The output data will be close (+-1) but not equal due to the
    #       inappropriate rounding done by record.adc()
    #       For example...
    #           Mismatched elements: 862881 / 24168000 (3.57%)
    #           Max absolute difference: 1
    #           Max relative difference: 0.0212766
    #            x: array([ 53, -28,  14, ..., 884, 898, 898], dtype=int16)
    #            y: array([ 53, -28,  14, ..., 884, 898, 898], dtype=int16)
    if record.e_p_signal is not None:
        temp_data = record.adc(expanded=True)
    else:
        temp_data = record.adc()
        temp_data = [v for v in np.transpose(temp_data)]

    out_data = []
    for i in range(record.sig_len):
        for j, sig in enumerate(temp_data):
            ind_start = i * samps_per_record[j]
            ind_stop = (i + 1) * samps_per_record[j]
            out_data.extend(sig[ind_start:ind_stop].tolist())
    out_data = np.array(out_data, dtype=np.int16)

    # Start writing the file
    if output_filename == "":
        output_filename = record_name_out + ".edf"

    with open(output_filename, "wb") as f:
        print(
            "Converting record {} to {} ({} mode)".format(
                record_name, output_filename, "EDF+" if edf_plus else "EDF"
            )
        )

        # Version of this data format (8 bytes)
        f.write(struct.pack("<8s", b"0").replace(b"\x00", b"\x20"))

        # Local patient identification (80 bytes)
        f.write(
            struct.pack(
                "<80s", "{}".format(record_name_write).encode("ascii")
            ).replace(b"\x00", b"\x20")
        )

        # Local recording identification (80 bytes)
        # Bob Kemp recommends using this field to encode the start date
        # including an abbreviated month name in English and a full (4-digit)
        # year, as is done here if this information is available in the input
        # record. EDF+ requires this.
        if hasattr("record", "base_datetime"):
            f.write(
                struct.pack(
                    "<80s",
                    "Startdate {}-{}-{}".format(
                        start_day, month_names[start_month - 1], start_year
                    ).encode("ascii"),
                ).replace(b"\x00", b"\x20")
            )
        else:
            f.write(
                struct.pack("<80s", b"Startdate not recorded").replace(
                    b"\x00", b"\x20"
                )
            )
        if edf_plus:
            print("WARNING: EDF+ requires start date (not specified)")

        # Start date of recording (dd.mm.yy) (8 bytes)
        f.write(
            struct.pack(
                "<8s",
                "{:02d}.{:02d}.{:02d}".format(
                    start_day, start_month, start_year % 100
                ).encode("ascii"),
            ).replace(b"\x00", b"\x20")
        )

        # Start time of recording (hh.mm.ss) (8 bytes)
        f.write(
            struct.pack(
                "<8s",
                "{:02d}.{:02d}.{:02d}".format(
                    start_hour, start_minute, start_second
                ).encode("ascii"),
            ).replace(b"\x00", b"\x20")
        )

        # Number of bytes in header (8 bytes)
        f.write(
            struct.pack(
                "<8s", "{:d}".format(header_bytes).encode("ascii")
            ).replace(b"\x00", b"\x20")
        )

        # Reserved (44 bytes)
        if edf_plus:
            f.write(struct.pack("<44s", b"EDF+C").replace(b"\x00", b"\x20"))
        else:
            f.write(struct.pack("<44s", b"").replace(b"\x00", b"\x20"))

        # Number of blocks (-1 if unknown) (8 bytes)
        f.write(
            struct.pack(
                "<8s", "{:d}".format(num_blocks).encode("ascii")
            ).replace(b"\x00", b"\x20")
        )

        # Duration of a block, in seconds (8 bytes)
        f.write(
            struct.pack(
                "<8s", "{:g}".format(seconds_per_block_write).encode("ascii")
            ).replace(b"\x00", b"\x20")
        )

        # Number of signals (4 bytes)
        f.write(
            struct.pack(
                "<4s", "{:d}".format(record.n_sig).encode("ascii")
            ).replace(b"\x00", b"\x20")
        )

        # Label (e.g., EEG FpzCz or Body temp) (16 bytes each)
        for i in sig_name_write:
            f.write(
                struct.pack("<16s", "{}".format(i).encode("ascii")).replace(
                    b"\x00", b"\x20"
                )
            )

        # Transducer type (e.g., AgAgCl electrode) (80 bytes each)
        for _ in range(record.n_sig):
            f.write(
                struct.pack("<80s", b"transducer type not recorded").replace(
                    b"\x00", b"\x20"
                )
            )

        # Physical dimension (e.g., uV or degreeC) (8 bytes each)
        for i in units_write:
            f.write(
                struct.pack("<8s", "{}".format(i).encode("ascii")).replace(
                    b"\x00", b"\x20"
                )
            )

        # Physical minimum (e.g., -500 or 34) (8 bytes each)
        for pmin in physical_min:
            f.write(
                struct.pack("<8s", "{:g}".format(pmin).encode("ascii")).replace(
                    b"\x00", b"\x20"
                )
            )

        # Physical maximum (e.g., 500 or 40) (8 bytes each)
        for pmax in physical_max:
            f.write(
                struct.pack("<8s", "{:g}".format(pmax).encode("ascii")).replace(
                    b"\x00", b"\x20"
                )
            )

        # Digital minimum (e.g., -2048) (8 bytes each)
        for dmin in digital_min:
            f.write(
                struct.pack("<8s", "{:d}".format(dmin).encode("ascii")).replace(
                    b"\x00", b"\x20"
                )
            )

        # Digital maximum (e.g., 2047) (8 bytes each)
        for dmax in digital_max:
            f.write(
                struct.pack("<8s", "{:d}".format(dmax).encode("ascii")).replace(
                    b"\x00", b"\x20"
                )
            )

        # Prefiltering (e.g., HP:0.1Hz LP:75Hz) (80 bytes each)
        for _ in range(record.n_sig):
            f.write(
                struct.pack("<80s", b"prefiltering not recorded").replace(
                    b"\x00", b"\x20"
                )
            )

        # Number of samples per block (8 bytes each)
        for spr in samps_per_record:
            f.write(
                struct.pack("<8s", "{:d}".format(spr).encode("ascii")).replace(
                    b"\x00", b"\x20"
                )
            )

        # The last 32*nsig bytes in the header are unused
        for _ in range(record.n_sig):
            f.write(struct.pack("<32s", b"").replace(b"\x00", b"\x20"))

        # Write the data blocks
        out_data.tofile(f, format="%d")

        # Add the buffer
        correct_bytes = num_blocks * sum(samps_per_record)
        current_bytes = len(out_data)
        num_to_write = correct_bytes - current_bytes
        for i in range(num_to_write):
            f.write(b"\x00\x80")

    print("Header block size: {:d} bytes".format((record.n_sig + 1) * 256))
    print(
        "Data block size: {:g} seconds ({:d} frames or {:d} bytes)".format(
            seconds_per_block, int(frames_per_block), int(bytes_per_block)
        )
    )
    print(
        "Recording length: {:d} ({:d} data blocks, {:d} frames, {:d} bytes)".format(
            sum(
                [
                    num_blocks,
                    num_blocks * int(frames_per_block),
                    num_blocks * bytes_per_block,
                ]
            ),
            num_blocks,
            num_blocks * int(frames_per_block),
            num_blocks * bytes_per_block,
        )
    )
    print(
        "Total length of file to be written: {:d} bytes".format(
            int((record.n_sig + 1) * 256 + num_blocks * bytes_per_block)
        )
    )

    if edf_plus:
        print(
            (
                "WARNING: EDF+ requires the subject's gender, birthdate, and name, as "
                "well as additional information about the recording that is not usually "
                "available. This information is not saved in the output file even if "
                "available. EDF+ also requires the use of standard names for signals and "
                "for physical units;  these requirements are not enforced by this program. "
                "To make the output file fully EDF+ compliant, its header must be edited "
                "manually."
            )
        )

        if "EDF-Annotations" not in record.sig_name:
            print(
                "WARNING: The output file does not include EDF annotations, which are required for EDF+."
            )

    # Check that all characters in the header are valid (printable ASCII
    # between 32 and 126 inclusive). Note that this test does not prevent
    # generation of files containing invalid characters; it merely warns
    # the user if this has happened.
    header_test = open(output_filename, "rb").read((record.n_sig + 1) * 256)
    for i, val in enumerate(header_test):
        if (val < 32) or (val > 126):
            print(
                "WARNING: output contains an invalid character, {}, at byte {}".format(
                    val, i
                )
            )


def rdedfann(
    record_name,
    pn_dir=None,
    delete_file=True,
    info_only=True,
    record_only=False,
    verbose=False,
    encoding="iso8859-1",
):
    """
    This program returns the annotation information from an EDF+ file
    containing annotations (with the signal name given as 'EDF Annotations').
    The information that is returned if `info_only` is set to True is:
        {
            'onset_time': list of %H:%M:%S.fff strings denoting the annotation
                          onset times,
            'sample_num': list of integers denoting the annotation onset
                          sample numbers,
            'comment': list of comments (`aux_note`) for the annotations,
            'duration': list of floats denoting the duration of the event
        }
    Else, this function will return either the WFDB Annotation format of the
    information of the file if `record_only` is set to True, or nothing if
    neither are set to True though a WFDB Annotation file will be created.

    Parameters
    ----------
    record_name : str
        The name of the input EDF record to be read.
    pn_dir : str, optional
        Option used to stream data from Physionet. The Physionet
        database directory from which to find the required record files.
        eg. For record '100' in 'http://physionet.org/content/mitdb'
        pn_dir='mitdb'.
    delete_file : bool, optional
        Whether to delete the saved EDF file (False) or not (True)
        after being imported.
    info_only : bool, optional
        Return, strictly, the information contained in the file as formatted
        by the original WFDB package. Must not be True if `record_only` is
        True.
    record_only : bool, optional
        Whether to only return the annotation information (True) or not
        (False). If False, this function will generate a WFDB-formatted
        annotation file. If True, it will return the object returned if that
        file were read with `rdann`. Must not be True if `info_only` is True.
    verbose : bool, optional
        Whether to print all the information read about the file (True) or
        not (False).
    encoding : str, optional
        The encoding to use for strings in the header. Although the edf
        specification requires ascii strings, some files do not adhere to it.

    Returns
    -------
    N/A : dict, Annotation, optional
        If 'info_only' is set to True, return all of the annotation
        information needed to generate WFDB-formatted annotation files.
        If 'record_only' is set to True, return the WFDB-formatted annotation
        object generated by the `rdann` output. If none are set to True, write
        the WFDB-formatted annotation file.

    Notes
    -----
    The entire file is composed of (seen here:
    https://www.edfplus.info/specs/edfplus.html#edfplusannotations):

    HEADER RECORD (we suggest to also adopt the 12 simple additional EDF+ specs)
    8 ascii : version of this data format (0)
    80 ascii : local patient identification (mind item 3 of the additional EDF+ specs)
    80 ascii : local recording identification (mind item 4 of the additional EDF+ specs)
    8 ascii : startdate of recording (dd.mm.yy) (mind item 2 of the additional EDF+ specs)
    8 ascii : starttime of recording (hh.mm.ss)
    8 ascii : number of bytes in header record
    44 ascii : reserved
    8 ascii : number of data records (-1 if unknown, obey item 10 of the additional EDF+ specs)
    8 ascii : duration of a data record, in seconds
    4 ascii : number of signals (ns) in data record
    ns * 16 ascii : ns * label (must be 'EDF Annotations')
    ns * 80 ascii : ns * transducer type (must be whitespace)
    ns * 8 ascii : ns * physical dimension (must be whitespace)
    ns * 8 ascii : ns * physical minimum (e.g. -500 or 34, different than physical maximum)
    ns * 8 ascii : ns * physical maximum (e.g. 500 or 40, different than physical minimum)
    ns * 8 ascii : ns * digital minimum (must be -32768)
    ns * 8 ascii : ns * digital maximum (must be 32767)
    ns * 80 ascii : ns * prefiltering (must be whitespace)
    ns * 8 ascii : ns * nr of samples in each data record
    ns * 32 ascii : ns * reserved

    ANNOTATION RECORD

    Examples
    --------
    >>> ann_info = wfdb.rdedfann('sample-data/test_edfann.edf')

    """
    # Some preliminary checks
    if info_only and record_only:
        raise Exception(
            "Both `info_only` and `record_only` are set. Only one "
            "can be set at a time."
        )

    # According to the EDF+ docs:
    #   "The coding is EDF compatible in the sense that old EDF software would
    #    simply treat this 'EDF Annotations' signal as if it were a (strange-
    #    looking) ordinary signal"
    rec = read_edf(
        record_name,
        pn_dir=pn_dir,
        delete_file=delete_file,
        record_only=True,
        rdedfann_flag=True,
    )

    # Convert from array of integers to ASCII strings
    annotation_string = ""
    for chunk in rec.p_signal.flatten().astype(np.int64):
        if chunk + 1 == 0:
            continue
        else:
            adjusted_hex = hex(
                struct.unpack("<H", struct.pack(">H", chunk + 1))[0]
            )
            annotation_string += bytes.fromhex(adjusted_hex[2:]).decode(
                encoding
            )
            # Remove all of the whitespace
            for rep in ["\x00", "\x14", "\x15"]:
                annotation_string = annotation_string.replace(rep, " ")

    # Parse the resulting annotation string
    onsets = []
    onset_times = []
    sample_nums = []
    comments = []
    durations = []
    all_anns = annotation_string.split("+")
    for ann in all_anns:
        if ann == "":
            continue
        try:
            ann_split = ann.strip().split(" ")
            onset = float(ann_split[0])
            hours, rem = divmod(onset, 3600)
            minutes, seconds = divmod(rem, 60)
            onset_time = f"{hours:02.0f}:{minutes:02.0f}:{seconds:06.3f}"
            sample_num = int(onset * rec.sig_len)
            duration = float(ann_split[1])
            comment = " ".join(ann_split[2:])
            if verbose:
                print(
                    f"{onset_time}\t{sample_num}\t{comment}\t\tduration: {duration}"
                )
            onsets.append(onset)
            onset_times.append(onset_time)
            sample_nums.append(sample_num)
            comments.append(comment)
            durations.append(duration)
        except IndexError:
            continue

    if info_only:
        return {
            "onset_time": onset_times,
            "sample_num": sample_nums,
            "comment": comments,
            "duration": durations,
        }
    else:
        df_in = pd.DataFrame(
            data={
                "onset": onsets,
                "duration": durations,
                "description": comments,
            }
        )
        df_out = format_ann_from_df(df_in)
        # Remove extension from input file name
        record_name = record_name.split(os.sep)[-1].split(".")[0]
        extension = "atr"
        fs = rec.fs
        sample = (df_out["onset"].to_numpy() * fs).astype(np.int64)
        # Assume each annotation is a comment
        symbol = ['"'] * len(df_out.index)
        subtype = np.array([22] * len(df_out.index))
        # Assume each annotation belongs with the 1st channel
        chan = np.array([0] * len(df_out.index))
        num = np.array([0] * len(df_out.index))
        aux_note = df_out["description"].tolist()

        if record_only:
            return Annotation(
                record_name=record_name,
                extension=extension,
                sample=sample,
                symbol=symbol,
                subtype=subtype,
                chan=chan,
                num=num,
                aux_note=aux_note,
                fs=fs,
            )
        else:
            wrann(
                record_name,
                extension,
                sample=sample,
                symbol=symbol,
                subtype=subtype,
                chan=chan,
                num=num,
                aux_note=aux_note,
                fs=fs,
            )