<commit_before># -*- coding: utf-8 -*-
"""
Class for reading data from pCLAMP and AxoScope
files (.abf version 1 and 2), developed by Molecular device/Axon technologies.

- abf = Axon binary file
- atf is a text file based format from axon that could be
  read by AsciiIO (but this file is less efficient.)


This code is a port of abfload and abf2load
written in Matlab (BSD-2-Clause licence) by :
 - Copyright (c) 2009, <NAME>, <EMAIL>
 - Copyright (c) 2004, <NAME>
and available here:
http://www.mathworks.com/matlabcentral/fileexchange/22114-abf2load

Information on abf 1 and 2 formats is available here:
http://www.moleculardevices.com/pages/software/developer_info.html

This file supports the old (ABF1) and new (ABF2) format.
ABF1 (clampfit <=9) and ABF2 (clampfit >10)

All possible mode are possible :
    - event-driven variable-length mode 1 -> return several Segments per Block
    - event-driven fixed-length mode 2 or 5 -> return several Segments
    - gap free mode -> return one (or sevral) Segment in the Block

Supported : Read

Author: <NAME>, <NAME>

Note: <EMAIL> has a C++ library with SWIG bindings which also
reads abf files - would be good to cross-check

"""
from __future__ import print_function, division, absolute_import
# from __future__ import unicode_literals is not compatible with numpy.dtype both py2 py3

from .baserawio import (BaseRawIO, _signal_channel_dtype, _unit_channel_dtype,
                        _event_channel_dtype)

import numpy as np

import struct
import datetime
import os
from io import open, BufferedReader

import numpy as np


