<commit_before>#!/usr/bin/env python

# Copyright (c) 2017, DIANA-HEP
# All rights reserved.
# 
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 
# * Redistributions of source code must retain the above copyright notice, this
#   list of conditions and the following disclaimer.
# 
# * Redistributions in binary form must reproduce the above copyright notice,
#   this list of conditions and the following disclaimer in the documentation
#   and/or other materials provided with the distribution.
# 
# * Neither the name of the copyright holder nor the names of its
#   contributors may be used to endorse or promote products derived from
#   this software without specific prior written permission.
# 
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

import ast
import keyword
import sys
import math
from collections import namedtuple

import numpy

import uproot.tree
import uproot.interp.auto

def ifinstalled(f):
    try:
        import numba
    except ImportError:
        return f
    else:
        return numba.jit()(f)

if sys.version_info[0] <= 2:
    parsable = (unicode, str)
else:
    parsable = (str,)

stringenv = dict(math.__dict__.items() + [(x, eval(x)) for x in ("round", "min", "max")])

class ChainStep(object):
    NEW_ARRAY_DTYPE = numpy.dtype(numpy.float64)

    def __init__(self, previous):
        self.previous = previous

    def define(self, **exprs):
        return Define(self, exprs)

    def intermediate(self, cache=None, **exprs):
        return Intermediate._create(self, cache, exprs)

    def filter(self, expr):
        return Filter(self, expr)

    @property
    def source(self):
        return self.previous.source

    @property
    def tree(self):
        return self.previous.tree

    @staticmethod
    def _makefcn(code, env, name, source):
        env = dict(env)
        exec(code, env)
        env[name].source = source
        return env[name]

    @staticmethod
    def _generatenames(want, avoid=set()):
        disambigifier = 0
        out = {}
        for name in want:
            if ChainStep._isidentifier(name):
                newname = name
            else:
                newname = "tmp"
            while newname in avoid or newname in out or keyword.iskeyword(newname):
                newname = "{0}_{1}".format(name, disambigifier)
                disambigifier += 1
            out[name] = newname
        return out

    @staticmethod
    def _string2fcn(string):
        insymbols = []
        outsymbols = set()

        def recurse(node):
            if isinstance(node, ast.FunctionDef):
                raise TypeError("function definitions are not allowed in a function parsed from a string")

            elif isinstance(node, ast.Name):
                if isinstance(node.ctx, ast.Load):
                    if node.id not in outsymbols and node.id not in stringenv:
                        insymbols.append(node.id)
                elif isinstance(node.ctx, ast.Store):
                    outsymbols.add(node.id)

            elif isinstance(node, ast.AST):
                for field in node._fields:
                    recurse(getattr(node, field))

            elif isinstance(node, list):
                for x in node:
                    recurse(x)

        body = ast.parse(string).body
        recurse(body)

        if len(body) == 0:
            raise TypeError("string contains no expressions")
        elif isinstance(body[-1], ast.Expr):
            body[-1] = ast.Return(body[-1].value)
            body[-1].lineno = body[-1].value.lineno
            body[-1].col_offset = body[-1].value.col_offset

        names = ChainStep._generatenames(["fcn"], set(insymbols).union(outsymbols))

        module = ast.parse("def {fcn}({args}): pass".format(fcn=names["fcn"], args=", ".join(insymbols)))
        module.body[0].body = body
        return ChainStep._makefcn(compile(module, string, "exec"), stringenv, names["fcn"], string)

    @staticmethod
    def _isfcn(expr):
        return isinstance(expr, parsable) or (callable(expr) and hasattr(expr, "__code__"))

    @staticmethod
    def _tofcn(expr):
        identifier = None
        if isinstance(expr, parsable):
            if ChainStep._isidentifier(expr):
                identifier = expr
            expr = ChainStep._string2fcn(expr)

        return expr, ChainStep._params(expr), identifier

    @staticmethod
    def _params(fcn):
        return fcn.__code__.co_varnames

    @staticmethod
    def _tofcns(exprs):
        used = {}
        def getname(fcn):
            trial = getattr(fcn, "__name__", "<lambda>")
            if trial == "<lambda>":
                trial = "fcn"
                if "fcn" not in used:
                    used["fcn"] = 1
            out = trial
            while out in used:
                out = "{0}_{1}".format(trial, used[trial])
                used[trial] += 1
            if trial not in used:
                used[trial] = 2
            return out

        if isinstance(exprs, parsable):
            return [ChainStep._tofcn(exprs) + (exprs, exprs)]

        elif callable(exprs) and hasattr(exprs, "__code__"):
            return [ChainStep._tofcn(exprs) + (id(exprs), getname(exprs))]

        elif isinstance(exprs, dict) and all(ChainStep._isfcn(x) for x in exprs.values()):
            return [ChainStep._tofcn(x) + (x if isinstance(x, parsable) else id(x), n) for n, x in exprs.items()]

        else:
            try:
                assert all(ChainStep._isfcn(x) for x in exprs)
            except (TypeError, AssertionError):
                raise TypeError("exprs must be a dict of strings or functions, an iterable of strings or functions, a single string, or a single function")
            else:
                return [ChainStep._tofcn(x) + ((x, x) if isinstance(x, parsable) else (id(x), getname(x))) for i, x in enumerate(exprs)]

    @staticmethod
    def _isidentifier(dictname):
        try:
            assert not keyword.iskeyword(dictname)
            x = ast.parse(dictname)
            assert len(x.body) == 1 and isinstance(x.body[0], ast.Expr) and isinstance(x.body[0].value, ast.Name)
        except (SyntaxError, TypeError, AssertionError):
            return False
        else:
            return True

    @staticmethod
    def _compilefcn(numba):
        if callable(numba):
            return numba

        elif numba is None or numba is False:
            return lambda f: f

        elif numba is True:
            import numba as nb
            return lambda f: nb.njit()(f)

        else:
            import numba as nb
            return lambda f: nb.jit(**numba)(f)

    def _prepare(self, exprs, aliases, entryvar, numba):
        compilefcn = self._compilefcn(numba)

        dictnames = []
        nonidentifiers = []
        toresolve = []
        for fcn, requirements, identifier, cacheid, dictname in self._tofcns(exprs):
            dictnames.append(dictname)
            if identifier is None:
                nonidentifiers.append((fcn, requirements, identifier, cacheid, dictname))
                toresolve.append(dictname)
            else:
                toresolve.append(identifier)

        tmpnode = Intermediate(self, None, nonidentifiers)

        # find all the dependencies and put unique ones in lists
        sourcenames = []
        intermediates = []
        entryvars = set()
        for name in toresolve:
            tmpnode._satisfy(name, sourcenames, intermediates, entryvars, entryvar, aliases)

        # reorder the (Intermediate, name) pairs in order of increasing dependency (across all expressions)
        intermediates = Intermediate._dependencyorder(sourcenames, intermediates, entryvar, aliases)

        # now compile them, using the established "sourcenames" and "intermediates" order to get arguments by tuple index (hard-compiled into functions)
        fcncache = {}
        compiled = []
        for name in toresolve:
            compiled.append(tmpnode._argfcn(name, sourcenames, intermediates, entryvar, aliases, compilefcn, fcncache))

        # compile the intermediates in the same way
        compiledintermediates = [intermediate._compileintermediate(requirement, sourcenames, intermediates, entryvar, aliases, compilefcn, fcncache) for intermediate, requirement in intermediates]

        return tmpnode, dictnames, compiled, sourcenames, intermediates, compiledintermediates, entryvars, compilefcn

    def _wouldsatisfy(self, requirement, entryvar, aliases):
        return self.previous._wouldsatisfy(requirement, entryvar, aliases)

    def _satisfy(self, requirement, sourcenames, intermediates, entryvars, entryvar, aliases):
        self.previous._satisfy(requirement, sourcenames, intermediates, entryvars, entryvar, aliases)

    def _argfcn(self, requirement, sourcenames, intermediates, entryvar, aliases, compilefcn, fcncache):
        return self.previous._argfcn(requirement, sourcenames, intermediates, entryvar, aliases, compilefcn, fcncache)

    def _chain(self, sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, entrysteps, entrystart, entrystop, cache, basketcache, keycache, readexecutor, numba):
        return self.previous._chain(sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, entrysteps, entrystart, entrystop, cache, basketcache, keycache, readexecutor, numba)

    def _endchain(self, compiled, sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, entrysteps, entrystart, entrystop, cache, basketcache, keycache, readexecutor, calcexecutor, numba):
        awaits = self.source._chain(sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, entrysteps, entrystart, entrystop, cache, basketcache, keycache, readexecutor, numba)

        def calculate(await):
            try:
                start, stop, numentries, arrays = await()

                for compiledintermediate in compiledintermediates:
                    compiledintermediate(arrays)

                out = start, stop, numentries, [x(arrays) for x in compiled]

            except:
                return sys.exc_info(), None
            else:
                return None, out

        if calcexecutor is None:
            for await in awaits:
                excinfo, result = calculate(await)
                uproot.tree._delayedraise(excinfo)
                yield result

        else:
            for excinfo, result in calcexecutor.map(calculate, awaits):
                uproot.tree._delayedraise(excinfo)
                yield result
            
    def iterate_newarrays(self, exprs, entrysteps=None, entrystart=None, entrystop=None, aliases={}, interpretations={}, entryvar=None, outputtype=dict, reportentries=False, cache=None, basketcache=None, keycache=None, readexecutor=None, calcexecutor=None, numba=ifinstalled):
        tmpnode, dictnames, compiled, sourcenames, intermediates, compiledintermediates, entryvars, compilefcn = self._prepare(exprs, aliases, entryvar, numba)

        if outputtype == namedtuple:
            for dictname in dictnames:
                if not self._isidentifier(dictname):
                    raise ValueError("illegal field name for namedtuple: {0}".format(repr(dictname)))
            outputtype = namedtuple("Arrays", dictnames)

        if issubclass(outputtype, dict):
            def finish(arrays):
                return outputtype(zip(dictnames, arrays))
        elif outputtype == tuple or outputtype == list:
            def finish(arrays):
                return outputtype(arrays)
        else:
            def finish(arrays):
                return outputtype(*arrays)

        for start, stop, numentries, arrays in tmpnode._endchain(compiled, sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, entrysteps, entrystart, entrystop, cache, basketcache, keycache, readexecutor, calcexecutor, numba):
            if reportentries:
                yield start, stop, numentries, finish(arrays)
            else:
                yield finish(arrays)

    def newarrays(self, exprs, entrystart=None, entrystop=None, aliases={}, interpretations={}, entryvar=None, outputtype=dict, cache=None, basketcache=None, keycache=None, readexecutor=None, calcexecutor=None, numba=ifinstalled):
        tmpnode, dictnames, compiled, sourcenames, intermediates, compiledintermediates, entryvars, compilefcn = self._prepare(exprs, aliases, entryvar, numba)

        if outputtype == namedtuple:
            for dictname in dictnames:
                if not self._isidentifier(dictname):
                    raise ValueError("illegal field name for namedtuple: {0}".format(repr(dictname)))
            outputtype = namedtuple("Arrays", dictnames)

        partitions = []
        totalentries = 0
        for start, stop, numentries, arrays in tmpnode._endchain(compiled, sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, None, entrystart, entrystop, cache, basketcache, keycache, readexecutor, calcexecutor, numba):
            assert len(dictnames) == len(arrays)
            if len(arrays) > 0:
                for array in arrays:
                    assert numentries == len(array)

            partitions.append((totalentries, totalentries + numentries, arrays))
            totalentries += numentries

        if len(partitions) == 0:
            outarrays = [numpy.ones(0, dtype=self.NEW_ARRAY_DTYPE)*222 for i in range(len(dictnames))]
        else:
            outarrays = [numpy.ones(totalentries, dtype=array.dtype)*333 for array in partitions[0][2]]

        for start, stop, arrays in partitions:
            for outarray, array in zip(outarrays, arrays):
                outarray[start:stop] = array
            
        if issubclass(outputtype, dict):
            return outputtype(zip(dictnames, outarrays))
        elif outputtype == tuple or outputtype == list:
            return outputtype(outarrays)
        else:
            return outputtype(*outarrays)

    def newarray(self, expr, entrystart=None, entrystop=None, aliases={}, interpretations={}, entryvar=None, cache=None, basketcache=None, keycache=None, readexecutor=None, numba=ifinstalled):
        if not ChainStep._isfcn(expr):
            raise TypeError("expr must be a single string or function")

        return self.newarrays(expr, entrystart=entrystart, entrystop=entrystop, aliases=aliases, interpretations=interpretations, entryvar=entryvar, outputtype=tuple, cache=cache, basketcache=basketcache, keycache=keycache, readexecutor=readexecutor, numba=numba)[0]

    def _normalize_reduceargs(self, identity, increment, combine):
        if isinstance(identity, parsable):
            identity = self._string2fcn(identity)
        if isinstance(increment, parsable):
            increment = self._string2fcn(increment)
        if isinstance(combine, parsable):
            combine = self._string2fcn(combine)

        if self._isfcn(identity) and self._isfcn(increment) and (combine is None or self._isfcn(combine)):
            if len(self._params(increment)) == 0:
                raise TypeError("increment function must have at least one parameter")

            if combine is None:
                combine = lambda x, y: x + y

            monoidvar = self._params(increment)[0]
            identity  = {monoidvar: identity}
            increment = {monoidvar: increment}
            combine   = {monoidvar: combine}
            order     = [monoidvar]

        if isinstance(increment, dict):
            if not isinstance(identity, dict):
                raise TypeError("if increment is a dict of functions, identity must be as well (to match up argument lists)")
            if not combine is None and not isinstance(combine, dict):
                raise TypeError("if increment is a dict of functions, combine must be as well (to match up argument lists)")

            if len(increment) == 0 or not all(self._isfcn(x) for n, x in increment.items()):
                raise TypeError("increment must be a (non-empty) dict of strings or functions, a (non-empty) iterable of strings or functions, a single string, or a single function")
            increment = dict((n, string2fcn(x) if isinstance(x, parsable) else x) for n, x in increment.items())
            if increment == dict:
                order = sorted(increment)
            else:
                order = list(increment)   # may be an OrderedDict; preserve whatever order it has
        else:
            try:
                assert len(increment) > 0 and all(self._isfcn(x) for x in increment)
            except (TypeError, AssertionError):
                raise TypeError("increment must be a (non-empty) dict of strings or functions, a (non-empty) iterable of strings or functions, a single string, or a single function")
            else:
                # define as a list first so that we get the order
                increment = [string2fcn(x) if isinstance(x, parsable) else x for x in increment]
                order = []
                for x in increment:
                    params = self._params(x)
                    if len(params) == 0:
                        raise TypeError("increment functions must have at least one argument (the aggregator)")
                    order.append(params[0])
                if len(order) != len(set(order)):
                    raise TypeError("if providing a list of increment functions, the aggregator (first argument) of each must be distinct")
                # redefine as a dict
                increment = dict(zip(order, increment))

        if isinstance(identity, dict):
            if not all(self._isfcn(x) for n, x in identity.items()):
                raise TypeError("identity must be a (non-empty) dict of strings or functions, a (non-empty) iterable of strings or functions, a single string, or a single function")
            identity = dict((n, string2fcn(x) if isinstance(x, parsable) else x) for n, x in identity.items())
        else:
            try:
                assert all(self._isfcn(x) for x in identity)
            except (TypeError, AssertionError):
                raise TypeError("identity must be a (non-empty) dict of strings or functions, a (non-empty) iterable of strings or functions, a single string, or a single function")
            else:
                identity = dict(zip(order, [string2fcn(x) if isinstance(x, parsable) else x for x in identity]))

        if not all(len(self._params(x)) == 0 for x in identity.values()):
            raise TypeError("identity functions must have zero arguments")

        if combine is None:
            combine = dict((n, lambda x, y: x + y) for n in order)
        elif isinstance(combine, dict):
            if not all(self._isfcn(x) for n, x in combine.items()):
                raise TypeError("combine must be None, a (non-empty) dict of strings or functions, a (non-empty) iterable of strings or functions, a single string, or a single function")
            combine = dict((n, string2fcn(x) if isinstance(x, parsable) else x) for n, x in combine.items())
        else:
            try:
                assert all(self._isfcn(x) for x in combine)
            except (TypeError, AssertionError):
                raise TypeError("combine must be None, a (non-empty) dict of strings or functions, a (non-empty) iterable of strings or functions, a single string, or a single function")
            else:
                combine = dict(zip(order, [string2fcn(x) if isinstance(x, parsable) else x for x in combine]))
        if not all(len(self._params(x)) == 2 for x in combine.values()):
            raise TypeError("combine functions must have two arguments")

        monoidvars = dict((n, self._params(x)[0]) for n, x in increment.items())

        if not set(identity.keys()) == set(increment.keys()) == set(combine.keys()):
            raise TypeError("if identity, increment, and combine are provided as dicts, they must have the same set of keys (to match up argument lists)")

        return identity, increment, combine, order, monoidvars

    def reduceall(self, identity, increment, combine=None, entrystart=None, entrystop=None, aliases={}, interpretations={}, entryvar=None, outputtype=dict, cache=None, basketcache=None, keycache=None, readexecutor=None, calcexecutor=None, numba=ifinstalled):
        identity, increment, combine, order, monoidvars = self._normalize_reduceargs(identity, increment, combine)

        if outputtype == namedtuple:
            for name in order:
                if not self._isidentifier(name):
                    raise ValueError("illegal field name for namedtuple: {0}".format(repr(name)))
            outputtype = namedtuple("Reduced", order)

        compilefcn = self._compilefcn(numba)

        dependencies = []
        for inc in increment.values():
            dependencies.extend(self._params(inc)[1:])

        # normal preparations for calculating dependencies
        sourcenames = []
        intermediates = []
        entryvars = set()
        for name in dependencies:
            self._satisfy(name, sourcenames, intermediates, entryvars, entryvar, aliases)

        intermediates = Intermediate._dependencyorder(sourcenames, intermediates, entryvar, aliases)
        compiledintermediates = [intermediate._compileintermediate(requirement, sourcenames, intermediates, entryvar, aliases, compilefcn, fcncache) for intermediate, requirement in intermediates]

        # dependencies are unique strings
        avoid = set(dependencies)

        # unique names for dependency getters
        getternames = self._generatenames(dependencies, avoid)
        avoid = avoid.union(getternames.values())

        # unique names for dependency items
        itemnames = self._generatenames(dependencies, avoid)
        avoid = avoid.union(itemnames.values())

        # unique names for monoids
        monoidnames = self._generatenames(monoidvars.values(), avoid)
        avoid = avoid.union(monoidnames.values())

        # unique names for increment functions
        incnames = self._generatenames(increment, avoid)
        avoid = avoid.union(incnames.values())

        # unique names for builtins and dummy variables
        builtins = self._generatenames(["rfcn", "arrays", "numentries", "i", "range"], avoid)
        avoid = avoid.union(builtins.values())

        env = dict([("range", range)] + [(incnames[n], compilefcn(x)) for n, x in increment.items()])

        # getter -> item for each dependency
        fcncache = {}
        itemdefs = []
        for n in dependencies:
            argfcn = self._argfcn(n, sourcenames, intermediates, entryvar, aliases, compilefcn, fcncache)
            env[getternames[n]] = argfcn
            itemdefs.append("{0} = {1}({2})".format(itemnames[n], getternames[n], builtins["arrays"]))

        # call each increment function on its parameters (per item), in declaration or sorted order (not that it matters)
        incfcns = ["{0} = {1}({0}{2})".format(monoidnames[monoidvars[n]], incnames[n], "".join(", " + itemnames[x] for x in self._params(increment[n])[1:])) for n in order]

        # input parameters and output tuple
        monoidargs = [monoidnames[monoidvars[n]] for n in order]

        source = """
def {rfcn}({arrays}, {numentries}, {monoidargs}):
    for {i} in {range}({numentries}):
        {itemdefs}
        {incfcns}
    return ({monoidargs},)
""".format(rfcn=builtins["rfcn"], arrays=builtins["arrays"], numentries=builtins["numentries"], monoidargs=", ".join(monoidargs), i=builtins["i"], range=builtins["range"], itemdefs="\n        ".join(itemdefs), incfcns="\n        ".join(incfcns))

        print
        print source
        print
        print env
        print

        rfcn = compilefcn(self._makefcn(compile(ast.parse(source), "<reduce>", "exec"), env, builtins["rfcn"], source))

        awaits = self._chain(sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, None, entrystart, entrystop, cache, basketcache, keycache, readexecutor, numba)

        results = [[None for j in range(len(order))] for i in range(len(awaits))]

        def calculate(i):
            try:
                start, stop, numentries, arrays = awaits[i]()

                for compiledintermediate in compiledintermediates:
                    compiledintermediate(arrays)

                inmonoids  = [identity[n]() for n in order]

                outmonoids = rfcn(arrays, numentries, *inmonoids)

                for j, monoid in enumerate(outmonoids):
                    results[i][j] = monoid

            except:
                return sys.exc_info()
            else:
                return None

        if calcexecutor is None:
            for i in range(len(awaits)):
                uproot.tree._delayedraise(calculate(i))
        else:
            excinfos = calcexecutor.map(calculate, range(len(awaits)))
            for excinfo in excinfos:
                uproot.tree._delayedraise(excinfo)

        # MAYBEFIXME: could combine in O(log_2(N)) steps rather than O(N) if combining is ever resource-heavy
        for i in range(1, len(awaits)):
            for j in range(len(order)):
                results[0][j] = combine(results[0][j], results[i][j])

        if issubclass(outputtype, dict):
            return outputtype(zip(order, results[0]))
        elif outputtype == tuple or outputtype == list:
            return outputtype(results[0])
        else:
            return outputtype(*results[0])

    def reduce(self, identity, increment, combine=None, entrystart=None, entrystop=None, aliases={}, interpretations={}, entryvar=None, cache=None, basketcache=None, keycache=None, readexecutor=None, calcexecutor=None, numba=ifinstalled):
        if not self._isfcn(identity):
            raise TypeError("identity must be a string or a function")
        if not self._isfcn(increment):
            raise TypeError("increment must be a string or a function")
        if not (combine is None or self._isfcn(combine)):
            raise TypeError("combine must be None, a string, or a function")
        return self.reduceall(identity, increment, combine=combine, entrystart=entrystart, entrystop=entrystop, aliases=aliases, interpretations=interpretations, entryvar=entryvar, outputtype=tuple, cache=cache, basketcache=basketcache, keycache=keycache, readexecutor=readexecutor, calcexecutor=calcexecutor, numba=numba)[0]

