260 lines
10 KiB
Python
260 lines
10 KiB
Python
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:
|
|
print(f'File not found: {file_path}')
|
|
return None
|
|
except OSError as e: # noqa: BLE001
|
|
print(f'Error reading file: {e}')
|
|
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'))
|
|
else:
|
|
print(f"Section with id '{section_id}' not found in the main HTML template.")
|
|
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)
|