<commit_before>from collections import deque
from .specs import SYSEX_START, SYSEX_END
from .specs import SPEC_BY_STATUS
from .specs import CHANNEL_MESSAGES, REALTIME_MESSAGES
from .specs import VALID_BYTES, VALID_DATA_BYTES
from .specs import MIN_PITCHWHEEL, MAX_PITCHWHEEL
from .checks import check_data_byte, check_data
from ..py2 import convert_py2_bytes

def _decode_sysex_data(data):
    return {'data': tuple(data)}


def _decode_quarter_frame_data(data):
    return {'frame_type': data[0] >> 4,
            'frame_value' : data[0] & 15}


def _decode_songpos_data(data):
    return {'pos': data[0] | (data[1] << 7)}


def _decode_pitchwheel_data(data):
    return {'pitch': data[0] | ((data[1] << 7) + MIN_PITCHWHEEL)}


def _make_special_cases():
    cases = {
        0xe0: _decode_pitchwheel_data,
        0xf0: _decode_sysex_data,
        0xf1: _decode_quarter_frame_data,
        0xf2: _decode_songpos_data,
    }

    for i in range(16):
        cases[0xe0 | i] = _decode_pitchwheel_data

    return cases

_SPECIAL_CASES = _make_special_cases()


def _decode_data_bytes(status_byte, data, spec):
    # Subtract 1 for status byte.
    if len(data) != (spec['length'] - 1):
        raise ValueError(
            'wrong number of bytes for {} message'.format(spec['type']))

    # Todo: better name than args?
    names = [name for name in spec['value_names'] if name != 'channel']
    args = {name: value for name, value in zip(names, data)}

    if status_byte in CHANNEL_MESSAGES:
        # Channel is stored in the lower nibble of the status byte.
        args['channel'] = status_byte & 0x0f

    return args


def decode_msg(midi_bytes, time=0, check=True):
    """Decode message bytes and return messages as a dictionary.

    Raises ValueError if the bytes are out of range or the message is
    invalid.
    """
    # Todo: this function is getting long.
    midi_bytes = convert_py2_bytes(midi_bytes)

    if len(midi_bytes) == 0:
        raise ValueError('message is 0 bytes long')

    status_byte = midi_bytes[0]
    data = midi_bytes[1:]

    try:
        spec = SPEC_BY_STATUS[status_byte]
    except KeyError:
        raise ValueError('invalid status byte {!r}'.format(status_byte))

    msg = {
        'type': spec['type'],
        'time': time,
    }

    # Sysex.
    if status_byte == SYSEX_START:
        end = data[-1]
        data = data[:-1]
        if end != SYSEX_END:
            raise ValueError('invalid sysex end byte {!r}'.format(end))

    if check:
        check_data(data)

    if status_byte in _SPECIAL_CASES:
        msg.update(_SPECIAL_CASES[status_byte](data))
    else:
        msg.update(_decode_data_bytes(status_byte, data, spec))

    return msg


class Decoder(object):
    """
    Encodes MIDI message bytes to dictionaries.
    """
    def __init__(self, data=None):
        """Create a new decoder."""
        self.messages = deque()
        self._reset()
        if data is not None:
            self.feed(data)

    def _reset(self):
        self._bytes = []
        self._len = -1
        self._in_msg = False
        self._in_sysex = False

    def _deliver(self, msg=None):
        if msg is None:
            msg = decode_msg(self._bytes, check=False)
            print(msg)
        self.messages.append(msg)


    def _handle_status_byte(self, byte):
        spec = SPEC_BY_STATUS.get(byte)

        if self._in_sysex and byte in REALTIME_MESSAGES:
            if spec:
                self._deliver(decode_msg([byte]))
        elif byte == SYSEX_END:
            if self._in_sysex:
                self._bytes.append(SYSEX_END)
                self._deliver()
                self._reset()
            else:
                self._reset()
        elif spec:
            # Start new message.
            self._in_msg = True<commit_msg>Forgot to remove a print.<commit_after>from collections import deque
