<commit_before>"""Conversion tool from Brain Vision EEG to FIF"""

# Authors: <NAME> <<EMAIL>>
#          <NAME> <<EMAIL>>
#
# License: BSD (3-clause)

import os
import time
import re
import warnings
from ...externals.six import StringIO, u
from ...externals.six.moves import configparser

import numpy as np

from ...coreg import get_ras_to_neuromag_trans, read_elp
from ...transforms import als_ras_trans, apply_trans
from ...utils import verbose, logger
from .. import pick_types
from ..constants import FIFF
from ..meas_info import Info
from ..raw import Raw


class RawBrainVision(Raw):
    """Raw object from Brain Vision EEG file

    Parameters
    ----------
    vdhr_fname : str
        Path to the EEG header file.
    elp_fname : str | None
        Path to the elp file containing electrode positions.
        If None, sensor locations are (0,0,0).
    elp_names : list | None
        A list of channel names in the same order as the points in the elp
        file. Electrode positions should be specified with the same names as
        in the vhdr file, and fiducials should be specified as "lpa" "nasion",
        "rpa". ELP positions with other names are ignored. If elp_names is not
        None and channels are missing, a KeyError is raised.
    preload : bool
        If True, all data are loaded at initialization.
        If False, data are not read until save.
    fix_ptb_events : bool
        Fix events produced by psychtoolbox and recorded through a stim
        tracker (KIT setup).
    verbose : bool, str, int, or None
        If not None, override default verbose level (see mne.verbose).

    See Also
    --------
    mne.fiff.Raw : Documentation of attribute and methods.
    """
    @verbose
    def __init__(self, vhdr_fname, elp_fname=None, elp_names=None,
                 preload=False, fix_ptb_events=False, verbose=None):
        logger.info('Extracting eeg Parameters from %s...' % vhdr_fname)
        vhdr_fname = os.path.abspath(vhdr_fname)
        self.info, self._eeg_info, self._events = _get_eeg_info(vhdr_fname,
                                                                elp_fname,
                                                                elp_names,
                                                                fix_ptb_events)
        logger.info('Creating Raw.info structure...')

        # Raw attributes
        self.verbose = verbose
        self._preloaded = False
        self.fids = list()
        self._projector = None
        self.comp = None  # no compensation for EEG
        self.proj = False
        self.first_samp = 0
        f = open(self.info['file_id'])
        f.seek(0, os.SEEK_END)
        n_samples = f.tell()
        dtype = int(self._eeg_info['dtype'][-1])
        n_chan = self.info['nchan']
        self.last_samp = (n_samples // (dtype * (n_chan - 1))) - 1

        if preload:
            self._preloaded = preload
            logger.info('Reading raw data from %s...' % vhdr_fname)
            self._data, _ = self._read_segment()
            assert len(self._data) == self.info['nchan']

            # Add time info
            self._times = np.arange(self.first_samp, self.last_samp + 1,
                                    dtype=np.float64)
            self._times /= self.info['sfreq']
            logger.info('    Range : %d ... %d =  %9.3f ... %9.3f secs'
                        % (self.first_samp, self.last_samp,
                           float(self.first_samp) / self.info['sfreq'],
                           float(self.last_samp) / self.info['sfreq']))
        logger.info('Ready.')

    def __repr__(self):
        n_chan = self.info['nchan']
        data_range = self.last_samp - self.first_samp + 1
        s = ('%r' % os.path.basename(self.info['file_id']),
             "n_channels x n_times : %s x %s" % (n_chan, data_range))
        return "<RawEEG  |  %s>" % ', '.join(s)

    def _read_segment(self, start=0, stop=None, sel=None, verbose=None,
                      projector=None):
        """Read a chunk of raw data

        Parameters
        ----------
        start : int, (optional)
            first sample to include (first is 0). If omitted, defaults to the
            first sample in data.
        stop : int, (optional)
            First sample to not include.
            If omitted, data is included to the end.
        sel : array, optional
            Indices of channels to select.
        projector : array
            SSP operator to apply to the data.
        verbose : bool, str, int, or None
            If not None, override default verbose level (see mne.verbose).

        Returns
        -------
        data : array, [channels x samples]
           the data matrix (channels x samples).
        times : array, [samples]
            returns the time values corresponding to the samples.
        """
        if sel is not None:
            if len(sel) == 1 and sel[0] == 0 and start == 0 and stop == 1:
                return (666, 666)
        if projector is not None:
            raise NotImplementedError('Currently does not handle projections.')
        if stop is None:
            stop = self.last_samp + 1
        elif stop > self.last_samp + 1:
            stop = self.last_samp + 1

        #  Initial checks
        start = int(start)
        stop = int(stop)

        eeg_info = self._eeg_info
        sfreq = self.info['sfreq']
        n_chan = self.info['nchan']
        cals = np.array([chan_info['cal'] for chan_info in self.info['chs']])
        mults = np.array([chan_info['unit_mul'] for chan_info
                          in self.info['chs']])
        picks = pick_types(self.info, meg=False, eeg=True, exclude=[])
        n_eeg = picks.size
        cals = np.atleast_2d(cals[picks])
        mults = np.atleast_2d(mults[picks])

        if start >= stop:
            raise ValueError('No data in this range')

        logger.info('Reading %d ... %d  =  %9.3f ... %9.3f secs...' %
                    (start, stop - 1, start / float(sfreq),
                     (stop - 1) / float(sfreq)))

        with open(self.info['file_id'], 'rb') as f:
            buffer_size = (stop - start)
            pointer = start * n_chan
            f.seek(pointer)
            # extract data
            data = np.fromfile(f, dtype=eeg_info['dtype'],
                               count=buffer_size * n_eeg)
        if eeg_info['data_orientation'] == 'MULTIPLEXED':
            data = data.reshape((n_eeg, -1), order='F')
        elif eeg_info['data_orientation'] == 'VECTORIZED':
            data = data.reshape((n_eeg, -1), order='C')

        gains = cals * mults
        data = data * gains.T

        if len(self._events):
            stim_channel = _synthesize_stim_channel(self._events, start, stop)
            data = np.vstack((data, stim_channel))

        if sel is not None:
            data = data[sel]

        logger.info('[done]')
        times = np.arange(start, stop, dtype=float) / sfreq

        return data, times


def _read_vmrk_events(fname):
    """Read events from a vmrk file<commit_msg>FIX:  initial file offset when reading from brainvision eeg<commit_after>"""Conversion tool from Brain Vision EEG to FIF"""

# Authors: <NAME> <<EMAIL>>
#          <NAME> <<EMAIL>>
#
# License: BSD (3-clause)

import os
import time
import re
import warnings
from ...externals.six import StringIO, u
from ...externals.six.moves import configparser

import numpy as np

from ...coreg import get_ras_to_neuromag_trans, read_elp
from ...transforms import als_ras_trans, apply_trans
from ...utils import verbose, logger
from .. import pick_types
from ..constants import FIFF
from ..meas_info import Info
from ..raw import Raw


class RawBrainVision(Raw):
    """Raw object from Brain Vision EEG file

    Parameters
    ----------
    vdhr_fname : str
        Path to the EEG header file.
    elp_fname : str | None
        Path to the elp file containing electrode positions.
        If None, sensor locations are (0,0,0).
    elp_names : list | None
        A list of channel names in the same order as the points in the elp
        file. Electrode positions should be specified with the same names as
        in the vhdr file, and fiducials should be specified as "lpa" "nasion",
        "rpa". ELP positions with other names are ignored. If elp_names is not
        None and channels are missing, a KeyError is raised.
    preload : bool
        If True, all data are loaded at initialization.
        If False, data are not read until save.
    fix_ptb_events : bool
        Fix events produced by psychtoolbox and recorded through a stim
        tracker (KIT setup).
    verbose : bool, str, int, or None
        If not None, override default verbose level (see mne.verbose).

    See Also
    --------
    mne.fiff.Raw : Documentation of attribute and methods.
    """
    @verbose
    def __init__(self, vhdr_fname, elp_fname=None, elp_names=None,
                 preload=False, fix_ptb_events=False, verbose=None):
        logger.info('Extracting eeg Parameters from %s...' % vhdr_fname)
        vhdr_fname = os.path.abspath(vhdr_fname)
        self.info, self._eeg_info, self._events = _get_eeg_info(vhdr_fname,
                                                                elp_fname,
                                                                elp_names,
                                                                fix_ptb_events)
        logger.info('Creating Raw.info structure...')

        # Raw attributes
        self.verbose = verbose
        self._preloaded = False
        self.fids = list()
        self._projector = None
        self.comp = None  # no compensation for EEG
        self.proj = False
        self.first_samp = 0
        f = open(self.info['file_id'])
        f.seek(0, os.SEEK_END)
        n_samples = f.tell()
        dtype = int(self._eeg_info['dtype'][-1])
        n_chan = self.info['nchan']
        self.last_samp = (n_samples // (dtype * (n_chan - 1))) - 1

        if preload:
            self._preloaded = preload
            logger.info('Reading raw data from %s...' % vhdr_fname)
            self._data, _ = self._read_segment()
            assert len(self._data) == self.info['nchan']

            # Add time info
            self._times = np.arange(self.first_samp, self.last_samp + 1,
                                    dtype=np.float64)
            self._times /= self.info['sfreq']
            logger.info('    Range : %d ... %d =  %9.3f ... %9.3f secs'
                        % (self.first_samp, self.last_samp,
                           float(self.first_samp) / self.info['sfreq'],
                           float(self.last_samp) / self.info['sfreq']))
        logger.info('Ready.')

    def __repr__(self):
        n_chan = self.info['nchan']
        data_range = self.last_samp - self.first_samp + 1
        s = ('%r' % os.path.basename(self.info['file_id']),
             "n_channels x n_times : %s x %s" % (n_chan, data_range))
        return "<RawEEG  |  %s>" % ', '.join(s)

    def _read_segment(self, start=0, stop=None, sel=None, verbose=None,
                      projector=None):
        """Read a chunk of raw data

        Parameters
        ----------
        start : int, (optional)
            first sample to include (first is 0). If omitted, defaults to the
            first sample in data.
        stop : int, (optional)
            First sample to not include.
            If omitted, data is included to the end.
        sel : array, optional
            Indices of channels to select.
        projector : array
            SSP operator to apply to the data.
        verbose : bool, str, int, or None
            If not None, override default verbose level (see mne.verbose).

        Returns
        -------
        data : array, [channels x samples]
           the data matrix (channels x samples).
        times : array, [samples]
            returns the time values corresponding to the samples.
        """
        if sel is not None:
            if len(sel) == 1 and sel[0] == 0 and start == 0 and stop == 1:
                return (666, 666)
        if projector is not None:
            raise NotImplementedError('Currently does not handle projections.')
        if stop is None:
            stop = self.last_samp + 1
        elif stop > self.last_samp + 1:
            stop = self.last_samp + 1

        #  Initial checks
        start = int(start)
        stop = int(stop)

        eeg_info = self._eeg_info
        sfreq = self.info['sfreq']
        n_chan = self.info['nchan']
        cals = np.array([chan_info['cal'] for chan_info in self.info['chs']])
        mults = np.array([chan_info['unit_mul'] for chan_info
                          in self.info['chs']])
        picks = pick_types(self.info, meg=False, eeg=True, exclude=[])
        n_eeg = picks.size
        cals = np.atleast_2d(cals[picks])
        mults = np.atleast_2d(mults[picks])

        if start >= stop:
            raise ValueError('No data in this range')

        logger.info('Reading %d ... %d  =  %9.3f ... %9.3f secs...' %
                    (start, stop - 1, start / float(sfreq),
                     (stop - 1) / float(sfreq)))

        dtype = np.dtype(eeg_info['dtype'])
        buffer_size = (stop - start)
        pointer = start * n_eeg * dtype.itemsize
        with open(self.info['file_id'], 'rb') as f:
            f.seek(pointer)
            # extract data
            data = np.fromfile(f, dtype=dtype, count=buffer_size * n_eeg)
        if eeg_info['data_orientation'] == 'MULTIPLEXED':
            data = data.reshape((n_eeg, -1), order='F')
        elif eeg_info['data_orientation'] == 'VECTORIZED':
            data = data.reshape((n_eeg, -1), order='C')

        gains = cals * mults
        data = data * gains.T

        if len(self._events):
            stim_channel = _synthesize_stim_channel(self._events, start, stop)
            data = np.vstack((data, stim_channel))

        if sel is not None:
            data = data[sel]

        logger.info('[done]')
        times = np.arange(start, stop, dtype=float) / sfreq

        return data, times


def _read_vmrk_events(fname):
    """Read events from a vmrk file
{'fraction_non_alphanumeric': 0.10786596119929454, 'fraction_numerical': 0.0060670194003527335, 'mean_word_length': 1.7806983130639467, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7539438', 'n_tokens_mistral': 4302, 'n_tokens_neox': 4132, 'n_words': 1258}