Function message

runtime/python/websocket/funasr_wss_client.py:317–471

接收服务端识别结果 + 打印实时文本 + 打印延迟 + 写验收日志(events.jsonl)

(id, writer: MeetingWriter)

async def message(id, writer: MeetingWriter):
    """接收服务端识别结果 + 打印实时文本 + 打印延迟 + 写验收日志(events.jsonl)"""
    import websockets
    global websocket, voices, offline_msg_done
    global latency_first_audio_time, latency_last_audio_time, latency_first_text_printed

    multi_mode = args.thread_num > 1  # 多路并发时，打印风格更简洁
    text_print = ""
    text_print_2pass_online = ""
    text_print_2pass_offline = ""

    if args.output_dir is not None:
        ibest_writer = open(
            os.path.join(args.output_dir, "text.{}".format(id)), "a", encoding="utf-8"
        )
    else:
        ibest_writer = None

    try:
        while True:
            meg = await websocket.recv()
            meg = json.loads(meg)

            wav_name = meg.get("wav_name", "demo")
            text = meg.get("text", "")
            mode = meg.get("mode", "")
            spk_name = meg.get("spk_name", "")
            spk_score = meg.get("spk_score", None)
            now_ts = time.time()

            # === 延迟统计：仅在首条 online/2pass-online 文本时计算并打印一次 ===
            latency_last_ms = None
            latency_first_ms = None
            if text and mode in ("online", "2pass-online"):
                if not latency_first_text_printed.get(wav_name, False):
                    t_last = latency_last_audio_time.get(wav_name, None)
                    t_first = latency_first_audio_time.get(wav_name, None)
                    latency_last_ms = (now_ts - t_last) * 1000.0 if t_last is not None else None
                    latency_first_ms = (now_ts - t_first) * 1000.0 if t_first is not None else None

                    latency_first_text_printed[wav_name] = True

                    if multi_mode:
                        parts = [f"[MEETING {id}][LATENCY] wav={wav_name}, mode={mode}"]
                        if latency_last_ms is not None:
                            parts.append(f"from_last_chunk={latency_last_ms:.1f} ms")
                        if latency_first_ms is not None:
                            parts.append(f"from_first_chunk={latency_first_ms:.1f} ms")
                        print(" ".join(parts))
                    else:
                        print(
                            f"[LATENCY] wav={wav_name}, mode={mode}, "
                            f"from_last_chunk={(latency_last_ms or 0):.1f} ms, "
                            f"from_first_chunk={(latency_first_ms or 0):.1f} ms"
                        )

            timestamp = meg.get("timestamp", "")
            offline_msg_done = meg.get("is_final", False)