init project code

app_runner.py

@@ -0,0 +1,80 @@
+import signal
+from typing import Literal
+import subprocess
+import time
+import threading
+import os
+import signal
+
+from environment import *
+
+
+class AppRunner:
+    def __init__(self, run_type: RunType):
+        self.run_type = run_type
+        self.sub_proc = None
+
+    def start(self):
+        print(f"start app by type: {self.run_type}")
+        if self.run_type==RunType.electron:
+            cmd_args = [APP_PATH, f"--remote-debugging-port={DEBUG_PORT}"]
+            cwd = None
+            log_file = APP_LOG
+        elif self.run_type == RunType.code:
+            cmd_args = ["python", str(CODE_PATH)]
+            cwd = CODE_DIR
+            log_file = CODE_LOG
+        else:
+            raise TypeError(f"invalid run_type: {self.run_type}")
+        self.clear_log(log_file)
+        self.sub_proc = subprocess.Popen(
+            cmd_args,
+            cwd=cwd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            preexec_fn= os.setsid
+        )
+        print(f"translator started, PID: {self.sub_proc.pid}")
+        time.sleep(10) # 等待app 启动
+        self.wait_for_app_start(log_file)
+        print(f"translator is ready.")
+        return self.sub_proc
+    def clear_log(self, log_path: Path):
+        if log_path.exists():
+            log_path.unlink(missing_ok=True)
+
+    def wait_for_app_start(self, log_file, timeout = 30, interval=3):
+        for i in range(int(timeout/interval)):
+            ret = subprocess.run(f"tail -n 10 {log_file}", check=True, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                             universal_newlines=True)
+            if "Pipeline is ready" in ret.stdout:
+                print(ret.stdout)
+                return True
+            # print(ret.stdout)
+            print(f"app is not started yet, retry {i+1}...")
+            time.sleep(interval)
+        return False
+
+    def stop(self):
+        pgid = os.getpgid(self.sub_proc.pid)
+        os.killpg(pgid, signal.SIGTERM)
+        try:
+            time.sleep(3)
+            return_code = self.sub_proc.wait(timeout=10)
+            print(f"进程组已终止，进程退出码: {return_code}")
+        except subprocess.TimeoutExpired:
+            print("超时，进程组可能未完全终止，尝试 SIGKILL...")
+            os.killpg(pgid, signal.SIGKILL)
+            self.sub_proc.wait(timeout=10)
+            print("进程组已被强制杀死。")
+        # stdout, stderr = self.sub_proc.communicate()
+        # print("\n最终 STDOUT:")
+        # print(stdout)
+        # print("\n最终 STDERR:")
+        # print(stderr)
+
+if __name__ == '__main__':
+    app = AppRunner(RunType.code)
+    app.start()
+    print(app.wait_for_app_start())
+    app.stop()

audio.py

@@ -0,0 +1,37 @@
+import pygame
+import librosa
+
+def play_audio(audio_path, volume=0.7):
+    pygame.mixer.init()  # 初始化 pygame 的音频模块
+    try:
+        pygame.mixer.music.load(audio_path)  # 加载音频文件
+        pygame.mixer.music.set_volume(volume)
+        pygame.mixer.music.play()
+        print(f"开始播放: {audio_path}")
+    except pygame.error as e:
+        print(f"播放音频文件时发生错误: {e}")
+    # finally:
+    #     pygame.mixer.quit()  # 退出 pygame 的音频模块
+
+def play_audio_until_end(audio_path, volume=0.7):
+    pygame.mixer.init()  # 初始化 pygame 的音频模块
+    try:
+        pygame.mixer.music.load(audio_path)  # 加载音频文件
+        pygame.mixer.music.set_volume(volume)
+        pygame.mixer.music.play()
+        print(f"开始播放: {audio_path}")
+        while pygame.mixer.music.get_busy():  # 等待音频播放结束
+            pygame.time.Clock().tick(100)  # 控制循环速度，避免 CPU 占用过高
+        print("音频播放结束")
+    except pygame.error as e:
+        print(f"播放音频文件时发生错误: {e}")
+    # finally:
+    #     pygame.mixer.quit()  # 退出 pygame 的音频模块
+
+def get_length(audio_path):
+    try:
+        duration = librosa.get_duration(path=audio_path)
+        print(f"音频时长: {duration} 秒")
+        return duration
+    except Exception as e:
+        print(f"获取音频时长时发生错误: {e}")

environment.py

@@ -0,0 +1,23 @@
+from pathlib import Path
+from enum import Enum
+
+
+
+PROJECT_DIR = Path("/Users/jeqin/work/code/TestTranslator")
+APP_PATH = Path("/Applications/YoYo Translator.app/Contents/MacOS/YoYo Translator")
+APP_LOG = Path('/tmp/translator.log')
+CODE_DIR = Path("/Users/jeqin/work/code/Translator")
+CODE_PATH = CODE_DIR / "main.py"
+CODE_LOG = CODE_DIR / "translator.log"
+
+DEBUG_PORT = 9222
+TEST_DATA = PROJECT_DIR / "tests" / "test_data"
+TEST_AUDIOS_DIR = TEST_DATA / "test_audios"
+
+REPORTS_DIR  = PROJECT_DIR / "reports"
+SCREENSHOT_DIR = PROJECT_DIR / "screenshots"
+
+class RunType(Enum):
+    code = 0
+    electron = 1
+RUN_TYPE = RunType.electron # electron or web

log_parser.py

