<commit_before># -*- coding: utf-8 -*-

"""main ingredient class"""

__all__ = ['Ingredient', 'ConceptIngredient', 'EntityIngredient', 'DataPointIngredient',
           'ingredient_from_dict', 'key_to_list', 'get_ingredient_class']


import fnmatch
import logging
import os
from collections.abc import Mapping, Sequence
from itertools import product
from functools import singledispatch
from typing import Optional, Union, Dict

import shutil

import numpy as np
import pandas as pd
import dask.dataframe as dd

import attr
from abc import ABC, abstractmethod

from ddf_utils.model.package import DDFcsv
from ddf_utils.model.ddf import DDF
from ddf_utils.model.repo import Repo, is_url
from ddf_utils.str import format_float_digits
from ..exceptions import IngredientError
from ..helpers import gen_sym, query, read_opt, sort_df, read_local_ddf, create_dsk


logger = logging.getLogger('Ingredient')


@attr.s
class Ingredient(ABC):
    """Protocol class for all ingredients.

    all ingredients should have following format:

    .. code-block:: yaml

        id: example-ingredient
        dataset: ddf--example--dataset
        key: "geo,time"  # key columns of ingredient
        value:  # only include concepts listed here
          - concept_1
          - concept_2
        filter:  # select rows by column values
          geo:  # only keep datapoint where `geo` is in [swe, usa, chn]
            - swe
            - usa
            - chn

    The other way to define the ingredient data is using the ``data``
    keyword to include external csv file, or inline the data in the
    ingredient definition. Example:

    .. code-block:: yaml

       id: example-ingredient
       key: concept
       data: external_concepts.csv

    On-the-fly ingredient:

    .. code-block:: yaml

       id: example-ingredient
       key: concept
       data:
           - concept: concept_1
             name: concept_name_1
             concept_type: string
             description: concept_description_1
           - concept: concept_2
             name: concept_name_2
             concept_type: measure
             description: concept_description_2
    """
    # TODO: make some of these attributes _Frozen_
    # currently attrs doesn't support frozen attrs, check issue 133 for attrs
    id: str = attr.ib()
    key: Union[list, str] = attr.ib()
    value: Union[list, dict, str] = attr.ib(default='*')
    dataset: str = attr.ib(default=None)
    data: dict = attr.ib(default=None)
    row_filter: dict = attr.ib(default=None)
    base_dir: str = attr.ib(default='./')

    is_procedure_result: bool = attr.ib(default=False, init=False)  # if data is created by a procedure
    data_computed: dict = attr.ib(default=None, init=False)  # cached result from get_data() or a procedure.

    dtype = 'abc'

    @property
    def ingredient_type(self):
        if self.is_procedure_result:
            return 'procedure_result'
        if self.dataset is not None:
            return 'ddf'
        else:
            if isinstance(self.data, str):
                return 'external'
            else:
                return 'inline'

    @property
    def dataset_path(self):
        """return the full path to ingredient's dataset if the ingredient is from local ddf dataset."""
        if self.ingredient_type == 'ddf':
            if os.path.isabs(self.dataset):
                return self.dataset
            return os.path.join(self.base_dir, self.dataset)
        return None

    @property
    def ddf_id(self):
        if self.ingredient_type == 'ddf':
            ddf = read_local_ddf(self.dataset_path)
            return ddf.props.get('name', self.dataset)
        return None

    @classmethod
    def from_procedure_result(cls, id, key, data_computed: dict):
        res = cls(id, key)
        for _, df in data_computed.items():
            if isinstance(key, list):
                for k in key:
                    assert k in df.columns, "the key {} not in data!".format(k)
            else:
                assert key in df.columns, "the key {} not in data!".format(key)
        res.data_computed = data_computed
        res.is_procedure_result = True
        return res

    @abstractmethod
    def get_data(self):
        ...

    @abstractmethod
    def serve(self, *args, **kwargs):
        """serving data to disk"""
        ...

    @staticmethod
    def filter_row(data: dict, row_filter):
        """return the rows selected by row_filter."""
        if row_filter is None:
            return data
        for k, df in data.items():<commit_msg>add DDF property of ingredient<commit_after># -*- coding: utf-8 -*-

"""main ingredient class"""

