Writer

openseize.file_io.edf.Writer

Bases: bases.Writer

A writer of European Data Format (EDF/EDF+) files.

This Writer is a context manager for writing EEG data and metadata to an EDF binary file. Unlike Readers it must be opened under the context management protocol. Importantly, this writer does not currently support writing annotations to an EDF file.

Attributes:

Name	Type	Description
path	Path	A python path instance to target file to write data to.

Examples:

>>> from openseize.demos import paths
>>> filepath = paths.locate('recording_001.edf')
>>> # Create a reader that will read only channels [0, 1]
>>> # and write out these channels to a new file
>>> writepath = paths.data_dir.joinpath('subset_001.edf')
>>> with Reader(filepath) as reader:
>>>     with Writer(writepath) as writer:
>>>         writer.write(reader.header, reader, channels=[0, 1])

Source code in openseize/file_io/edf.py

class Writer(bases.Writer):
    """A writer of European Data Format (EDF/EDF+) files.

    This Writer is a context manager for writing EEG data and metadata to an
    EDF binary file. Unlike Readers it must be opened under the context
    management protocol. Importantly, this writer does not currently support
    writing annotations to an EDF file.

    Attributes:
        path (Path):
            A python path instance to target file to write data to.

    Examples:
        >>> from openseize.demos import paths
        >>> filepath = paths.locate('recording_001.edf')
        >>> # Create a reader that will read only channels [0, 1]
        >>> # and write out these channels to a new file
        >>> writepath = paths.data_dir.joinpath('subset_001.edf')
        >>> with Reader(filepath) as reader:
        >>>     with Writer(writepath) as writer:
        >>>         writer.write(reader.header, reader, channels=[0, 1])
    """

    def __init__(self, path: Union[str, Path]) -> None:
        """Initialize this Writer. See base class for further details."""

        super().__init__(path, mode='wb')

    def _write_header(self, header: Header) -> None:
        """Writes a dict of EDF header metadata to this Writer's opened
        file.

        Args:
            header:
                A dict of EDF compliant metadata. Please see Header for
                further details.
        """

        # the header should be added during write not initialization
        # pylint: disable-next=attribute-defined-outside-init
        self.header = header
        bytemap = header.bytemap(header.num_signals)

        # Move to file start and write each ascii encoded byte string
        self._fobj.seek(0)
        for items, (nbytes, _) in zip(header.values(), bytemap.values()):
            items = [items] if not isinstance(items, list) else items

            for item, nbyte in zip(items, nbytes):
                bytestr = bytes(str(item), encoding='ascii').ljust(nbyte)
                self._fobj.write(bytestr)

    def _records(self,
                 data: Union[np.ndarray, Reader],
                 channels: Sequence[int]
    ) -> Generator[List[np.ndarray], None, None]:
        """Yields 1-D arrays, one per channel, to write to a data record.

        Args:
            data:
                An 2D array, memmap or Reader instance with samples along
                the last axis.
            channels:
                A sequence of channels to write to each data record.

        Yields:
            A list of single row 2-D arrays of samples for a single channel
            for a single data record.
        """

        for n in range(self.header.num_records):
            result = []
            # The number of samples per record is channel dependent if
            # sample rates are not equal across channels.
            starts = n * np.array(self.header.samples_per_record)
            stops = (n+1) * np.array(self.header.samples_per_record)

            for channel, start, stop in zip(channels, starts, stops):
                if isinstance(data, np.ndarray):
                    x = np.atleast_2d(data[channel][start:stop])
                    result.append(x)
                    #result.append(data[channel][start:stop])
                else:
                    data.channels = [channel]
                    result.append(data.read(start, stop))

            yield result

    def _encipher(self, arrs: Sequence):
        """Converts float arrays to 2-byte little-endian integer arrays
        using the EDF specification.

        Args:
            arrs:
                A sequence of 1-D arrays of float dtype.

        Returns:
            A sequence of 1-D arrays in 2-byte little-endian format.
        """

        slopes = self.header.slopes
        offsets = self.header.offsets
        results = []
        for ch, x in enumerate(arrs):
            arr = np.rint((x - offsets[ch]) / slopes[ch])
            arr = arr.astype('<i2')
            results.append(arr)
        return results

    def _validate(self, header: Header, data: np.ndarray) -> None:
        """Ensures the number of samples is divisible by the number of
        records.

        EDF files must have an integer number of records (i.e. the number of
        samples must fill the records). This may require appending 0's to
        the end of data to ensure this.

        Args:
            header:
                A Header instance of EDF metadata.
            data:
                The 2-D data with samples along the last axis.

        Raises:
            A ValueError if not divisible.
        """

        if data.shape[1] % header.num_records != 0:
            msg=('Number of data samples must be divisible by '
                 'the number of records; {} % {} != 0')
            msg.format(data.shape[1], header.num_records)
            raise ValueError(msg)

    def _progress(self, record_idx: int) -> None:
        """Relays write progress during file writing."""

        msg = 'Writing data: {:.1f}% complete'
        perc = record_idx / self.header.num_records * 100
        print(msg.format(perc), end='\r', flush=True)

    # override of general abstract method requires setting specific args
    # pylint: disable-next=arguments-differ
    def write(self,
              header: Header,
              data: Union[np.ndarray, Reader],
              channels: Sequence[int],
              verbose: bool = True,
              ) -> None:
        """Write header metadata and data for channel in channels to this
        Writer's file instance.

        Args:
            header:
                A mapping of EDF compliant fields and values.
            data:
                An array with shape (channels, samples) or Reader instance.
            channels:
                Channel indices to write to this Writer's open file.
            verbose:
                An option to print progress of write.

        Raises:
            ValueErrror: An error occurs if samples to be written is not
                         divisible by the number of records in the Header
                         instance.
        """

        header = Header.from_dict(header)
        header = header.filter(channels)
        self._validate(header, data)

        self._write_header(header) #and store header to instance
        self._fobj.seek(header.header_bytes)
        for idx, record in enumerate(self._records(data, channels)):
            samples = self._encipher(record) # floats to '<i2'
            samples = np.concatenate(samples, axis=1)
            #concatenate data bytes to str and write
            byte_str = samples.tobytes()
            self._fobj.write(byte_str)
            if verbose:
                self._progress(idx)

