sientia-dataops-model-manager/model_manager/sientia/reports.py

import os
from collections.abc import Sequence
from typing import Any

from bs4 import BeautifulSoup, Tag
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.pipeline.column_mapping import ColumnMapping
from evidently.report import Report

COLOR_DISCRETE_SEQUENCE = (
    '#ed0400',
    '#0a5f38',
    '#6c3461',
    '#71aa34',
    '#d8dcd6',
    '#6b8ba4',
)


def load_html_from_file(file_path):
    with open(file_path, encoding='utf-8') as file:
        return file.read()


def inject_content(main_html, section_id, content):
    soup = BeautifulSoup(main_html, 'html.parser')
    section = soup.find(id=section_id)

    # Verifica se a seção foi encontrada E se ela é uma Tag (não uma string)
    if section and isinstance(section, Tag):
        section.clear()
        # Converte o conteúdo para um fragmento de BeautifulSoup e anexa
        new_content = BeautifulSoup(content, 'html.parser')
        section.append(new_content)

    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,
        target_name: str,
        base_path: str | None = None,
        template_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.target_name = target_name
        self.set_color_options(primary_color='#0F4C81', secondary_color='#001E60')
        self.base_path = base_path
        self.template_path = template_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:
            mapping = ColumnMapping()
            mapping.target = self.target_name
            report = Report(metrics=metrics, options=self.options)
            report.run(
                reference_data=self.ref_data,
                current_data=self.cur_data,
                column_mapping=mapping,
            )
            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:
            mapping = ColumnMapping()
            mapping.target = self.target_name
            report = Report(metrics=[DataDriftPreset(columns=columns)], options=self.options)
            report.run(
                reference_data=self.ref_data,
                current_data=self.cur_data,
                column_mapping=mapping,
            )
            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:
            mapping = ColumnMapping()

            mapping.target = self.target_name
            mapping.prediction = 'prediction'

            report = Report(metrics=metrics, options=self.options)
            report.run(
                reference_data=self.ref_data,
                current_data=self.cur_data,
                column_mapping=mapping,
            )
            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')

        if not self.template_path:
            raise ValueError('template_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)

        print(f'Output directory: {output_dir}')
        print(f'Report path: {report_path}')
        print(f'Base path: {self.base_path}')
        print(f'Template path: {self.template_path}')

        # Load main HTML template
        main_html_path = os.path.join(self.template_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)