@@ -0,0 +1,51 @@
+from dataclasses import dataclass
+from enum import Enum
+from datetime import datetime
+import re
+
+
+class LogTag(Enum):
+    load_start:str = "start initial TranslatePipe"
+    load_end:str = "Pipeline is ready."
+    audio_end:str = "Audio buffer length"
+    transcribe_end:str = "transcribe output"
+    transcribe_cost:str = "transcribe cost"
+    translate_start:str = "Translation input"
+    translate_end:str = "Translation out"
+    translate_cost:str = "Translate cost"
+    translate_large_end:str = "Translation large model output"
+    translate_large_cost:str = "Translate-large cost"
+
+
+@dataclass
+class LogItem:
+    timestamp: datetime
+    tag: LogTag
+    content: str = ""
+
+    @classmethod
+    def from_log(cls, log_tag, log_line):
+        try:
+            time_str = re.match("^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}", log_line).group(0)
+            timestamp = datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S,%f")
+            # time_str = re.match("^\d{2}:\d{2}:\d{2}", log_line).group(0)
+            # timestamp = datetime.strptime(time_str, "%H:%M:%S")
+            res = re.match(".*?]:\s*(.*)", log_line)
+            content = ""
+            if res:
+                content = res.group(1)
+        except Exception as e:
+            print(e)
+            print(log_line)
+        return cls(timestamp=timestamp,
+                   tag=log_tag,
+                   content=content)
+@dataclass
+class WebItem:
+    timestamp: datetime
+    src_text: str
+    dst_text: str
+
+if __name__ == '__main__':
+    a = LogItem.from_log(LogTag.translate_finish, "2025-05-08 16:17:28,468 - INFO - [  📝 transcribe output      ]: Today is Friday.")
+    print(a)

main.py

@@ -0,0 +1,55 @@
+from report import LogReport
+from audio import play_audio
+import time
+import os
+import subprocess
+from playwright.sync_api import sync_playwright
+from utils import cmd
+from environment import APP_PATH, DEBUG_PORT
+from page_runner import PageRunner
+from audio import play_audio_until_end
+from tests.test_data import test_audios
+
+if __name__ == '__main__':
+    # report = Report()
+    # report.from_logfile("/Users/jeqin/work/code/TestTranslator/logs/0508_zhen.log")
+    # report.to_csv("/Users/jeqin/work/code/TestTranslator/reports/test.csv")
+    # electron_app_path = '/Applications/YoYo Translator.app'
+
+    # electron_app_path = "/Applications/YoYo\ Translator.app/Contents/MacOS/YoYo\ Translator"
+    # pros = subprocess.Popen(
+    #     [APP_PATH, f"--remote-debugging-port={DEBUG_PORT}"],
+    #     stdout=subprocess.PIPE,
+    #     stderr=subprocess.PIPE,
+    #     text=True
+    # )
+    # time.sleep(15)  # 等待页面加载
+    # print(f"{APP_PATH} started")
+    # # cmd(f"{electron_app_path} --remote-debugging-port=9222")
+    # # print("cmd finished")
+    # with sync_playwright() as p:
+    #     # 连接到已开启的 Electron 调试端口
+    #     browser = p.chromium.connect_over_cdp("http://localhost:9222")
+    #     print(browser.is_connected())
+    #     context = browser.contexts[0]
+    #     page = context.pages[0]
+    #     print(f"page title: {page.title()}")
+    #     # page.screenshot(path="electron2.png")
+    #     page.get_by_role("switch").click()
+    #     print("clicked switch")
+    #     print(f"page title: {page.title()}")
+    #     time.sleep(2)
+    #     page.screenshot(path="electron1.png")
+    #
+    # pros.terminate()
+    # stdout, stderr = pros.communicate()
+    # print("\n最终 STDOUT:")
+    # print(stdout)
+    # print("\n最终 STDERR:")
+    # print(stderr)
+    # p = PageRunner("electron").start()
+    # p.start_en2zh()
+    audios = test_audios.get("zh")
+    for a in audios:
+        play_audio_until_end(a)
+        time.sleep(5)

page_runner.py

@@ -0,0 +1,48 @@
+import time
+from typing import Literal
+
+from playwright.sync_api import Playwright, sync_playwright
+
+from pages import TranslatorPage
+from environment import DEBUG_PORT, RunType
+
+class PageRunner:
+    def __init__(self, run_type: RunType):
+        self.run_type = run_type
+
+    def start(self, p: Playwright)-> TranslatorPage:
+        if self.run_type == RunType.electron:
+            return TranslatorPage(self._start_electron(p))
+        elif self.run_type == RunType.code:
+            return TranslatorPage(self._start_web(p))
+        else:
+            raise TypeError(f"invalid run_type: {self.run_type}")
+
+    def _start_web(self, p: Playwright):
+        url = "http://127.0.0.1:9191/app"
+        browser = p.chromium.launch(headless=False)
+        context = browser.new_context(permissions=['microphone'])
+        page = context.new_page()
+        for i in range(20):
+            try:
+                page.goto(url)
+                break
+            except Exception as e:
+                print(f"Exception happened: {e}. retry {i+1}...")
+                time.sleep(3)
+        return page
+
+    def _start_electron(self, p: Playwright):
+        # 连接到已开启的 Electron 调试端口
+        browser = p.chromium.connect_over_cdp(f"http://localhost:{DEBUG_PORT}")
+        assert browser.is_connected()
+        print("connected to electron debugging port")
+        context = browser.contexts[0]
+        page = context.pages[0]
+        return page
+
+if __name__ == '__main__':
+    with sync_playwright() as p:
+        page = PageRunner(RunType.code).start(p)
+        page.start_zh2en()
+        page.set_off()

pages.py

