import json
import os
import shutil
import tempfile
import uuid
from pathlib import Path
from typing import Tuple, List, Dict, Union, Any, Optional
import mlflow
import pandas as pd
from mlflow.exceptions import MlflowException
from mlflow.tracking import MlflowClient
from .compare.accuracy import CompareAccuracyTable
from .compare.base import Comparer
from .compare.fairness import CompareFairness
from .compare.reliability import CompareReliability
from .compare.residual_cluster import CompareResidualCluster
from .compare.resilience import CompareResilience
from .compare.robustness import CompareRobustness
from .compare.slicing_accuracy import CompareSlicingAccuracy
from .compare.slicing_fairness import CompareSlicingFairness
from .compare.slicing_overfit import CompareSlicingOverfit
from .compare.slicing_reliability import CompareSlicingReliability
from .compare.slicing_robustness import CompareSlicingRobustness
from .diagnose.accuracy import AccuracyTable
from .diagnose.base import Diagnoser
from .diagnose.fairness import FairnessTest
from .diagnose.mitigate_unfair_binning import MitigateUnFairBinning
from .diagnose.mitigate_unfair_thresholding import MitigateUnFairThresholding
from .diagnose.reliability import ReliabilityTest
from .diagnose.residual import ResidualAnalysis
from .diagnose.residual_cluster import ResidualClusterTest
from .diagnose.residual_interpret import ResidualInterpret
from .diagnose.weakness_region import WeaknessRegion
from .diagnose.resilience import ResilienceTest
from .diagnose.robustness import RobustnessTest
from .diagnose.slicing_accuracy import SlicingAccuracy
from .diagnose.slicing_fairness import SlicingFairness
from .diagnose.slicing_overfit import SlicingOverfit
from .diagnose.slicing_reliability import SlicingReliability
from .diagnose.slicing_robustness import SlicingRobustness
from .explain.ale import ALEExplainer
from .explain.base import Explainer
from .explain.hstatistic import HStatistics
from .explain.lime import LIMEExplainer
from .explain.pdp import PDPExplainer
from .explain.pfi import PFIExplainer
from .explain.shap import SHAPExplainer
from .interpret.aletheia.base import InterpretReLUDNN
from .interpret.base import Interpreter
from .interpret.decision_tree import InterpretDecisionTree
from .interpret.fanova.base import InterpretFANOVA
from .interpret.fanova.global_fi import GlobalFeatureImportance
from .interpret.fanova.local_fi import LocalFeatureImportance
from .interpret.fanova.local_linear_fi import LocalLinearFeatureImportance
from .interpret.linear_model.base import InterpretLinearModel
from .interpret.moe import InterpretMoeFANOVA
from ..auth import auth
from ..dashboard.report_template import ReportTemplate
from ..dashboard.utils.report import create_html_reprt
from ..data import local_dataset
from ..exceptions import EmptyDataset, EmptyModel, ComparisonNotEnable, InvalidLicenseExecption
from ..models import base
from ..utils.constants import MODEVA_TEST_TAG
from ..utils.helper import hash_list_dict, add_docs, convert_timestamp_to_datetime, NumpyEncoder
from ..utils.mlflow import get_mlflow_config
from ..utils.registry import list_registered_tests_
from ..utils.results import ValidationResult
[docs]
class LocalTestSuite:
"""A comprehensive model evaluation and analysis toolkit that provides methods for explaining, diagnosing, comparing, and interpreting machine learning models.
This class serves as the main interface for model evaluation, offering capabilities to:
- Explain model behavior using various techniques (PFI, SHAP, LIME, etc.)
- Diagnose model performance (accuracy, reliability, robustness, fairness, etc.)
- Compare multiple models across different metrics
- Interpret model decisions and feature importance
- Register and manage test results using MLflow
Parameters
----------
dataset : LocalDataSet, optional
The dataset object to be used for evaluation
model : ModelBase, optional
The primary model to be evaluated
models : List[ModelBase], optional
List of models for comparison purposes
name : str, default="testsuite"
Name of the testsuite, used for MLflow experiment tracking
Examples
--------
>>> # Create a testsuite for single model evaluation
>>> testsuite = LocalTestSuite(dataset=my_dataset, model=my_model, name="model_evaluation")
>>>
>>> # Run accuracy diagnosis
>>> result = testsuite.diagnose_accuracy_table()
>>>
>>> # Compare multiple models
>>> comparison_sheet = LocalTestSuite(dataset=my_dataset, models=[model1, model2])
>>> comparison = comparison_sheet.compare_accuracy_table()
"""
def __init__(self,
dataset: local_dataset.LocalDataSet = None,
model: base.ModelBase = None,
models: List[base.ModelBase] = None,
name: str = "testsuite"):
auth.run()
self.name = name
self.__dataset = dataset
self.__model = model
self.__models = models
self._setup()
self._check_if_trail()
self.experiment_name = f"testsuite-{str(uuid.uuid4())[:8]}" if name is None else name
self._init_mlflow()
def _setup(self):
if self.__dataset is not None and self.__model is not None:
self.__explainer = Explainer(self.__dataset, self.__model)
self.__diagnoser = Diagnoser(self.__dataset, self.__model)
self.__interpreter = Interpreter(self.__dataset, self.__model)
if self.__dataset is not None and self.__models is not None:
self.__comparer = Comparer(self.__dataset, self.__models)
def _check_if_trail(self):
if auth.is_trial():
if not isinstance(self.__dataset, local_dataset.LocalDataSet):
raise TypeError("Testsuite only accept DataSet object as dataset.")
if not self.__dataset.is_built_in():
raise InvalidLicenseExecption("This function is not supported for trial version.") from None
def _validate(self):
if self.__dataset is None:
raise EmptyDataset(
"Dataset cannot be empty. Set the dataset using set_dataset or load registered dataset using load_dataset.")
if self.__model is None:
raise EmptyModel(
"Model cannot be empty. Set the model using set_model or load registered model using load_model.")
[docs]
def get_dataset(self):
"""Return the dataset object.
"""
return self.__dataset
[docs]
def set_dataset(self, dataset: local_dataset.LocalDataSet):
"""Set dataset for test suite.
Parameters
----------
dataset : Dataset
Dataset object
"""
self.__dataset = dataset
self._setup()
[docs]
def get_model(self):
"""Return the model object.
"""
return self.__model
[docs]
def set_model(self, model: base.ModelBase):
"""Set model for test suite.
Parameters
----------
model : ModelBase
modeva built-in models or external models that implement ModelBase
"""
self.__model = model
self._setup()
# ------------------
# Model Explain
# ------------------
[docs]
@add_docs(PFIExplainer.run.__doc__)
def explain_pfi(self,
dataset: str = "test",
sample_size: int = 5000,
n_repeats: int = 10,
random_state: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__explainer.explain_pfi(**kwargs)
[docs]
@add_docs(HStatistics.run.__doc__)
def explain_hstatistic(self,
features: Union[Tuple, List] = None,
dataset: str = "test",
sample_size: int = 5000,
percentiles: Tuple = (0, 1),
grid_resolution: int = 10,
response_method: str = "auto",
random_state: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__explainer.explain_hstatistic(**kwargs)
[docs]
@add_docs(PDPExplainer.run.__doc__)
def explain_pdp(self,
features: Union[str, Tuple[str]] = None,
dataset: str = "test",
sample_size: int = 5000,
percentiles: Tuple = (0, 1),
grid_resolution: int = 20,
response_method: str = "auto",
random_state: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__explainer.explain_pdp(**kwargs)
[docs]
@add_docs(ALEExplainer.run.__doc__)
def explain_ale(self,
features: Union[str, Tuple[str]] = None,
dataset: str = "test",
sample_size: int = 5000,
grid_resolution: int = 20,
response_method: str = "auto",
random_state: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__explainer.explain_ale(**kwargs)
[docs]
@add_docs(LIMEExplainer.run.__doc__)
def explain_lime(self,
dataset: str = "test",
sample_index: int = 0,
centered: bool = True,
random_state: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__explainer.explain_lime(**kwargs)
[docs]
@add_docs(SHAPExplainer.run.__doc__)
def explain_shap(self,
dataset: str = "test",
sample_index: int = 0,
baseline_dataset: str = "train",
baseline_sample_index: int = None,
baseline_sample_size: int = 500,
random_state: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__explainer.explain_shap(**kwargs)
# ------------------
# Model Diagnose
# ------------------
[docs]
@add_docs(AccuracyTable.run.__doc__)
def diagnose_accuracy_table(self,
train_dataset: str = "train",
test_dataset: str = "test",
metric: Union[str, Tuple] = None):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_accuracy_table(**kwargs)
[docs]
@add_docs(ResidualAnalysis.run.__doc__)
def diagnose_residual_analysis(self,
features: str = None,
use_prediction: bool = False,
dataset: str = "test",
sample_size: int = 2000,
random_state: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_residual_analysis(**kwargs)
[docs]
@add_docs(ResidualInterpret.run.__doc__)
def diagnose_residual_interpret(self,
dataset: str = "test",
n_estimators: int = 100,
max_depth: int = 2,
**xgb_kwargs):
self._validate()
xgb_kwargs = hash_list_dict(xgb_kwargs)
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_residual_interpret(**kwargs)
[docs]
@add_docs(ResidualClusterTest.run.__doc__)
def diagnose_residual_cluster(self,
dataset: str = "test",
response_type: str = "abs_residual",
metric: str = None,
n_clusters: int = 10,
cluster_method: str = "ltc",
kmedoids_method: str = "pam",
sample_size: int = 2000,
n_estimators: int = 100,
max_depth: int = 5,
random_state: int = 0,
n_repeats: int = 10,
perturb_features: Union[str, Tuple] = None,
perturb_method: str = "normal",
noise_level: Union[float, int] = 0.1,
alpha: float = 0.1
):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_residual_cluster(**kwargs)
[docs]
@add_docs(WeaknessRegion.run.__doc__)
def diagnose_weakness_region(self,
train_dataset="train",
test_dataset="test",
metric=None,
geometry_method="arf",
bins=10,
weak_fraction=0.2,
top_n_features=10,
min_count=20,
geometry_n_estimators=120,
geometry_max_depth=12,
geometry_min_samples_leaf=20,
geometry_num_trees=30,
geometry_max_iters=10,
mi_n_estimators=200,
mi_max_depth=0,
mi_min_samples_leaf=10,
mi_n_splits=5,
random_state=0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_weakness_region(**kwargs)
[docs]
@add_docs(SlicingAccuracy.run.__doc__)
def diagnose_slicing_accuracy(self,
features: Union[str, Tuple] = None,
dataset: str = "test",
metric: str = None,
method: str = "uniform",
bins: Union[int, Dict] = 10,
n_estimators: int = 1000,
threshold: Union[float, int] = None):
# TODO: find a better way to solve the unhashable dict error.
self._validate()
features = hash_list_dict(features)
bins = hash_list_dict(bins)
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_slicing_accuracy(**kwargs)
[docs]
@add_docs(SlicingOverfit.run.__doc__)
def diagnose_slicing_overfit(self,
features: Union[str, Tuple] = None,
train_dataset: str = "train",
test_dataset: str = "test",
metric: str = None,
method: str = "uniform",
bins: Union[int, Dict] = 10,
n_estimators: int = 1000,
threshold: Union[float, int] = None):
self._validate()
features = hash_list_dict(features)
bins = hash_list_dict(bins)
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_slicing_overfit(**kwargs)
[docs]
@add_docs(SlicingReliability.run.__doc__)
def diagnose_slicing_reliability(self,
features: Union[str, Tuple] = None,
train_dataset: str = "test",
test_dataset: str = "test",
test_size: float = 0.5,
method: str = "uniform",
bins: Union[int, Dict] = 10,
n_estimators: int = 1000,
threshold: Union[float, int] = None,
metric: str = "width",
alpha: float = 0.1,
max_depth: int = 5,
random_state: int = 0):
self._validate()
features = hash_list_dict(features)
bins = hash_list_dict(bins)
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_slicing_reliability(**kwargs)
[docs]
@add_docs(SlicingRobustness.run.__doc__)
def diagnose_slicing_robustness(self,
features: Union[str, Tuple] = None,
dataset: str = "test",
method: str = "uniform",
bins: Union[int, Dict] = 10,
metric: str = None,
n_estimators: int = 1000,
threshold: Union[float, int] = None,
n_repeats: int = 10,
perturb_features: Union[str, Tuple] = None,
perturb_method: str = "normal",
noise_levels: Union[float, int] = 0.1,
random_state: int = 0):
self._validate()
features = hash_list_dict(features)
bins = hash_list_dict(bins)
perturb_features = hash_list_dict(perturb_features)
noise_levels = hash_list_dict(noise_levels)
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_slicing_robustness(**kwargs)
[docs]
@add_docs(SlicingFairness.run.__doc__)
def diagnose_slicing_fairness(self,
group_config,
features: Union[str, Tuple] = None,
favorable_label: int = 1,
dataset: str = "test",
metric: str = None,
method: str = "uniform",
bins: Union[int, Dict] = 10,
n_estimators: int = 1000,
threshold: Union[float, int] = None):
self._validate()
features = hash_list_dict(features)
group_config = hash_list_dict(group_config)
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_slicing_fairness(**kwargs)
[docs]
@add_docs(RobustnessTest.run.__doc__)
def diagnose_robustness(self,
dataset: str = "test",
threshold: float = 0.1,
metric: str = None,
n_repeats: int = 10,
perturb_features: Union[str, Tuple] = None,
perturb_method: str = "normal",
noise_levels: Union[float, int, Tuple] = 0.1,
random_state: int = 0):
self._validate()
perturb_features = hash_list_dict(perturb_features)
noise_levels = hash_list_dict(noise_levels)
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_robustness(**kwargs)
[docs]
@add_docs(ReliabilityTest.run.__doc__)
def diagnose_reliability(self,
train_dataset: str = "test",
test_dataset: str = "test",
test_size: float = 0.5,
alpha: float = 0.1,
max_depth: int = 5,
width_threshold: float = 0.1,
random_state: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_reliability(**kwargs)
[docs]
@add_docs(ResilienceTest.run.__doc__)
def diagnose_resilience(self,
dataset: str = "test",
method: str = "worst-sample",
metric: str = None,
alphas: tuple = None,
n_clusters: int = 10,
random_state: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_resilience(**kwargs)
[docs]
@add_docs(FairnessTest.run.__doc__)
def diagnose_fairness(self,
group_config,
favorable_label: int = 1,
dataset: str = "test",
metric: str = None,
threshold: Union[float, int] = None):
self._validate()
group_config = hash_list_dict(group_config)
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_fairness(**kwargs)
[docs]
@add_docs(MitigateUnFairThresholding.run.__doc__)
def diagnose_mitigate_unfair_thresholding(self,
group_config,
favorable_label: int = 1,
dataset: str = "test",
metric: str = None,
performance_metric: str = None,
proba_cutoff: Union[int, Tuple] = None):
self._validate()
group_config = hash_list_dict(group_config)
proba_cutoff = hash_list_dict(proba_cutoff)
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_mitigate_unfair_thresholding(**kwargs)
[docs]
@add_docs(MitigateUnFairBinning.run.__doc__)
def diagnose_mitigate_unfair_binning(self,
group_config,
favorable_label: int = 1,
dataset: str = "test",
metric: str = None,
performance_metric: str = None,
binning_features: Union[str, Tuple] = None,
binning_method: str = "uniform",
bins: Union[int, Dict] = 10):
self._validate()
group_config = hash_list_dict(group_config)
bins = hash_list_dict(bins)
kwargs = locals()
kwargs.pop('self')
return self.__diagnoser.diagnose_mitigate_unfair_binning(**kwargs)
# ------------------
# Model Compare
# ------------------
def __check_compare(self):
try:
self.__comparer
except:
raise ComparisonNotEnable("Please enable model comparison by running the compare_models function.")
[docs]
@add_docs(CompareAccuracyTable.run.__doc__)
def compare_accuracy_table(self,
train_dataset: str = "train",
test_dataset: str = "test",
metric: Union[str, Tuple] = None):
self.__check_compare()
metric = hash_list_dict(metric)
kwargs = locals()
kwargs.pop('self')
return self.__comparer.compare_accuracy_table(**kwargs)
[docs]
@add_docs(CompareSlicingAccuracy.run.__doc__)
def compare_slicing_accuracy(self,
features: str,
dataset: str = "test",
metric: str = None,
method: str = "uniform",
bins: Union[int, Dict] = 10,
n_estimators: int = 1000,
threshold: Union[float, int] = None):
self.__check_compare()
bins = hash_list_dict(bins)
kwargs = locals()
kwargs.pop('self')
return self.__comparer.compare_slicing_accuracy(**kwargs)
[docs]
@add_docs(CompareSlicingOverfit.run.__doc__)
def compare_slicing_overfit(self,
features: str,
train_dataset: str = "train",
test_dataset: str = "test",
metric: str = None,
method: str = "uniform",
bins: Union[int, Dict] = 10,
n_estimators: int = 1000,
threshold: Union[float, int] = None):
self.__check_compare()
bins = hash_list_dict(bins)
kwargs = locals()
kwargs.pop('self')
return self.__comparer.compare_slicing_overfit(**kwargs)
[docs]
@add_docs(CompareSlicingReliability.run.__doc__)
def compare_slicing_reliability(self,
features: str,
train_dataset: str = "test",
test_dataset: str = "test",
test_size: float = 0.5,
method: str = "uniform",
bins: Union[int, Dict] = 10,
n_estimators: int = 1000,
threshold: Union[float, int] = None,
metric: str = "width",
alpha: float = 0.1,
max_depth: int = 5,
random_state: int = 0):
self.__check_compare()
bins = hash_list_dict(bins)
kwargs = locals()
kwargs.pop('self')
return self.__comparer.compare_slicing_reliability(**kwargs)
[docs]
@add_docs(CompareSlicingRobustness.run.__doc__)
def compare_slicing_robustness(self,
features: str,
dataset: str = "test",
metric: str = None,
method: str = "uniform",
bins: Union[int, Dict] = 10,
n_estimators: int = 1000,
threshold: Union[float, int] = None,
n_repeats: int = 10,
perturb_features: Union[str, Tuple] = None,
perturb_method: str = "normal",
noise_levels: Union[float, int] = 0.1,
random_state: int = 0):
self.__check_compare()
bins = hash_list_dict(bins)
perturb_features = hash_list_dict(perturb_features)
noise_levels = hash_list_dict(noise_levels)
kwargs = locals()
kwargs.pop('self')
return self.__comparer.compare_slicing_robustness(**kwargs)
[docs]
@add_docs(CompareSlicingFairness.run.__doc__)
def compare_slicing_fairness(self,
group_config,
features: str,
favorable_label: int = 1,
dataset: str = "test",
metric: str = None,
method: str = "uniform",
bins: Union[int, Dict] = 10,
n_estimators: int = 1000,
threshold: Union[float, int] = None):
self.__check_compare()
group_config = hash_list_dict(group_config)
bins = hash_list_dict(bins)
kwargs = locals()
kwargs.pop('self')
return self.__comparer.compare_slicing_fairness(**kwargs)
[docs]
@add_docs(CompareRobustness.run.__doc__)
def compare_robustness(self,
dataset: str = "test",
metric: str = None,
n_repeats: int = 10,
perturb_features: Union[str, Tuple] = None,
perturb_method: str = "normal",
noise_levels: Union[float, int, Tuple] = 0.1,
random_state: int = 0):
self.__check_compare()
perturb_features = hash_list_dict(perturb_features)
noise_levels = hash_list_dict(noise_levels)
kwargs = locals()
kwargs.pop('self')
return self.__comparer.compare_robustness(**kwargs)
[docs]
@add_docs(CompareReliability.run.__doc__)
def compare_reliability(self,
train_dataset: str = "test",
test_dataset: str = "test",
test_size: float = 0.5,
alpha: float = 0.1,
max_depth: int = 5,
random_state: int = 0):
self.__check_compare()
kwargs = locals()
kwargs.pop('self')
return self.__comparer.compare_reliability(**kwargs)
[docs]
@add_docs(CompareResilience.run.__doc__)
def compare_resilience(self,
dataset: str = "test",
method: str = "worst-sample",
metric: str = None,
alphas: tuple = None,
n_clusters: int = 10,
random_state: int = 0):
self.__check_compare()
kwargs = locals()
kwargs.pop('self')
return self.__comparer.compare_resilience(**kwargs)
[docs]
@add_docs(CompareResidualCluster.run.__doc__)
def compare_residual_cluster(self,
dataset: str = "test",
response_type: str = "abs_residual",
metric: str = None,
n_clusters: int = 10,
cluster_method: str = "ltc",
kmedoids_method: str = "pam",
sample_size: int = 2000,
n_estimators: int = 100,
max_depth: int = 5,
random_state: int = 0,
n_repeats: int = 10,
perturb_features: Union[str, Tuple] = None,
perturb_method: str = "normal",
noise_level: Union[float, int] = 0.1,
alpha: float = 0.1):
self.__check_compare()
kwargs = locals()
kwargs.pop('self')
return self.__comparer.compare_residual_cluster(**kwargs)
[docs]
@add_docs(CompareFairness.run.__doc__)
def compare_fairness(self,
group_config,
favorable_label: int = 1,
dataset: str = "test",
metric: str = None,
threshold: Union[float, int] = None):
self.__check_compare()
group_config = hash_list_dict(group_config)
kwargs = locals()
kwargs.pop('self')
return self.__comparer.compare_fairness(**kwargs)
# ------------------
# Model Interpret
# ------------------
[docs]
@add_docs(InterpretFANOVA.get_main_effects.__doc__)
def get_main_effects(self):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.get_main_effects(**kwargs)
[docs]
@add_docs(InterpretFANOVA.get_interactions.__doc__)
def get_interactions(self):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.get_interactions(**kwargs)
[docs]
@add_docs(InterpretFANOVA.run_effects.__doc__)
def interpret_effects(self,
features: Union[str, Tuple] = None,
dataset: str = "test",
grid_size: int = 200):
self._validate()
features = hash_list_dict(features)
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_effects(**kwargs)
[docs]
@add_docs(InterpretFANOVA.run_ei.__doc__)
def interpret_ei(self,
dataset: str = "test"):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_ei(**kwargs)
[docs]
@add_docs(InterpretFANOVA.run_local_ei.__doc__)
def interpret_local_ei(self,
dataset: str = "test",
sample_index: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_local_ei(**kwargs)
[docs]
@add_docs(InterpretLinearModel.run_coef.__doc__)
def interpret_coef(self,
features: Union[str, Tuple[str]] = None):
self._validate()
features = hash_list_dict(features)
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_coef(**kwargs)
[docs]
@add_docs(GlobalFeatureImportance.run_fi.__doc__)
def interpret_fi(self,
dataset: str = "test"):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_fi(**kwargs)
[docs]
@add_docs(LocalFeatureImportance.run_local_fi.__doc__)
def interpret_local_fi(self,
dataset: str = "test",
sample_index: int = 0,
centered: bool = True):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_local_fi(**kwargs)
[docs]
@add_docs(LocalLinearFeatureImportance.run_local_linear_fi.__doc__)
def interpret_local_linear_fi(self,
dataset: str = "test",
sample_index: int = 0,
centered: bool = True):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_local_linear_fi(**kwargs)
[docs]
@add_docs(InterpretReLUDNN.run_llm_summary.__doc__)
def interpret_llm_summary(self,
dataset: str = "test"):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_llm_summary(**kwargs)
[docs]
@add_docs(InterpretReLUDNN.run_llm_pc.__doc__)
def interpret_llm_pc(self,
dataset: str = "test"):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_llm_pc(**kwargs)
[docs]
@add_docs(InterpretReLUDNN.run_llm_profile.__doc__)
def interpret_llm_profile(self,
feature: str = None,
dataset: str = "test"):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_llm_profile(**kwargs)
[docs]
@add_docs(InterpretReLUDNN.run_llm_violin.__doc__)
def interpret_llm_violin(self,
feature: str = None,
dataset: str = "test"):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_llm_violin(**kwargs)
[docs]
@add_docs(InterpretDecisionTree.run_global_tree.__doc__)
def interpret_global_tree(self):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_global_tree(**kwargs)
[docs]
@add_docs(InterpretDecisionTree.run_local_tree.__doc__)
def interpret_local_tree(self,
dataset: str = "test",
sample_index: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_local_tree(**kwargs)
[docs]
@add_docs(InterpretMoeFANOVA.run_local_moe_weights.__doc__)
def interpret_local_moe_weights(self,
dataset: str = "test",
sample_index: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_local_moe_weights(**kwargs)
[docs]
@add_docs(InterpretMoeFANOVA.run_effects_moe_average.__doc__)
def interpret_effects_moe_average(self,
features: Union[str, Tuple],
dataset: str = "test",
grid_size: int = 100,
sample_size: int = 5000,
random_state: int = 0):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_effects_moe_average(**kwargs)
[docs]
@add_docs(InterpretMoeFANOVA.run_cluster_analysis.__doc__)
def interpret_moe_cluster_analysis(self,
dataset: str = "test",
metric: str = None):
self._validate()
kwargs = locals()
kwargs.pop('self')
return self.__interpreter.interpret_moe_cluster_analysis(**kwargs)
# ------------------
# MLFLOW
# ------------------
def _init_mlflow(self):
mlflow_home, local_mlflow_uri, mlflow_folder = get_mlflow_config()
mlflow.set_tracking_uri(local_mlflow_uri)
self.mlflow_client = MlflowClient()
self.mlflow_tag = MODEVA_TEST_TAG
try:
# Set type tag: MODEVA_TEST_TAG for experiment retrieval
self.experiment_id = mlflow.create_experiment(self.experiment_name, tags={"type": MODEVA_TEST_TAG,
"register_purpose": True,
})
except MlflowException:
mlflow.set_experiment(self.experiment_name)
self.experiment_id = dict(mlflow.get_experiment_by_name(self.experiment_name))["experiment_id"]
[docs]
@classmethod
def list(cls):
"""
List all the experiments saved in database.
Returns
-------
pd.DataFrame
A table showing the experiments details in database.
"""
exps = mlflow.search_experiments(filter_string=f"tags.type = '{MODEVA_TEST_TAG}'")
return pd.DataFrame([[exp.name, convert_timestamp_to_datetime(exp.creation_time)] for exp in exps],
columns=["Name", "Creation Time"])
def _register_test(
self,
name: str,
test_result: ValidationResult,
register_name: str = None,
description: str = None,
tags: Optional[Dict[str, Any]] = None,
experiment_id: str = None,
run_id: str = None
):
# use default model name if no name provided
if register_name is None:
register_name = name
if tags is None:
tags = dict()
# Create a new run
if run_id is None:
run = self.mlflow_client.create_run(self.experiment_id, run_name=register_name, tags=tags)
run_id = run.info.run_id
if isinstance(test_result.table, pd.DataFrame):
tmp_table = test_result.table.to_dict()
new_table = {}
for i, v in tmp_table.items():
if isinstance(i, tuple):
i = [str(single) for single in i]
new_table['_-_-_'.join(list(i))] = v
else:
new_table[i] = v
elif isinstance(test_result.table, dict):
new_table = {}
for key, value in test_result.table.items():
tmp_table = value.to_dict()
new_tmp_table = {}
for i, v in tmp_table.items():
if isinstance(i, tuple):
i = [str(single) for single in i]
new_tmp_table['_-_-_'.join(list(i))] = v
else:
new_tmp_table[i] = v
new_table[key] = new_tmp_table
else:
new_table = test_result.table
test_dict = {
'key': test_result.key,
'inputs': test_result.inputs,
'model': test_result.model,
'data': test_result.data,
'value': test_result.value,
'options': test_result.options,
'table': new_table
}
# Save test and metainfo in temp directory and log them in current new run
with tempfile.TemporaryDirectory() as tmpdirname:
with open(f'{tmpdirname}/{register_name}.json', 'w') as fp:
json.dump(test_dict, fp, cls=NumpyEncoder)
self.mlflow_client.log_artifact(run_id, f'{tmpdirname}/{register_name}.json')
[docs]
def register(
self,
name: str,
test_result: ValidationResult,
register_name: str = None,
description: str = None,
tags: Optional[Dict[str, Any]] = None,
run_id: str = None,
):
"""Register a test into MLFlow.
Parameters
----------
name : str
The current name of the test to be registered.
test_result : ValidationResult
The validation result object of test.
register_name : str, default=None
The register name of the test in MLFlow.
If None, will be the same as name.
description : str, default=None
The description of this test.
tags : dict, default=None
The tags.
run_id : str, default=None
The run id in MLFLow.
"""
# Run id is needed to keep test registered under experiment
if tags is None:
tags = dict()
if isinstance(tags, list):
# Convert tag from list to dict, required by mlflow
ts = dict()
for tag in tags:
ts[tag] = tag
tags = ts
self._register_test(
name=name, # name
test_result=test_result,
register_name=register_name,
description=description,
tags=tags,
experiment_id=self.experiment_id,
run_id=run_id)
[docs]
def delete_registed_test(self, name, run_id: str = None):
"""Load config and result of registered tests.
Parameters
----------
name : str
The name of test used for filtering.
run_id : str, default=None
run id of the registered test.
"""
mlflow_home, local_mlflow_uri, mlflow_folder = get_mlflow_config()
if run_id is None:
runs = self.mlflow_client.search_runs(self.experiment_id, filter_string=f"run_name = '{name}'")
else:
runs = self.mlflow_client.search_runs(self.experiment_id,
filter_string=f"run_name = '{name}' and run_id = '{run_id}' ")
if not len(runs) > 0:
return None
run = runs[0] # Get the latest run.
file_path = Path(mlflow_folder, f"{self.experiment_id}", run.info.run_id)
shutil.rmtree(file_path, ignore_errors=True)
self.mlflow_client.delete_run(run_id)
[docs]
def load_registered_test(self, name: str, run_id: str = None):
"""Load config and result of registered tests.
Parameters
----------
name : str
The name of test used for filtering.
run_id : str, default=None
run id of the registered test.
"""
mlflow_home, local_mlflow_uri, mlflow_folder = get_mlflow_config()
if run_id is None:
runs = self.mlflow_client.search_runs(self.experiment_id, filter_string=f"run_name = '{name}'")
else:
runs = self.mlflow_client.search_runs(self.experiment_id,
filter_string=f"run_name = '{name}' and run_id = '{run_id}' ")
if not len(runs) > 0:
return None
run = runs[0] # Get the latest run.
file_path = Path(mlflow_folder, f"{self.experiment_id}", run.info.run_id, "artifacts", f"{name}.json")
test_result = json.load(open(file_path, "rb"))
def get_depth(d):
if isinstance(d, dict):
return 1 + (max(map(get_depth, d.values())) if d else 0)
return 0
if test_result['table'] is not None:
new_dict = {}
for i, v in test_result['table'].items():
if get_depth(v) == 2:
single_table_dict = {}
for single_key, single_table in v.items():
if '_-_-_' in single_key:
single_table_dict[tuple(i.split('_-_-_'))] = single_table
else:
single_table_dict[single_key] = single_table
single_table_dict = pd.DataFrame(single_table_dict)
new_dict[i] = single_table_dict
else:
if '_-_-_' in i:
new_dict[tuple(i.split('_-_-_'))] = v
else:
new_dict[i] = v
new_dict = pd.DataFrame(new_dict)
test_result['table'] = new_dict
test_result['name'] = f"{name}-{run.info.start_time}" # Update model name to starttime name
return test_result
[docs]
def list_registered_tests(self, name: str = None):
"""Return the list all registered tests.
Parameters
----------
name : str
The name of test used for filtering.
"""
return list_registered_tests_(mlflow_client=self.mlflow_client,
name=name,
experiment_id=self.experiment_id)
[docs]
def display_test_results(self, testsuite_name, test_list: list = None):
"""Get ValidationResult object of registered test.
Parameters
----------
testsuite_name : str
The testsuite to display.
test_list : list
The content list of test, e.g.,
.. code-block:: python
[{'name': 'test1', 'run_id': 'xxx', 'display_table': True, 'display_plot': True},
{'name': 'test2', 'display_table': True, 'display_plot': False}]
or
.. code-block:: python
['test1', 'test2']
"""
tmp_ts = LocalTestSuite(name=testsuite_name)
template = ReportTemplate(ts=tmp_ts)
if test_list is None:
test_list = tmp_ts.list_registered_tests().Name.unique().tolist()
return template.show_panel(component_list=test_list)
else:
return template.show_panel(component_list=test_list)
[docs]
def export_report(self, path: str = "report.html"):
"""Export report to html
Parameters
----------
path : str, optional
The export path, by default "report.html"
"""
names = self.list_registered_tests().Name.unique().tolist()
rs = []
for name in reversed(names):
f = self.load_registered_test(name=name)
plots = []
plot = f['options']
if plot:
# single plot
if 'chart_id' in plot:
plots.append(plot)
else:
# multiple plots
# options['MSE']
# options['MAE']
for n, option in plot.items():
if 'chart_id' in option:
plots.append(option)
else:
# deeply nested plots
# options['summary']
# options['density']['MedInc']
# options['density']['MedInc']
for nm, op in option.items():
plots.append(op)
tables = []
if f['table'] is not None:
if isinstance(f['table'], dict):
# multiple tables
for k, v in f['table'].items():
tables.append(v.round(6).to_dict(orient="split"))
else:
# single table
tables.append(f['table'].round(6).to_dict(orient="split"))
rs.append({
"name": name,
"data": json.dumps(f['data']),
"model": json.dumps(f['model']),
"inputs": json.dumps(f['inputs']),
"tables": json.dumps(tables).replace("nan", "null").replace("NaN", "null"),
"plots": json.dumps(plots).replace("nan", "null").replace("NaN", "null").replace('"nameGap": "auto"',
'"nameGap": 40')
})
html_str = create_html_reprt(self.name, rs)
directory_path = os.path.dirname(path)
if directory_path != '':
os.makedirs(directory_path, exist_ok=True)
with open(path, 'w', encoding='utf-8') as f:
f.write(html_str)