class AxonRawIO(BaseRawIO):
    extensions = ['abf']
    rawmode = 'one-file'

    def __init__(self, filename=''):
        BaseRawIO.__init__(self)
        self.filename = filename

    def _parse_header(self):
        info = self._axon_info = parse_axon_soup(self.filename)

        version = info['fFileVersionNumber']

        # file format
        if info['nDataFormat'] == 0:
            sig_dtype = np.dtype('i2')
        elif info['nDataFormat'] == 1:
            sig_dtype = np.dtype('f4')

        if version < 2.:
            nbchannel = info['nADCNumChannels']
            head_offset = info['lDataSectionPtr'] * BLOCKSIZE + info[
                                                'nNumPointsIgnored'] * sig_dtype.itemsize
            totalsize = info['lActualAcqLength']
        elif version >= 2.:
            nbchannel = info['sections']['ADCSection']['llNumEntries']
            head_offset = info['sections']['DataSection'][
                              'uBlockIndex'] * BLOCKSIZE
            totalsize = info['sections']['DataSection']['llNumEntries']

        self._raw_data = np.memmap(self.filename, dtype=sig_dtype, mode='r',
                                   shape=(totalsize,), offset=head_offset)

        # 3 possible modes
        if version < 2.:
            mode = info['nOperationMode']
        elif version >= 2.:
            mode = info['protocol']['nOperationMode']

        assert mode in [1, 2, 3, 5], 'Mode {} is not supported'.formagt(mode)
        # event-driven variable-length mode (mode 1)
        # event-driven fixed-length mode (mode 2 or 5)
        # gap free mode (mode 3) can be in several episodes

        # read sweep pos
        if version < 2.:
            nbepisod = info['lSynchArraySize']
            offset_episode = info['lSynchArrayPtr'] * BLOCKSIZE
        elif version >= 2.:
            nbepisod = info['sections']['SynchArraySection'][
                'llNumEntries']
            offset_episode = info['sections']['SynchArraySection'][
                                 'uBlockIndex'] * BLOCKSIZE
        if nbepisod > 0:
            episode_array = np.memmap(
                self.filename, [('offset', 'i4'), ('len', 'i4')], 'r',
                shape=nbepisod, offset=offset_episode)
        else:
            episode_array = np.empty(1, [('offset', 'i4'), ('len', 'i4')])
            episode_array[0]['len'] = self._raw_data.size
            episode_array[0]['offset'] = 0

        # sampling_rate
        if version < 2.:
            self._sampling_rate = 1. / (info['fADCSampleInterval'] *
                                        nbchannel * 1.e-6)
        elif version >= 2.:
            self._sampling_rate = 1.e6 / \
                                  info['protocol']['fADCSequenceInterval']

        # one sweep = one segment
        nb_segment = episode_array.size

        # Get raw data by segment
        self._raw_signals = {}
        self._t_starts = {}
        pos = 0
        for seg_index in range(nb_segment):
            length = episode_array[seg_index]['len']

            if version < 2.:
                fSynchTimeUnit = info['fSynchTimeUnit']
            elif version >= 2.:
                fSynchTimeUnit = info['protocol']['fSynchTimeUnit']

            if (fSynchTimeUnit != 0) and (mode == 1):
                length /= fSynchTimeUnit

            self._raw_signals[seg_index] = self._raw_data[pos:pos + length].reshape(-1, nbchannel)
            pos += length

            t_start = float(episode_array[seg_index]['offset'])
            if (fSynchTimeUnit == 0):
                t_start = t_start / self._sampling_rate
            else:
                t_start = t_start * fSynchTimeUnit * 1e-6
            self._t_starts[seg_index] = t_start

        # Create channel header
        if version < 2.:
            channel_ids = [chan_num for chan_num in
                           info['nADCSamplingSeq'] if chan_num >= 0]
        else:
            channel_ids = list(range(nbchannel))

        sig_channels = []
        adc_nums = []
        for chan_index, chan_id in enumerate(channel_ids):
            if version < 2.:
                name = info['sADCChannelName'][chan_id].replace(b' ', b'')
                units = info['sADCUnits'][chan_id].replace(b'\xb5', b'u'). \
                    replace(b' ', b'').decode('utf-8')  # \xb5 is µ
                adc_num = info['nADCPtoLChannelMap'][chan_id]
            elif version >= 2.:
                ADCInfo = info['listADCInfo'][chan_id]
                name = ADCInfo['ADCChNames'].replace(b' ', b'')
                units = ADCInfo['ADCChUnits'].replace(b'\xb5', b'u'). \
                    replace(b' ', b'').decode('utf-8')
                adc_num = ADCInfo['nADCNum']
            adc_nums.append(adc_num)

            if info['nDataFormat'] == 0:
                # int16 gain/offset
                if version < 2.:
                    gain = info['fADCRange']
                    gain /= info['fInstrumentScaleFactor'][chan_id]
                    gain /= info['fSignalGain'][chan_id]
                    gain /= info['fADCProgrammableGain'][chan_id]
                    gain /= info['lADCResolution']
                    if info['nTelegraphEnable'][chan_id] == 1:
                        gain /= info['fTelegraphAdditGain'][chan_id]
                    offset = info['fInstrumentOffset'][chan_id]
                    offset -= info['fSignalOffset'][chan_id]
                elif version >= 2.:
                    gain = info['protocol']['fADCRange']
                    gain /= info['listADCInfo'][chan_id]['fInstrumentScaleFactor']
                    gain /= info['listADCInfo'][chan_id]['fSignalGain']
                    gain /= info['listADCInfo'][chan_id]['fADCProgrammableGain']
                    gain /= info['protocol']['lADCResolution']
                    if info['listADCInfo'][chan_id]['nTelegraphEnable']:
                        gain /= info['listADCInfo'][chan_id]['fTelegraphAdditGain']
                    offset = info['listADCInfo'][chan_id]['fInstrumentOffset']
                    offset -= info['listADCInfo'][chan_id]['fSignalOffset']
            else:
                gain, offset = 1., 0.
            group_id = 0
            sig_channels.append((name, chan_id, self._sampling_rate,
                                 sig_dtype, units, gain, offset, group_id))

        sig_channels = np.array(sig_channels, dtype=_signal_channel_dtype)

        # only one events channel : tag
        if mode in [3, 5]:  # TODO check if tags exits in other mode
            # In ABF timstamps are not attached too any particular segment
            # so each segment acess all event
            timestamps = []
            labels = []
            comments = []
            for i, tag in enumerate(info['listTag']):
                timestamps.append(tag['lTagTime'])<commit_msg>warn if buggy nTelegraphEnable detected<commit_after># -*- coding: utf-8 -*-