@@ -0,0 +1,82 @@
+from datetime import datetime
+from time import sleep
+from typing import List, Literal
+from playwright.sync_api import Page
+
+from log_parser import WebItem
+
+
+class TranslatorPage:
+    def __init__(self, page:Page):
+        self.page = page
+
+    @property
+    def _switch(self):
+        return self.page.get_by_role("switch")
+
+    @property
+    def _trans_selector(self):
+        return self.page.locator(".ant-select-selector")
+
+    def start(self, translation_lang):
+        if translation_lang == "zh2en":
+            self.start_zh2en()
+        elif translation_lang == "en2zh":
+            self.start_en2zh()
+        else:
+            raise TypeError(f"invalid translation_lang: {translation_lang}")
+
+    def start_zh2en(self):
+        self._trans_selector.click()
+        self.page.locator(".ant-select-item").get_by_text("Chinese -> English").click()
+        self.set_on()
+        print("page started zh2en translation")
+
+    def start_en2zh(self):
+        self._trans_selector.click()
+        self.page.locator(".ant-select-item").get_by_text("English -> Chinese").click()
+        self.set_on()
+        print("page started en2zh translation")
+
+    def _is_on(self):
+        state = self._switch.get_attribute("aria-checked")
+        return True if state=="true" else False
+
+    def set_on(self):
+        if not self._is_on():
+            self._switch.click()
+            print("click button to set translation on")
+
+    def set_off(self):
+        if self._is_on():
+            self._switch.click()
+            print("click button to set translation off")
+
+
+    def get_current_node_text(self, duration=0, interval=0.1)-> List[WebItem]:
+        """在一定时间内持续读取页面最新的 node的内容"""
+        print(f"capture page latest content for duration: {duration}s")
+        translate_items = []
+        last_src, last_dst = None, None
+        for i in range(int(duration // interval) + 1):
+            current_node = self.page.locator(".trans-list").locator(".current_node")
+            src_lang = current_node.locator(".trans-src-lang").inner_text()
+            dst_lang = current_node.locator(".trans-dst-lang").inner_text()
+            tsp = datetime.now()
+
+            sleep(interval)
+            if src_lang == last_src and dst_lang == last_dst:
+                continue
+            # print("src lang:", src_lang)
+            # print("dst lang:", dst_lang)
+            translate_items.append(WebItem(tsp, src_lang, dst_lang))
+            last_src, last_dst = src_lang, dst_lang
+        print(f"capture page latest content finished")
+        return translate_items
+
+    def get_translated_texts(self):
+        src_all = self.page.locator(".trans-list").locator(".trans-src-lang").all_inner_texts()
+        dst_all = self.page.locator(".trans-list").locator(".trans-dst-lang").all_inner_texts()
+        return "".join(src_all), "".join(dst_all)
+
+

report.py

@@ -0,0 +1,207 @@
+from typing import List
+from dataclasses import dataclass, astuple
+
+from tabulate import tabulate
+from log_parser import LogTag, LogItem, WebItem
+from utils import save_csv, run_textdistance, highlight_diff
+
+@dataclass
+class Row:
+    audio_end_tsp:str = ""
+    audio_length:str =""
+    tsb_end_tsp:str =""
+    tsb_opt:str =""
+    tsb_cost:str =""
+    tsl_ipt:str =""
+    tsl_end_tsp:str =""
+    tsl_opt:str =""
+    tsl_cost:str =""
+    web_tsp:str =""
+    web_src:str =""
+    web_dst:str =""
+    def __repr__(self):
+        return f"Row(audio_length={self.audio_length}, tsb_opt={self.tsb_opt})"
+
+
+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 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的结果合并成 csv行的形式
+        返回 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}
+
+        row_0 = [self.translation_type, self.audio, self.audio_length]
+        rows = []
+        current_row = Row()
+        for i in self.log_items:
+            if i.tag in [LogTag.load_start, LogTag.load_end]:
+                row_0 += [i.tag.name, i.timestamp]
+            elif i.tag == LogTag.audio_end:
+                # 每次到 audio_end就是新的一行
+                rows.append(current_row)
+                current_row = Row()
+                current_row.audio_end_tsp = i.timestamp
+                current_row.audio_length = i.content.replace(" s", "")
+            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 = i.content.replace(" s", "")
+            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
+                    # 删除 dict已匹配过的内容，避免多次匹配
+                    web_items_dict.pop(current_row.tsb_opt+current_row.tsl_opt)
+            elif i.tag in [LogTag.translate_cost, LogTag.translate_large_cost]:
+                current_row.tsl_cost = i.content.replace(" s", "")
+        # print("rows value in DelayItem:",rows)
+
+        return row_0, rows # [astuple(i) for i in rows]
+
+class DelayReport:
+    """存储delay 报告中所有 case的结果"""
+    start_line = 0
+    items: List[DelayItem] = []
+    # summary_items = {
+    #     "translation_type": "",
+    #     "audio length": "",
+    #     "load_model": "",
+    #     "total_transcribe": "",
+    #     "average_transcribe": "",
+    #     "total_translate": "",
+    #     "average_translate": "",
+    #     "asr accuracy": "",
+    #     "llm translation score": "",
+    #     "delay": "",
+    # }
+    def print_summary(self, data):
+        header = ["audio", "load", "total audio len","total tsb","total tsl"]
+        print(tabulate(data, header))
+
+    def to_csv(self, csv_path):
+        all_rows = []
+        summaries = []
+        for i in self.items:
+            row_0, rows = i.to_rows()
+            all_rows.append(row_0)
+            all_rows += [astuple(i) for i in rows]
+            all_rows += [] # 每个 case后加一个空行
+
+            audios = [float(r.audio_length) for r in rows if r.audio_length]
+            transcribes = [float(r.tsb_cost) for r in rows if r.tsb_cost]
+            translates = [float(r.tsl_cost) for r in rows if r.tsl_cost]
+            if len(row_0) >=7:
+                summaries.append([row_0[1], row_0[6]-row_0[4], sum(audios), sum(transcribes), sum(translates)])
+            else:
+                summaries.append([row_0[1], 0, sum(audios), sum(transcribes), sum(translates)])
+        save_csv(csv_path, [], all_rows)
+        self.print_summary(summaries)
+
+@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):
+        self.asr_accuracy = run_textdistance(self.audio_text, self.src_text)
+        self.text_compare = highlight_diff(self.audio_text, self.src_text)
+    def to_list(self):
+        return [self.translation_type, self.audio, self.audio_length, self.src_text,
+                # self.dst_text,
+                self.asr_accuracy, 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):
+        save_csv(csv_path, [], [i.to_list() for i in self.items])
+        self.print_summary()
+
+

