<commit_before>from __future__ import print_function, division

import pandas as pd
import numpy as np
import re
import fastparquet
import logging
import dask.dataframe as dd
import dask.array as da
import dask
import time

from .catalog import Catalog, MatchedCatalog

class Functor(object):
    """Performs computation on catalog(s), read from disk

    Catalog is a Catalog object

    Subclasses must define _columns attribute that is read from table
    """
    _allow_difference = True

    def __init__(self, allow_difference=None):
        if allow_difference is not None:
            self.allow_difference = allow_difference
        else:
            self.allow_difference = self._allow_difference

    @property
    def columns(self):
        try:
            return self._columns
        except AttributeError:
            raise NotImplementedError('Must define columns property or _columns attribute')

    def _func(self, df, dropna=True):
        raise NotImplementedError('Must define calculation on dataframe')

    def __call__(self, catalog, query=None, dropna=True, dask=False):

        if isinstance(catalog, MatchedCatalog):
            df1, df2 = catalog.get_columns(self.columns, query=query)
            if self.allow_difference:
                vals = self._func(df1) - self._func(df2)
            else:<commit_msg>make able to just pass dataframe<commit_after>from __future__ import print_function, division

import pandas as pd
import numpy as np
import re
import fastparquet
import logging
import dask.dataframe as dd
import dask.array as da
import dask
import time

from .catalog import Catalog, MatchedCatalog

class Functor(object):
    """Performs computation on catalog(s), read from disk

    Catalog is a Catalog object

    Subclasses must define _columns attribute that is read from table
    """
    _allow_difference = True

    def __init__(self, allow_difference=None):
        if allow_difference is not None:
            self.allow_difference = allow_difference
        else:
            self.allow_difference = self._allow_difference

    @property
    def columns(self):
        try:
            return self._columns
        except AttributeError:
            raise NotImplementedError('Must define columns property or _columns attribute')

    def _func(self, df, dropna=True):
        raise NotImplementedError('Must define calculation on dataframe')

    def __call__(self, catalog, query=None, dropna=True, dask=False):

        if isinstance(catalog, dd.DataFrame):
            vals = self._func(catalog)
            
        elif isinstance(catalog, MatchedCatalog):
            df1, df2 = catalog.get_columns(self.columns, query=query)
            if self.allow_difference:
                vals = self._func(df1) - self._func(df2)
            else:
{'fraction_non_alphanumeric': 0.06533380935380222, 'fraction_numerical': 0.002856122813280971, 'mean_word_length': 2.88626907073509, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '617333', 'n_tokens_mistral': 765, 'n_tokens_neox': 724, 'n_words': 249}