lib/report.py

from datetime import datetime, timedelta
from typing import List
from dataclasses import dataclass, astuple

from tabulate import tabulate
from lib.log_parser import LogTag, LogItem, WebItem
from lib.utils import *

class LogReport:
    """用于处理 log文件"""
    def __init__(self):
        self.items:List[LogItem] = []

    def from_logfile(self, log_file, start_line=0):
        """将log文件中有效的行转换成 LogItem，返回 LogItem的列表
        和当前文件的行数（用于下一个 case读取 log文件的起始行数）"""
        print(f"generate LogReport from logfile: {log_file}")
        with open(log_file, "r") as f:
            lines = f.readlines()
        print(f"read log file lines {start_line}:{len(lines)}")
        for l in lines[start_line:]:
            for item in LogTag:
                if item.value in l:
                    log_item = LogItem.from_log(item, l)
                    self.items.append(log_item)
        return self.items, len(lines)

    def item_to_rows(self):
        """将 LogItem 列表转换成 csv行的格式，每行以 audio开始"""
        rows = []
        current_line = []
        for index, item in enumerate(self.items):
            if item.tag in [LogTag.load_start, LogTag.load_end]:
                continue
            # 每次检查到 audio_end就另起一行
            if item.tag == LogTag.audio_end:
                rows.append(current_line)
                current_line = []
            current_line += [item.tag.name, item.timestamp, item.content]
        return  rows

    def to_csv(self, csv_path=None):
        header_mapping = {
            # 注释掉header，在 csv中就不保留对应的列
            "audio_end_tag": 0,
            "audio_end_tsp": 1,
            "audio_length": 2,
            "transcribe_cost_tag": 3,
            "transcribe_cost_tsp": 4,
            "transcribe_cost": 5,
            "transcribe_end_tag": 6,
            "transcribe_end_tsp": 7,
            "transcribe_output": 8,
            "translate_start_tag": 9,
            "translate_start_tsp": 10,
            "translate_input": 11,
            "translate_cost_tag": 12,
            "translate_cost_tsp": 13,
            "translate_cost": 14,
            "translate_end_tag": 15,
            "translate_end_tsp": 16,
            "translate_output": 17,
        }
        rows = self.item_to_rows()
        header = list(header_mapping.keys())
        rows = [[row[i] for i in header_mapping.values() if i < len(row)] for row in rows]
        save_csv(csv_path, header, rows)

@dataclass
class DelaySummary:
    audio_name:str = ""
    trans_type: str = ""
    audio_length:str = ""
    load_start: datetime =None
    load_end: datetime=None
    load: float = 0
    avg_audio_len: float = 0
    total_tsb: float = 0
    avg_tsb_per_second: float = 0
    total_tsl: float = 0
    avg_tsl_per_second: float = 0
    total_web: float = 0
    avg_web_per_second: float = 0
    avg_web_freq: float = 0

@dataclass
class DelayDetailRow:
    audio_end_tsp:datetime = ""
    audio_length:float =0
    tsb_end_tsp:datetime =""
    tsb_opt:str =""
    tsb_cost:float = 0
    tsb_cost_per_second: float = 0
    tsl_ipt:str =""
    tsl_end_tsp:datetime =""
    tsl_opt:str =""
    tsl_cost:float =0
    tsl_cost_per_second: float = 0
    web_tsp:datetime =""
    web_src:str =""
    web_dst:str =""
    web_delay: float = 0
    web_delay_per_second: float = 0
    web_freq: float = 0
    def __repr__(self):
        return f"Row(audio_length={self.audio_length}, tsb_opt={self.tsb_opt})"

