<commit_before>"""

>>> from blaze.expr import symbol
>>> from blaze.compute.pandas import compute

>>> accounts = symbol('accounts', 'var * {name: string, amount: int}')
>>> deadbeats = accounts[accounts['amount'] < 0]['name']

>>> from pandas import DataFrame
>>> data = [['Alice', 100], ['Bob', -50], ['Charlie', -20]]
>>> df = DataFrame(data, columns=['name', 'amount'])
>>> compute(deadbeats, df)
1        Bob
2    Charlie
Name: name, dtype: object
"""
from __future__ import absolute_import, division, print_function

import fnmatch
import itertools
from distutils.version import LooseVersion
import warnings

import numpy as np

import pandas as pd

from pandas.core.generic import NDFrame
from pandas import DataFrame, Series
from pandas.core.groupby import DataFrameGroupBy, SeriesGroupBy

from toolz import merge as merge_dicts
from toolz.curried import pipe, filter, map, concat

import datashape

from datashape import to_numpy_dtype
from datashape.predicates import isscalar

from odo import into
try:
    import dask.dataframe as dd
    DaskDataFrame = dd.DataFrame
    DaskSeries = dd.Series
except ImportError:
    DaskDataFrame = pd.DataFrame
    DaskSeries = pd.Series

from ..dispatch import dispatch

from .core import compute, compute_up, base

from ..expr import (Projection, Field, Sort, Head, Tail, Broadcast, Selection,
                    Reduction, Distinct, Join, By, Summary, Label, ReLabel,
                    Map, Apply, Merge, std, var, Like, Slice, summary,
                    ElemWise, DateTime, Millisecond, Expr, Symbol, IsIn,
                    UTCFromTimestamp, nelements, DateTimeTruncate, count,
                    UnaryStringFunction, nunique, Coerce, Concat,
                    isnan, notnull, Shift)
from ..expr import UnaryOp, BinOp, Interp
from ..expr import symbol, common_subexpression

from ..compatibility import _inttypes

__all__ = []


@dispatch(Projection, (DataFrame, DaskDataFrame))
def compute_up(t, df, **kwargs):
    return df[list(t.fields)]


@dispatch(Field, (DataFrame, DataFrameGroupBy, DaskDataFrame))
def compute_up(t, df, **kwargs):
    assert len(t.fields) == 1
    return df[t.fields[0]]


@dispatch(Field, (Series, DaskSeries))
def compute_up(t, data, **kwargs):
    assert len(t.fields) == 1
    if t.fields[0] == data.name:
        return data
    else:
        raise ValueError("Fieldname %r does not match Series name %r"
                         % (t.fields[0], data.name))


@dispatch(Broadcast, (DataFrame, DaskDataFrame))
def compute_up(t, df, **kwargs):
    d = dict((t._child[c]._expr, df[c]) for c in t._child.fields)
    return compute(t._expr, d)


@dispatch(Broadcast, Series)
def compute_up(t, s, **kwargs):
    return compute_up(t, s.to_frame(), **kwargs)


@dispatch(Interp, Series)
def compute_up(t, data, **kwargs):
    if isinstance(t.lhs, Expr):
        return data % t.rhs
    else:
        return t.lhs % data


@compute_up.register(Interp, Series, (Series, base))
@compute_up.register(Interp, base, Series)
def compute_up_pd_interp(t, lhs, rhs, **kwargs):
    return lhs % rhs



@dispatch(BinOp, (Series, DaskSeries))
def compute_up(t, data, **kwargs):
    if isinstance(t.lhs, Expr):
        return t.op(data, t.rhs)
    else:
        return t.op(t.lhs, data)


@dispatch(BinOp, (Series, DaskSeries), (Series, base, DaskSeries))
def compute_up(t, lhs, rhs, **kwargs):
    return t.op(lhs, rhs)


@dispatch(BinOp, (Series, base, DaskSeries), (Series, DaskSeries))
def compute_up(t, lhs, rhs, **kwargs):
    return t.op(lhs, rhs)


@dispatch(UnaryOp, NDFrame)
def compute_up(t, df, **kwargs):
    f = getattr(t, 'op', getattr(np, t.symbol, None))
    if f is None:
        raise ValueError('%s is not a valid operation on %s objects' %
                         (t.symbol, type(df).__name__))
    return f(df)


@dispatch(Selection, (Series, DataFrame, DaskSeries, DaskDataFrame))
def compute_up(expr, df, **kwargs):
    return compute_up(
        expr,
        df,
        compute(expr.predicate, {expr._child: df}),
        **kwargs
    )


@dispatch(Selection, (Series, DataFrame, DaskSeries, DaskDataFrame), 
                     (Series, DaskSeries))
def compute_up(expr, df, predicate, **kwargs):
    return df[predicate]