class Define(ChainStep):
    def __init__(self, previous, exprs):
        self.previous = previous

        self.fcn = {}
        self.requirements = {}
        self.order = []
        for fcn, requirements, identifier, cacheid, dictname in self._tofcns(exprs):
            self.fcn[dictname] = fcn
            self.requirements[dictname] = requirements
            self.order.append(dictname)

    def _wouldsatisfy(self, requirement, entryvar, aliases):
        if requirement in self.fcn:<commit_msg>now it seems to be working in general<commit_after>#!/usr/bin/env python

# Copyright (c) 2017, DIANA-HEP
# All rights reserved.
# 
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 
# * Redistributions of source code must retain the above copyright notice, this
#   list of conditions and the following disclaimer.
# 
# * Redistributions in binary form must reproduce the above copyright notice,
#   this list of conditions and the following disclaimer in the documentation
#   and/or other materials provided with the distribution.
# 
# * Neither the name of the copyright holder nor the names of its
#   contributors may be used to endorse or promote products derived from
#   this software without specific prior written permission.
# 
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

import ast
import keyword
import sys
import math
from collections import namedtuple

import numpy

import uproot.tree
import uproot.interp.auto

def ifinstalled(f):
    try:
        import numba
    except ImportError:
        return f
    else:
        return numba.jit(nogil=True)(f)

