| File was renamed from funasr/metrics/compute_wer.py |
| | |
| | | import os |
| | | import numpy as np |
| | | import sys |
| | | import hydra |
| | | |
| | | def compute_wer(ref_file, |
| | | hyp_file, |
| | | cer_detail_file): |
| | | cer_file, |
| | | cn_postprocess=False, |
| | | ): |
| | | rst = { |
| | | 'Wrd': 0, |
| | | 'Corr': 0, |
| | |
| | | for line in hyp_reader: |
| | | key = line.strip().split()[0] |
| | | value = line.strip().split()[1:] |
| | | if cn_postprocess: |
| | | value = value.replace(" ", "") |
| | | value = [x for x in value] |
| | | value = " ".join(value) |
| | | hyp_dict[key] = value |
| | | with open(ref_file, 'r') as ref_reader: |
| | | for line in ref_reader: |
| | | key = line.strip().split()[0] |
| | | value = line.strip().split()[1:] |
| | | if cn_postprocess: |
| | | value = value.replace(" ", "") |
| | | value = [x for x in value] |
| | | value = " ".join(value) |
| | | ref_dict[key] = value |
| | | |
| | | cer_detail_writer = open(cer_detail_file, 'w') |
| | | cer_detail_writer = open(cer_file, 'w') |
| | | for hyp_key in hyp_dict: |
| | | if hyp_key in ref_dict: |
| | | out_item = compute_wer_by_line(hyp_dict[hyp_key], ref_dict[hyp_key]) |
| | |
| | | cer_detail_writer.write(hyp_key + print_cer_detail(out_item) + '\n') |
| | | cer_detail_writer.write("ref:" + '\t' + " ".join(list(map(lambda x: x.lower(), ref_dict[hyp_key]))) + '\n') |
| | | cer_detail_writer.write("hyp:" + '\t' + " ".join(list(map(lambda x: x.lower(), hyp_dict[hyp_key]))) + '\n') |
| | | cer_detail_writer.flush() |
| | | |
| | | if rst['Wrd'] > 0: |
| | | rst['Err'] = round(rst['wrong_words'] * 100 / rst['Wrd'], 2) |
| | |
| | | cer_detail_writer.write("%SER " + str(rst['S.Err']) + " [ " + str(rst['wrong_sentences']) + " / " + str(rst['Snt']) + " ]" + '\n') |
| | | cer_detail_writer.write("Scored " + str(len(hyp_dict)) + " sentences, " + str(len(hyp_dict) - rst['Snt']) + " not present in hyp." + '\n') |
| | | |
| | | cer_detail_writer.close() |
| | | |
| | | def compute_wer_by_line(hyp, |
| | | ref): |
| | |
| | | + str(rst['sub']) + ") corr:" + '{:.2%}'.format(rst['cor']/rst['nwords']) |
| | | + ",cer:" + '{:.2%}'.format(rst['wrong']/rst['nwords'])) |
| | | |
| | | if __name__ == '__main__': |
| | | if len(sys.argv) != 4: |
| | | print("usage : python compute-wer.py test.ref test.hyp test.wer") |
| | | sys.exit(0) |
| | | |
| | | ref_file = sys.argv[1] |
| | | hyp_file = sys.argv[2] |
| | | cer_detail_file = sys.argv[3] |
| | | compute_wer(ref_file, hyp_file, cer_detail_file) |
| | | @hydra.main(config_name=None, version_base=None) |
| | | def main_hydra(cfg: DictConfig): |
| | | ref_file = cfg.get("ref_file", None) |
| | | hyp_file = cfg.get("hyp_file", None) |
| | | cer_file = cfg.get("cer_file", None) |
| | | cn_postprocess = cfg.get("cn_postprocess", False) |
| | | if ref_file is None or hyp_file is None or cer_file is None: |
| | | print("usage : python -m funasr.metrics.wer ++ref_file=test.ref ++hyp_file=test.hyp ++cer_file=test.wer ++cn_postprocess=false") |
| | | sys.exit(0) |
| | | |
| | | compute_wer(ref_file, hyp_file, cer_file, cn_postprocess) |
| | | |
| | | if __name__ == '__main__': |
| | | main_hydra() |
| | | |
| | | |
| | | |