"""
Class for reading data from pCLAMP and AxoScope
files (.abf version 1 and 2), developed by Molecular device/Axon technologies.

- abf = Axon binary file
- atf is a text file based format from axon that could be
  read by AsciiIO (but this file is less efficient.)


This code is a port of abfload and abf2load
written in Matlab (BSD-2-Clause licence) by :
 - Copyright (c) 2009, <NAME>, <EMAIL>
 - Copyright (c) 2004, <NAME>
and available here:
http://www.mathworks.com/matlabcentral/fileexchange/22114-abf2load

Information on abf 1 and 2 formats is available here:
http://www.moleculardevices.com/pages/software/developer_info.html

This file supports the old (ABF1) and new (ABF2) format.
ABF1 (clampfit <=9) and ABF2 (clampfit >10)

All possible mode are possible :
    - event-driven variable-length mode 1 -> return several Segments per Block
    - event-driven fixed-length mode 2 or 5 -> return several Segments
    - gap free mode -> return one (or sevral) Segment in the Block

Supported : Read

Author: <NAME>, <NAME>

Note: j.s.nowacki<EMAIL> has a C++ library with SWIG bindings which also
reads abf files - would be good to cross-check

"""
from __future__ import print_function, division, absolute_import
# from __future__ import unicode_literals is not compatible with numpy.dtype both py2 py3

from .baserawio import (BaseRawIO, _signal_channel_dtype, _unit_channel_dtype,
                        _event_channel_dtype)

import numpy as np

import struct
import datetime
import os
from io import open, BufferedReader

import numpy as np