@dispatch(Join, DataFrame, DataFrame)
def compute_up(t, lhs, rhs, **kwargs):
    """ Join two pandas data frames on arbitrary columns

    The approach taken here could probably be improved.

    To join on two columns we force each column to be the index of the
    dataframe, perform the join, and then reset the index back to the left
    side's original index.
    """
    result = pd.merge(
        lhs,
        rhs,
        left_on=t.on_left,
        right_on=t.on_right,
        how=t.how,<commit_msg>Cleanup some trailing whitespace<commit_after>"""

>>> from blaze.expr import symbol
>>> from blaze.compute.pandas import compute

>>> accounts = symbol('accounts', 'var * {name: string, amount: int}')
>>> deadbeats = accounts[accounts['amount'] < 0]['name']

>>> from pandas import DataFrame
>>> data = [['Alice', 100], ['Bob', -50], ['Charlie', -20]]
>>> df = DataFrame(data, columns=['name', 'amount'])
>>> compute(deadbeats, df)
1        Bob
2    Charlie
Name: name, dtype: object
"""
from __future__ import absolute_import, division, print_function

import fnmatch
import itertools
from distutils.version import LooseVersion
import warnings

import numpy as np

import pandas as pd

from pandas.core.generic import NDFrame
from pandas import DataFrame, Series
from pandas.core.groupby import DataFrameGroupBy, SeriesGroupBy

from toolz import merge as merge_dicts
from toolz.curried import pipe, filter, map, concat

import datashape

from datashape import to_numpy_dtype
from datashape.predicates import isscalar

from odo import into
try:
    import dask.dataframe as dd
    DaskDataFrame = dd.DataFrame
    DaskSeries = dd.Series
except ImportError:
    DaskDataFrame = pd.DataFrame
    DaskSeries = pd.Series

from ..dispatch import dispatch

from .core import compute, compute_up, base

from ..expr import (Projection, Field, Sort, Head, Tail, Broadcast, Selection,
                    Reduction, Distinct, Join, By, Summary, Label, ReLabel,
                    Map, Apply, Merge, std, var, Like, Slice, summary,
                    ElemWise, DateTime, Millisecond, Expr, Symbol, IsIn,
                    UTCFromTimestamp, nelements, DateTimeTruncate, count,
                    UnaryStringFunction, nunique, Coerce, Concat,
                    isnan, notnull, Shift)
from ..expr import UnaryOp, BinOp, Interp
from ..expr import symbol, common_subexpression

from ..compatibility import _inttypes

__all__ = []


@dispatch(Projection, (DataFrame, DaskDataFrame))
def compute_up(t, df, **kwargs):
    return df[list(t.fields)]


@dispatch(Field, (DataFrame, DataFrameGroupBy, DaskDataFrame))
def compute_up(t, df, **kwargs):
    assert len(t.fields) == 1
    return df[t.fields[0]]


@dispatch(Field, (Series, DaskSeries))
def compute_up(t, data, **kwargs):
    assert len(t.fields) == 1
    if t.fields[0] == data.name:
        return data
    else:
        raise ValueError("Fieldname %r does not match Series name %r"
                         % (t.fields[0], data.name))


@dispatch(Broadcast, (DataFrame, DaskDataFrame))
def compute_up(t, df, **kwargs):
    d = dict((t._child[c]._expr, df[c]) for c in t._child.fields)
    return compute(t._expr, d)


@dispatch(Broadcast, Series)
def compute_up(t, s, **kwargs):
    return compute_up(t, s.to_frame(), **kwargs)


@dispatch(Interp, Series)
def compute_up(t, data, **kwargs):
    if isinstance(t.lhs, Expr):
        return data % t.rhs
    else:
        return t.lhs % data


@compute_up.register(Interp, Series, (Series, base))
@compute_up.register(Interp, base, Series)
def compute_up_pd_interp(t, lhs, rhs, **kwargs):
    return lhs % rhs



@dispatch(BinOp, (Series, DaskSeries))
def compute_up(t, data, **kwargs):
    if isinstance(t.lhs, Expr):
        return t.op(data, t.rhs)
    else:
        return t.op(t.lhs, data)


@dispatch(BinOp, (Series, DaskSeries), (Series, base, DaskSeries))
def compute_up(t, lhs, rhs, **kwargs):
    return t.op(lhs, rhs)


@dispatch(BinOp, (Series, base, DaskSeries), (Series, DaskSeries))
def compute_up(t, lhs, rhs, **kwargs):
    return t.op(lhs, rhs)


@dispatch(UnaryOp, NDFrame)
def compute_up(t, df, **kwargs):
    f = getattr(t, 'op', getattr(np, t.symbol, None))
    if f is None:
        raise ValueError('%s is not a valid operation on %s objects' %
                         (t.symbol, type(df).__name__))
    return f(df)


@dispatch(Selection, (Series, DataFrame, DaskSeries, DaskDataFrame))
def compute_up(expr, df, **kwargs):
    return compute_up(
        expr,
        df,
        compute(expr.predicate, {expr._child: df}),
        **kwargs
    )


@dispatch(Selection, (Series, DataFrame, DaskSeries, DaskDataFrame),
                     (Series, DaskSeries))
def compute_up(expr, df, predicate, **kwargs):
    return df[predicate]


@dispatch(Join, DataFrame, DataFrame)
def compute_up(t, lhs, rhs, **kwargs):
    """ Join two pandas data frames on arbitrary columns

    The approach taken here could probably be improved.

    To join on two columns we force each column to be the index of the
    dataframe, perform the join, and then reset the index back to the left
    side's original index.
    """
    result = pd.merge(
        lhs,
        rhs,
        left_on=t.on_left,
        right_on=t.on_right,
        how=t.how,
{'fraction_non_alphanumeric': 0.1218381112984823, 'fraction_numerical': 0.0031618887015177066, 'mean_word_length': 3.3910226746876444, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 51, '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': '5180790', 'n_tokens_mistral': 3121, 'n_tokens_neox': 3025, 'n_words': 936}