How to use pyEDFlib - 10 common examples

# if self.isActivated:
            #     self.fevent.write('#Streaming restarted at ' + self.time_stamp + '\n')
            # self.fevent.write('#Streaming started at ' + self.time_stamp + '\n')
            if self.writeEDF:

                EDF_format_extention = ".edf"
                EDF_format_filetype = pyedflib.FILETYPE_EDFPLUS
                temp_filterStringFileIndicator = "_prefiltered"
                temp_filterStringHeader = 'HP ' + str(self.prefilterEDF_hp) + ' Hz'
                if self.prefilterEDF_hp is None:
                    EDF_format_extention = ".bdf"
                    EDF_format_filetype = pyedflib.FILETYPE_BDFPLUS
                    temp_filterStringFileIndicator = "_unfiltered"
                    temp_filterStringHeader = 'none'

                self.edfWriter = pyedflib.EdfWriter(self.path_and_file_name + "_missing_samples_corrected" + temp_filterStringFileIndicator + EDF_format_extention, self.nChannels+3, file_type=EDF_format_filetype)

                """
                 Only when the number of annotations you want to write is more than the number of seconds of the duration of the recording, you can use this function to increase the storage space for annotations */
                /* Minimum is 1, maximum is 64 */
                """
                if self.writeEDFAnnotations:
                    self.edfWriter.set_number_of_annotation_signals(self.edfAnnotationChannels) #7*60 = 420 annotations per minute on average
                channel_info = {'label': 'ch', 'dimension': 'uV', 'sample_rate': int(round(self.fs)),
                                'physical_max': self.EDF_Physical_max_microVolt, 'physical_min': self.EDF_Physical_min_microVolt,
                                'digital_max': 32767, 'digital_min': -32767,
                                'prefilter': temp_filterStringHeader, 'transducer': 'none'}
                channel_info_accel = {'label': 'acc', 'dimension': 'G', 'sample_rate': int(round(self.fs)),
                                      'physical_max': 4, 'physical_min': -4,
                                      'digital_max': 32767, 'digital_min': -32767,
                                      'prefilter': 'none', 'transducer': 'none'}

if self.doRealTimeStreaming:
            self.path_and_file_name_event = self.path_and_file_name_event
            print "Will export Events CSV to:", self.path_and_file_name_event + '.csv'
            # Open in append mode
            self.fevent = open(self.path_and_file_name_event + '.csv', 'a', buffering=100000)
            if self.exportEDFAnnotations:
                print "Will export Events from EDF to CSV in:", self.path_and_file_name_event + '.edf' + '.csv'
                self.feventEDF = open(self.path_and_file_name_event + '.edf' + '.csv', 'a', buffering=100000)
                self.feventEDF.write('onset' + self.delim + 'annotation' + '\n')
            # if self.isActivated:
            #     self.fevent.write('#Streaming restarted at ' + self.time_stamp + '\n')
            # self.fevent.write('#Streaming started at ' + self.time_stamp + '\n')
            if self.writeEDF:

                EDF_format_extention = ".edf"
                EDF_format_filetype = pyedflib.FILETYPE_EDFPLUS
                temp_filterStringFileIndicator = "_prefiltered"
                temp_filterStringHeader = 'HP ' + str(self.prefilterEDF_hp) + ' Hz'
                if self.prefilterEDF_hp is None:
                    EDF_format_extention = ".bdf"
                    EDF_format_filetype = pyedflib.FILETYPE_BDFPLUS
                    temp_filterStringFileIndicator = "_unfiltered"
                    temp_filterStringHeader = 'none'

                self.edfWriter = pyedflib.EdfWriter(self.path_and_file_name + "_missing_samples_corrected" + temp_filterStringFileIndicator + EDF_format_extention, self.nChannels+3, file_type=EDF_format_filetype)

                """
                 Only when the number of annotations you want to write is more than the number of seconds of the duration of the recording, you can use this function to increase the storage space for annotations */
                /* Minimum is 1, maximum is 64 */
                """
                if self.writeEDFAnnotations:
                    self.edfWriter.set_number_of_annotation_signals(self.edfAnnotationChannels) #7*60 = 420 annotations per minute on average

