import os
import torch
from atk.common.log import Logger
from atk.common.utils import get_file_md5
from atk.configs.results_config import AccuracyConfig
from atk.configs.single_benchmark_config import (
SingleBenchmarkCompareStandard,
SingleBenchSummary
)
from atk.tasks.post_process import ACCURACY_REGISTRY
from atk.tasks.post_process.base_compare import BaseAccuracyCompare
from atk.tasks.post_process.utils import check_invalid_value
from atk.tasks.post_process.equal_compare import Compare as equal_compare
logging = Logger().get_logger()
@ACCURACY_REGISTRY.register("single_bm")
@ACCURACY_REGISTRY.register("default")
class SingleBenchmarkAccuracyCompare(BaseAccuracyCompare):
def __init__(self, case_config, **kwargs):
super(SingleBenchmarkAccuracyCompare, self).__init__(case_config, **kwargs)
logging.debug(f"SingleBenchmark accuracy compare kwargs = {kwargs}", )
self.benchmark_standard = SingleBenchmarkCompareStandard()
self.benchmark_standard.update(**kwargs)
@staticmethod
def check_data_equality(local_output, remote_output):
return torch.equal(local_output, remote_output)
@staticmethod
def compute_quantize_accuracy(local_output, remote_output, data_file):
'''
量化计算：int8输出，绝对误差小于等于1
'''
diff_value = torch.subtract(local_output.to(torch.int64), remote_output)
diff_abs = torch.abs(diff_value)
flat_diff_abs = diff_abs.view(-1)
max_diff, max_diff_idx = torch.max(flat_diff_abs, dim=0)
result = torch.all(diff_abs <= 1)
acc_result = AccuracyConfig(
result=result.item(),
filename=data_file,
max_diff=max_diff.item(),
max_diff_idx=max_diff_idx.item(),
)
return acc_result
def compare_file_md5(self, local_data_path, remote_data_path):
acc_result = None
local_digest = get_file_md5(local_data_path)
remote_digest = self.remote_manager.get_file_md5(remote_data_path)
filename = os.path.basename(local_data_path)
info = f"The data comparison between {local_data_path} and {remote_data_path} by md5(hashlib)"
if local_digest == remote_digest:
logging.debug(info + "is passed.")
acc_result = AccuracyConfig(result=True, filename=filename)
else:
logging.debug(info + "is failed.")
return acc_result
def check_output_size(self, local_output, remote_output, data_file):
acc_result = None
if local_output.numel() == 0 and remote_output.numel() == 0:
info = "The npu_output is [], and it is same as bm_output, the result of data_compare is Pass"
logging.debug(info)
acc_result = AccuracyConfig(filename=data_file, result=True, error_info=info)
if local_output.size() != remote_output.size():
error_info = (f"the size of npu output[{local_output.size()}] and "
f"benchmark[{remote_output.size()}] is not equal.")
logging.error(error_info)
acc_result = AccuracyConfig(filename=data_file, result=False, error_info=error_info)
return acc_result
def compute_accuracy_result(self, local_output, remote_output, data_file):
if torch.is_complex(local_output):
real_ret = self.compute_mere_mare(local_output.real, remote_output.real, data_file)
imag_ret = self.compute_mere_mare(local_output.imag, remote_output.imag, data_file)
acc_ret = AccuracyConfig(
result=real_ret.result and imag_ret.result,
filename=data_file,
error_info=f"real: MERE={real_ret.mean_rel_err}, MARE={real_ret.max_rel_err}\n"
f"imag: MERE={imag_ret.mean_rel_err}, MARE={imag_ret.max_rel_err}",
)
else:
acc_ret = self.compute_mere_mare(local_output, remote_output, data_file)
return acc_ret
def compute_mere_mare(self, local_output, remote_output, data_file):
dtype = local_output.dtype
if local_output.dtype in [
torch.float16,
torch.bfloat16,
] and remote_output.dtype in [torch.float32]:
local_output = local_output.to(torch.float32)
if check_invalid_value(remote_output) or check_invalid_value(local_output):
acc_result = AccuracyConfig(result=True, filename=data_file)
return acc_result
if check_invalid_value(remote_output):
acc_result = AccuracyConfig(result=True, filename=data_file)
return acc_result
if check_invalid_value(local_output):
error_info = f"The NPU result file {data_file} contains nan/inf value"
acc_result = AccuracyConfig(result=False, filename=data_file, error_info=error_info)
return acc_result
if dtype in [torch.int8]:
logging.info(f"output is {dtype}, use quantize standard.")
acc_result = self.compute_quantize_accuracy(local_output, remote_output, data_file)
return acc_result
if dtype in [torch.uint8, torch.int16, torch.int32, torch.int64]:
logging.info(f"output is {dtype}, use binary consistency standard.")
instance_equal_compare = equal_compare(self.case_config)
acc_result = instance_equal_compare.compute_accuracy_result(
local_output, remote_output, data_file
)
return acc_result
threshold = self.benchmark_standard.get_threshold(dtype)
mare_ratio = self.benchmark_standard.mare_ratio
if threshold is None:
acc_result = AccuracyConfig(filename=data_file)
if self.check_data_equality(local_output, remote_output):
acc_result.update(result=True)
else:
error_info = f"Actual value is different from benchmark value with equal comparison in {data_file}"
acc_result.update(result=False, error_info=error_info)
return acc_result
acc_result = AccuracyConfig(filename=data_file)
self.compute_mare(local_output, remote_output, threshold, mare_ratio, acc_result)
benchmark_summary = SingleBenchSummary(acc_result, threshold, mare_ratio)
error_info = benchmark_summary.get_result_msg()
acc_result.update(result=benchmark_summary.check_result, error_info=error_info)
return acc_result
def compute_mare(self, local_output, remote_output, threshold, mare_ratio, acc_result: AccuracyConfig):
eps = 1e-7
diff = torch.abs(local_output - remote_output)
denom = torch.abs(remote_output) + eps
rel_err = diff / denom
mean_rel_err = torch.mean(rel_err).item()
max_rel_err = torch.max(rel_err).item()
acc_result.update(
mean_rel_err=float(mean_rel_err),
max_rel_err=float(max_rel_err),
)