1  from typing import Callable
 2  from typing import Tuple
 3  
 4  import pandas as pd
 5  
 6  from evidently._pydantic_compat import Field
 7  from evidently.legacy.base_metric import ColumnName
 8  from evidently.legacy.core import ColumnType
 9  from evidently.legacy.core import new_id
10  from evidently.legacy.features.generated_features import FeatureTypeFieldMixin
11  from evidently.legacy.features.generated_features import GeneratedFeature
12  from evidently.legacy.utils.data_preprocessing import DataDefinition
13  from evidently.pydantic_utils import FingerprintPart
14  
15  
16  class CustomFeature(FeatureTypeFieldMixin, GeneratedFeature):
17      class Config:
18          type_alias = "evidently:feature:CustomFeature"
19  
20      display_name: str
21      name: str = Field(default_factory=lambda: str(new_id()))
22      func: Callable[[pd.DataFrame, DataDefinition], pd.Series]
23      feature_type: ColumnType = ColumnType.Numerical
24  
25      def generate_feature(self, data: pd.DataFrame, data_definition: DataDefinition) -> pd.DataFrame:
26          result = self.func(data, data_definition)
27          return pd.DataFrame({self.name: result})
28  
29      def _as_column(self) -> "ColumnName":
30          return self._create_column(self.name)
31  
32  
33  class CustomSingleColumnFeature(FeatureTypeFieldMixin, GeneratedFeature):
34      class Config:
35          type_alias = "evidently:feature:CustomSingleColumnFeature"
36  
37      display_name: str
38      func: Callable[[pd.Series], pd.Series]
39      name: str = Field(default_factory=lambda: str(new_id()))
40      column_name: str
41      feature_type: ColumnType = ColumnType.Numerical
42  
43      def generate_feature(self, data: pd.DataFrame, data_definition: DataDefinition) -> pd.DataFrame:
44          result = self.func(data[self.column_name])
45          return pd.DataFrame({self.name: result}, index=data.index)
46  
47      def _as_column(self) -> "ColumnName":
48          return self._create_column(self.name)
49  
50      def get_fingerprint_parts(self) -> Tuple[FingerprintPart, ...]:
51          return tuple(
52              (name, self.get_field_fingerprint(name))
53              for name, field in sorted(self.__fields__.items())
54              if (field.required or getattr(self, name) != field.get_default()) and field.name != "func"
55          )
56  
57  
58  class CustomPairColumnFeature(FeatureTypeFieldMixin, GeneratedFeature):
59      class Config:
60          type_alias = "evidently:feature:CustomPairColumnFeature"
61  
62      display_name: str
63      func: Callable[[pd.Series, pd.Series], pd.Series]
64      name: str = Field(default_factory=lambda: str(new_id()))
65      first_column: str
66      second_column: str
67      feature_type: ColumnType = ColumnType.Numerical
68  
69      def generate_feature(self, data: pd.DataFrame, data_definition: DataDefinition) -> pd.DataFrame:
70          result = self.func(data[self.first_column], data[self.second_column])
71          return pd.DataFrame({self.name: result})
72  
73      def _as_column(self) -> "ColumnName":
74          return self._create_column(self.name)