<commit_before>from __future__ import absolute_import
from itertools import chain, count, islice, takewhile
from functools import reduce
from multiprocessing import Pool, cpu_count

import collections
try:
    import dill as serializer
except ImportError:
    import pickle as serializer
import future.builtins as builtins
import six



if six.PY2:
    CSV_WRITE_MODE = 'wb'
    PROTOCOL = 2
else:
    CSV_WRITE_MODE = 'w'
    PROTOCOL = 4
CPU_COUNT = cpu_count()


def is_primitive(val):
    """
    Checks if the passed value is a primitive type.

    >>> is_primitive(1)
    True

    >>> is_primitive("abc")
    True

    >>> is_primitive(True)
    True

    >>> is_primitive({})
    False

    >>> is_primitive([])
    False

    >>> is_primitive(set([]))

    :param val: value to check
    :return: True if value is a primitive, else False
    """
    return isinstance(val, str) \
        or isinstance(val, bool) \
        or isinstance(val, six.string_types + (six.text_type,)) \
        or isinstance(val, six.integer_types) \
        or isinstance(val, float) \
        or isinstance(val, complex) \
        or isinstance(val, bytes)


def is_namedtuple(val):
    """
    Use Duck Typing to check if val is a named tuple. Checks that val is of type tuple and contains
    the attribute _fields which is defined for named tuples.
    :param val: value to check type of
    :return: True if val is a namedtuple
    """
    val_type = type(val)
    bases = val_type.__bases__
    if len(bases) != 1 or bases[0] != tuple:
        return False
    fields = getattr(val_type, '_fields', None)
    return all(isinstance(n, str) for n in fields)


def identity(arg):
    """
    Function which returns the argument. Used as a default lambda function.

    >>> obj = object()
    >>> obj is identity(obj)
    True

    :param arg: object to take identity of
    :return: return arg
    """
    return arg


def is_iterable(val):
    """
    Check if val is not a list, but is a collections.Iterable type. This is used to determine
    when list() should be called on val

    >>> l = [1, 2]
    >>> is_iterable(l)
    False
    >>> is_iterable(iter(l))
    True

    :param val: value to check
    :return: True if it is not a list, but is a collections.Iterable
    """
    if isinstance(val, list):
        return False
    return isinstance(val, collections.Iterable)


def split_every(parts, iterable):
    """
    Split an iterable in n parts

    >>> l = [1, 2, 3, 4]
    >>> split_every(l, 2)
    [[1, 2], [3, 4]]

    :param iterable: iterable to split
    :param parts: number of chunks
    :return: return the iterable split in parts
    """
    return takewhile(bool, (list(islice(iterable, parts)) for _ in count(0)))


def unpack(packed):
    func, args = serializer.loads(packed)
    result = func(*args)
    if isinstance(result, collections.Iterable):
        return list(result)


def pack(func, args):
    return serializer.dumps((func, args), PROTOCOL)


def is_serializable(func, raise_errors=True):
    if raise_errors is None:
        raise_errors = True
    try:
        serializer.dumps(func, PROTOCOL)
        return True
    except (AttributeError, serializer.PicklingError):
        if raise_errors:
            raise serializer.PicklingError(
                "Function {} is not serializable. "
                "Try installing dill or passing raise_errors to False "
                "for non-parallel execution when serialization fails."
                .format(str(func)))
        return False


def parallelize(func, result, processes=None, raise_errors=True):
    if not is_serializable(func, raise_errors=raise_errors):
        return func(result)
    parallel_iter = lazy_parallelize(func, result, processes=processes)
    return chain.from_iterable(parallel_iter)