requirement.txt

@@ -0,0 +1,6 @@
+pytest
+textdistance
+pytest-playwright
+pygame
+librosa
+tabulate

scripts/audio_utils.py

@@ -0,0 +1,30 @@
+from pathlib import Path
+import subprocess
+from subprocess import CompletedProcess
+
+
+def cmd(command: str, check=True, capture_output=False) -> CompletedProcess:
+    print(command)
+    if capture_output:
+        ret = subprocess.run(command, shell=True, check=check, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                             universal_newlines=True)
+    else:
+        ret = subprocess.run(command, shell=True, check=check)
+    print(ret.stdout)
+    return ret
+
+
+current = Path("/Users/moyoyo/code/tests/audios")
+audios_5s = current/"5s"
+audios_10s = current/"10s"
+for f in current.glob("*.wav"):
+    file_name = f.name
+    print(file_name)
+    for i in [0, 5, 10, 15]:
+        new_name = f"{f.name.split('.')[0]}-{i}.wav"
+        command=f"ffmpeg -i {f} -ss 00:00:{str(i).zfill(2)} -t 00:00:05 {audios_5s/new_name}"
+        cmd(command)
+    for i in [0, 10, 20, 30]:
+        new_name = f"{f.name.split('.')[0]}-{i}.wav"
+        command = f"ffmpeg -i {f} -ss 00:00:{str(i).zfill(2)} -t 00:00:10 {audios_10s/new_name}"
+        cmd(command)

scripts/compare_text.py

@@ -0,0 +1,64 @@
+import csv
+import re
+import textdistance
+import difflib
+
+
+def replace_symbol(text):
+    symbol_pattern = "[,.，。!?\n]"
+    to = ""
+    return re.sub(symbol_pattern, to, text)
+
+
+def run_textdistance(text1, text2):
+    # text1 = replace_symbol(text1)
+    # text2 = replace_symbol(text2)
+    d = textdistance.levenshtein.distance(text1, text2)
+    nd = d / len(text1)
+    # print("Levenshtein distance of texts:", d, "normalized distance is:", nd)
+    return d, nd
+
+def highlight_diff(a, b):
+    matcher = difflib.SequenceMatcher(None, a, b)
+    output = []
+    for tag, a_start, a_end, b_start, b_end in matcher.get_opcodes():
+        if tag == 'equal':
+            output.append(a[a_start:a_end])
+        elif tag == 'delete':
+            output.append(f"[-{a[a_start:a_end]}-]")
+        elif tag == 'insert':
+            output.append(f"{{+{b[b_start:b_end]}+}}")
+        elif tag == 'replace':
+            output.append(f"[-{a[a_start:a_end]}-]{{+{b[b_start:b_end]}+}}")
+    return ''.join(output)
+
+def read_csv(file_path):
+    res ={}
+    with open(file_path, 'r', encoding='utf-8') as csvfile:
+        reader = csv.reader(csvfile)
+        for row in reader:
+            res[row[0]] = row[-1]
+    return res
+
+def save_csv(file_path, rows):
+    with open(file_path, "w", encoding="utf-8") as f:
+        writer = csv.writer(f)
+        writer.writerows(rows)
+        print(f"write csv to {file_path}")
+
+def main():
+    funasr_text = read_csv("run_funasr.csv")
+    quant_text = read_csv("run_quant.csv")
+    print(funasr_text)
+    print(quant_text)
+    rows = [["file_name", "diff", "distance", "normalized_d"]]
+    for key, v in funasr_text.items():
+        d, normalized_d = run_textdistance(v, quant_text[key])
+        opt = highlight_diff(v, quant_text[key])
+        print(key,opt, d, normalized_d)
+        rows.append([key,opt, d, normalized_d])
+    save_csv("compare_asr.csv", rows)
+
+
+if __name__ == '__main__':
+    main()

scripts/demo_playwright.py

@@ -0,0 +1,24 @@
+import re
+from playwright.sync_api import Playwright, sync_playwright, expect
+
+
+def run(playwright: Playwright) -> None:
+    browser = playwright.chromium.launch(headless=False)
+    context = browser.new_context()
+    page = context.new_page()
+    page.goto("http://127.0.0.1:9191/app/")
+    page.get_by_text("English -> Chinese").click()
+    page.get_by_text("Chinese -> English").click()
+    page.locator("#app").get_by_title("Chinese -> English").click()
+    page.get_by_text("Chinese -> English").nth(1).click()
+    page.get_by_role("switch", name="ON OFF").click()
+    page.locator("#app").get_by_text("Chinese -> English").click()
+    page.get_by_text("Chinese -> English").nth(1).click()
+
+    # ---------------------
+    context.close()
+    browser.close()
+
+
+with sync_playwright() as playwright:
+    run(playwright)

scripts/funasr_utils.py

