ExpTestProcess(
test_list,
csv_out_name="exp_tests.csv",
raise_exception=True,
store_results=True,
)
Bases: object
Class to execute a list of ExperimentTests
Initialize the ExpTestProcess
Source code in niceml/experiments/experimenttests/testinitializer.py
| def __init__(
self,
test_list: List[ExperimentTest],
csv_out_name: str = "exp_tests.csv",
raise_exception: bool = True,
store_results: bool = True,
):
"""Initialize the ExpTestProcess"""
self.test_list = test_list
self.csv_out_name = csv_out_name
self.raise_exception = raise_exception
self.store_results = store_results
|
Functions
__call__
__call__(
input_folder, output_folder=None, file_system=None
)
Execute the list of tests and return a list of ExpTestResults
Source code in niceml/experiments/experimenttests/testinitializer.py
| def __call__(
self,
input_folder: str,
output_folder: Optional[str] = None,
file_system: Optional[AbstractFileSystem] = None,
):
"""Execute the list of tests and return a list of ExpTestResults"""
file_system = file_system or LocalFileSystem()
if output_folder is None:
output_folder = input_folder
test_result_list: List[ExpTestResult] = []
failed: List[ExpTestResult] = []
for test_obj in self.test_list:
test_result = test_obj(input_folder, file_system)
if test_result.status == TestStatus.FAILED:
failed.append(test_result)
test_result_list.append(test_result)
out_path = join(output_folder, self.csv_out_name)
if self.store_results:
dataframe: pd.DataFrame = pd.DataFrame(
[x.to_dict() for x in test_result_list]
)
write_csv(
dataframe, out_path, file_system=file_system, sep=";", decimal=","
)
df_dict = dataframe.to_dict(orient="records")
mlflow.log_dict(df_dict, "exp_tests.yaml")
if len(failed) > 0 and self.raise_exception:
for failed_exp in failed:
logging.getLogger(__name__).error(str(failed_exp))
raise ExperimentTestFailedError(
f"Experiment Tests Failed: {failed} Info at: {out_path}"
)
|