__init__(path)

Initialize this Writer. See base class for further details.

Source code in openseize/file_io/edf.py

def __init__(self, path: Union[str, Path]) -> None:
    """Initialize this Writer. See base class for further details."""

    super().__init__(path, mode='wb')

write(header, data, channels, verbose=True)

Write header metadata and data for channel in channels to this Writer's file instance.

Parameters:

Name	Type	Description	Default
header	Header	A mapping of EDF compliant fields and values.	required
data	Union[np.ndarray, Reader]	An array with shape (channels, samples) or Reader instance.	required
channels	Sequence[int]	Channel indices to write to this Writer's open file.	required
verbose	bool	An option to print progress of write.	True

Raises:

Type	Description
ValueErrror	An error occurs if samples to be written is not divisible by the number of records in the Header instance.

Source code in openseize/file_io/edf.py

def write(self,
          header: Header,
          data: Union[np.ndarray, Reader],
          channels: Sequence[int],
          verbose: bool = True,
          ) -> None:
    """Write header metadata and data for channel in channels to this
    Writer's file instance.

    Args:
        header:
            A mapping of EDF compliant fields and values.
        data:
            An array with shape (channels, samples) or Reader instance.
        channels:
            Channel indices to write to this Writer's open file.
        verbose:
            An option to print progress of write.

    Raises:
        ValueErrror: An error occurs if samples to be written is not
                     divisible by the number of records in the Header
                     instance.
    """

    header = Header.from_dict(header)
    header = header.filter(channels)
    self._validate(header, data)

    self._write_header(header) #and store header to instance
    self._fobj.seek(header.header_bytes)
    for idx, record in enumerate(self._records(data, channels)):
        samples = self._encipher(record) # floats to '<i2'
        samples = np.concatenate(samples, axis=1)
        #concatenate data bytes to str and write
        byte_str = samples.tobytes()
        self._fobj.write(byte_str)
        if verbose:
            self._progress(idx)

Bases and Mixins

Writer Base

Bases: abc.ABC, mixins.ViewInstance

Abstract base class for all writers of EEG data.

This ABC defines all EEG writers as context managers that write data to a file path. Inheritors must override the 'write' method.

Attributes:

Name	Type	Description
path		Python path instance to an EEG data file.
mode		A str mode for writing the eeg file. Must be 'r' for plain text files and 'rb' for binary files.