128 lines
3.2 KiB
Python
128 lines
3.2 KiB
Python
"""TTS 状态监控窗口
|
|
|
|
独立小黑窗显示 TTS 引擎状态、最近合成文本和耗时。
|
|
启动参数:
|
|
tts_monitor.py --state-file data/tts_state.json
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
APP_DIR = Path(__file__).resolve().parent
|
|
if str(APP_DIR) not in sys.path:
|
|
sys.path.insert(0, str(APP_DIR))
|
|
|
|
from core.runtime_paths import DATA_DIR, ensure_runtime_dirs
|
|
|
|
|
|
def clear():
|
|
os.system("cls" if os.name == "nt" else "clear")
|
|
|
|
|
|
def default_state() -> dict:
|
|
return {
|
|
"enabled": False,
|
|
"provider": "none",
|
|
"model_loaded": False,
|
|
"model_name": "",
|
|
"last_text": "",
|
|
"last_duration_ms": 0,
|
|
"last_error": "",
|
|
"total_synthesized": 0,
|
|
"total_errors": 0,
|
|
"recent_events": [],
|
|
"updated_at": 0,
|
|
}
|
|
|
|
|
|
def load_state(state_file: Path) -> dict:
|
|
if not state_file.exists():
|
|
return default_state()
|
|
try:
|
|
data = json.loads(state_file.read_text(encoding="utf-8"))
|
|
if isinstance(data, dict):
|
|
state = default_state()
|
|
state.update(data)
|
|
return state
|
|
except Exception:
|
|
pass
|
|
return default_state()
|
|
|
|
|
|
def fmt_duration(ms: int) -> str:
|
|
if ms <= 0:
|
|
return "-"
|
|
if ms < 1000:
|
|
return f"{ms}ms"
|
|
return f"{ms/1000:.2f}s"
|
|
|
|
|
|
def render(state: dict):
|
|
clear()
|
|
print("=" * 60)
|
|
print(" TTS 状态监控")
|
|
print("=" * 60)
|
|
print()
|
|
|
|
enabled = state.get("enabled", False)
|
|
provider = state.get("provider", "none")
|
|
model_loaded = state.get("model_loaded", False)
|
|
|
|
print(f" TTS 启用: {'是' if enabled else '否'}")
|
|
print(f" 引擎: {provider}")
|
|
print(f" 模型加载: {'完成' if model_loaded else '未加载/加载中'}")
|
|
print(f" 模型名: {state.get('model_name', '') or '-'}")
|
|
print()
|
|
print(f" 累计合成: {state.get('total_synthesized', 0)} 次")
|
|
print(f" 累计失败: {state.get('total_errors', 0)} 次")
|
|
print(f" 上次合成: {state.get('last_text', '') or '-'}")
|
|
print(f" 合成耗时: {fmt_duration(state.get('last_duration_ms', 0))}")
|
|
print()
|
|
|
|
last_error = state.get("last_error", "")
|
|
if last_error:
|
|
print(f" [错误] {last_error}")
|
|
print()
|
|
|
|
print("-" * 60)
|
|
print(" 最近事件")
|
|
print("-" * 60)
|
|
events = state.get("recent_events", [])
|
|
if not events:
|
|
print(" (无)")
|
|
else:
|
|
for ev in events[-8:]:
|
|
ts = ev.get("time", "")
|
|
msg = ev.get("msg", "")
|
|
print(f" {ts} {msg}")
|
|
print()
|
|
print(" 按 Ctrl+C 关闭本窗口")
|
|
|
|
|
|
def main():
|
|
ensure_runtime_dirs()
|
|
parser = argparse.ArgumentParser(description="TTS 状态监控")
|
|
parser.add_argument("--state-file", default=str(DATA_DIR / "tts_state.json"), help="TTS 状态文件路径")
|
|
args = parser.parse_args()
|
|
|
|
state_file = Path(args.state_file)
|
|
print("等待 TTS 状态更新...")
|
|
time.sleep(0.5)
|
|
|
|
try:
|
|
while True:
|
|
state = load_state(state_file)
|
|
render(state)
|
|
time.sleep(0.5)
|
|
except KeyboardInterrupt:
|
|
print("\nTTS 监控已关闭")
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|