@@ -0,0 +1,101 @@
+from pathlib import Path
+import csv
+import time
+from funasr import AutoModel
+from funasr_onnx import SeacoParaformer, CT_Transformer, Fsmn_vad
+
+
+model_dir = Path("/Users/jeqin/work/code/Translator/moyoyo_asr_models")
+
+def save_csv(file_path, rows):
+    with open(file_path, "w", encoding="utf-8") as f:
+        writer = csv.writer(f)
+        writer.writerows(rows)
+        print(f"write csv to {file_path}")
+
+def export_onnx():
+    asr_model_path = model_dir / 'speech_seaco_paraformer_large_asr_nat-zh-cn-16k-common-vocab8404-pytorch'
+    vad_model_path = model_dir / '/speech_fsmn_vad_zh-cn-16k-common-pytorch'
+    punc_model_path = model_dir / '/punc_ct-transformer_cn-en-common-vocab471067-large'
+
+    model = AutoModel(model=asr_model_path)
+    output = model.export(type="onnx", quantize=True, disable_update=True)
+    print(output)
+
+    model = AutoModel(model=vad_model_path)
+    output = model.export(type="onnx", quantize=True, disable_update=True)
+    print(output)
+
+    model = AutoModel(model=punc_model_path)
+    output = model.export(type="onnx", quantize=True, disable_update=True)
+    print(output)
+
+def run_funasr():
+    asr_model_path = model_dir / 'speech_seaco_paraformer_large_asr_nat-zh-cn-16k-common-vocab8404-pytorch'
+    vad_model_path = model_dir / 'speech_fsmn_vad_zh-cn-16k-common-pytorch'
+    punc_model_path = model_dir / 'punc_ct-transformer_cn-en-common-vocab471067-large'
+    t0 = time.time()
+    model = AutoModel(
+        model=asr_model_path.as_posix(),
+        vad_model=vad_model_path.as_posix(),
+        punc_model=punc_model_path.as_posix(),
+        log_level="ERROR",
+        disable_update=True
+    )
+    t1 = time.time()
+    print("load model: ", t1 - t0)
+    audios = Path("/Users/jeqin/work/code/TestTranslator/tests/test_data/test_audios")
+    rows = [["file_name", "inference_time", "inference_result"]]
+    for audio in sorted(audios.glob("Chinese-mayun-part2.mp3")):
+        print(audio)
+        t1 = time.time()
+        try:
+            result = model.generate(input=str(audio), disable_pbar=True,
+                                    hotword="")
+        except Exception as e:
+            print(audio)
+            print(e)
+        t2 = time.time()
+        t = t2 - t1
+        print("inference time:", t)
+        text = result[0]["text"]
+        print("inference result", text)
+        rows.append([f"{audio.parent.name}/{audio.name}", t, text])
+    save_csv(f"run_funasr.csv", rows)
+
+def run_onnx(quant=True):
+    asr_model_path = model_dir / 'speech_seaco_paraformer_large_asr_nat-zh-cn-16k-common-vocab8404-pytorch'
+    vad_model_path = model_dir / 'speech_fsmn_vad_zh-cn-16k-common-pytorch'
+    punc_model_path = model_dir / 'punc_ct-transformer_cn-en-common-vocab471067-large'
+    t0 = time.time()
+    vad_model = Fsmn_vad(vad_model_path, quantize=quant)
+    asr_model = SeacoParaformer(asr_model_path, quantize=quant)
+    punc_model = CT_Transformer(punc_model_path, quantize=quant)
+    t1 = time.time()
+    print("load model: ", t1 - t0)
+    audios = Path("/Users/moyoyo/code/tests/audios")
+    rows = [["file_name", "inference_time", "inference_result"]]
+    for audio in sorted(audios.glob("*s/*.wav")):
+        t1 = time.time()
+        vad_res = vad_model(str(audio))
+        t2 = time.time()
+        # print("vad time:", t2 - t1)
+        asr_res = asr_model(str(audio), hotwords="")
+        asr_text = asr_res[0]["preds"]
+        t3 = time.time()
+        # print("asr time:", t3 - t2)
+        # print("asr text:", asr_text)
+        result = punc_model(asr_text)
+        text = result[0]
+        t4 = time.time()
+        # print("punc time:", t4 - t3)
+        # print("punc text:", text)
+        print(text)
+        t = t4 - t1
+        print("inference time:", t)
+        rows.append([f"{audio.parent.name}/{audio.name}", t, text])
+    file_name = "run_quant.csv" if quant else "run_onnx.csv"
+    save_csv(file_name, rows)
+
+if __name__ == '__main__':
+    run_funasr()

scripts/run_llm.py

@@ -0,0 +1,42 @@
+from logging import getLogger
+from pathlib import Path
+from llama_cpp import Llama
+from functools import lru_cache
+
+logger = getLogger(__name__)
+
+class QwenTranslator:
+    def __init__(self, model_path, system_prompt_en="", system_prompt_zh="") -> None:
+        self.llm = Llama(
+        model_path=model_path,
+        chat_format="chatml",
+        verbose=False)
+        self.sys_prompt_en = system_prompt_en
+        self.sys_prompt_zh = system_prompt_zh
+
+    def to_message(self, prompt, src_lang, dst_lang):
+        """构造提示词"""
+        return [
+            {"role": "system", "content": self.sys_prompt_en if src_lang == "en" else self.sys_prompt_zh},
+            {"role": "user", "content": prompt},
+        ]
+
+    @lru_cache(maxsize=10)
+    def translate(self, prompt, src_lang, dst_lang) -> str:
+        message = self.to_message(prompt, src_lang, dst_lang)
+        output = self.llm.create_chat_completion(messages=message, temperature=0)
+        return output['choices'][0]['message']['content']
+
+    def __call__(self, prompt,*args, **kwargs):
+        return self.llm(
+            prompt,
+            *args,
+            **kwargs
+        )
+
+if __name__ == '__main__':
+    model_dir = Path("/Users/jeqin/work/code/Translator/moyoyo_asr_models")
+    qwen2 = (model_dir / "qwen2.5-1.5b-instruct-q5_0.gguf").as_posix()
+    qwen3 = (model_dir / "Qwen_Qwen3-0.6B-Q4_K_M.gguf").as_posix()
+
+    # translator = QwenTranslator(qwen2, )

