Data storage handler¶

This data handler use [PyTables](https://www.pytables.org/) that is built on top of the HDF5 library, using the Python language and the NumPy package. It features an object-oriented interface that, combined with C extensions for the performance-critical parts of the code (generated using Cython), makes it a fast, yet extremely easy to use tool for interactively browse, process and search very large amounts of data. One important feature of [PyTables](https://www.pytables.org/) is that it optimizes memory and disk resources so that data takes much less space (specially if on-flight compression is used) than other solutions such as relational or object oriented databases.

For examples and descriptions refers to documentation: Data storage handler

class openbci_stream.utils.hdf5.HDF5Reader(filename: str)[source]¶

Objects created with HDF5Writer can be opened with HDF5Reader.

This class support export to other formmats like MNE epochs and EDF.

Parameters: filename – Path with the location of the hdf file.

property annotations: list¶

A list of annotations.

The HDF5Writer write the annotations with timestamps, but EDF needs the relative time from start in seconds.

property aux: numpy.ndarray¶: The AUX data of the hdf file in the shape of (aux, time).

property aux_timestamp: List[openbci_stream.utils.hdf5.timesamp]¶: A list of timestamps for EEG data.

property classes¶: A list with the same length of EEG with markers as numbers.

property classes_indexes: Dict[str, int]¶: The standard for classes and indexes.

property eeg: numpy.ndarray¶: The EEG data of the hdf file in the shape of (channels, time).

get_data(tmax: int, tmin: Optional[int] = 0, ref=None, markers: Union[None, List[str]] = None, eeg=None, preprocess=None, **kwargs) → Tuple[numpy.ndarray][source]¶

Create an EpochsArray object with the MNE library.

This method auto crop the data in regard to markers also will drop channels that no correspond with the montage. For an example of use refer to `Data storage handler - MNE objects<../notebooks/07-data_storage_handler.html#MNE-objects>`_

Parameters

duration – The duration of the trial.
tmin – The time to take previous to the marker.
markers – A filter of markers for crop the signal.
kwargs – Optional arguments passed to EpochsArray

Returns

trials – Dataset with the shape (trials, channels, time)
classes – List of classes

get_epochs(tmax: int, tmin: Optional[int] = 0, ref=None, markers: Union[None, List[str]] = None, preprocess=None, eeg=None, **kwargs) → mne.EpochsArray[source]¶

Create an EpochsArray object with the MNE library.

This method auto crop the data in regard to markers also will drop channels that no correspond with the montage. For an example of use refer to `Data storage handler - MNE objects<../notebooks/07-data_storage_handler.html#MNE-objects>`_

Parameters

duration – The duration of the trial, in seconds.
tmin – The time to take previous to the marker, in seconds.
markers – A filter of markers for crop the signal.
kwargs –
Optional arguments passed to EpochsArray

Returns

An MNE Epochs object.

Return type

epochs

property header: Dict[str, Any]¶: The header of the hdf file.

property markers: Dict[str, List[openbci_stream.utils.hdf5.timesamp]]¶: A dictionary with the markers and timestamps as values.

property sample_id: numpy.ndarray¶: The EEG data of the hdf file in the shape of (channels, time).

property timestamp: List[openbci_stream.utils.hdf5.timesamp]¶: A list of timestamps for EEG data.

to_edf(filename: str, eeg=None) → None[source]¶: Export to EDF file.

class openbci_stream.utils.hdf5.HDF5Writer(filename: str)[source]¶

This HDF5 data handler was pre-configured for the architecture of acquired EEG data.

This module can be used like an instance e.g.

>>> writer = HDF5Writer('file.h5')
>>> writer.add_marker('LEFT', datetime.now().timestamp())
>>> writer.close()

or can be used with the with control-flow structure e.g.

>>> with HDF5Writer('file.h5') as write:
        writer.add_marker('LEFT', datetime.now().timestamp())

Parameters: filename – Path where the edf file will be created.

add_annotation(onset: openbci_stream.utils.hdf5.timesamp, duration: int = 0, description: str = '') → None[source]¶

Add EDF annotations to the hdf5 file.

These annotations will be exported with EDF file and follow the format defined by pyedflib.

There is some difference between markers and annotations:

Markers are writed as time series.

Annotations are writed as a list of events.

Markers are mainly for repetitions of the same event.

Annotations can describe a complex event with a custom duration and long description, e.g artifacts.

Parameters

onset – Timestamp for annotation.
duration – The duration of the event.
description – The description of the annotation.

add_aux(aux_data: numpy.ndarray, timestamp: numpy.ndarray) → None[source]¶

Write AUX data into the hdf5 file.

The shape of aux data cannot be changed after the first write.

Parameters

aux_data – OpenBCI aux data defined in board modes
timestamp – The timestamp for this data.

add_aux_timestamp(timestamp: openbci_stream.utils.hdf5.timesamp) → None[source]¶

Add a list of timestamps to the hdf5 file.

The use of this method is not recommended, instead, it must be used add_aux that includes a validation.

add_eeg(eeg_data: numpy.ndarray, timestamp: numpy.ndarray) → None[source]¶

Add EEG data to hdf5 file, optionally adds timestamps.

The first time this method is called the number of channels of EEG is configured, and cannot be changed.

Parameters

eeg_data – An array of shape (channels, time)
timestamp – The timestamp for this data.

add_header(header: Dict[str, Any], host: Optional[str] = None) → None[source]¶

Set the header for hdf5 file.

A header is basically a dictionary with all kinds of useful information. There are required keys for some specific methods.

MNE objects requiere:

montage: str with montage name, e.g. ‘standard_1020’.
channels: dict with keys as channel index and values as channel name, e.g {1: ‘FP1’, 2: ‘FP2’, 3: ‘F7’}.
sample_rate: int sample rate for acuiered signal, e.g `1000
channels_by_board: list fo ints with the number of channels generated by each board (if multiple boards has been used).

EDF objects requiere (In addition to the above):

admincode: str with the admincode.
birthdate: date object with the the birthdate of the patient.
equipment: str thats describes the measurement equpipment.
gender: int with the the gender, 1 is male, 0 is female.
patientcode: str with the patient code.
patientname: str with the patient name.
patient_additional: str with the additional patient information.
recording_additional: str wit the additional recording information.
technician: str with the technicians name.

add_marker(marker: Any, timestamp: openbci_stream.utils.hdf5.timesamp) → None[source]¶

Add a pair of marker-timestamp to the hdf5 file.

There is some difference between markers and annotations:

Markers are writed as time series.

Annotations are writed as a list of events.

Markers are mainly for repetitions of the same event.

Annotations can describe a complex event with a custom duration and long description, e.g artifacts.

add_markers(markers: Dict[str, List[openbci_stream.utils.hdf5.timesamp]]) → None[source]¶

Add a set of markers to the hdf5 file.

This method is used to write a set of markers at the same time, works with a dictionary object, with keys as marker and values as a list of timestamps

Example

>>> markes = {'LEFT': [1603898187.226709,
                       1603898197.226709,
                       1603898207.226709],
              'RIGHT': [1603898192.226709,
                        1603898202.226709,
                        1603898212.226709]
              }
>>> add_markers(markers)

add_sampleid(aux_data: numpy.ndarray) → None[source]¶

Write AUX data into the hdf5 file.

The shape of aux data cannot be changed after the first write.

Parameters

aux_data –
OpenBCI aux data defined in board modes
timestamp – The timestamp for this data.

add_timestamp(timestamp: openbci_stream.utils.hdf5.timesamp) → None[source]¶

Add a list of timestamps to the hdf5 file.

The use of this method is not recommended, instead, it must be used add_eeg that includes a validation.

close() → None[source]¶

Close the file handler.

Before to close, add some extra values into the header.

openbci_stream.utils.hdf5.interpolate_datetime(timestamp: List[openbci_stream.utils.hdf5.timesamp], length: Optional[int] = None) → List[openbci_stream.utils.hdf5.timesamp][source]¶

Interpolate uncomplete timestamp list.

The input timestamp list must be a list of timestamps separated by zeros, this script will complete the missing timestamps. This primary purpose is to complete the list generated from stream data when only it has a sample rate and acquired times.

Parameters

timestamp – An array with timestamps and zeros.
length – The length of the final timestamp array, if not defined then will be the same size of the input timestamp.

Returns

An array with interpolated timestamps.

Return type

timestamp

openbci_stream.utils.hdf5.np2json_serializer(obj)[source]¶: hdf5 handler needs Python classic data types.