import collections
import datetime
import json
import os
import shutil
from copy import copy
from typing import Optional
import pandas as pd
from openpyxl import load_workbook, Workbook
from openpyxl.utils.dataframe import dataframe_to_rows
from atk.configs.results_config import RunStatus, TaskResult
from atk.configs.base_config import TaskTypesSelect, NONE_RET
from atk.common.log import Logger
from atk.configs.report_config import CsvReport, BaseReportConfig
from atk.configs.standard_config import PerfStandardConfig, PassStandard
from atk.tasks.post_process.perf_compare_tool import PerfCompareExecutor
from atk.tasks.report.report_title.summary_report_title import SummaryReportTitleFactory
from atk.tasks.report.report_title.base_report_title import TitleType
from atk.tasks.report.report_title import summary_report_title
from atk.common.output_manager import OutputManager
from atk.common.utils import get_ttk_report_dir, chmod_output_path
logging = Logger().get_logger()
class SuccessInfo:
def __init__(self):
self.succ = set()
self.fail = set()
self.null = set()
@property
def length(self):
return len(self.succ.union(self.fail).union(self.null))
@property
def succ_length(self):
return len(self.succ)
@property
def fail_length(self):
return len(self.fail)
@property
def null_length(self):
return len(self.null)
def add(self, id, succ):
if succ == NONE_RET:
func = self.null
other_funcs = [self.succ, self.fail]
elif succ:
func = self.succ
other_funcs = [self.null, self.fail]
else:
func = self.fail
other_funcs = [self.null, self.succ]
func.add(id)
for other_fun in other_funcs:
if id in other_fun:
other_fun.remove(id)
class SummaryCsv(BaseReportConfig):
SUCCESS_INFO = {
"accuracy": {"key": "accuracy_pass"},
"perf": {"key": "performance_e2e_result_node"},
"perf_device": {"key": "performance_device_result_node"},
"memory": {"key": "memory_result_node"},
}
def __init__(self, nodes_config):
self.report_title_factory = None
super(SummaryCsv, self).__init__(nodes_config)
self.total = 0
self.csv_reports = {}
self.success_infos = {}
self.e2e_perf_total = {}
self.device_perf_total = {}
self.perf_standard = None
self.acc_pass_standard = PassStandard().get_acc_pass_ratio()
self.mem_pass_standard = PassStandard().get_memory_pass_ratio()
self.init()
self.failed_number = set()
def __str__(self):
report_data = self.report_title_factory.get_report_data(TitleType.API_TITLE)
data = report_data["data_en"]
return str(json.dumps(data, ensure_ascii=False))
@property
def case_num(self):
return len(self.csv_reports.keys())
@property
def success_num(self):
return self.case_num - self.failed_num
@property
def failed_num(self):
return len(self.failed_number)
@staticmethod
def get_avg_perf_ratio(node, perf_total):
_input = perf_total.get(node.get_backend_name())
data = _input.values()
data = list(filter(lambda x: x is not None, data))
if data:
ratio = sum(data) / len(data)
return "{:.4f}".format(ratio)
else:
return NONE_RET
def update_standard(self, pass_standard: PassStandard):
self.acc_pass_standard = pass_standard.get_acc_pass_ratio()
self.mem_pass_standard = pass_standard.get_memory_pass_ratio()
def init_report_title_factory(self):
self.report_title_factory = SummaryReportTitleFactory(self)
def init(self):
for node in self.compare_nodes:
backend_name = node.get_backend_name()
for info, _ in self.SUCCESS_INFO.items():
key = f"{backend_name}_{info}"
self.success_infos[key] = SuccessInfo()
self.e2e_perf_total[backend_name] = {}
self.device_perf_total[backend_name] = {}
def add(self, report: CsvReport):
case_id = report.id
self.csv_reports[case_id] = report
if len(self.csv_reports) == 1:
psd = report.task_result.get_pass_standard()
self.update_standard(psd)
main_node = self.nodes_config.get_main_node()
if report.run_status != RunStatus.SUCCESS:
self.add_success_info_false(case_id)
self.failed_number.add(case_id)
return
if case_id in self.failed_number:
self.failed_number.remove(case_id)
for node in self.compare_nodes:
backend_name = node.get_backend_name()
for info, info_data in self.SUCCESS_INFO.items():
key = f"{backend_name}_{info}"
if key in self.success_infos:
self.success_infos[key].add(
case_id, report.get_value(info_data.get("key"), node)
)
self.e2e_perf_total[backend_name][case_id] = report.get_value(
"one_div_one_e2e_perf", node
)
self.device_perf_total[backend_name][case_id] = report.get_value(
"one_div_one_device_perf", node
)
def add_success_info_false(self, case_id):
main_node = self.nodes_config.get_main_node()
for node in self.compare_nodes:
if node == main_node:
continue
backend_name = node.get_backend_name()
for info, _ in self.SUCCESS_INFO.items():
key = f"{backend_name}_{info}"
if key in self.success_infos:
self.success_infos[key].add(case_id, False)
def add_csv_json_failed(self, case_id):
self.csv_reports[case_id] = {}
self.add_success_info_false(case_id)
def add_csv_json(self, csv_json):
if self.perf_standard is None:
perf_standard = PerfStandardConfig()
perf_standard.update(**csv_json["perf_standard"])
self.perf_standard = perf_standard
self.add(csv_json)
def acc_case_num(self, node, info):
main_node = self.nodes_config.get_main_node()
key = f"{main_node.get_backend_name()}_{info}"
_input = self.success_infos[key]
return _input.length
def error_case_num(self, node, info):
main_node = self.nodes_config.get_main_node()
key = f"{main_node.get_backend_name()}_{info}"
_input = self.success_infos[key]
if _input.length == 0:
return NONE_RET
return _input.length - _input.null_length
def fail_case_num(self, node, info):
main_node = self.nodes_config.get_main_node()
main_node_key = f"{main_node.get_backend_name()}_{info}"
key = f"{node.get_backend_name()}_{info}"
main_node_input = self.success_infos.get(main_node_key)
_input = self.success_infos.get(key)
return _input.length - main_node_input.length
def acc_pass_num(self, node, info):
main_node = self.nodes_config.get_main_node()
main_node_key = f"{main_node.get_backend_name()}_{info}"
key = f"{node.get_backend_name()}_{info}"
main_node_input = self.success_infos.get(main_node_key)
_input = self.success_infos.get(key)
return _input.succ_length + _input.null_length - main_node_input.fail_length
def pass_ratio(self, node, info):
key = f"{node.get_backend_name()}_{info}"
_input = self.success_infos.get(key)
if _input.length == 0:
return NONE_RET
main_node = self.nodes_config.get_main_node()
main_node_key = f"{main_node.get_backend_name()}_{info}"
main_node_key_input = self.success_infos.get(main_node_key)
pass_case = _input.succ_length + _input.null_length - main_node_key_input.fail_length
pass_ratio = 100.0 * pass_case / main_node_key_input.length
return round(pass_ratio, 4)
def get_avg_perf_e2e_ratio(self, node):
return SummaryCsv.get_avg_perf_ratio(node, self.e2e_perf_total)
def get_avg_perf_device_ratio(self, node):
return SummaryCsv.get_avg_perf_ratio(node, self.device_perf_total)
def get_acc_pass_result(self, node) -> Optional[str]:
_input = self.pass_ratio(node, "accuracy")
if _input == NONE_RET or all([task not in node.task for task in TaskTypesSelect.ACCURACY_TASKS]):
return NONE_RET
return "Pass" if _input >= 100.0 * self.acc_pass_standard else "Failed"
def get_acc_dc_pass_result(self, node):
_input = self.pass_ratio(node, "accuracy")
if _input == NONE_RET or all([task not in node.task for task in TaskTypesSelect.ACCURACY_DC_TASKS]):
return "-"
return "Pass" if _input >= 100.0 * self.acc_pass_standard else "Failed"
def get_is_perf_pass(self, node, node_key):
backend_name = node.get_backend_name()
node_perf = self.success_infos.get(f"{backend_name}_perf")
node_avg_perf_ratio = self.get_value(node_key, node)
data = PerfCompareExecutor.compare_summary(
node_perf, node_avg_perf_ratio, self.perf_standard
)
return data
def get_is_memory_pass(self, node) -> Optional[str]:
_input = self.get_value("memory_pass_ratio_node", node)
if _input == NONE_RET:
return NONE_RET
return "Pass" if _input >= 100.0 * self.mem_pass_standard else "Failed"
def get_mutil_perf_ratio(self, node):
backend = node.backend
_input = self.e2e_perf_total.get(node.get_backend_name())
mutil_perf_ratio = ""
perf_ratios = [0.25, 0.30, 0.80, 1.00]
for ratio in perf_ratios:
filtered_dict = {k: v for k, v in _input.items() if v and v >= ratio}
num = (
"{:.4f}".format(100.0 * len(filtered_dict) / len(_input))
if len(filtered_dict)
else "NA"
)
mutil_perf_ratio += f"{ratio:.2f}倍{backend}的占比: {num}\n"
return mutil_perf_ratio.strip()
class CSVWriter:
def __init__(self, nodes_config):
self.all_cases = {}
self.nodes_config = nodes_config
self.compare_nodes = nodes_config.get_compare_nodes()
self.summary_info = SummaryCsv(nodes_config)
self.now_time = datetime.datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
self.output = OutputManager.get_dir_by_env("report")
os.makedirs(self.output, exist_ok=True)
self.api_name = None
self.benchmark_path = None
self.csv_path = None
self.reports_path = None
self.csv_report = None
@staticmethod
def get_column_index(sheet, column_name):
"""
在 Excel 工作表中查找列名对应的列号（从1开始）。
"""
for cell in sheet[1]:
if cell.value == column_name:
return cell.column
raise ValueError(f"can not find column：{column_name}")
@staticmethod
def merge_ttk_files(input_folder, merged_sheet):
"""
合并指定文件夹中的所有 Excel 文件。
第一个 Excel 文件完全复制，后续文件只复制非标题数据。
"""
files = [f for f in os.listdir(input_folder) if f.endswith('.xlsx') or f.endswith('.xls')]
if not files:
return
first_file = os.path.join(input_folder, files[0])
first_workbook = load_workbook(first_file)
first_sheet = first_workbook.active
def set_cell_style(cell, new_cell):
if cell.has_style:
new_cell.font = copy(cell.font)
new_cell.border = copy(cell.border)
new_cell.fill = copy(cell.fill)
new_cell.number_format = copy(cell.number_format)
new_cell.protection = copy(cell.protection)
new_cell.alignment = copy(cell.alignment)
for row in first_sheet.iter_rows():
for cell in row:
new_cell = merged_sheet.cell(row=cell.row, column=cell.column, value=cell.value)
set_cell_style(cell, new_cell)
merged_cells = first_sheet.merged_cells.ranges
for merged_cell in merged_cells:
merged_sheet.merge_cells(str(merged_cell))
title_row = 1
for merged_cell in merged_cells:
if merged_cell.min_col == 1 and merged_cell.min_row == 1:
title_row = merged_cell.max_row - merged_cell.min_row + 1
for file in files[1:]:
file_path = os.path.join(input_folder, file)
workbook = load_workbook(file_path)
sheet = workbook.active
max_row = merged_sheet.max_row
for row in sheet.iter_rows(min_row=title_row + 1):
max_row = max_row + 1
for cell in row:
new_cell = merged_sheet.cell(row=max_row, column=cell.column, value=cell.value)
set_cell_style(cell, new_cell)
def set_api_name(self, api_name):
self.api_name = api_name
self.benchmark_path = os.path.join(
self.output, f"{self.api_name}_benchmark_{self.now_time}.csv"
)
self.csv_path = os.path.join(
self.output, f"{self.api_name}_statistic_{self.now_time}.csv"
)
self.reports_path = os.path.join(
self.output, f"{self.api_name}_reports_{self.now_time}.xlsx"
)
def add_and_write_case(self, task_result: TaskResult):
try:
self.csv_report = CsvReport(task_result)
if self.summary_info.perf_standard is None:
self.summary_info.perf_standard = self.csv_report.perf_standard
self.summary_info.add(self.csv_report)
data = collections.OrderedDict()
for en_title, zh_title in self.csv_report.get_print_name_list():
data[zh_title] = getattr(self.csv_report, en_title)
self.all_cases[self.csv_report.id] = data
except Exception as e:
logging.error(f'{task_result.case_config.name}_{task_result.case_config.id} add task report failed')
logging.exception(e)
def add_report(self, csv_report):
try:
if self.summary_info.perf_standard is None:
self.summary_info.perf_standard = self.csv_report.perf_standard
self.summary_info.add(csv_report)
data = collections.OrderedDict()
for en_title, zh_title in csv_report.get_print_name_list():
data[zh_title] = getattr(csv_report, en_title)
self.all_cases[csv_report.id] = data
except Exception as e:
logging.error(f'{csv_report.name}_{csv_report.id} add task report failed')
logging.exception(e)
def save_csv(self):
if len(self.all_cases) == 0:
logging.error('All cases is running failed.')
return
df_data = pd.DataFrame(self.all_cases).transpose()
header = [row[-1] for row in self.csv_report.get_print_name_list()]
df_data = df_data[header]
df_data.to_csv(self.csv_path, index=False, header=True, encoding="utf_8_sig")
logging.debug(f'save result csv success on {self.csv_path}')
def save_benchmark(self, benchmark_ret_list: list):
for benchmark_ret in benchmark_ret_list:
df_data = pd.DataFrame(benchmark_ret, index=[0])
if not os.path.exists(self.benchmark_path):
df_data.to_csv(
self.benchmark_path,
index=False,
header=True,
encoding="utf_8_sig",
)
else:
df_data.to_csv(self.benchmark_path, mode="a", index=False, header=False)
def print_summary(self):
report_title_type = summary_report_title.ReportTitlesType
print_titles = {
"acc_pass_result": {"key": report_title_type.ACC_PASS_RESULT_NODE},
"is_acc_dc_pass": {"key": report_title_type.IS_ACC_DC_PASS_NODE},
"is_e2e_perf_pass": {"key": report_title_type.IS_PERF_E2E_PASS_NODE},
"is_device_perf_pass": {"key": report_title_type.IS_PERF_DEVICE_PASS_NODE},
"is_device_memory_pass": {"key": report_title_type.IS_MEMORY_PASS_NODE},
}
data_str = ""
for en_title, info in print_titles.items():
sum_value = None
for node in self.compare_nodes:
value = self.summary_info.get_value(info.get("key"), node=node)
if value is None:
continue
if value == "Failed":
sum_value = "Failed"
break
elif sum_value is None:
sum_value = "Pass"
data_str += f"{en_title}:{sum_value}\t"
success_len = self.summary_info.success_num
failed_len = self.summary_info.failed_num
logging.info(f"Total Task: {self.summary_info.case_num}, success {success_len}, failed {failed_len}")
logging.info(f"Summary info: {data_str.strip()}.")
def summary(self):
self.save_csv()
print_info = self.summary_info.get_print_table()
table = print_info["table"]
zh_title = print_info["zh_title"]
data_list = print_info["data_list"]
logging.info(f"\n{table}")
self.print_summary()
merged_workbook = Workbook()
merged_workbook.remove(merged_workbook.active)
if os.path.exists(self.csv_path):
df_statistic = pd.read_csv(self.csv_path)
new_ws = merged_workbook.create_sheet(title="statistic")
for r in dataframe_to_rows(df_statistic, index=False, header=True):
new_ws.append(r)
os.remove(self.csv_path)
summary_frame = pd.DataFrame(data_list, columns=zh_title)
new_ws = merged_workbook.create_sheet(title="summary")
for r in dataframe_to_rows(summary_frame, index=False, header=True):
new_ws.append(r)
failed_cases_ws = merged_workbook.create_sheet(title="failed cases")
failed_pd = df_statistic[df_statistic['运行结果'] == 'FAILED']
for r in dataframe_to_rows(failed_pd, index=False, header=True):
failed_cases_ws.append(r)
accuracy_false_ws = merged_workbook.create_sheet(title="accuracy false cases")
matched_cols = [col for col in df_statistic.columns if "精度通过" in col]
if matched_cols:
target_col = matched_cols[-1]
acc_false_pd = df_statistic[(df_statistic['运行结果'] == 'SUCCESS') & (df_statistic[target_col] is False)]
for r in dataframe_to_rows(acc_false_pd, index=False, header=True):
accuracy_false_ws.append(r)
task_dir_name = OutputManager.get_task_output_file_name_by_env()
ttk_report_dir = get_ttk_report_dir(self.output, task_dir_name, create=False)
if os.path.exists(ttk_report_dir):
md5_sheet = None
for folder in os.listdir(ttk_report_dir):
folder_dir = os.path.join(ttk_report_dir, folder)
merged_sheet = merged_workbook.create_sheet(title=folder)
self.merge_ttk_files(folder_dir, merged_sheet)
if folder == "BinaryCompare":
md5_sheet = merged_sheet
shutil.rmtree(ttk_report_dir)
if md5_sheet:
self.merge_benchmark_files_to_ttk(md5_sheet)
else:
self.merge_benchmark_files(merged_workbook)
else:
self.merge_benchmark_files(merged_workbook)
merged_workbook.save(filename=self.reports_path)
chmod_output_path(os.path.join(self.reports_path))
logging.info(f"save result excel file: {os.path.join(self.reports_path)}")
def merge_benchmark_files(self, merged_workbook):
if self.benchmark_path and os.path.exists(self.benchmark_path):
logging.debug("start merge benchmark_file")
merged_sheet = merged_workbook.create_sheet(title="BinaryCompare")
df_benchmark = pd.read_csv(self.benchmark_path)
for r in dataframe_to_rows(df_benchmark, index=False, header=True):
merged_sheet.append(r)
os.remove(self.benchmark_path)
def merge_benchmark_files_to_ttk(self, sheet):
if self.benchmark_path and os.path.exists(self.benchmark_path):
logging.debug("start merge benchmark_file to ttk")
csv_column_map = {
"用例ID": "用例",
"用例名称": "算子名称",
"错误信息": "message",
"WARNING状态": "精度结果",
}
csv_data = pd.read_csv(self.benchmark_path)
excel_column_indices = {}
for csv_column, excel_column_name in csv_column_map.items():
excel_column_indices[csv_column] = self.get_column_index(sheet, excel_column_name)
max_row = sheet.max_row
for index, row in csv_data.iterrows():
new_row = max_row + index + 1
for csv_column, excel_column_index in excel_column_indices.items():
cell = sheet.cell(row=new_row, column=excel_column_index)
cell.value = row[csv_column]
os.remove(self.benchmark_path)