from .specs import SYSEX_START, SYSEX_END
from .specs import SPEC_BY_STATUS
from .specs import CHANNEL_MESSAGES, REALTIME_MESSAGES
from .specs import VALID_BYTES, VALID_DATA_BYTES
from .specs import MIN_PITCHWHEEL, MAX_PITCHWHEEL
from .checks import check_data_byte, check_data
from ..py2 import convert_py2_bytes

def _decode_sysex_data(data):
    return {'data': tuple(data)}


def _decode_quarter_frame_data(data):
    return {'frame_type': data[0] >> 4,
            'frame_value' : data[0] & 15}


def _decode_songpos_data(data):
    return {'pos': data[0] | (data[1] << 7)}


def _decode_pitchwheel_data(data):
    return {'pitch': data[0] | ((data[1] << 7) + MIN_PITCHWHEEL)}


def _make_special_cases():
    cases = {
        0xe0: _decode_pitchwheel_data,
        0xf0: _decode_sysex_data,
        0xf1: _decode_quarter_frame_data,
        0xf2: _decode_songpos_data,
    }

    for i in range(16):
        cases[0xe0 | i] = _decode_pitchwheel_data

    return cases

_SPECIAL_CASES = _make_special_cases()


def _decode_data_bytes(status_byte, data, spec):
    # Subtract 1 for status byte.
    if len(data) != (spec['length'] - 1):
        raise ValueError(
            'wrong number of bytes for {} message'.format(spec['type']))

    # Todo: better name than args?
    names = [name for name in spec['value_names'] if name != 'channel']
    args = {name: value for name, value in zip(names, data)}

    if status_byte in CHANNEL_MESSAGES:
        # Channel is stored in the lower nibble of the status byte.
        args['channel'] = status_byte & 0x0f

    return args


def decode_msg(midi_bytes, time=0, check=True):
    """Decode message bytes and return messages as a dictionary.

    Raises ValueError if the bytes are out of range or the message is
    invalid.
    """
    # Todo: this function is getting long.
    midi_bytes = convert_py2_bytes(midi_bytes)

    if len(midi_bytes) == 0:
        raise ValueError('message is 0 bytes long')

    status_byte = midi_bytes[0]
    data = midi_bytes[1:]

    try:
        spec = SPEC_BY_STATUS[status_byte]
    except KeyError:
        raise ValueError('invalid status byte {!r}'.format(status_byte))

    msg = {
        'type': spec['type'],
        'time': time,
    }

    # Sysex.
    if status_byte == SYSEX_START:
        end = data[-1]
        data = data[:-1]
        if end != SYSEX_END:
            raise ValueError('invalid sysex end byte {!r}'.format(end))

    if check:
        check_data(data)

    if status_byte in _SPECIAL_CASES:
        msg.update(_SPECIAL_CASES[status_byte](data))
    else:
        msg.update(_decode_data_bytes(status_byte, data, spec))

    return msg


class Decoder(object):
    """
    Encodes MIDI message bytes to dictionaries.
    """
    def __init__(self, data=None):
        """Create a new decoder."""
        self.messages = deque()
        self._reset()
        if data is not None:
            self.feed(data)

    def _reset(self):
        self._bytes = []
        self._len = -1
        self._in_msg = False
        self._in_sysex = False

    def _deliver(self, msg=None):
        if msg is None:
            msg = decode_msg(self._bytes, check=False)
        self.messages.append(msg)


    def _handle_status_byte(self, byte):
        spec = SPEC_BY_STATUS.get(byte)

        if self._in_sysex and byte in REALTIME_MESSAGES:
            if spec:
                self._deliver(decode_msg([byte]))
        elif byte == SYSEX_END:
            if self._in_sysex:
                self._bytes.append(SYSEX_END)
                self._deliver()
                self._reset()
            else:
                self._reset()
        elif spec:
            # Start new message.
            self._in_msg = True
{'fraction_non_alphanumeric': 0.10233109442907941, 'fraction_numerical': 0.010009219017516134, 'mean_word_length': 2.478699038021072, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6851851', 'n_tokens_mistral': 2555, 'n_tokens_neox': 2427, 'n_words': 646}