scripts/run_whisper.py

@@ -0,0 +1,26 @@
+from pywhispercpp.model import Model
+from pathlib import Path
+import time
+
+from silero_vad.utils_vad import languages
+
+models_dir = Path("/Users/jeqin/work/code/Translator/moyoyo_asr_models")
+whisper_model = 'large-v3-turbo-q5_0'
+model = Model(
+            model=whisper_model,
+            models_dir=models_dir,
+            print_realtime=False,
+            print_progress=False,
+            print_timestamps=False,
+            translate=False,
+            # beam_search=1,
+            temperature=0.,
+            no_context=True
+        )
+audios = Path("/Users/jeqin/work/code/TestTranslator/tests/test_data/test_audios")
+for audio in sorted(audios.glob("English*")):
+    print(audio)
+    t1 = time.time()
+    output = model.transcribe(str(audio), language="en")
+    print("inference time:", time.time()-t1)
+    print(" ".join([a.text for a in output]))

scripts/test_pages.py

@@ -0,0 +1,35 @@
+import re
+import time
+from time import sleep
+from playwright.sync_api import Page, expect
+from audio import play_audio, get_length
+
+
+def test_has_title(page: Page, context):
+    context.grant_permissions(permissions=['microphone'])
+    page.goto("http://127.0.0.1:9191/")
+    page.get_by_text("English -> Chinese").click()
+    page.get_by_text("Chinese -> English").click()
+    page.get_by_role("switch").click()
+
+    audio = "/Users/jeqin/work/code/TestTranslator/test_audios/Chinese-economics-part1.wav"
+    play_audio(audio)
+
+    audio_length = get_length(audio)
+    interval = 0.1
+    last_src, last_dst = None, None
+    for i in range(int(audio_length//interval)+1):
+        print(i)
+        current_node = page.locator(".trans-list").locator(".current_node")
+        src_lang = current_node.locator(".trans-src-lang").inner_text()
+        dst_lang = current_node.locator(".trans-dst-lang").inner_text()
+        time.sleep(interval)
+        if src_lang == last_src and dst_lang == last_dst:
+            continue
+        print("src lang:", src_lang)
+        print("dst lang:", dst_lang)
+        last_src, last_dst = src_lang, dst_lang
+
+    all_texts = page.locator(".trans-list").locator(".trans-src-lang").all_inner_texts()
+    print("all_texts:", all_texts)
+    print("run_finished")

tests/conftest.py

@@ -0,0 +1,56 @@
+import subprocess
+import time
+
+from numpy.core.numeric import True_
+from pytest import fixture
+from playwright.sync_api import Page, expect
+from playwright.sync_api import sync_playwright
+
+from environment import *
+from app_runner import AppRunner
+from page_runner import PageRunner
+from report import AccuracyReport, DelayReport, LogReport
+from utils import get_time_str
+
+@fixture(scope="session")
+def log_file():
+    if RUN_TYPE == RunType.electron:
+        log_file = APP_LOG
+    elif RUN_TYPE == RunType.code:
+        log_file = CODE_LOG
+    else:
+        raise TypeError(f"invalid run_type: {RUN_TYPE}")
+    return log_file
+
+@fixture(scope="session")
+def app():
+    app = AppRunner(RUN_TYPE)
+    app.start()
+    yield app
+    app.stop()
+
+@fixture(scope="module")
+def page():
+    with sync_playwright() as p:
+        page = PageRunner(RUN_TYPE).start(p)
+        yield page
+
+@fixture(scope="module")
+def accuracy_report(request):
+    report = AccuracyReport()
+    yield report
+    report.to_csv(REPORTS_DIR/f"accuracy_report_{get_time_str()}.csv")
+
+@fixture(scope="module")
+def delay_report(request):
+    report = DelayReport()
+    yield report
+    report.to_csv(REPORTS_DIR/f"delay_report_{get_time_str()}.csv")
+
+@fixture(scope="session")
+def log_report(request):
+    report = LogReport()
+    yield report
+    report.to_csv(REPORTS_DIR/f"log_report_{get_time_str()}.csv")
+
+

tests/test_accuracy.py

@@ -0,0 +1,37 @@
+import time
+from pathlib import Path
+import pytest
+
+from audio import play_audio_until_end, get_length
+from pages import TranslatorPage
+from tests.test_data import test_audios, audio_texts
+from report import AccuracyReport, AccuracyItem
+
+@pytest.mark.parametrize("audio", test_audios.get("zh"))
+def test_accuracy_zh2en(app, page: TranslatorPage, accuracy_report: AccuracyReport, audio:Path):
+    page.start_zh2en()
+    play_audio_until_end(audio)
+    time.sleep(4)
+    page.set_off()
+    zh, en = page.get_translated_texts()
+    accuracy_report.items.append(
+        AccuracyItem(translation_type="zh2en", audio=audio.name,
+                     audio_length=get_length(audio), audio_text=audio_texts.get(audio.stem),
+                     src_text=zh, dst_text=en)
+    )
+
+
+@pytest.mark.parametrize("audio", test_audios.get("en"))
+def test_accuracy_en2zh(app, page: TranslatorPage, accuracy_report: AccuracyReport, audio):
+    page.start_en2zh()
+    play_audio_until_end(audio)
+    time.sleep(4)
+    page.set_off()
+    en, zh = page.get_translated_texts()
+    accuracy_report.items.append(
+        AccuracyItem(translation_type="en2zh", audio=audio.name,
+                     audio_length=get_length(audio), audio_text=audio_texts.get(audio.stem),
+                     src_text=en, dst_text=zh)
+    )
+
+

tests/test_data/.gitattributes

@@ -0,0 +1 @@
+*.wav filter=lfs diff=lfs merge=lfs -text

tests/test_data/__init__.py

@@ -0,0 +1,22 @@
+from environment import TEST_AUDIOS_DIR
+
+test_audios = {
+    "zh": [
+        TEST_AUDIOS_DIR/"Chinese-calculus-part1.mp3",
+        TEST_AUDIOS_DIR/"Chinese-economics-part1.wav",
+        # TEST_AUDIOS_DIR / "Chinese-mayun-part2.mp3"
+
+    ],
+    "en": [
+        TEST_AUDIOS_DIR/"English-chaos-part2.wav",
+        TEST_AUDIOS_DIR/"English-internet-part20.mp3",
+    ]
+}
+
+audio_texts = {
+    "Chinese-calculus-part1": "后来我自己总结啊，微积分这么难，入门主要有几个原因。首先呢大部分的教材为了追求严谨，从一开始就使用了现代数学的这个所谓极限的概念。在它的基础之上向你介绍微积分。呃，问题是它是一个非常抽象的概念。对于大部分在接触微积分之前啊，主要的学习经验就是刷题啊，甚至是连题也不刷的。同学们来说呢，这种抽象语言会很陌生。而且如果你去了解微积分的历史的时候，你会发现极限这个概念啊是微积分创立之后大概一两百年才出现的这么个东西。你等于说我们现在公认的这些微积分的创始人，这些大佬们、牛顿啊、莱布尼茨、欧拉啊，连他们都不知道极限是什么。但是人家就是凭着直觉创建了微积分，当然作为教科书嘛，追求严谨无可厚非啊。虽然对于我来讲，过早的追求这种严谨，导致学习的人入门困难甚至入不了门。",
+    "Chinese-economics-part1": "经济就像一部简单的机器那样运行，但很多人不懂得这一点，或是对经济的运行方式持有不同观点，于是导致很多不必要的经济损失。我深感有责任与大家分享我的简单，但是实用的经济分析模式。这个模式虽然不符合常规传统经济学，但是已经帮助我预测和躲避了全球金融危机。三十多年来对我一直很有用。我们开始吧，经济虽然可能看起来复杂，但是其实是以简单和机械的方式运行。经济由几个简单的零部件和无数次重复的简单交易组成。这些交易首先是由人的天性所驱动，因而形成三股主要的经济动力，一、生产率的提高。二短期债务周期。三、长期债务周期。",
+    "Chinese-mayun-part2":"第二个公司我们成立了中国黄页。在中国黄业的创业经验中有很多的经验，也是可以要跟他这儿跟大家进行分享。九五年做互联网是最艰难的时候，就中国那时候还没联通互联网，我就到美国去了一趟。回来以后我们要做互联网，我请了二十四个朋友在我们家开会，说了两个小时，没人听懂。我在说什么。最后二十三个人反对一个人同意这一个人就说马云你这样做，你就试试看，不行的话，赶紧逃回来，还来得及。那我自己想了一个晚上，第二天早上我决定还是做下去。中国人很多创业是晚上想想千条路，早上起来走原路，晚上想想是热血沸腾，真好，第二天早上骑个自行车又上班去了，对吧？这是我们很多创业者所碰上的问题。说那一天我觉得因为我看见过互联网，我觉得互联网会将来会好，但这些人没看见过，但是没看见过的机会，就是你怎么把它变成现实。",
+    "English-chaos-part2": "This sequence of events is an example of what is known as the Butterfly Effect, a manifestation of Chaos Theory. For many centuries, the world was explained through the laws of Isaac Newton in classical physics. According to these laws, if the current state of an object is known, its future behavior can be predicted with relative ease. Chaos Theory questions this deterministic vision. Not everything is predictable anymore, nor does it work like a quirk. Since the 1800s, mathematicians have raised the idea that not all phenomena could be predicted by Newtonian laws. But a meteorologist named Edward Lawrence made Chaos Theory a visible phenomenon. It all started in 1961, when he was working on a mathematical model to forecast the weather. Lawrence entered data such as temperature, humidity, pressure, and wind direction into his computer. His computer would draw a graph modeling what the weather would be like. Not always accurate, but very close to reality.",
+    "English-internet-part20": "Many, many years ago in the early 1970s, my partner Bob Kahn and I began working on the design of what we now call the Internet. Bob and I had the responsibility and the opportunity to design the Internet's protocols and its architecture. So we persisted in participating in the Internet's growth and evolution for all of this time up to and including the present. The way information gets transferred from one computer to another is pretty interesting. It need not follow a fixed path. In fact, your path may change in the midst of a computer-to-computer conversation. Information on the Internet goes from one computer to another in what we call a packet of information. And a packet travels from one place to another on the Internet a lot like how you might get from one place to another in a car. Depending on traffic congestion or road conditions, you might choose or be forced to take a different route to get to the same place each time you travel."
+}

tests/test_data/test_audios/Chinese-calculus-part1.mp3

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:43b9658edd50a6a69bb509960d5ee90e33078a3f84e116eb586fd2d903f6009a
+size 1041064

tests/test_data/test_audios/Chinese-economics-part1.mp3

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:60caa6584b37308ce048cc2bbbac1b48a37518a728683057fcfbe49aea0d0b44
+size 977023

tests/test_data/test_audios/Chinese-mayun-part2.mp3

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:47ac644833dac81de0efa7d1784924a293f4b07a8008c821c37c1255c96ca158
+size 992999

tests/test_data/test_audios/English-chaos-part2.wav

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:9548f6d69ca750eb965ee1e8b45fdf7f5c9060639c686c003f8f24ea78a23a09
+size 4032078

tests/test_data/test_audios/English-internet-part20.mp3

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:c481a27b50e5ce2487a9556becd22b5ab9bf33e136b59f0973f5e6cd5c1ae0c6
+size 992906

tests/test_data/test_audios/English-legalsystem-part1.mp3

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:cbca4d430e4451b4c2aebd70a21b6e19ba9670cf609d4e88f795e9862cb5268b
+size 817038

tests/test_data/test_audios/English-literarytheory-part1.mp3

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:20af50212f5e60170c27973a29870cb6615d33e7de9465f4bd5b6239b5c23cb8
+size 880986

tests/test_delay.py

@@ -0,0 +1,34 @@
+from pathlib import Path
+import pytest
+
+from audio import get_length, play_audio
+from pages import TranslatorPage
+from report import DelayReport, DelayItem, LogReport
+from tests.test_data import test_audios
+
+@pytest.mark.parametrize("audio", test_audios.get("zh"))
+def test_delay_zh2en(log_file, app, delay_report: DelayReport,page: TranslatorPage, audio:Path):
+    page.start_zh2en()
+    audio_length = get_length(audio)
+    play_audio(audio)
+    web_records = page.get_current_node_text(duration=audio_length)
+    log_records, delay_report.start_line = LogReport().from_logfile(log_file, delay_report.start_line)
+    delay_report.items.append(
+        DelayItem(translation_type="zh2en", audio=audio.name, audio_length=audio_length,
+                  web_items=web_records, log_items=log_records,)
+    )
+    page.set_off()
+
+
+@pytest.mark.parametrize("audio", test_audios.get("en"))
+def test_delay_en2zh(log_file,app, delay_report: DelayReport,page: TranslatorPage, audio:Path):
+    page.start_en2zh()
+    audio_length = get_length(audio)
+    play_audio(audio)
+    web_records = page.get_current_node_text(duration=audio_length)
+    log_records, delay_report.start_line = LogReport().from_logfile(log_file, delay_report.start_line)
+    delay_report.items.append(
+        DelayItem(translation_type="zh2en", audio=audio.name, audio_length=audio_length,
+                  web_items=web_records, log_items=log_records, )
+    )
+    page.set_off()

tests/test_logfile.py

@@ -0,0 +1,18 @@
+from pathlib import Path
+import pytest
+
+from report import LogReport, DelayItem
+from environment import REPORTS_DIR
+from utils import get_time_str
+
+test_files = [
+    # Path("/Users/jeqin/work/code/Translator/translator.log"),
+    Path("/Users/jeqin/Downloads/translator.log"),
+    # Path("/Users/jeqin/Downloads/translator-web-0527.log"),
+    # Path("/tmp/translator.log"),
+]
+
+@pytest.mark.parametrize("test_file", test_files)
+def test_zh2en_logfile(test_file: Path, log_report):
+    log_report.from_logfile(test_file)
+    # log_report.to_csv(REPORTS_DIR/f"logfile_{get_time_str()}.csv")

utils.py

@@ -0,0 +1,65 @@
+import csv
+from datetime import datetime
+import subprocess
+from subprocess import CompletedProcess
+from typing import Literal
+import re
+import difflib
+
+import textdistance
+
+
+def get_time_str(level:Literal["d","s","ms"]="d"):
+    time = datetime.now()
+    if level == "d":
+        return time.strftime("%Y-%m-%d")
+    if level == "s":
+        return time.strftime("%H%M%S")
+    if level == "ms":
+        return time.strftime("%H%M%S.%f")
+
+
+def save_csv(file_path, header, rows):
+    with open(file_path, "w", encoding="utf-8") as f:
+        writer = csv.writer(f)
+        writer.writerow(header)
+        writer.writerows(rows)
+        print(f"write csv to {file_path}")
+
+def cmd(command: str, check=True, capture_output=False) -> CompletedProcess:
+    print(command)
+    if capture_output:
+        ret = subprocess.run(command, shell=True, check=check, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                             universal_newlines=True)
+    else:
+        ret = subprocess.run(command, shell=True, check=check)
+    print(ret.stdout)
+    return ret
+
+def replace_symbol(text):
+    symbol_pattern = "[,.，。!?\n]"
+    to = ""
+    return re.sub(symbol_pattern, to, text)
+
+
+def run_textdistance(text1, text2):
+    text1 = replace_symbol(text1)
+    text2 = replace_symbol(text2)
+    d = textdistance.levenshtein.distance(text1, text2)
+    nd = d / len(text1)
+    # print("Levenshtein distance of texts:", d, "normalized distance is:", nd)
+    return d, nd
+
+def highlight_diff(a, b):
+    matcher = difflib.SequenceMatcher(None, a, b)
+    output = []
+    for tag, a_start, a_end, b_start, b_end in matcher.get_opcodes():
+        if tag == 'equal':
+            output.append(a[a_start:a_end])
+        elif tag == 'delete':
+            output.append(f"[-{a[a_start:a_end]}-]")
+        elif tag == 'insert':
+            output.append(f"{{+{b[b_start:b_end]}+}}")
+        elif tag == 'replace':
+            output.append(f"[-{a[a_start:a_end]}-]{{+{b[b_start:b_end]}+}}")
+    return ''.join(output)