import os from collections.abc import Sequence from typing import Any from bs4 import BeautifulSoup from evidently.metric_preset import DataDriftPreset from evidently.metrics import ( ColumnSummaryMetric, ConflictTargetMetric, DatasetCorrelationsMetric, DatasetSummaryMetric, RegressionAbsPercentageErrorPlot, RegressionDummyMetric, RegressionErrorDistribution, RegressionErrorPlot, RegressionPerformanceMetrics, RegressionPredictedVsActualPlot, RegressionPredictedVsActualScatter, ) from evidently.metrics.base_metric import generate_column_metrics from evidently.options import ColorOptions from evidently.report import Report COLOR_DISCRETE_SEQUENCE = ( '#ed0400', '#0a5f38', '#6c3461', '#71aa34', '#d8dcd6', '#6b8ba4', ) def load_html_from_file(file_path): try: with open(file_path, encoding='utf-8') as file: return file.read() except FileNotFoundError: return None except OSError: # noqa: BLE001 return None def inject_content(main_html, section_id, content): soup = BeautifulSoup(main_html, 'html.parser') section = soup.find(id=section_id) if section: section.clear() section.append(BeautifulSoup(content, 'html.parser')) return str(soup) class Reports: """ Report generator using Evidently library. Thread-safety: This class is NOT thread-safe. Multiple threads should not call add_*_section() methods on the same instance simultaneously as they modify shared state (self.metrics, self.sections, self.options). For multi-threaded environments: - Create separate Reports instances per thread - Or synchronize access using locks - After generation, instances are safe for read-only operations I/O Note: This class relies on Evidently's report.save_html() method for file operations. Ensure Evidently properly manages file handles. """ def __init__( self, reference_data: Any, current_data: Any, base_path: str | None = None ) -> None: """ Initializes an instance of the AigReport class. Args: reference_data: The reference data for the report. current_data: The current data for the report. base_path: The base path for the report. """ self.metrics: list[Any] = [] self.options: list[Any] | None = None self.sections: dict[str, Any] = {} self.report: Any = None self.ref_data = reference_data self.cur_data = current_data self.set_color_options(primary_color='#0F4C81', secondary_color='#001E60') self.base_path = base_path def add_data_quality_section(self, columns: list[str] | None = None, run: bool = True) -> None: """ Adds a data quality section to the report. Args: columns: The list of columns to include in the data quality section. If None, all columns will be included. run: Indicates whether to run the report immediately after adding the section. """ metrics = [ DatasetSummaryMetric(), generate_column_metrics(ColumnSummaryMetric, columns=columns, skip_id_column=True), ConflictTargetMetric(), DatasetCorrelationsMetric(), ] self.metrics.extend(metrics) if run: report = Report(metrics=metrics, options=self.options) report.run(reference_data=self.ref_data, current_data=self.cur_data) self.sections['data_quality'] = report.as_dict() if self.base_path: # Note: Relies on Evidently's save_html() to properly manage file I/O report.save_html(os.path.join(self.base_path, 'data_quality.html')) def add_data_drift_section(self, columns: list[str] | None = None, run: bool = True) -> None: """ Adds a data drift section to the report. Args: columns: The list of columns to include in the data drift section. If None, all columns will be included. run: Indicates whether to run the report immediately after adding the section. """ self.metrics.append(DataDriftPreset(columns=columns)) if run: report = Report(metrics=[DataDriftPreset(columns=columns)], options=self.options) report.run(reference_data=self.ref_data, current_data=self.cur_data) self.sections['data_drift'] = report.as_dict() if self.base_path: # Note: Relies on Evidently's save_html() to properly manage file I/O report.save_html(os.path.join(self.base_path, 'data_drift.html')) def add_regression_section(self, run: bool = True) -> None: """ Adds a regression section to the report. Args: run: Indicates whether to run the report immediately after adding the section. """ metrics = [ RegressionPerformanceMetrics(), RegressionDummyMetric(), RegressionPredictedVsActualScatter(), RegressionPredictedVsActualPlot(), RegressionErrorPlot(), RegressionAbsPercentageErrorPlot(), RegressionErrorDistribution(), ] self.metrics.extend(metrics) if run: report = Report(metrics=metrics, options=self.options) report.run(reference_data=self.ref_data, current_data=self.cur_data) self.sections['regression'] = report.as_dict() if self.base_path: # Note: Relies on Evidently's save_html() to properly manage file I/O report.save_html(os.path.join(self.base_path, 'regression.html')) def set_color_options( self, primary_color: str = '#0F4C81', secondary_color: str = '#001E60', current_data_color: str | None = None, reference_data_color: str | None = None, additional_data_color: str = '#0a5f38', color_sequence: Sequence[str] = COLOR_DISCRETE_SEQUENCE, fill_color: str = 'LightGreen', zero_line_color: str = 'green', non_visible_color: str = 'white', underestimation_color: str = '#6574f7', overestimation_color: str = '#ee5540', majority_color: str = '#1acc98', vertical_lines: str = 'green', heatmap: str = 'RdBu_r', ) -> None: """ Sets the color options for the report. Args: primary_color: The primary color for the report. secondary_color: The secondary color for the report. current_data_color: The color for the current data. reference_data_color: The color for the reference data. additional_data_color: The color for additional data. color_sequence: The color sequence for discrete values. fill_color: The fill color for visualizations. zero_line_color: The color for the zero line. non_visible_color: The color for non-visible elements. underestimation_color: The color for underestimation. overestimation_color: The color for overestimation. majority_color: The color for majority elements. vertical_lines: The color for vertical lines. heatmap: The color map for heatmaps. """ color_scheme = ColorOptions( primary_color=primary_color, secondary_color=secondary_color, current_data_color=current_data_color, reference_data_color=reference_data_color, additional_data_color=additional_data_color, color_sequence=color_sequence, fill_color=fill_color, zero_line_color=zero_line_color, non_visible_color=non_visible_color, underestimation_color=underestimation_color, overestimation_color=overestimation_color, majority_color=majority_color, vertical_lines=vertical_lines, heatmap=heatmap, ) if self.options is None: self.options = [color_scheme] else: self.options.append(color_scheme) def save_all_sections_html(self, report_path): """ Saves the report with all sections as HTML. Args: report_path: The path to save the report HTML file. Raises: ValueError: If base_path is not set OSError: If directory creation or file writing fails Note: This method uses context manager (with open) to ensure file is properly closed. Creates parent directories if they don't exist. """ if not self.base_path: raise ValueError('base_path is required to save all sections HTML') # Ensure output directory exists output_dir = os.path.dirname(report_path) if output_dir and not os.path.exists(output_dir): os.makedirs(output_dir, exist_ok=True) # Load main HTML template main_html_path = os.path.join(self.base_path, 'header.html') main_html = load_html_from_file(main_html_path) # Load content from data_drift.html, data_quality.html, and regression.html data_drift_content = load_html_from_file(os.path.join(self.base_path, 'data_drift.html')) data_quality_content = load_html_from_file( os.path.join(self.base_path, 'data_quality.html') ) regression_content = load_html_from_file(os.path.join(self.base_path, 'regression.html')) # Inject content into the main HTML template main_html = inject_content(main_html, 'data_drift', data_drift_content) main_html = inject_content(main_html, 'data_quality', data_quality_content) main_html = inject_content(main_html, 'regression', regression_content) # Save the final HTML to a new file (report.html) # Context manager ensures file is properly closed even if an error occurs with open(report_path, 'w', encoding='utf-8') as report_file: report_file.write(main_html)