container.py
1 import abc 2 import itertools 3 from typing import TYPE_CHECKING 4 from typing import ClassVar 5 from typing import Generator 6 from typing import List 7 from typing import Optional 8 from typing import Sequence 9 from typing import Tuple 10 from typing import Union 11 12 from evidently.core.metric_types import Metric 13 from evidently.core.metric_types import MetricId 14 from evidently.core.metric_types import convert_tests 15 from evidently.legacy.model.widget import BaseWidgetInfo 16 from evidently.pydantic_utils import AutoAliasMixin 17 from evidently.pydantic_utils import EvidentlyBaseModel 18 19 if TYPE_CHECKING: 20 from evidently.core.report import Context 21 22 MetricOrContainer = Union[Metric, "MetricContainer"] 23 24 25 class MetricContainer(AutoAliasMixin, EvidentlyBaseModel, abc.ABC): 26 """Base class for containers that generate multiple metrics. 27 28 Metric containers are used to programmatically create multiple related metrics, 29 such as generating the same metric for multiple columns or creating metric combinations. 30 Examples include `ColumnMetricGenerator` and preset classes like `DataDriftPreset`. 31 """ 32 33 __alias_type__: ClassVar[str] = "metric_container" 34 35 class Config: 36 is_base_type = True 37 38 include_tests: bool = True 39 """Whether to include default tests for generated metrics.""" 40 41 def __init__(self, include_tests: bool = True, **data): 42 """Initialize a metric container. 43 44 Args: 45 * `include_tests`: If `True`, generated metrics will include default tests. 46 """ 47 self.include_tests = include_tests 48 super().__init__(**data) 49 50 @abc.abstractmethod 51 def generate_metrics(self, context: "Context") -> Sequence[MetricOrContainer]: 52 """Generate metrics based on the container configuration. 53 54 Args: 55 * `context`: `Context` containing datasets and configuration. 56 57 Returns: 58 * Sequence of `Metric` or `MetricContainer` objects to compute. 59 """ 60 raise NotImplementedError() 61 62 def metrics(self, context: "Context") -> List[MetricOrContainer]: 63 """Get all metrics generated by this container. 64 65 Results are cached in the context to avoid regenerating on subsequent calls. 66 67 Args: 68 * `context`: `Context` containing datasets and configuration. 69 70 Returns: 71 * List of `Metric` or `MetricContainer` objects. 72 """ 73 metric_container_fp = self.get_fingerprint() 74 metrics = context.metrics_container(metric_container_fp) 75 if metrics is None: 76 metrics = list(self.generate_metrics(context)) 77 context.set_metric_container_data(metric_container_fp, metrics) 78 return metrics 79 80 def render( 81 self, 82 context: "Context", 83 child_widgets: Optional[List[Tuple[Optional[MetricId], List[BaseWidgetInfo]]]] = None, 84 ) -> List[BaseWidgetInfo]: 85 """Render visualization widgets for this container. 86 87 Combines widgets from all child metrics/containers. 88 89 Args: 90 * `context`: `Context` containing datasets and configuration. 91 * `child_widgets`: Optional list of (metric_id, widgets) tuples from child metrics. 92 93 Returns: 94 * List of `BaseWidgetInfo` objects for visualization. 95 """ 96 return list(itertools.chain(*[widget[1] for widget in (child_widgets or [])])) 97 98 def list_metrics(self, context: "Context") -> Generator[Metric, None, None]: 99 """Iterate over all leaf metrics in this container. 100 101 Recursively yields all `Metric` objects, flattening nested containers. 102 103 Args: 104 * `context`: `Context` containing datasets and configuration. 105 106 Yields: 107 * `Metric` objects from this container and nested containers. 108 109 Raises: 110 * `ValueError`: If metrics haven't been generated yet. 111 """ 112 metrics = context.metrics_container(self.get_fingerprint()) 113 if metrics is None: 114 raise ValueError("Metrics weren't composed in container") 115 for item in metrics: 116 if isinstance(item, Metric): 117 yield item 118 elif isinstance(item, MetricContainer): 119 yield from item.list_metrics(context) 120 else: 121 raise ValueError(f"invalid metric type {type(item)}") 122 123 def _get_tests(self, tests): 124 """Get tests list, handling None and include_tests flag. 125 126 Args: 127 * `tests`: Optional list of tests. 128 129 Returns: 130 * Converted tests list, or None if default tests should be used, or empty list if tests disabled. 131 """ 132 if tests is not None: 133 return convert_tests(tests) 134 if self.include_tests: 135 return None 136 return [] 137 138 139 MetricOrContainer = Union[Metric, MetricContainer] 140 141 142 class ColumnMetricContainer(MetricContainer, abc.ABC): 143 """Base class for metric containers that operate on a specific column. 144 145 Simplifies container implementation for containers that generate metrics 146 for a single column. Subclasses only need to implement `generate_metrics()`. 147 """ 148 149 column: str 150 """Name of the column to generate metrics for.""" 151 152 def __init__(self, column: str, include_tests: bool = True): 153 """Initialize a column metric container. 154 155 Args: 156 * `column`: Name of the column to generate metrics for. 157 * `include_tests`: If `True`, generated metrics will include default tests. 158 """ 159 self.column = column 160 super().__init__(include_tests=include_tests)