@dataclass
class DelayItem:
    """存储delay 报告中每一个 case的结果"""
    translation_type: str = ''
    audio: str = ""
    audio_length: str = ""
    web_items: List[WebItem] = None
    log_items: List[LogItem] = None

    def to_rows(self):
        """将 log和 web的结果合并, 返回 DelaySummary和 DelayDetail的列表
        返回 row_0包含音频信息和 load 时间
        rows 是每次推理的详细信息"""
        print(f"length of log_items: {len(self.log_items)}")
        web_items_dict = {i.src_text + i.dst_text: i for i in self.web_items}

        summary = DelaySummary(audio_name=self.audio,trans_type=self.translation_type,
                               audio_length=self.audio_length)
        detail_rows = []
        current_row = DelayDetailRow()
        for i in self.log_items:
            if i.tag == LogTag.load_start:
                summary.load_start = i.timestamp
            elif i.tag == LogTag.load_end:
                summary.load_end = i.timestamp
                summary.load = (summary.load_end-summary.load_start).total_seconds()
            elif i.tag == LogTag.audio_end:
                if current_row.audio_length > 0:
                    detail_rows.append(current_row)
                # 每次到 audio_end就是新的一行
                current_row = DelayDetailRow()
                current_row.audio_end_tsp = i.timestamp
                current_row.audio_length = time_to_float(i.content)
            elif i.tag == LogTag.transcribe_end:
                current_row.tsb_end_tsp = i.timestamp
                current_row.tsb_opt = i.content
            elif i.tag == LogTag.transcribe_cost:
                current_row.tsb_cost = time_to_float(i.content)
                current_row.tsb_cost_per_second = current_row.tsb_cost/current_row.audio_length if current_row.audio_length else 0
            elif i.tag == LogTag.translate_start:
                current_row.tsl_ipt = i.content
            elif i.tag in [LogTag.translate_end, LogTag.translate_large_end]:
                current_row.tsl_end_tsp = i.timestamp
                current_row.tsl_opt = i.content
                # 假设一行有翻译结果时，就一定已经有asr的结果
                if web_item:=web_items_dict.get(current_row.tsb_opt+current_row.tsl_opt):
                    current_row.web_tsp = web_item.timestamp
                    current_row.web_src = web_item.src_text
                    current_row.web_dst = web_item.dst_text
                    current_row.web_delay = (current_row.web_tsp - current_row.audio_end_tsp).total_seconds()
                    current_row.web_delay_per_second = current_row.web_delay / current_row.audio_length if current_row.audio_length else 0
                    # 删除 dict已匹配过的内容，避免多次匹配
                    web_items_dict.pop(current_row.tsb_opt+current_row.tsl_opt)
                    if len(detail_rows)>=1 and detail_rows[-1].web_tsp:
                        current_row.web_freq = (current_row.web_tsp - detail_rows[-1].web_tsp).total_seconds()

            elif i.tag in [LogTag.translate_cost, LogTag.translate_large_cost]:
                current_row.tsl_cost = time_to_float(i.content)
                current_row.tsl_cost_per_second = current_row.tsl_cost/current_row.audio_length if current_row.audio_length else 0
        summary = self.get_summary(summary, detail_rows)
        return summary, detail_rows # [astuple(i) for i in rows]
    def get_summary(self,summary: DelaySummary, detail_rows):
        audio_len = []
        total_tsb = []
        avg_tsb_per_second = []
        total_tsl = []
        avg_tsl_per_second = []
        total_web = []
        avg_web_per_second = []
        web_freq = []
        for row in detail_rows:
            if row.audio_length:
                audio_len.append(row.audio_length)
            if row.tsb_cost:
                total_tsb.append(row.tsb_cost)
            if row.tsb_cost_per_second:
                avg_tsb_per_second.append(row.tsb_cost_per_second)
            if row.tsl_cost:
                total_tsl.append(row.tsl_cost)
            if row.tsl_cost_per_second:
                avg_tsl_per_second.append(row.tsl_cost_per_second)
            if row.web_delay:
                total_web.append(row.web_delay)
            if row.web_delay_per_second:
                avg_web_per_second.append(row.web_delay_per_second)
            if row.web_freq:
                web_freq.append(row.web_freq)

        summary.avg_audio_len = sum(audio_len) / len(audio_len) if len(audio_len)>0 else 0
        summary.total_tsb = sum(total_tsb)
        summary.avg_tsb_per_second = sum(avg_tsb_per_second) / len(avg_tsb_per_second) if len(avg_tsb_per_second)>0 else 0
        summary.total_tsl = sum(total_tsl)
        summary.avg_tsl_per_second = sum(avg_tsl_per_second) / len(avg_tsl_per_second) if len(avg_tsl_per_second)>0 else 0
        summary.total_web = sum(total_web)
        summary.avg_web_per_second = sum(avg_web_per_second) / len(avg_web_per_second) if len(avg_web_per_second)>0 else 0
        summary.avg_web_freq = sum(web_freq) /len(web_freq) if len(web_freq)>0 else 0
        return summary


class DelayReport:
    """存储delay 报告中所有 case的结果"""
    start_line = 0
    items: List[DelayItem] = []
    def print_summary(self, data):
        print(tabulate(data))

    def to_csv(self, csv_path):
        summaries = [["audio_name", "translation", "audio_length",
                      "load_start", "load_end", "load", "avg_audio_len",
                      "total_tsb", "avg_tsb_per_sec", "total_tsl", "avg_tsl_per_sec",
                      "total_web", "avg_web_per_sec", "avg_web_freq"]]
        details = [["audio_end_tsp", "audio_length",
                    "tsb_end_tsp", "tsp_opt", "tsb_cost", "tsb_cost_per_sec",
                    "tsl_ipt", "tsl_end_tsp", "tsl_opt", "tsl_cost", "tsl_cost_per_sec",
                    "web_tsp", "web_src", "web_dst", "web_delay", "web_delay_per_sec", "web_freq"]]
        for i in self.items:
            summary, detail_rows = i.to_rows()
            summaries.append(astuple(summary))
            details += [astuple(i) for i in detail_rows]
            details.append([])
        self.print_summary(summaries)
        save_csv(csv_path, [], summaries+[[]]+details)

@dataclass
class AccuracyItem:
    """存储accuracy 报告中每一个 case的结果"""
    translation_type: str = ''
    audio: str = ""
    audio_length: str = ""
    audio_text: str = ""
    src_text: str = ""
    dst_text: str = ""
    asr_accuracy: tuple= (0,1)
    text_compare: str = ""
    def __post_init__(self):

        if self.translation_type == "en2zh":
            text1 = clean_text_for_comparison_en(self.audio_text)
            text2 = clean_text_for_comparison_en(self.src_text)
            spliter = " "
        else:
            text1 = clean_text_for_comparison_zh(self.audio_text)
            text2 = clean_text_for_comparison_zh(self.src_text)
            spliter = ""
        self.asr_accuracy = run_textdistance(text1, text2)
        self.text_compare = highlight_diff(text1, text2, spliter)

    def to_list(self):
        return [self.audio, self.translation_type, self.audio_length,
                self.asr_accuracy[0], self.asr_accuracy[1],
                self.src_text, self.audio_text, self.text_compare]

class AccuracyReport:
    items:List[AccuracyItem] = []

    def print_summary(self):
        header = ["audio", "distance", "normalized distance"]
        rows = [[i.audio, i.asr_accuracy[0], i.asr_accuracy[1]] for i in self.items]
        print(tabulate(rows, header))

    def to_csv(self, csv_path):
        print("accuracy item length: ", len(self.items))
        self.print_summary()
        header = ["audio_name", "translation", "audio_length",
                  "distance", "normalized distance",
                  "src text", "audio text", "text compare"]
        save_csv(csv_path, header, [i.to_list() for i in self.items])