class AxonRawIO(BaseRawIO):
    extensions = ['abf']
    rawmode = 'one-file'

    def __init__(self, filename=''):
        BaseRawIO.__init__(self)
        self.filename = filename

    def _parse_header(self):
        info = self._axon_info = parse_axon_soup(self.filename)

        version = info['fFileVersionNumber']

        # file format
        if info['nDataFormat'] == 0:
            sig_dtype = np.dtype('i2')
        elif info['nDataFormat'] == 1:
            sig_dtype = np.dtype('f4')

        if version < 2.:
            nbchannel = info['nADCNumChannels']
            head_offset = info['lDataSectionPtr'] * BLOCKSIZE + info[
                                                'nNumPointsIgnored'] * sig_dtype.itemsize
            totalsize = info['lActualAcqLength']
        elif version >= 2.:
            nbchannel = info['sections']['ADCSection']['llNumEntries']
            head_offset = info['sections']['DataSection'][
                              'uBlockIndex'] * BLOCKSIZE
            totalsize = info['sections']['DataSection']['llNumEntries']

        self._raw_data = np.memmap(self.filename, dtype=sig_dtype, mode='r',
                                   shape=(totalsize,), offset=head_offset)

        # 3 possible modes
        if version < 2.:
            mode = info['nOperationMode']
        elif version >= 2.:
            mode = info['protocol']['nOperationMode']

        assert mode in [1, 2, 3, 5], 'Mode {} is not supported'.formagt(mode)
        # event-driven variable-length mode (mode 1)
        # event-driven fixed-length mode (mode 2 or 5)
        # gap free mode (mode 3) can be in several episodes

        # read sweep pos
        if version < 2.:
            nbepisod = info['lSynchArraySize']
            offset_episode = info['lSynchArrayPtr'] * BLOCKSIZE
        elif version >= 2.:
            nbepisod = info['sections']['SynchArraySection'][
                'llNumEntries']
            offset_episode = info['sections']['SynchArraySection'][
                                 'uBlockIndex'] * BLOCKSIZE
        if nbepisod > 0:
            episode_array = np.memmap(
                self.filename, [('offset', 'i4'), ('len', 'i4')], 'r',
                shape=nbepisod, offset=offset_episode)
        else:
            episode_array = np.empty(1, [('offset', 'i4'), ('len', 'i4')])
            episode_array[0]['len'] = self._raw_data.size
            episode_array[0]['offset'] = 0

        # sampling_rate
        if version < 2.:
            self._sampling_rate = 1. / (info['fADCSampleInterval'] *
                                        nbchannel * 1.e-6)
        elif version >= 2.:
            self._sampling_rate = 1.e6 / \
                                  info['protocol']['fADCSequenceInterval']

        # one sweep = one segment
        nb_segment = episode_array.size

        # Get raw data by segment
        self._raw_signals = {}
        self._t_starts = {}
        pos = 0
        for seg_index in range(nb_segment):
            length = episode_array[seg_index]['len']

            if version < 2.:
                fSynchTimeUnit = info['fSynchTimeUnit']
            elif version >= 2.:
                fSynchTimeUnit = info['protocol']['fSynchTimeUnit']

            if (fSynchTimeUnit != 0) and (mode == 1):
                length /= fSynchTimeUnit

            self._raw_signals[seg_index] = self._raw_data[pos:pos + length].reshape(-1, nbchannel)
            pos += length

            t_start = float(episode_array[seg_index]['offset'])
            if (fSynchTimeUnit == 0):
                t_start = t_start / self._sampling_rate
            else:
                t_start = t_start * fSynchTimeUnit * 1e-6
            self._t_starts[seg_index] = t_start

        # Create channel header
        if version < 2.:
            channel_ids = [chan_num for chan_num in
                           info['nADCSamplingSeq'] if chan_num >= 0]
        else:
            channel_ids = list(range(nbchannel))

        sig_channels = []
        adc_nums = []
        for chan_index, chan_id in enumerate(channel_ids):
            if version < 2.:
                name = info['sADCChannelName'][chan_id].replace(b' ', b'')
                units = info['sADCUnits'][chan_id].replace(b'\xb5', b'u'). \
                    replace(b' ', b'').decode('utf-8')  # \xb5 is µ
                adc_num = info['nADCPtoLChannelMap'][chan_id]
            elif version >= 2.:
                ADCInfo = info['listADCInfo'][chan_id]
                name = ADCInfo['ADCChNames'].replace(b' ', b'')
                units = ADCInfo['ADCChUnits'].replace(b'\xb5', b'u'). \
                    replace(b' ', b'').decode('utf-8')
                adc_num = ADCInfo['nADCNum']
            adc_nums.append(adc_num)

            if info['nDataFormat'] == 0:
                # int16 gain/offset
                if version < 2.:
                    gain = info['fADCRange']
                    gain /= info['fInstrumentScaleFactor'][chan_id]
                    gain /= info['fSignalGain'][chan_id]
                    gain /= info['fADCProgrammableGain'][chan_id]
                    gain /= info['lADCResolution']
                    if info['nTelegraphEnable'][chan_id] == 0:
                        pass
                    elif info['nTelegraphEnable'][chan_id] == 1:
                        gain /= info['fTelegraphAdditGain'][chan_id]
                    else:
                        logger.warning('ignoring buggy nTelegraphEnable')
                    offset = info['fInstrumentOffset'][chan_id]
                    offset -= info['fSignalOffset'][chan_id]
                elif version >= 2.:
                    gain = info['protocol']['fADCRange']
                    gain /= info['listADCInfo'][chan_id]['fInstrumentScaleFactor']
                    gain /= info['listADCInfo'][chan_id]['fSignalGain']
                    gain /= info['listADCInfo'][chan_id]['fADCProgrammableGain']
                    gain /= info['protocol']['lADCResolution']
                    if info['listADCInfo'][chan_id]['nTelegraphEnable']:
                        gain /= info['listADCInfo'][chan_id]['fTelegraphAdditGain']
                    offset = info['listADCInfo'][chan_id]['fInstrumentOffset']
                    offset -= info['listADCInfo'][chan_id]['fSignalOffset']
            else:
                gain, offset = 1., 0.
            group_id = 0
            sig_channels.append((name, chan_id, self._sampling_rate,
                                 sig_dtype, units, gain, offset, group_id))

        sig_channels = np.array(sig_channels, dtype=_signal_channel_dtype)

        # only one events channel : tag
        if mode in [3, 5]:  # TODO check if tags exits in other mode
            # In ABF timstamps are not attached too any particular segment
            # so each segment acess all event
            timestamps = []
            labels = []
            comments = []
            for i, tag in enumerate(info['listTag']):
                timestamps.append(tag['lTagTime'])
{'fraction_non_alphanumeric': 0.11404997366727135, 'fraction_numerical': 0.011410849084206214, 'mean_word_length': 1.9228664272276381, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, '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': '5411501', 'n_tokens_mistral': 5112, 'n_tokens_neox': 4819, 'n_words': 1207}