if self.exportEDFAnnotations:
                print "Will export Events from EDF to CSV in:", self.path_and_file_name_event + '.edf' + '.csv'
                self.feventEDF = open(self.path_and_file_name_event + '.edf' + '.csv', 'a', buffering=100000)
                self.feventEDF.write('onset' + self.delim + 'annotation' + '\n')
            # if self.isActivated:
            #     self.fevent.write('#Streaming restarted at ' + self.time_stamp + '\n')
            # self.fevent.write('#Streaming started at ' + self.time_stamp + '\n')
            if self.writeEDF:

                EDF_format_extention = ".edf"
                EDF_format_filetype = pyedflib.FILETYPE_EDFPLUS
                temp_filterStringFileIndicator = "_prefiltered"
                temp_filterStringHeader = 'HP ' + str(self.prefilterEDF_hp) + ' Hz'
                if self.prefilterEDF_hp is None:
                    EDF_format_extention = ".bdf"
                    EDF_format_filetype = pyedflib.FILETYPE_BDFPLUS
                    temp_filterStringFileIndicator = "_unfiltered"
                    temp_filterStringHeader = 'none'

                self.edfWriter = pyedflib.EdfWriter(self.path_and_file_name + "_missing_samples_corrected" + temp_filterStringFileIndicator + EDF_format_extention, self.nChannels+3, file_type=EDF_format_filetype)

                """
                 Only when the number of annotations you want to write is more than the number of seconds of the duration of the recording, you can use this function to increase the storage space for annotations */
                /* Minimum is 1, maximum is 64 */
                """
                if self.writeEDFAnnotations:
                    self.edfWriter.set_number_of_annotation_signals(self.edfAnnotationChannels) #7*60 = 420 annotations per minute on average
                channel_info = {'label': 'ch', 'dimension': 'uV', 'sample_rate': int(round(self.fs)),
                                'physical_max': self.EDF_Physical_max_microVolt, 'physical_min': self.EDF_Physical_min_microVolt,
                                'digital_max': 32767, 'digital_min': -32767,
                                'prefilter': temp_filterStringHeader, 'transducer': 'none'}
                channel_info_accel = {'label': 'acc', 'dimension': 'G', 'sample_rate': int(round(self.fs)),

def loadEDF(self):
        if not self.edf:

            try:
                self.edf = pyedflib.EdfReader(self.edf_pathname)
            except OSError as osErr:
                print("OSError:", "Loading", self.edf_pathname)
                raise (osErr)

        for ch in self.channels:  # ['C3','C4','O1','O2','EOG-L','EOG-R','EMG','A1','A2']
            myprint('Loading', ch)
            if isinstance(self.channels_used[ch], int):

                self.loaded_channels[ch] = self.edf.readSignal(self.channels_used[ch])
                if self.edf.getPhysicalDimension(self.channels_used[ch]).lower() == 'mv':
                    myprint('mv')
                    self.loaded_channels[ch] *= 1e3
                elif self.edf.getPhysicalDimension(self.channels_used[ch]).lower() == 'v':
                    myprint('v')
                    self.loaded_channels[ch] *= 1e6

def read_pyedf(edf_file, ch_nrs=None, ch_names=None, digital=False, verbose=True):
    """
    Reads an EDF file using pyEDFlib and returns the data, header and signalheaders
    
    :param edf_file: link to an edf file
    :param ch_nrs: The numbers of channels to read
    :param ch_names: The names of channels to read
    :returns: signals, signal_headers, header
    """      
    assert (ch_nrs is  None) or (ch_names is None), 'names xor numbers should be supplied'
    if ch_nrs is not None and not isinstance(ch_nrs, list): ch_nrs = [ch_nrs]
    if ch_names is not None and not isinstance(ch_names, list): ch_names = [ch_names]

    with pyedflib.EdfReader(edf_file) as f:
        # see which channels we want to load
        channels = [ch.upper() for ch in f.getSignalLabels()]
        if ch_names is not None:
            ch_nrs = []
            for ch in ch_names:
                if not ch.upper() in channels:
                    warnings.warn('{} is not in source file (contains {})'.format(ch.upper(), channels))
                    print('will be ignored.')
                else:    
                    ch_nrs.append(channels.index(ch.upper()))
        if ch_nrs is None: # no numbers means we load all
            n = f.signals_in_file
            ch_nrs = np.arange(n)
        # load headers, signal information and 
        header = f.getHeader()
        signal_headers = [f.getSignalHeaders()[c] for c in ch_nrs]

def read_edfrecord(filename, channel, shhs='1'):

    # SHHS1 is at 125 Hz, SHHS2 at 128Hz

    assert channel=='EEG' or channel=='EEG(sec)', "Channel must be EEG or EEG(sec)"
    f = pyedflib.EdfReader(filename)
    print("Startdatetime: ", f.getStartdatetime())
    signal_labels = f.getSignalLabels()
    assert channel in signal_labels
    idx_chan = [i for i, x in enumerate(signal_labels) if x==channel][0]
    sigbuf = np.zeros((f.getNSamples()[idx_chan]))
    sigbuf[:] = f.readSignal(idx_chan)
    samplingrate = len(sigbuf) / f.file_duration
    print("sampling rate: ", samplingrate)
    if shhs=='1':
        print("30s * 125Hz divides signal length ?: ", len(sigbuf)%(30*125)==0)
        assert samplingrate == 125., "Sampling rate is not 125Hz on this record"
    else:
        print("30s * 128Hz divides signal length ?: ", len(sigbuf)%(30*128)==0)
        assert samplingrate == 128., "Sampling rate is not 128Hz on this record"
    return sigbuf

def loadHeader(self):
        if not self.edf:
            print(self.edf_pathname)
            self.edf = pyedflib.EdfReader(self.edf_pathname)

        signal_labels = self.edf.getSignalLabels()
        return signal_labels

-------
    f : EdfReader object
       object containing the handle to the file

    Examples
    --------
    >>> import pyedflib.data
    >>> f = pyedflib.data.test_generator()
    >>> f.signals_in_file == 11
    True
    >>> f._close()
    >>> del f

    """
    fname = os.path.join(os.path.dirname(__file__), 'test_generator.edf')
    f = pyedflib.EdfReader(fname)
    return f

def add_view_from_file(self, file: Path, panel_index: int=None):
        # TODO: both edf and xdf should be moved to tracking!!!
        if file.suffix == '.edf':
            import pyedflib
            with pyedflib.EdfReader(str(file)) as f:
                labels = f.getSignalLabels()
                for label in labels:
                    index = labels.index(label)
                    wav = tracking.Wave(f.readSignal(index), f.getSampleFrequency(index))
                    wav.label = label
                    wav.path = file.with_name(file.stem + '-' + label + '.wav')
                    wav.min = f.getPhysicalMinimum(index)
                    wav.max = f.getPhysicalMaximum(index)
                    wav.unit = f.getPhysicalDimension(index)
                    self.add_view(wav, panel_index=panel_index, y_min=wav.min, y_max=wav.max)
        elif file.suffix == '.xdf':
            import openxdf
            xdf = openxdf.OpenXDF(file)
            signals = openxdf.Signal(xdf, file.with_suffix('.nkamp'))
            # TODO: automate this, why are the xdf.header names different from signals.list_channels?
            for label in ['ECG', 'Chin']:

#    4    pulse 2           100 uV    1 Hz    256 Hz
#    5    pulse 3           100 uV    1 Hz    217 Hz
#    6    noise             100 uV    - Hz    200 Hz
#    7    sine 1 Hz         100 uV    1 Hz    200 Hz
#    8    sine 8 Hz         100 uV    8 Hz    200 Hz
#    9    sine 8.1777 Hz    100 uV    8.25 Hz 200 Hz
#    10    sine 8.5 Hz       100 uV    8.5Hz   200 Hz
#    11    sine 15 Hz        100 uV   15 Hz    200 Hz
#    12    sine 17 Hz        100 uV   17 Hz    200 Hz
#    13    sine 50 Hz        100 uV   50 Hz    200 Hz


if __name__ == '__main__':
    test_data_file = os.path.join('.', 'test_generator2.edf')
    file_duration = 600
    f = pyedflib.EdfWriter(test_data_file, 13,
                           file_type=pyedflib.FILETYPE_EDFPLUS)
    channel_info = []
    data_list = []

    ch_dict = {'label': 'squarewave', 'dimension': 'uV', 'sample_rate': 200, 'physical_max': 100, 'physical_min': -100, 'digital_max': 32767, 'digital_min': -32768, 'transducer': '', 'prefilter':''}
    channel_info.append(ch_dict)
    time = np.linspace(0, file_duration, file_duration*200)
    xtemp = np.sin(2*np.pi*0.1*time)
    x1 = xtemp.copy()
    x1[np.where(xtemp > 0)[0]] = 100
    x1[np.where(xtemp < 0)[0]] = -100
    data_list.append(x1)

    ch_dict = {'label': 'ramp', 'dimension': 'uV', 'sample_rate': 200, 'physical_max': 100, 'physical_min': -100, 'digital_max': 32767, 'digital_min': -32768, 'transducer': '', 'prefilter':''}
    channel_info.append(ch_dict)
    time = np.linspace(0, file_duration, file_duration*200)