def lazy_parallelize(func, result, processes=None):
    if processes is None or processes < 1:
        processes = CPU_COUNT
    else:
        processes = min(processes, CPU_COUNT)
    try:
        chunk_size = (len(result) // processes) or processes
    except TypeError:
        chunk_size = processes
    with Pool(processes=processes) as pool:
        chunks = split_every(chunk_size, iter(result))
        packed_chunks = (pack(func, (chunk, )) for chunk in chunks)
        for pool_result in pool.imap(unpack, packed_chunks):
            yield pool_result


def compose(*functions):
    # pylint: disable=undefined-variable
    return reduce(lambda f, g: lambda x: f(g(x)), functions, lambda x: x)


class ReusableFile(object):
    """
    Class which emulates the builtin file except that calling iter() on it will return separate
    iterators on different file handlers (which are automatically closed when iteration stops). This
    is useful for allowing a file object to be iterated over multiple times while keep evaluation
    lazy.
    """
    # pylint: disable=too-many-instance-attributes<commit_msg>Change pickling protocol to HIGHEST_PROTOCOL for Python3. Take out Pool from the context manager for Python2.<commit_after>from __future__ import absolute_import
from itertools import chain, count, islice, takewhile
from functools import reduce
from multiprocessing import Pool, cpu_count

import collections
try:
    import dill as serializer
except ImportError:
    import pickle as serializer
import future.builtins as builtins
import six



if six.PY2:
    CSV_WRITE_MODE = 'wb'
    PROTOCOL = 2
else:
    CSV_WRITE_MODE = 'w'
    PROTOCOL = serializer.HIGHEST_PROTOCOL
CPU_COUNT = cpu_count()


def is_primitive(val):
    """
    Checks if the passed value is a primitive type.

    >>> is_primitive(1)
    True

    >>> is_primitive("abc")
    True

    >>> is_primitive(True)
    True

    >>> is_primitive({})
    False

    >>> is_primitive([])
    False

    >>> is_primitive(set([]))

    :param val: value to check
    :return: True if value is a primitive, else False
    """
    return isinstance(val, str) \
        or isinstance(val, bool) \
        or isinstance(val, six.string_types + (six.text_type,)) \
        or isinstance(val, six.integer_types) \
        or isinstance(val, float) \
        or isinstance(val, complex) \
        or isinstance(val, bytes)


def is_namedtuple(val):
    """
    Use Duck Typing to check if val is a named tuple. Checks that val is of type tuple and contains
    the attribute _fields which is defined for named tuples.
    :param val: value to check type of
    :return: True if val is a namedtuple
    """
    val_type = type(val)
    bases = val_type.__bases__
    if len(bases) != 1 or bases[0] != tuple:
        return False
    fields = getattr(val_type, '_fields', None)
    return all(isinstance(n, str) for n in fields)


def identity(arg):
    """
    Function which returns the argument. Used as a default lambda function.

    >>> obj = object()
    >>> obj is identity(obj)
    True

    :param arg: object to take identity of
    :return: return arg
    """
    return arg


def is_iterable(val):
    """
    Check if val is not a list, but is a collections.Iterable type. This is used to determine
    when list() should be called on val

    >>> l = [1, 2]
    >>> is_iterable(l)
    False
    >>> is_iterable(iter(l))
    True

    :param val: value to check
    :return: True if it is not a list, but is a collections.Iterable
    """
    if isinstance(val, list):
        return False
    return isinstance(val, collections.Iterable)


def split_every(parts, iterable):
    """
    Split an iterable in n parts

    >>> l = [1, 2, 3, 4]
    >>> split_every(l, 2)
    [[1, 2], [3, 4]]

    :param iterable: iterable to split
    :param parts: number of chunks
    :return: return the iterable split in parts
    """
    return takewhile(bool, (list(islice(iterable, parts)) for _ in count(0)))


def unpack(packed):
    func, args = serializer.loads(packed)
    result = func(*args)
    if isinstance(result, collections.Iterable):
        return list(result)


def pack(func, args):
    return serializer.dumps((func, args), PROTOCOL)


def is_serializable(func, raise_errors=True):
    if raise_errors is None:
        raise_errors = True
    try:
        serializer.dumps(func, PROTOCOL)
        return True
    except (AttributeError, serializer.PicklingError):
        if raise_errors:
            raise serializer.PicklingError(
                "Function {} is not serializable. "
                "Try installing dill or passing raise_errors to False "
                "for non-parallel execution when serialization fails."
                .format(str(func)))
        return False


def parallelize(func, result, processes=None, raise_errors=True):
    if not is_serializable(func, raise_errors=raise_errors):
        return func(result)
    parallel_iter = lazy_parallelize(func, result, processes=processes)
    return chain.from_iterable(parallel_iter)


def lazy_parallelize(func, result, processes=None):
    if processes is None or processes < 1:
        processes = CPU_COUNT
    else:
        processes = min(processes, CPU_COUNT)
    try:
        chunk_size = (len(result) // processes) or processes
    except TypeError:
        chunk_size = processes
    pool = Pool(processes=processes)
    chunks = split_every(chunk_size, iter(result))
    packed_chunks = (pack(func, (chunk, )) for chunk in chunks)
    for pool_result in pool.imap(unpack, packed_chunks):
        yield pool_result
    pool.terminate()


def compose(*functions):
    # pylint: disable=undefined-variable
    return reduce(lambda f, g: lambda x: f(g(x)), functions, lambda x: x)


class ReusableFile(object):
    """
    Class which emulates the builtin file except that calling iter() on it will return separate
    iterators on different file handlers (which are automatically closed when iteration stops). This
    is useful for allowing a file object to be iterated over multiple times while keep evaluation
    lazy.
    """
    # pylint: disable=too-many-instance-attributes
{'fraction_non_alphanumeric': 0.0927710843373494, 'fraction_numerical': 0.00391566265060241, 'mean_word_length': 2.909340659340659, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 81, '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': '4357321', 'n_tokens_mistral': 3009, 'n_tokens_neox': 2923, 'n_words': 1015}