__all__ = ['Ingredient', 'ConceptIngredient', 'EntityIngredient', 'DataPointIngredient',
           'ingredient_from_dict', 'key_to_list', 'get_ingredient_class']


import fnmatch
import logging
import os
from collections.abc import Mapping, Sequence
from itertools import product
from functools import singledispatch
from typing import Optional, Union, Dict

import shutil

import numpy as np
import pandas as pd
import dask.dataframe as dd

import attr
from abc import ABC, abstractmethod

from ddf_utils.model.package import DDFcsv
from ddf_utils.model.ddf import DDF
from ddf_utils.model.repo import Repo, is_url
from ddf_utils.str import format_float_digits
from ..exceptions import IngredientError
from ..helpers import gen_sym, query, read_opt, sort_df, read_local_ddf, create_dsk


logger = logging.getLogger('Ingredient')


@attr.s
class Ingredient(ABC):
    """Protocol class for all ingredients.

    all ingredients should have following format:

    .. code-block:: yaml

        id: example-ingredient
        dataset: ddf--example--dataset
        key: "geo,time"  # key columns of ingredient
        value:  # only include concepts listed here
          - concept_1
          - concept_2
        filter:  # select rows by column values
          geo:  # only keep datapoint where `geo` is in [swe, usa, chn]
            - swe
            - usa
            - chn

    The other way to define the ingredient data is using the ``data``
    keyword to include external csv file, or inline the data in the
    ingredient definition. Example:

    .. code-block:: yaml

       id: example-ingredient
       key: concept
       data: external_concepts.csv

    On-the-fly ingredient:

    .. code-block:: yaml

       id: example-ingredient
       key: concept
       data:
           - concept: concept_1
             name: concept_name_1
             concept_type: string
             description: concept_description_1
           - concept: concept_2
             name: concept_name_2
             concept_type: measure
             description: concept_description_2
    """
    # TODO: make some of these attributes _Frozen_
    # currently attrs doesn't support frozen attrs, check issue 133 for attrs
    id: str = attr.ib()
    key: Union[list, str] = attr.ib()
    value: Union[list, dict, str] = attr.ib(default='*')
    dataset: str = attr.ib(default=None)
    data: dict = attr.ib(default=None)
    row_filter: dict = attr.ib(default=None)
    base_dir: str = attr.ib(default='./')

    is_procedure_result: bool = attr.ib(default=False, init=False)  # if data is created by a procedure
    data_computed: dict = attr.ib(default=None, init=False)  # cached result from get_data() or a procedure.

    dtype = 'abc'

    @property
    def ingredient_type(self):
        if self.is_procedure_result:
            return 'procedure_result'
        if self.dataset is not None:
            return 'ddf'
        else:
            if isinstance(self.data, str):
                return 'external'
            else:
                return 'inline'

    @property
    def dataset_path(self):
        """return the full path to ingredient's dataset if the ingredient is from local ddf dataset."""
        if self.ingredient_type == 'ddf':
            if os.path.isabs(self.dataset):
                return self.dataset
            return os.path.join(self.base_dir, self.dataset)
        return None

    @property
    def ddf_id(self):
        if self.ingredient_type == 'ddf':
            return self.ddf.props.get('name', self.dataset)
        return None

    @property
    def ddf(self):
        if self.ingredient_type == 'ddf':
            ddf = read_local_ddf(self.dataset_path)
            return ddf
        return None

    @classmethod
    def from_procedure_result(cls, id, key, data_computed: dict):
        res = cls(id, key)
        for _, df in data_computed.items():
            if isinstance(key, list):
                for k in key:
                    assert k in df.columns, "the key {} not in data!".format(k)
            else:
                assert key in df.columns, "the key {} not in data!".format(key)
        res.data_computed = data_computed
        res.is_procedure_result = True
        return res

    @abstractmethod
    def get_data(self):
        ...

    @abstractmethod
    def serve(self, *args, **kwargs):
        """serving data to disk"""
        ...

    @staticmethod
    def filter_row(data: dict, row_filter):
        """return the rows selected by row_filter."""
        if row_filter is None:
            return data
        for k, df in data.items():
{'fraction_non_alphanumeric': 0.0889487870619946, 'fraction_numerical': 0.0025876010781671157, 'mean_word_length': 2.302242791028836, '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': '3848673', 'n_tokens_mistral': 2756, 'n_tokens_neox': 2659, 'n_words': 849}