import os import subprocess from collections import defaultdict from typing import Dict, List from utils import asr_ter from utils.logger import logger log_mid_result = int(os.getenv("log", 0)) == 1 class AsrEvaluator: def __init__(self) -> None: self.query_count = 0 # query 数目（语音数目） self.voice_count = 0 # 有开始和结束时间的语音条数（用于 RTF 计算） self.cut_punc = [] # 切分标点符号，需要注意切分的时候根据列表中的顺序进行切分，比如 ... 应该放到 . 之前。 # cer 属性 self.one_minus_cer = 0 # 每个 query 的 1 - cer 和 self.token_count = 0 # 每个 query 的字数/词数和 # 句子切分率属性 self.miss_count = 0 # 每个 query miss-count 和 self.more_count = 0 # 每个 query more-count 和 self.cut_count = 0 # 每个 query cut-count 和 self.rate = 0 # 每个 query 的 cut-rate 和 # detail case self.result = [] def evaluate(self, eval_result): pass def post_evaluate(self): pass def gen_result(self) -> Dict: output_result = dict() output_result["query_count"] = self.query_count output_result["voice_count"] = self.voice_count output_result["token_cnt"] = self.token_count output_result["one_minus_cer"] = self.one_minus_cer output_result["one_minus_cer_metrics"] = self.one_minus_cer / self.query_count output_result["miss_count"] = self.miss_count output_result["more_count"] = self.more_count output_result["cut_count"] = self.cut_count output_result["cut_rate"] = self.rate output_result["cut_rate_metrics"] = self.rate / self.query_count output_result["rtf"] = self.rtf output_result["rtf_end"] = self.rtf_end output_result["rtf_metrics"] = self.rtf / self.voice_count output_result["rtf_end_metrics"] = self.rtf_end / self.voice_count detail_case = self.result return output_result, detail_case def _get_predict_final_sentences(self, predict_data: List[Dict]) -> List[str]: """ 获取 predict data 数据，然后将其中 final 的句子拿出来，放到列表里。 """ return [ item["recoginition_results"]["text"] for item in predict_data if item["recoginition_results"]["final_result"] ] def _sentence_final_index(self, sentences: List[str], tokens: List[str], tokenizer="word") -> List[int]: """ 获取 sentence 结束的字对应的 token 索引值。 """ token_index_list = [] token_idx = 0 for sentence in sentences: for token in Tokenizer.tokenize(sentence, tokenizer): if token not in tokens: continue while tokens[token_idx] != token: token_idx += 1 token_index_list.append(token_idx) return token_index_list def _voice_to_cut_sentence(self, voice_sentences: List[str]) -> Dict: """ 将数据集的语音片段转换为最小切分单元列表。 使用 cut_punc 中的所有 punc 进行依次切分，最后去除掉完全空的内容 示例： ["你好，你好呀", "你好，我在写抽象的代码逻辑"] -> cut_sentences: ["你好", "你好呀", "你好", "我在写抽象的代码逻辑"] cut_sentence_index_list: [1, 3] ("你好呀" 对应 1-idx, "我在写抽象的代码逻辑" 对应 3-idx) """ voice_sentences_result = defaultdict(list) for voice_sentence in voice_sentences: sentence_list = [voice_sentence] sentence_tmp_list = [] for punc in self.cut_punc: for sentence in sentence_list: sentence_tmp_list.extend(sentence.split(punc)) sentence_list, sentence_tmp_list = sentence_tmp_list, [] sentence_list = [item for item in sentence_list if item] # 切分后的句子单元 voice_sentences_result["cut_sentences"].extend(sentence_list) # 每个语音单元最后一个字对应的句子单元的索引 voice_sentences_result["cut_sentence_index_list"].append(len(voice_sentences_result["cut_sentences"]) - 1) return voice_sentences_result def _voice_bytes_index(self, timestamp, sample_rate=16000, bit_depth=16, channels=1): """ timestamp: 时间, 单位秒 """ bytes_per_sample = bit_depth // 8 return timestamp * sample_rate * bytes_per_sample * channels class AsrZhEvaluator(AsrEvaluator): """ 中文的评估方式 """ def __init__(self): super().__init__() self.cut_zh_punc = ["······", "......", "。", "，", "？", "！", "；", "："] self.cut_en_punc = ["...", ".", ",", "?", "!", ";", ":"] self.cut_punc = self.cut_zh_punc + self.cut_en_punc def evaluate(self, eval_result) -> Dict: self.query_count += 1 self.voice_count += len(eval_result["voice"]) # 获取，标注结果 & 语音单元（非句子单元） label_voice_sentences = [item["answer"] for item in eval_result["voice"]] # print("label_voice_sentences", label_voice_sentences) # 获取，标注结果 & 语音单元 -> 句子单元的转换情况 voice_to_cut_info = self._voice_to_cut_sentence(label_voice_sentences) # print("voice_to_cut_info", voice_to_cut_info) # 获取，标注结果 & 句子单元 label_sentences = voice_to_cut_info["cut_sentences"] # 获取，标注结果 & 语音单元 -> 句子单元的映射关系，每个语音单元最后一个字对应的句子单元的索引 cut_sentence_index_list = voice_to_cut_info["cut_sentence_index_list"] # 标注结果 & 句子单元 & norm 操作 label_sentences = [self._sentence_norm(sentence) for sentence in label_sentences] if log_mid_result: logger.info(f"label_sentences {label_sentences}") # print("label_sentences", label_sentences) # 预测结果 & 句子单元 predict_sentences_raw = self._get_predict_final_sentences(eval_result["predict_data"]) # print("predict_sentences_raw", predict_sentences_raw) # 预测结果 & 句子单元 & norm 操作 predict_sentences = [self._sentence_norm(sentence) for sentence in predict_sentences_raw] if log_mid_result: logger.info(f"predict_sentences {predict_sentences}") # print("predict_sentences", predict_sentences) # 基于最小编辑距离进行 token 匹配，获得匹配后的 token 列表 label_tokens, predict_tokens = self._sentence_transfer("".join(label_sentences), "".join(predict_sentences)) # cer 计算 cer_info = self.cer(label_sentences, predict_sentences) if log_mid_result: logger.info(f"cer_info {cer_info}") # print("cer_info", cer_info) self.one_minus_cer += cer_info["one_minus_cer"] self.token_count += cer_info["token_count"] # 句子切分准召率 cut_info = self.cut_rate(label_sentences, predict_sentences, label_tokens, predict_tokens) if log_mid_result: logger.info(f"{cut_info['miss_count']}, {cut_info['more_count']}, {cut_info['rate']}") # print("cut_info", cut_info) # print(cut_info["miss_count"], cut_info["more_count"], cut_info["rate"]) self.miss_count += cut_info["miss_count"] self.more_count += cut_info["more_count"] self.cut_count += cut_info["cut_count"] self.rate += cut_info["rate"] self.result.append( { "label_tokens": label_tokens, "predict_tokens": predict_tokens, "one_minus_cer": cer_info["one_minus_cer"], "token_count": cer_info["one_minus_cer"], "miss_count": cut_info["miss_count"], "more_count": cut_info["more_count"], "cut_count": cut_info["cut_count"], "rate": cut_info["rate"], } ) def cer(self, label_sentences, predict_sentences): pred_str = ''.join(predict_sentences) if predict_sentences is not None else '' label_str = ''.join(label_sentences) r = asr_ter.calc_ter_speechio(pred_str, label_str) one_minus_cer = max(1.0 - r['ter'], 0) token_count = r['ref_all_token_cnt'] return {"one_minus_cer": one_minus_cer, "token_count": token_count} def cut_rate(self, label_sentences, predict_sentences, label_tokens, predict_tokens): label_final_index_list = set(self._sentence_final_index(label_sentences, label_tokens)) pred_final_index_list = set(self._sentence_final_index(predict_sentences, predict_tokens)) label_sentence_count = len(label_final_index_list) miss_count = len(label_final_index_list - pred_final_index_list) more_count = len(pred_final_index_list - label_final_index_list) rate = max(1 - (miss_count + more_count * 2) / label_sentence_count, 0) return { "miss_count": miss_count, "more_count": more_count, "cut_count": label_sentence_count, "rate": rate, "label_final_index_list": label_final_index_list, "pred_final_index_list": pred_final_index_list, } def _sentence_norm(self, sentence, tokenizer="word"): """ 对句子进行 norm 操作 """ from utils.speechio import textnorm_zh as textnorm if tokenizer == "word": normalizer = textnorm.TextNorm( to_banjiao=True, to_upper=True, to_lower=False, remove_fillers=True, remove_erhua=False, # 这里同批量识别不同，改成了 False check_chars=False, remove_space=False, cc_mode="", ) return normalizer(sentence) else: logger.error("tokenizer error, not support.") def _sentence_transfer(self, label_sentence: str, predict_sentence: str, tokenizer="char"): """ 基于最小编辑距离，将 label 和 predict 进行字的位置匹配，并生成转换后的结果 args: label: "今天的通话质量不错呀昨天的呢" predict: "今天的通话质量不错昨天呢星期" tokenizer: 分词方式 return: label: ["今", "天", "的", "通", "话", "质", "量", "不", "错", "呀", "昨", "天", "的", "呢", None, None] predict: ["今", "天", "的", "通", "话", "质", "量", "不", "错", None, "昨", "天", None, "呢", "星", "期"] """ from utils.speechio import error_rate_zh as error_rate if tokenizer == "char": alignment, score = error_rate.EditDistance( error_rate.tokenize_text(label_sentence, tokenizer), error_rate.tokenize_text(predict_sentence, tokenizer), ) label_tokens, pred_tokens = [], [] for align in alignment: # print(align.__dict__) label_tokens.append(align.ref) pred_tokens.append(align.hyp) return (label_tokens, pred_tokens) else: logger.error("tokenizer 出错了，暂时不支持其它的") def _pred_data_transfer(self, predict_data, recv_time): """ predict_data = [ {"recoginition_results": {"text": "1", "final_result": False, "para_seq": 0}}, {"recoginition_results": {"text": "12", "final_result": False, "para_seq": 0}}, {"recoginition_results": {"text": "123", "final_result": True, "para_seq": 0}}, {"recoginition_results": {"text": "4", "final_result": False, "para_seq": 0}}, {"recoginition_results": {"text": "45", "final_result": False, "para_seq": 0}}, {"recoginition_results": {"text": "456", "final_result": True, "para_seq": 0}}, ] recv_time = [1, 3, 5, 6, 7, 8] -> [ [{'text': '1', 'time': 1}, {'text': '12', 'time': 3}, {'text': '123', 'time': 5}], [{'text': '4', 'time': 6}, {'text': '45', 'time': 7}, {'text': '456', 'time': 8}], ] """ pred_sentence_info = [] pred_sentence_index = 0 for predict_item, recv_time_item in zip(predict_data, recv_time): if len(pred_sentence_info) == pred_sentence_index: pred_sentence_info.append([]) pred_sentence_info[pred_sentence_index].append( { "text": predict_item["recoginition_results"]["text"], "time": recv_time_item, } ) if predict_item["recoginition_results"]["final_result"]: pred_sentence_index += 1 return pred_sentence_info class AsrEnEvaluator(AsrEvaluator): """ 英文的评估方式 """ def evaluate(self, eval_result) -> Dict: self.query_count += 1 self.voice_count += len(eval_result["voice"]) # 获取，标注结果 & 语音单元（非句子单元） label_voice_sentences = [item["answer"] for item in eval_result["voice"]] # print("label_voice_sentences", label_voice_sentences) # 获取，标注结果 & 语音单元 -> 句子单元的转换情况 voice_to_cut_info = self._voice_to_cut_sentence(label_voice_sentences) # print("voice_to_cut_info", voice_to_cut_info) # 获取，标注结果 & 句子单元 label_sentences = voice_to_cut_info["cut_sentences"] # 获取，标注结果 & 语音单元 -> 句子单元的映射关系，每个语音单元最后一个字对应的句子单元的索引 cut_sentence_index_list = voice_to_cut_info["cut_sentence_index_list"] # 标注结果 & 句子单元 & norm 操作 label_sentences = self._sentence_list_norm(label_sentences) # [self._sentence_norm(sentence) for sentence in label_sentences] # print("label_sentences", label_sentences) if log_mid_result: logger.info(f"label_sentences {label_sentences}") # 预测结果 & 句子单元 predict_sentences_raw = self._get_predict_final_sentences(eval_result["predict_data"]) # print("predict_sentences_raw", predict_sentences_raw) # 预测结果 & 句子单元 & norm 操作 predict_sentences = self._sentence_list_norm(predict_sentences_raw) # [self._sentence_norm(sentence) for sentence in predict_sentences_raw] # print("predict_sentences", predict_sentences) if log_mid_result: logger.info(f"predict_sentences {predict_sentences}") label_tokens, predict_tokens = self._sentence_transfer(" ".join(label_sentences), " ".join(predict_sentences)) # print(label_tokens) # print(predict_tokens) # cer 计算 cer_info = self.cer(label_tokens, predict_tokens) # print("cer_info", cer_info) if log_mid_result: logger.info(f"cer_info {cer_info}") self.one_minus_cer += cer_info["one_minus_cer"] self.token_count += cer_info["token_count"] # 句子切分准召率 cut_info = self.cut_rate(label_sentences, predict_sentences, label_tokens, predict_tokens) # print(cut_info["miss_count"], cut_info["more_count"], cut_info["rate"]) # print("cut_info", cut_info) if log_mid_result: logger.info(f"{cut_info['miss_count']}, {cut_info['more_count']}, {cut_info['rate']}") self.miss_count += cut_info["miss_count"] self.more_count += cut_info["more_count"] self.cut_count += cut_info["cut_count"] self.rate += cut_info["rate"] self.result.append( { "label_tokens": label_tokens, "predict_tokens": predict_tokens, "one_minus_cer": cer_info["one_minus_cer"], "token_count": cer_info["one_minus_cer"], "miss_count": cut_info["miss_count"], "more_count": cut_info["more_count"], "cut_count": cut_info["cut_count"], "rate": cut_info["rate"], } ) def cer(self, label_tokens, predict_tokens): s, d, i, c = 0, 0, 0, 0 for label_token, predict_token in zip(label_tokens, predict_tokens): if label_token == predict_token: c += 1 elif predict_token is None: d += 1 elif label_token is None: i += 1 else: s += 1 cer = (s + d + i) / (s + d + c) one_minus_cer = max(1.0 - cer, 0) token_count = s + d + c return {"one_minus_cer": one_minus_cer, "token_count": token_count} def cut_rate(self, label_sentences, predict_sentences, label_tokens, predict_tokens): label_final_index_list = set(self._sentence_final_index(label_sentences, label_tokens, "whitespace")) pred_final_index_list = set(self._sentence_final_index(predict_sentences, predict_tokens, "whitespace")) label_sentence_count = len(label_final_index_list) miss_count = len(label_final_index_list - pred_final_index_list) more_count = len(pred_final_index_list - label_final_index_list) rate = max(1 - (miss_count + more_count * 2) / label_sentence_count, 0) return { "miss_count": miss_count, "more_count": more_count, "cut_count": label_sentence_count, "rate": rate, "label_final_index_list": label_final_index_list, "pred_final_index_list": pred_final_index_list, } def _sentence_list_norm(self, sentence_list, tokenizer="whitespace"): pwd = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) with open('./predict.txt', 'w', encoding='utf-8') as fp: for idx, sentence in enumerate(sentence_list): fp.write('%s\t%s\n' % (idx, sentence)) subprocess.run( f'PYTHONPATH={pwd}/utils/speechio python {pwd}/utils/speechio/textnorm_en.py --has_key --to_upper ./predict.txt ./predict_norm.txt', shell=True, check=True, ) sentence_norm = [] with open('./predict_norm.txt', 'r', encoding='utf-8') as fp: for line in fp.readlines(): line_split_result = line.strip().split('\t', 1) if len(line_split_result) >= 2: sentence_norm.append(line_split_result[1]) # 有可能没有 norm 后就没了 return sentence_norm def _sentence_transfer(self, label_sentence: str, predict_sentence: str, tokenizer="whitespace"): """ 基于最小编辑距离，将 label 和 predict 进行字的位置匹配，并生成转换后的结果 args: label: "HELLO WORLD ARE U OK YEP" predict: "HELLO WORLD U ARE U OK YEP" tokenizer: 分词方式 return: label: ["HELLO", "WORLD", None, "ARE", "U", "OK", "YEP"] predict: ["HELLO", "WORLD", "U", "ARE", "U", "OK", "YEP"] """ from utils.speechio import error_rate_zh as error_rate if tokenizer == "whitespace": alignment, score = error_rate.EditDistance( error_rate.tokenize_text(label_sentence, tokenizer), error_rate.tokenize_text(predict_sentence, tokenizer), ) label_tokens, pred_tokens = [], [] for align in alignment: label_tokens.append(align.ref) pred_tokens.append(align.hyp) return (label_tokens, pred_tokens) else: logger.error("tokenizer 出错了，暂时不支持其它的") def post_evaluate(self) -> Dict: pass