if sys.version_info[0] <= 2:
    parsable = (unicode, str)
else:
    parsable = (str,)

stringenv = dict(math.__dict__.items() + [(x, eval(x)) for x in ("round", "min", "max")])

class ChainStep(object):
    NEW_ARRAY_DTYPE = numpy.dtype(numpy.float64)

    def __init__(self, previous):
        self.previous = previous

    def define(self, **exprs):
        return Define(self, exprs)

    def intermediate(self, cache=None, **exprs):
        return Intermediate._create(self, cache, exprs)

    def filter(self, expr):
        return Filter(self, expr)

    @property
    def source(self):
        return self.previous.source

    @property
    def tree(self):
        return self.previous.tree

    @staticmethod
    def _makefcn(code, env, name, source):
        env = dict(env)
        exec(code, env)
        env[name].source = source
        return env[name]

    @staticmethod
    def _generatenames(want, avoid=set()):
        disambigifier = 0
        out = {}
        for name in want:
            if ChainStep._isidentifier(name):
                newname = name
            else:
                newname = "tmp"
            while newname in avoid or newname in out or keyword.iskeyword(newname):
                newname = "{0}_{1}".format(name, disambigifier)
                disambigifier += 1
            out[name] = newname
        return out

    @staticmethod
    def _string2fcn(string):
        insymbols = []
        outsymbols = set()

        def recurse(node):
            if isinstance(node, ast.FunctionDef):
                raise TypeError("function definitions are not allowed in a function parsed from a string")

            elif isinstance(node, ast.Name):
                if isinstance(node.ctx, ast.Load):
                    if node.id not in outsymbols and node.id not in stringenv:
                        insymbols.append(node.id)
                elif isinstance(node.ctx, ast.Store):
                    outsymbols.add(node.id)

            elif isinstance(node, ast.AST):
                for field in node._fields:
                    recurse(getattr(node, field))

            elif isinstance(node, list):
                for x in node:
                    recurse(x)

        body = ast.parse(string).body
        recurse(body)

        if len(body) == 0:
            raise TypeError("string contains no expressions")
        elif isinstance(body[-1], ast.Expr):
            body[-1] = ast.Return(body[-1].value)
            body[-1].lineno = body[-1].value.lineno
            body[-1].col_offset = body[-1].value.col_offset

        names = ChainStep._generatenames(["fcn"], set(insymbols).union(outsymbols))

        module = ast.parse("def {fcn}({args}): pass".format(fcn=names["fcn"], args=", ".join(insymbols)))
        module.body[0].body = body
        return ChainStep._makefcn(compile(module, string, "exec"), stringenv, names["fcn"], string)

    @staticmethod
    def _isfcn(expr):
        return isinstance(expr, parsable) or (callable(expr) and hasattr(expr, "__code__"))

    @staticmethod
    def _tofcn(expr):
        identifier = None
        if isinstance(expr, parsable):
            if ChainStep._isidentifier(expr):
                identifier = expr
            expr = ChainStep._string2fcn(expr)

        return expr, ChainStep._params(expr), identifier

    @staticmethod
    def _params(fcn):
        return fcn.__code__.co_varnames

    @staticmethod
    def _tofcns(exprs):
        used = {}
        def getname(fcn):
            trial = getattr(fcn, "__name__", "<lambda>")
            if trial == "<lambda>":
                trial = "fcn"
                if "fcn" not in used:
                    used["fcn"] = 1
            out = trial
            while out in used:
                out = "{0}_{1}".format(trial, used[trial])
                used[trial] += 1
            if trial not in used:
                used[trial] = 2
            return out

        if isinstance(exprs, parsable):
            return [ChainStep._tofcn(exprs) + (exprs, exprs)]

        elif callable(exprs) and hasattr(exprs, "__code__"):
            return [ChainStep._tofcn(exprs) + (id(exprs), getname(exprs))]

        elif isinstance(exprs, dict) and all(ChainStep._isfcn(x) for x in exprs.values()):
            return [ChainStep._tofcn(x) + (x if isinstance(x, parsable) else id(x), n) for n, x in exprs.items()]

        else:
            try:
                assert all(ChainStep._isfcn(x) for x in exprs)
            except (TypeError, AssertionError):
                raise TypeError("exprs must be a dict of strings or functions, an iterable of strings or functions, a single string, or a single function")
            else:
                return [ChainStep._tofcn(x) + ((x, x) if isinstance(x, parsable) else (id(x), getname(x))) for i, x in enumerate(exprs)]

    @staticmethod
    def _isidentifier(dictname):
        try:
            assert not keyword.iskeyword(dictname)
            x = ast.parse(dictname)
            assert len(x.body) == 1 and isinstance(x.body[0], ast.Expr) and isinstance(x.body[0].value, ast.Name)
        except (SyntaxError, TypeError, AssertionError):
            return False
        else:
            return True

    @staticmethod
    def _compilefcn(numba):
        if callable(numba):
            return numba

        elif numba is None or numba is False:
            return lambda f: f

        elif numba is True:
            import numba as nb
            return lambda f: nb.njit(nogil=True)(f)

        else:
            import numba as nb
            return lambda f: nb.jit(**numba)(f)

    def _prepare(self, exprs, aliases, entryvar, numba):
        compilefcn = self._compilefcn(numba)

        dictnames = []
        nonidentifiers = []
        toresolve = []
        for fcn, requirements, identifier, cacheid, dictname in self._tofcns(exprs):
            dictnames.append(dictname)
            if identifier is None:
                nonidentifiers.append((fcn, requirements, identifier, cacheid, dictname))
                toresolve.append(dictname)
            else:
                toresolve.append(identifier)

        tmpnode = Intermediate(self, None, nonidentifiers)

        # find all the dependencies and put unique ones in lists
        sourcenames = []
        intermediates = []
        entryvars = set()
        for name in toresolve:
            tmpnode._satisfy(name, sourcenames, intermediates, entryvars, entryvar, aliases)

        # reorder the (Intermediate, name) pairs in order of increasing dependency (across all expressions)
        intermediates = Intermediate._dependencyorder(sourcenames, intermediates, entryvar, aliases)

        # now compile them, using the established "sourcenames" and "intermediates" order to get arguments by tuple index (hard-compiled into functions)
        fcncache = {}
        compiled = []
        for name in toresolve:
            compiled.append(tmpnode._argfcn(name, sourcenames, intermediates, entryvar, aliases, compilefcn, fcncache))

        # compile the intermediates in the same way
        compiledintermediates = [intermediate._compileintermediate(requirement, sourcenames, intermediates, entryvar, aliases, compilefcn, fcncache) for intermediate, requirement in intermediates]

        return tmpnode, dictnames, compiled, sourcenames, intermediates, compiledintermediates, entryvars, compilefcn

    def _wouldsatisfy(self, requirement, entryvar, aliases):
        return self.previous._wouldsatisfy(requirement, entryvar, aliases)

    def _satisfy(self, requirement, sourcenames, intermediates, entryvars, entryvar, aliases):
        self.previous._satisfy(requirement, sourcenames, intermediates, entryvars, entryvar, aliases)

    def _argfcn(self, requirement, sourcenames, intermediates, entryvar, aliases, compilefcn, fcncache):
        return self.previous._argfcn(requirement, sourcenames, intermediates, entryvar, aliases, compilefcn, fcncache)

    def _chain(self, sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, entrysteps, entrystart, entrystop, cache, basketcache, keycache, readexecutor, numba):
        return self.previous._chain(sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, entrysteps, entrystart, entrystop, cache, basketcache, keycache, readexecutor, numba)

    def _endchain(self, compiled, sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, entrysteps, entrystart, entrystop, cache, basketcache, keycache, readexecutor, calcexecutor, numba):
        awaits = self.source._chain(sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, entrysteps, entrystart, entrystop, cache, basketcache, keycache, readexecutor, numba)

        def calculate(await):
            try:
                start, stop, numentries, arrays = await()

                for compiledintermediate in compiledintermediates:
                    compiledintermediate(arrays)

                out = start, stop, numentries, [x(arrays) for x in compiled]

            except:
                return sys.exc_info(), None
            else:
                return None, out

        if calcexecutor is None:
            for await in awaits:
                excinfo, result = calculate(await)
                uproot.tree._delayedraise(excinfo)
                yield result

        else:
            for excinfo, result in calcexecutor.map(calculate, awaits):
                uproot.tree._delayedraise(excinfo)
                yield result
            
    def iterate_newarrays(self, exprs, entrysteps=None, entrystart=None, entrystop=None, aliases={}, interpretations={}, entryvar=None, outputtype=dict, reportentries=False, cache=None, basketcache=None, keycache=None, readexecutor=None, calcexecutor=None, numba=ifinstalled):
        tmpnode, dictnames, compiled, sourcenames, intermediates, compiledintermediates, entryvars, compilefcn = self._prepare(exprs, aliases, entryvar, numba)

        if outputtype == namedtuple:
            for dictname in dictnames:
                if not self._isidentifier(dictname):
                    raise ValueError("illegal field name for namedtuple: {0}".format(repr(dictname)))
            outputtype = namedtuple("Arrays", dictnames)

        if issubclass(outputtype, dict):
            def finish(arrays):
                return outputtype(zip(dictnames, arrays))
        elif outputtype == tuple or outputtype == list:
            def finish(arrays):
                return outputtype(arrays)
        else:
            def finish(arrays):
                return outputtype(*arrays)

        for start, stop, numentries, arrays in tmpnode._endchain(compiled, sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, entrysteps, entrystart, entrystop, cache, basketcache, keycache, readexecutor, calcexecutor, numba):
            if reportentries:
                yield start, stop, numentries, finish(arrays)
            else:
                yield finish(arrays)

    def newarrays(self, exprs, entrystart=None, entrystop=None, aliases={}, interpretations={}, entryvar=None, outputtype=dict, cache=None, basketcache=None, keycache=None, readexecutor=None, calcexecutor=None, numba=ifinstalled):
        tmpnode, dictnames, compiled, sourcenames, intermediates, compiledintermediates, entryvars, compilefcn = self._prepare(exprs, aliases, entryvar, numba)

        if outputtype == namedtuple:
            for dictname in dictnames:
                if not self._isidentifier(dictname):
                    raise ValueError("illegal field name for namedtuple: {0}".format(repr(dictname)))
            outputtype = namedtuple("Arrays", dictnames)

        partitions = []
        totalentries = 0
        for start, stop, numentries, arrays in tmpnode._endchain(compiled, sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, None, entrystart, entrystop, cache, basketcache, keycache, readexecutor, calcexecutor, numba):
            assert len(dictnames) == len(arrays)
            if len(arrays) > 0:
                for array in arrays:
                    assert numentries == len(array)

            partitions.append((totalentries, totalentries + numentries, arrays))
            totalentries += numentries

        if len(partitions) == 0:
            outarrays = [numpy.ones(0, dtype=self.NEW_ARRAY_DTYPE)*222 for i in range(len(dictnames))]
        else:
            outarrays = [numpy.ones(totalentries, dtype=array.dtype)*333 for array in partitions[0][2]]

        for start, stop, arrays in partitions:
            for outarray, array in zip(outarrays, arrays):
                outarray[start:stop] = array
            
        if issubclass(outputtype, dict):
            return outputtype(zip(dictnames, outarrays))
        elif outputtype == tuple or outputtype == list:
            return outputtype(outarrays)
        else:
            return outputtype(*outarrays)

    def newarray(self, expr, entrystart=None, entrystop=None, aliases={}, interpretations={}, entryvar=None, cache=None, basketcache=None, keycache=None, readexecutor=None, numba=ifinstalled):
        if not ChainStep._isfcn(expr):
            raise TypeError("expr must be a single string or function")

        return self.newarrays(expr, entrystart=entrystart, entrystop=entrystop, aliases=aliases, interpretations=interpretations, entryvar=entryvar, outputtype=tuple, cache=cache, basketcache=basketcache, keycache=keycache, readexecutor=readexecutor, numba=numba)[0]

    def _normalize_reduceargs(self, identity, increment, combine):
        if isinstance(identity, parsable):
            identity = self._string2fcn(identity)
        if isinstance(increment, parsable):
            increment = self._string2fcn(increment)
        if isinstance(combine, parsable):
            combine = self._string2fcn(combine)

        if self._isfcn(identity) and self._isfcn(increment) and (combine is None or self._isfcn(combine)):
            if len(self._params(increment)) == 0:
                raise TypeError("increment function must have at least one parameter")

            if combine is None:
                combine = lambda x, y: x + y

            monoidvar = self._params(increment)[0]
            identity  = {monoidvar: identity}
            increment = {monoidvar: increment}
            combine   = {monoidvar: combine}
            order     = [monoidvar]

        if isinstance(increment, dict):
            if not isinstance(identity, dict):
                raise TypeError("if increment is a dict of functions, identity must be as well (to match up argument lists)")
            if not combine is None and not isinstance(combine, dict):
                raise TypeError("if increment is a dict of functions, combine must be as well (to match up argument lists)")

            if len(increment) == 0 or not all(self._isfcn(x) for n, x in increment.items()):
                raise TypeError("increment must be a (non-empty) dict of strings or functions, a (non-empty) iterable of strings or functions, a single string, or a single function")
            increment = dict((n, string2fcn(x) if isinstance(x, parsable) else x) for n, x in increment.items())
            if increment == dict:
                order = sorted(increment)
            else:
                order = list(increment)   # may be an OrderedDict; preserve whatever order it has
        else:
            try:
                assert len(increment) > 0 and all(self._isfcn(x) for x in increment)
            except (TypeError, AssertionError):
                raise TypeError("increment must be a (non-empty) dict of strings or functions, a (non-empty) iterable of strings or functions, a single string, or a single function")
            else:
                # define as a list first so that we get the order
                increment = [string2fcn(x) if isinstance(x, parsable) else x for x in increment]
                order = []
                for x in increment:
                    params = self._params(x)
                    if len(params) == 0:
                        raise TypeError("increment functions must have at least one argument (the aggregator)")
                    order.append(params[0])
                if len(order) != len(set(order)):
                    raise TypeError("if providing a list of increment functions, the aggregator (first argument) of each must be distinct")
                # redefine as a dict
                increment = dict(zip(order, increment))

        if isinstance(identity, dict):
            if not all(self._isfcn(x) for n, x in identity.items()):
                raise TypeError("identity must be a (non-empty) dict of strings or functions, a (non-empty) iterable of strings or functions, a single string, or a single function")
            identity = dict((n, string2fcn(x) if isinstance(x, parsable) else x) for n, x in identity.items())
        else:
            try:
                assert all(self._isfcn(x) for x in identity)
            except (TypeError, AssertionError):
                raise TypeError("identity must be a (non-empty) dict of strings or functions, a (non-empty) iterable of strings or functions, a single string, or a single function")
            else:
                identity = dict(zip(order, [string2fcn(x) if isinstance(x, parsable) else x for x in identity]))

        if not all(len(self._params(x)) == 0 for x in identity.values()):
            raise TypeError("identity functions must have zero arguments")

        if combine is None:
            combine = dict((n, lambda x, y: x + y) for n in order)
        elif isinstance(combine, dict):
            if not all(self._isfcn(x) for n, x in combine.items()):
                raise TypeError("combine must be None, a (non-empty) dict of strings or functions, a (non-empty) iterable of strings or functions, a single string, or a single function")
            combine = dict((n, string2fcn(x) if isinstance(x, parsable) else x) for n, x in combine.items())
        else:
            try:
                assert all(self._isfcn(x) for x in combine)
            except (TypeError, AssertionError):
                raise TypeError("combine must be None, a (non-empty) dict of strings or functions, a (non-empty) iterable of strings or functions, a single string, or a single function")
            else:
                combine = dict(zip(order, [string2fcn(x) if isinstance(x, parsable) else x for x in combine]))
        if not all(len(self._params(x)) == 2 for x in combine.values()):
            raise TypeError("combine functions must have two arguments")

        monoidvars = dict((n, self._params(x)[0]) for n, x in increment.items())

        if not set(identity.keys()) == set(increment.keys()) == set(combine.keys()):
            raise TypeError("if identity, increment, and combine are provided as dicts, they must have the same set of keys (to match up argument lists)")

        return identity, increment, combine, order, monoidvars

    def reduceall(self, identity, increment, combine=None, entrystart=None, entrystop=None, aliases={}, interpretations={}, entryvar=None, outputtype=dict, cache=None, basketcache=None, keycache=None, readexecutor=None, calcexecutor=None, numba=ifinstalled):
        identity, increment, combine, order, monoidvars = self._normalize_reduceargs(identity, increment, combine)

        if outputtype == namedtuple:
            for name in order:
                if not self._isidentifier(name):
                    raise ValueError("illegal field name for namedtuple: {0}".format(repr(name)))
            outputtype = namedtuple("Reduced", order)

        compilefcn = self._compilefcn(numba)

        dependencies = []
        for inc in increment.values():
            dependencies.extend(self._params(inc)[1:])

        # normal preparations for calculating dependencies
        sourcenames = []
        intermediates = []
        entryvars = set()
        for name in dependencies:
            self._satisfy(name, sourcenames, intermediates, entryvars, entryvar, aliases)

        intermediates = Intermediate._dependencyorder(sourcenames, intermediates, entryvar, aliases)
        compiledintermediates = [intermediate._compileintermediate(requirement, sourcenames, intermediates, entryvar, aliases, compilefcn, fcncache) for intermediate, requirement in intermediates]

        # dependencies are unique strings
        avoid = set(dependencies)

        # unique names for dependency getters
        getternames = self._generatenames(dependencies, avoid)
        avoid = avoid.union(getternames.values())

        # unique names for dependency items
        itemnames = self._generatenames(dependencies, avoid)
        avoid = avoid.union(itemnames.values())

        # unique names for monoids
        monoidnames = self._generatenames(monoidvars.values(), avoid)
        avoid = avoid.union(monoidnames.values())

        # unique names for increment functions
        incnames = self._generatenames(increment, avoid)
        avoid = avoid.union(incnames.values())

        # unique names for builtins and dummy variables
        builtins = self._generatenames(["rfcn", "arrays", "numentries", "i", "range"], avoid)
        avoid = avoid.union(builtins.values())

        env = dict([("range", range)] + [(incnames[n], compilefcn(x)) for n, x in increment.items()])

        # getter -> item for each dependency
        fcncache = {}
        itemdefs = []
        for n in dependencies:
            argfcn = self._argfcn(n, sourcenames, intermediates, entryvar, aliases, compilefcn, fcncache)
            env[getternames[n]] = argfcn
            itemdefs.append("{0} = {1}({2})[{3}]".format(itemnames[n], getternames[n], builtins["arrays"], builtins["i"]))

        # call each increment function on its parameters (per item), in declaration or sorted order (not that it matters)
        incfcns = ["{0} = {1}({0}{2})".format(monoidnames[monoidvars[n]], incnames[n], "".join(", " + itemnames[x] for x in self._params(increment[n])[1:])) for n in order]

        # input parameters and output tuple
        monoidargs = [monoidnames[monoidvars[n]] for n in order]

        source = """
def {rfcn}({arrays}, {numentries}, {monoidargs}):
    for {i} in {range}({numentries}):
        {itemdefs}
        {incfcns}
    return ({monoidargs},)
""".format(rfcn=builtins["rfcn"], arrays=builtins["arrays"], numentries=builtins["numentries"], monoidargs=", ".join(monoidargs), i=builtins["i"], range=builtins["range"], itemdefs="\n        ".join(itemdefs), incfcns="\n        ".join(incfcns))

        rfcn = compilefcn(self._makefcn(compile(ast.parse(source), "<reduce>", "exec"), env, builtins["rfcn"], source))

        awaits = self._chain(sourcenames, compiledintermediates, entryvars, aliases, interpretations, entryvar, None, entrystart, entrystop, cache, basketcache, keycache, readexecutor, numba)

        results = [[None for j in range(len(order))] for i in range(len(awaits))]

        def calculate(i):
            try:
                start, stop, numentries, arrays = awaits[i]()

                for compiledintermediate in compiledintermediates:
                    compiledintermediate(arrays)

                inmonoids  = [identity[n]() for n in order]

                outmonoids = rfcn(arrays, numentries, *inmonoids)

                for j, monoid in enumerate(outmonoids):
                    results[i][j] = monoid

            except:
                return sys.exc_info()
            else:
                return None

        if calcexecutor is None:
            for i in range(len(awaits)):
                uproot.tree._delayedraise(calculate(i))
        else:
            excinfos = calcexecutor.map(calculate, range(len(awaits)))
            for excinfo in excinfos:
                uproot.tree._delayedraise(excinfo)

        # MAYBEFIXME: could combine in O(log_2(N)) steps rather than O(N) if combining is ever resource-heavy
        for i in range(1, len(awaits)):
            for j in range(len(order)):
                results[0][j] = combine[order[j]](results[0][j], results[i][j])

        if issubclass(outputtype, dict):
            return outputtype(zip(order, results[0]))
        elif outputtype == tuple or outputtype == list:
            return outputtype(results[0])
        else:
            return outputtype(*results[0])

    def reduce(self, identity, increment, combine=None, entrystart=None, entrystop=None, aliases={}, interpretations={}, entryvar=None, cache=None, basketcache=None, keycache=None, readexecutor=None, calcexecutor=None, numba=ifinstalled):
        if not self._isfcn(identity):
            raise TypeError("identity must be a string or a function")
        if not self._isfcn(increment):
            raise TypeError("increment must be a string or a function")
        if not (combine is None or self._isfcn(combine)):
            raise TypeError("combine must be None, a string, or a function")
        return self.reduceall(identity, increment, combine=combine, entrystart=entrystart, entrystop=entrystop, aliases=aliases, interpretations=interpretations, entryvar=entryvar, outputtype=tuple, cache=cache, basketcache=basketcache, keycache=keycache, readexecutor=readexecutor, calcexecutor=calcexecutor, numba=numba)[0]

class Define(ChainStep):
    def __init__(self, previous, exprs):
        self.previous = previous

        self.fcn = {}
        self.requirements = {}
        self.order = []
        for fcn, requirements, identifier, cacheid, dictname in self._tofcns(exprs):
            self.fcn[dictname] = fcn
            self.requirements[dictname] = requirements
            self.order.append(dictname)

    def _wouldsatisfy(self, requirement, entryvar, aliases):
        if requirement in self.fcn:
{'fraction_non_alphanumeric': 0.09691082113762527, 'fraction_numerical': 0.002973830293417922, 'mean_word_length': 2.6362146929680845, 'pattern_counts': {'":': 2, '<': 11, '<?xml version=': 0, '>': 15, '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': '7180885', 'n_tokens_mistral': 15650, 'n_tokens_neox': 14925, 'n_words': 4692}