import math import torch import numpy as np import subprocess import os import time import cv2 import glob import resampy import queue from queue import Queue from threading import Thread, Event from io import BytesIO import soundfile as sf import asyncio from av import AudioFrame, VideoFrame import av from fractions import Fraction from ttsreal import EdgeTTS,SovitsTTS,XTTS,CosyVoiceTTS,FishTTS,TencentTTS,DoubaoTTS,IndexTTS2,AzureTTS,State from logger import logger from videoplayer import RealtimeVideoPlayer from audioplayer import RealtimeAudioPlayer from tqdm import tqdm from threading import Lock def read_imgs(img_list): frames = [] logger.info('reading images...') for img_path in tqdm(img_list): frame = cv2.imread(img_path) frames.append(frame) return frames def play_audio(quit_event,queue): import pyaudio p = pyaudio.PyAudio() stream = p.open( rate=16000, channels=1, format=8, output=True, output_device_index=1, ) stream.start_stream() # while queue.qsize() <= 0: # time.sleep(0.1) while not quit_event.is_set(): stream.write(queue.get(block=True)) stream.close() class BaseReal: def __init__(self, opt,model,avatar): #原本的播放序列 (添加大小限制防止内存泄漏) self.msg_queue = queue.Queue(maxsize=1000) # 最多1000条消息 self.interrupted_queue = queue.Queue(maxsize=500) # 最多500条中断消息 self.user_question_queue = queue.Queue(maxsize=100) # 最多100个用户问题 self.is_speaking_flag = False self.current_text = "" # 当前正在播放的文本 self.current_pos = 0 # 当前正在播放的文本位置 self.interrupt_lock = Lock() #线程安全锁 self.opt = opt self.sample_rate = 16000 self.chunk = self.sample_rate // opt.fps # 320 samples per chunk (20ms * 16000 / 1000) self.sessionid = self.opt.sessionid if opt.tts == "edgetts": self.tts = EdgeTTS(opt,self) elif opt.tts == "qwen3tts": from qwen3tts import Qwen3TTS self.tts = Qwen3TTS(opt,self) elif opt.tts == "gpt-sovits": self.tts = SovitsTTS(opt,self) elif opt.tts == "xtts": self.tts = XTTS(opt,self) elif opt.tts == "cosyvoice": self.tts = CosyVoiceTTS(opt,self) elif opt.tts == "fishtts": self.tts = FishTTS(opt,self) elif opt.tts == "tencent": self.tts = TencentTTS(opt,self) elif opt.tts == "doubao": self.tts = DoubaoTTS(opt,self) elif opt.tts == "indextts2": self.tts = IndexTTS2(opt,self) elif opt.tts == "azuretts": self.tts = AzureTTS(opt,self) elif opt.tts == "voxcpm2": from ttsreal import VoxCPM2TTS self.tts = VoxCPM2TTS(opt,self) elif opt.tts == "voxcpm2api": from voxcpm2_api_tts import VoxCPM2APITTS self.tts = VoxCPM2APITTS(opt,self) elif opt.tts == "none": # 不使用 TTS，使用音频播放模式 self.tts = None self.video_player = None self.audio_player = None # 检查是否指定视频播放目录 if hasattr(opt, 'VIDEO_PLAYBACK_DIR') and opt.VIDEO_PLAYBACK_DIR: self.video_player = RealtimeVideoPlayer(opt.VIDEO_PLAYBACK_DIR) logger.info(f"🎬 视频播放模式已启用，监听目录: {opt.VIDEO_PLAYBACK_DIR}") # 检查是否指定音频播放目录 elif hasattr(opt, 'AUDIO_PLAYBACK_DIR') and opt.AUDIO_PLAYBACK_DIR: self.audio_player = RealtimeAudioPlayer(opt.AUDIO_PLAYBACK_DIR, sample_rate=16000) logger.info(f"🎵 音频播放模式已启用，监听目录: {opt.AUDIO_PLAYBACK_DIR}") else: logger.warning("⚠️ 未指定 VIDEO_PLAYBACK_DIR 或 AUDIO_PLAYBACK_DIR，播放模式不可用") else: # 默认使用 edgetts logger.warning(f"未知的 TTS 类型: {opt.tts}，使用 edgetts 作为默认") self.tts = EdgeTTS(opt,self) self.speaking = False self.recording = False self._record_video_pipe = None self._record_audio_pipe = None self.width = self.height = 0 # RTMP 推流支持 self._rtmp_pushing = False self._rtmp_video_pipe = None self._rtmp_audio_pipe = None self._rtmp_push_url = getattr(opt, 'push_url', '') if hasattr(opt, 'push_url') and opt.transport == 'rtcpush' and getattr(opt, 'push_url', '').startswith('rtmp://') else '' if self._rtmp_push_url: logger.info(f'RTMP 推流已配置: {self._rtmp_push_url}') self.curr_state=0 self.custom_img_cycle = {} self.custom_audio_cycle = {} self.custom_audio_index = {} self.custom_index = {} self.custom_opt = {} self.__loadcustom() def put_msg_txt(self,msg,datainfo:dict={}): if self.opt.tts == "none": # 音频播放模式：激活音频播放器 if self.audio_player and not self.audio_player.is_running: logger.info("🎵 音频播放模式已激活") self.audio_player.set_callbacks( on_audio=self._on_audio_data ) self.audio_player.start() # 视频播放模式：不需要处理文本消息 elif self.video_player and not self.video_player.is_running: logger.info("🎬 视频播放模式已激活") self.video_player.set_callbacks( on_frame=self._on_video_frame, on_audio=self._on_audio_frame ) self.video_player.start() else: # TTS 模式：处理文本消息 self.tts.put_msg_txt(msg,datainfo) def _on_video_frame(self, frame): """视频帧回调""" # 这里可以将帧传递给渲染管道 # 目前先简单处理，后续可以集成到渲染流程 pass def _on_audio_frame(self, audio): """音频帧回调（用于视频播放模式）""" # 这里可以将音频数据传递给音频播放 pass def _on_audio_data(self, audio_chunk): """音频数据回调（用于音频播放模式）""" # 将音频数据帧传递给渲染管道 # audio_chunk 是 16kHz 20ms 的 PCM 数据 (int16) try: # 转换为 float32 格式，这是 ASR 需要的格式 if audio_chunk.dtype == np.int16: audio_float = audio_chunk.astype(np.float32) / 32768.0 else: audio_float = audio_chunk.astype(np.float32) # 传递给 put_audio_frame 处理 self.put_audio_frame(audio_float, {}) except Exception as e: logger.error(f"处理音频数据回调时出错: {e}") def put_audio_frame(self,audio_chunk,datainfo:dict={}): #16khz 20ms pcm self.asr.put_audio_frame(audio_chunk,datainfo) def put_audio_file(self,filebyte,datainfo:dict={}): input_stream = BytesIO(filebyte) stream = self.__create_bytes_stream(input_stream) streamlen = stream.shape[0] idx=0 while streamlen >= self.chunk: #and self.state==State.RUNNING self.put_audio_frame(stream[idx:idx+self.chunk],datainfo) streamlen -= self.chunk idx += self.chunk def __create_bytes_stream(self,byte_stream): #byte_stream=BytesIO(buffer) stream, sample_rate = sf.read(byte_stream) # [T*sample_rate,] float64 logger.info(f'[INFO]put audio stream {sample_rate}: {stream.shape}') stream = stream.astype(np.float32) if stream.ndim > 1: logger.info(f'[WARN] audio has {stream.shape[1]} channels, only use the first.') stream = stream[:, 0] if sample_rate != self.sample_rate and stream.shape[0]>0: logger.info(f'[WARN] audio sample rate is {sample_rate}, resampling into {self.sample_rate}.') stream = resampy.resample(x=stream, sr_orig=sample_rate, sr_new=self.sample_rate) return stream def stop_tts(self): """停止当前TTS播放""" with self.interrupt_lock: # 添加锁保护 if hasattr(self.tts, 'state'): self.tts.state = State.PAUSE # 设置打断标志，强制中断当前播放 if hasattr(self.tts, 'set_interrupt_flag'): self.tts.set_interrupt_flag() # 清空TTS内部的普通队列，但保留高优先级队列（用于用户问题） if hasattr(self.tts, 'msgqueue'): with self.tts.msgqueue.mutex: self.tts.msgqueue.queue.clear() # 注意：不清空高优先级队列，以确保用户问题能够被处理 # 清空输入音频流缓冲区 if hasattr(self.tts, 'input_stream'): if hasattr(self.tts.input_stream, 'seek') and hasattr(self.tts.input_stream, 'truncate'): self.tts.input_stream.seek(0) self.tts.input_stream.truncate() # 在锁保护下更新状态 self.is_speaking_flag = False self.current_text = "" self.current_pos = 0 def flush_talk(self): """打断当前播放并保存未完成的内容""" with self.interrupt_lock: # 1. 保存当前正在播、但没播完的文本（关键） if self.current_text and self.current_pos < len(self.current_text): unfinished = self.current_text[self.current_pos:] if unfinished.strip(): try: self.interrupted_queue.put_nowait(unfinished) except queue.Full: logger.warning("中断队列已满，丢弃未完成的文本") # 2. 把 msg_queue 里剩余的全部移到 interrupted_queue（不丢弃） while not self.msg_queue.empty(): try: self.interrupted_queue.put_nowait(self.msg_queue.get_nowait()) except queue.Full: logger.warning("中断队列已满，丢弃剩余消息") break # 3.停止当前TTS播放 # 注意：stop_tts内部已经有锁，这里需要临时释放锁避免死锁 pass # 4. 在锁保护下清空当前播放状态 self.is_speaking_flag = False self.current_text = "" self.current_pos = 0 # 5. 不要重置打断标志，保持中断状态直到新内容开始播放 # 在锁外调用stop_tts避免死锁 self.stop_tts() # self.tts.flush_talk() # self.asr.flush_talk() def handle_interruption_during_intro(self, user_question, datainfo:dict={}): """处理在介绍过程中用户提问的逻辑，暂停介绍并优先回答问题""" with self.interrupt_lock: # 1. 保存当前正在播放的介绍内容到中断队列 if self.current_text and self.current_pos < len(self.current_text): unfinished_intro = self.current_text[self.current_pos:] if unfinished_intro.strip(): self.interrupted_queue.put(unfinished_intro) logger.info(f"保存未完成的介绍内容: {unfinished_intro[:50]}...") # 2. 将消息队列中剩余的介绍内容也保存到中断队列 remaining_items = [] while not self.msg_queue.empty(): remaining_items.append(self.msg_queue.get()) if remaining_items: for item in remaining_items: self.interrupted_queue.put(item) logger.info(f"保存了 {len(remaining_items)} 个剩余的介绍内容到中断队列") # 3. 停止TTS播放（注意：这会将TTS状态设为PAUSE，但我们随后会处理用户问题） self.stop_tts() # 4. 清空当前播放状态 self.is_speaking_flag = False self.current_text = "" self.current_pos = 0 # 5. 确保TTS状态设置为RUNNING以处理用户问题 if hasattr(self.tts, 'state'): self.tts.state = State.RUNNING # 6. 重置打断标志，允许播放新内容 if hasattr(self.tts, 'reset_interrupt_flag'): self.tts.reset_interrupt_flag() # 7. 优先播放用户问题 self.tts.put_high_priority_msg(user_question, datainfo) # 8. 立即尝试处理高优先级队列中的消息 # 通过直接调用TTS处理方法，绕过正常的队列处理延迟 logger.info("已暂停介绍并开始播放用户问题") # 9. 唤醒TTS处理线程，确保立即处理高优先级消息 # 通过向普通队列添加一个空消息来触发处理循环 if hasattr(self.tts, 'msgqueue'): self.tts.msgqueue.put(("", {})) def put_user_question(self, msg, datainfo:dict={}): """专门用于处理用户问题，优先级高于普通消息""" with self.interrupt_lock: # 强制中断当前播放 self.flush_talk() # 直接播放用户问题（使用高优先级队列） self.tts.put_high_priority_msg(msg, datainfo) # 确保TTS状态设置为RUNNING以处理用户问题 if hasattr(self.tts, 'state'): self.tts.state = State.RUNNING # 重置打断标志，允许播放新内容 if hasattr(self.tts, 'reset_interrupt_flag'): self.tts.reset_interrupt_flag() # 唤醒TTS处理线程，确保立即处理高优先级消息 # 通过向普通队列添加一个空消息来触发处理循环 if hasattr(self.tts, 'msgqueue'): self.tts.msgqueue.put(("", {})) def process_user_questions_and_resume(self): """处理用户问题并恢复之前的内容""" with self.interrupt_lock: # 检查是否有用户问题需要处理 while not self.user_question_queue.empty(): msg, datainfo = self.user_question_queue.get() # 直接播放用户问题 self.tts.put_msg_txt(msg, datainfo) # 检查是否有被中断的内容需要恢复 if not self.interrupted_queue.empty(): # 将被中断的内容放回主队列 while not self.interrupted_queue.empty(): msg = self.interrupted_queue.get() self.msg_queue.put(msg) def resume_interrupted(self): """把 interrupted_queue 里的内容放回播放队列""" with self.interrupt_lock: resumed = False items_to_resume = [] # 先把所有中断的内容取出 while not self.interrupted_queue.empty(): items_to_resume.append(self.interrupted_queue.get()) resumed = True # 再按顺序放回主队列 for item in items_to_resume: self.msg_queue.put(item) return resumed # """恢复播放被中断的消息""" # return self.tts.resume_interrupted() def start_intro_with_interrupt_capability(self, intro_text, datainfo:dict={}): """开始播放介绍内容，同时保持接收用户提问的能力""" # 将介绍内容放入主消息队列 self.put_msg_txt(intro_text, datainfo) def is_speaking(self)->bool: return self.speaking def __loadcustom(self): for item in self.opt.customopt: logger.info(item) input_img_list = glob.glob(os.path.join(item['imgpath'], '*.[jpJP][pnPN]*[gG]')) input_img_list = sorted(input_img_list, key=lambda x: int(os.path.splitext(os.path.basename(x))[0])) self.custom_img_cycle[item['audiotype']] = read_imgs(input_img_list) self.custom_audio_cycle[item['audiotype']], sample_rate = sf.read(item['audiopath'], dtype='float32') self.custom_audio_index[item['audiotype']] = 0 self.custom_index[item['audiotype']] = 0 self.custom_opt[item['audiotype']] = item def init_customindex(self): self.curr_state=0 for key in self.custom_audio_index: self.custom_audio_index[key]=0 for key in self.custom_index: self.custom_index[key]=0 def notify(self,eventpoint): logger.info("notify:%s",eventpoint) # 检查是否是用户问题回答的结束事件 if isinstance(eventpoint, dict) and eventpoint.get('status') == 'end': # 如果是用户问题的回答结束，检查是否需要恢复被中断的内容 # 检查这个结束事件是否与用户问题相关 if 'knowledge_base' in eventpoint or self.interrupted_queue.qsize() > 0: # 这是一个用户问题的回答结束，检查是否需要恢复被中断的内容 import threading # 使用线程延时一小段时间再恢复，确保当前事件处理完成 timer = threading.Timer(0.5, self._try_resume_after_qa) timer.start() # 注意：介绍播放的自动续播不再通过notify触发，而是在TTS处理完成后自动触发 # 这样可以确保前一条完全播放完成后再播放下一条 def _try_resume_after_qa(self): """尝试恢复问答后的内容""" with self.interrupt_lock: # 检查是否还有被中断的内容需要恢复 if not self.interrupted_queue.empty(): logger.info("检测到问答完成，恢复被中断的内容...") # 有被中断的内容需要恢复，调用恢复方法 self.resume_interrupted() def _continue_intro_play(self): """继续播放下一条介绍内容""" try: # 检查是否处于介绍播放状态 if not (hasattr(self, 'intro_play_state') and self.intro_play_state.get('is_playing', False) and not self.intro_play_state.get('is_paused', False)): logger.info("不处于介绍播放状态，跳过自动续播") return # 检查TTS队列是否为空，如果不为空说明还有消息在等待，不放入新消息 if hasattr(self, 'tts') and hasattr(self.tts, 'msgqueue'): if not self.tts.msgqueue.empty(): logger.info("TTS消息队列不为空，等待队列清空后再播放下一条") # 设置等待标志，并设置定时器再次检查 self.intro_play_state['is_waiting_next'] = True import threading timer = threading.Timer(2.0, self._continue_intro_play) timer.start() return # 设置等待标志，防止重复触发 if hasattr(self, 'intro_play_state'): self.intro_play_state['is_waiting_next'] = True if hasattr(self, 'knowledge_intro_instance') and self.knowledge_intro_instance: # 获取下一条介绍内容 next_content = self.knowledge_intro_instance._get_next_content() if next_content and next_content.get('text'): logger.info(f"自动续播介绍内容 - 序号：{next_content.get('play_index', 1)}/{next_content.get('total_count', 1)}") # 更新播放状态 if hasattr(self, 'intro_play_state'): self.intro_play_state["last_played_index"] = next_content.get("play_index", 1) # 计算停顿时间：根据文本长度估算播放时间 + 额外停顿 # 语速约 3 字/秒（较慢的播报速度） text = next_content.get('text', '') estimated_play_time = len(text) / 3 # 估算播放时间（秒） pause_time = estimated_play_time + 3 # 播放时间 + 3秒停顿 logger.info(f"当前文案长度：{len(text)}字，估算播放时间：{estimated_play_time:.1f}秒，播放下一条需等待：{pause_time:.1f}秒") # 播放下一条内容 self.put_msg_txt(next_content['text']) # 设置定时器，在播放完成后继续播放下一条 import threading timer = threading.Timer(pause_time, self._continue_intro_play) timer.start() else: # 没有更多内容了，标记播放完成 logger.info("介绍内容全部播放完成") if hasattr(self, 'intro_play_state'): self.intro_play_state["is_playing"] = False self.intro_play_state['is_waiting_next'] = False except Exception as e: logger.error(f"自动续播介绍内容时出错: {e}") # 确保错误时也重置等待标志 if hasattr(self, 'intro_play_state'): self.intro_play_state['is_waiting_next'] = False def _reset_waiting_flag(self): """重置等待标志，允许播放下一条""" if hasattr(self, 'intro_play_state'): self.intro_play_state['is_waiting_next'] = False logger.info("等待时间结束，可以播放下一条") def start_recording(self): """开始录制视频""" if self.recording: return command = ['ffmpeg', '-y', '-an', '-f', 'rawvideo', '-vcodec','rawvideo', '-pix_fmt', 'bgr24', #像素格式 '-s', "{}x{}".format(self.width, self.height), '-r', str(25), '-i', '-', '-pix_fmt', 'yuv420p', '-vcodec', "h264", #'-f' , 'flv', f'temp{self.opt.sessionid}.mp4'] self._record_video_pipe = subprocess.Popen(command, shell=False, stdin=subprocess.PIPE) acommand = ['ffmpeg', '-y', '-vn', '-f', 's16le', #'-acodec','pcm_s16le', '-ac', '1', '-ar', '16000', '-i', '-', '-acodec', 'aac', #'-f' , 'wav', f'temp{self.opt.sessionid}.aac'] self._record_audio_pipe = subprocess.Popen(acommand, shell=False, stdin=subprocess.PIPE) self.recording = True # self.recordq_video.queue.clear() # self.recordq_audio.queue.clear() # self.container = av.open(path, mode="w") # process_thread = Thread(target=self.record_frame, args=()) # process_thread.start() def record_video_data(self,image): if self.width == 0: logger.debug("image.shape: %s", image.shape) self.height,self.width,_ = image.shape if self.recording: self._record_video_pipe.stdin.write(image.tostring()) def record_audio_data(self,frame): if self.recording: self._record_audio_pipe.stdin.write(frame.tostring()) def start_rtmp_push(self): """启动 RTMP 推流（单进程 FFmpeg，视频+音频共用一套时钟）""" if not self._rtmp_push_url or self._rtmp_pushing: return if self.width == 0 or self.height == 0: logger.warning('RTMP 推流: 视频尺寸未初始化，等待第一帧') return self._rtmp_pushing = True # ffmpeg 命令：直接通过 stdin 管道接收视频和音频 # 使用 pipe:0 接收视频，pipe:1 接收音频（需要两个独立的 FFmpeg 进程或使用 filter_complex） # 但更简单的方式是：使用一个 FFmpeg 进程，视频通过 stdin，音频通过第二个管道 ffmpeg_cmd = [ 'ffmpeg', '-y', # 视频输入：通过管道 '-f', 'rawvideo', '-vcodec', 'rawvideo', '-pix_fmt', 'bgr24', '-s', f'{self.width}x{self.height}', '-r', '25', '-i', 'pipe:0', # 从 stdin 读取视频 # 音频输入：通过第二个管道 '-f', 's16le', '-acodec', 'pcm_s16le', '-ar', '16000', '-ac', '1', '-i', 'pipe:1', # 从 fd 1 读取音频（需要特殊处理） # 编码输出 '-c:v', 'libx264', '-preset', 'ultrafast', '-tune', 'zerolatency', '-pix_fmt', 'yuv420p', '-g', '50', '-keyint_min', '25', '-c:a', 'aac', '-b:a', '128k', '-ar', '44100', '-f', 'flv', self._rtmp_push_url ] # 注意：FFmpeg 不支持同时从两个 pipe 读取，需要使用 FIFO # 所以我们保留 FIFO 方案，但去掉手动速率控制，让 FFmpeg 自动处理 # 创建 FIFO import tempfile self._rtmp_fifo_dir = tempfile.mkdtemp(prefix='rtmp_fifo_') self._rtmp_video_fifo = os.path.join(self._rtmp_fifo_dir, 'video') self._rtmp_audio_fifo = os.path.join(self._rtmp_fifo_dir, 'audio') os.mkfifo(self._rtmp_video_fifo) os.mkfifo(self._rtmp_audio_fifo) # 视频和音频缓冲队列 self._rtmp_video_queue = queue.Queue(maxsize=30) self._rtmp_audio_queue = queue.Queue(maxsize=100) # 重写 ffmpeg 命令使用 FIFO # 关键：不使用 -re（会阻塞 FIFO），改用 -vf realtime/-af arealtime 对输出限速 # 再配合 -fflags +genpts 与 -vsync cfr 保证恒定 25fps，避免下游转推看到 2x 速度 ffmpeg_cmd = [ 'ffmpeg', '-y', '-fflags', '+genpts', # 视频输入（不使用 -re，让 FIFO 自然反压控制速率） '-f', 'rawvideo', '-vcodec', 'rawvideo', '-pix_fmt', 'bgr24', '-s', f'{self.width}x{self.height}', '-r', '25', '-thread_queue_size', '1024', '-i', self._rtmp_video_fifo, # 音频输入（16kHz 单声道 PCM） '-f', 's16le', '-acodec', 'pcm_s16le', '-ar', '16000', '-ac', '1', '-thread_queue_size', '1024', '-i', self._rtmp_audio_fifo, # 滤镜：按墙钟实时限速，视频+音频同步限速到 1x，避免 SRS 转推出现 speed=2x '-vf', 'realtime', '-af', 'arealtime,aresample=async=1:first_pts=0', # 编码输出（音频保持 16kHz 不重采样，避免时间戳错乱） '-c:v', 'libx264', '-preset', 'ultrafast', '-tune', 'zerolatency', '-pix_fmt', 'yuv420p', '-g', '50', '-keyint_min', '25', '-vsync', 'cfr', '-r', '25', # 强制输出恒定 25fps '-c:a', 'aac', '-b:a', '128k', '-ar', '16000', # 保持 16kHz！ '-f', 'flv', self._rtmp_push_url ] try: # 启动 ffmpeg 进程 self._rtmp_pipe = subprocess.Popen( ffmpeg_cmd, shell=False, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE ) logger.info(f'RTMP ffmpeg 进程已启动: PID={self._rtmp_pipe.pid}') # 视频 FIFO 写入线程（无速率控制，FFmpeg 自动处理） def video_fifo_writer(): try: vfd = os.open(self._rtmp_video_fifo, os.O_WRONLY) logger.info('RTMP 视频 FIFO 已打开') while self._rtmp_pushing: try: data = self._rtmp_video_queue.get(timeout=1.0) if data is None: break os.write(vfd, data) except queue.Empty: continue except Exception as e: logger.error(f'RTMP 视频写入线程异常: {e}') self._rtmp_pushing = False finally: try: os.close(vfd) except: pass # 音频 FIFO 写入线程（无速率控制，FFmpeg 自动处理） def audio_fifo_writer(): try: afd = os.open(self._rtmp_audio_fifo, os.O_WRONLY) logger.info('RTMP 音频 FIFO 已打开') # 静音帧：匹配一个视频帧（40ms）的 PCM 数据量 # 16kHz * 0.04s * 2字节(单声道 s16le) = 1280 字节 silence_frame = b'\x00' * 1280 # 40ms 静音（严格匹配视频帧率，避免音频过速） while self._rtmp_pushing: try: data = self._rtmp_audio_queue.get(timeout=0.04) if data is None: break os.write(afd, data) except queue.Empty: # 队列为空（TTS未说话），写入一帧 40ms 静音保持ffmpeg不卡顿 try: os.write(afd, silence_frame) except: break except Exception as e: logger.error(f'RTMP 音频写入线程异常: {e}') self._rtmp_pushing = False finally: try: os.close(afd) except: pass self._rtmp_video_thread = Thread(target=video_fifo_writer, daemon=True, name='rtmp_video_writer') self._rtmp_audio_thread = Thread(target=audio_fifo_writer, daemon=True, name='rtmp_audio_writer') self._rtmp_video_thread.start() self._rtmp_audio_thread.start() # stderr 读取线程 def read_stderr(): for line in iter(self._rtmp_pipe.stderr.readline, b''): msg = line.decode('utf-8', errors='ignore').strip() if msg: logger.debug(f'FFmpeg RTMP: {msg}') Thread(target=read_stderr, daemon=True, name='rtmp_stderr_reader').start() logger.info(f'RTMP 音视频推流已启动: {self._rtmp_push_url}') except Exception as e: logger.error(f'RTMP 推流启动失败: {e}') self._rtmp_pushing = False self._cleanup_rtmp_fifos() def rtmp_push_video_data(self, image): """将视频帧放入 RTMP 推流队列""" if not self._rtmp_pushing: return try: self._rtmp_video_queue.put_nowait(image.tobytes()) except queue.Full: pass # 丢弃旧帧，避免延迟累积 def rtmp_push_audio_data(self, frame): """将音频帧放入 RTMP 推流队列""" if not self._rtmp_pushing: return try: self._rtmp_audio_queue.put_nowait(frame.tobytes()) except queue.Full: pass # 丢弃旧帧，避免延迟累积 def stop_rtmp_push(self): """停止 RTMP 推流""" if not self._rtmp_pushing: return self._rtmp_pushing = False # 发送停止信号给写入线程 try: self._rtmp_video_queue.put(None, timeout=1) except: pass try: self._rtmp_audio_queue.put(None, timeout=1) except: pass # 等待写入线程结束 for t in [getattr(self, '_rtmp_video_thread', None), getattr(self, '_rtmp_audio_thread', None)]: if t and t.is_alive(): t.join(timeout=3) # 终止 ffmpeg if self._rtmp_pipe is not None: try: self._rtmp_pipe.wait(timeout=5) except: self._rtmp_pipe.kill() self._cleanup_rtmp_fifos() logger.info('RTMP 推流已停止') def _cleanup_rtmp_fifos(self): """清理 FIFO 文件和目录""" for path in [getattr(self, '_rtmp_video_fifo', ''), getattr(self, '_rtmp_audio_fifo', '')]: if path and os.path.exists(path): try: os.unlink(path) except: pass fifo_dir = getattr(self, '_rtmp_fifo_dir', '') if fifo_dir and os.path.isdir(fifo_dir): try: os.rmdir(fifo_dir) except: pass # def record_frame(self): # videostream = self.container.add_stream("libx264", rate=25) # videostream.codec_context.time_base = Fraction(1, 25) # audiostream = self.container.add_stream("aac") # audiostream.codec_context.time_base = Fraction(1, 16000) # init = True # framenum = 0 # while self.recording: # try: # videoframe = self.recordq_video.get(block=True, timeout=1) # videoframe.pts = framenum #int(round(framenum*0.04 / videostream.codec_context.time_base)) # videoframe.dts = videoframe.pts # if init: # videostream.width = videoframe.width # videostream.height = videoframe.height # init = False # for packet in videostream.encode(videoframe): # self.container.mux(packet) # for k in range(2): # audioframe = self.recordq_audio.get(block=True, timeout=1) # audioframe.pts = int(round((framenum*2+k)*0.02 / audiostream.codec_context.time_base)) # audioframe.dts = audioframe.pts # for packet in audiostream.encode(audioframe): # self.container.mux(packet) # framenum += 1 # except queue.Empty: # print('record queue empty,') # continue # except Exception as e: # print(e) # #break # for packet in videostream.encode(None): # self.container.mux(packet) # for packet in audiostream.encode(None): # self.container.mux(packet) # self.container.close() # self.recordq_video.queue.clear() # self.recordq_audio.queue.clear() # print('record thread stop') def stop_recording(self): """停止录制视频""" if not self.recording: return self.recording = False self._record_video_pipe.stdin.close() #wait() self._record_video_pipe.wait() self._record_audio_pipe.stdin.close() self._record_audio_pipe.wait() cmd_combine_audio = f"ffmpeg -y -i temp{self.opt.sessionid}.aac -i temp{self.opt.sessionid}.mp4 -c:v copy -c:a copy data/record.mp4" os.system(cmd_combine_audio) #os.remove(output_path) def mirror_index(self,size, index): #size = len(self.coord_list_cycle) turn = index // size res = index % size if turn % 2 == 0: return res else: return size - res - 1 def get_audio_stream(self,audiotype): idx = self.custom_audio_index[audiotype] stream = self.custom_audio_cycle[audiotype][idx:idx+self.chunk] self.custom_audio_index[audiotype] += self.chunk if self.custom_audio_index[audiotype]>=self.custom_audio_cycle[audiotype].shape[0]: self.curr_state = 1 #当前视频不循环播放，切换到静音状态 return stream def set_custom_state(self,audiotype, reinit=True): logger.debug('set_custom_state: %s', audiotype) if self.custom_audio_index.get(audiotype) is None: return self.curr_state = audiotype if reinit: self.custom_audio_index[audiotype] = 0 self.custom_index[audiotype] = 0 def process_frames(self,quit_event,loop=None,audio_track=None,video_track=None): enable_transition = False # 设置为False禁用过渡效果，True启用 if enable_transition: _last_speaking = False _transition_start = time.time() _transition_duration = 0.1 # 过渡时间 _last_silent_frame = None # 静音帧缓存 _last_speaking_frame = None # 说话帧缓存 if self.opt.transport=='virtualcam': import pyvirtualcam vircam = None audio_tmp = queue.Queue(maxsize=3000) audio_thread = Thread(target=play_audio, args=(quit_event,audio_tmp,), daemon=True, name="pyaudio_stream") audio_thread.start() # RTMP 推流: 延迟启动，等待第一帧确定视频尺寸 _rtmp_started = False _rtmp_attempted = False # 记录是否已经尝试过启动推流 while not quit_event.is_set(): try: res_frame,idx,audio_frames = self.res_frame_queue.get(block=True, timeout=1) except queue.Empty: continue if enable_transition: # 检测状态变化 current_speaking = not (audio_frames[0][1]!=0 and audio_frames[1][1]!=0) if current_speaking != _last_speaking: logger.info(f"状态切换：{'说话' if _last_speaking else '静音'} → {'说话' if current_speaking else '静音'}") _transition_start = time.time() _last_speaking = current_speaking if audio_frames[0][1]!=0 and audio_frames[1][1]!=0: #全为静音数据，只需要取fullimg self.speaking = False audiotype = audio_frames[0][1] if self.custom_index.get(audiotype) is not None: #有自定义视频 mirindex = self.mirror_index(len(self.custom_img_cycle[audiotype]),self.custom_index[audiotype]) target_frame = self.custom_img_cycle[audiotype][mirindex] self.custom_index[audiotype] += 1 else: target_frame = self.frame_list_cycle[idx] if enable_transition: # 说话→静音过渡 if time.time() - _transition_start < _transition_duration and _last_speaking_frame is not None: alpha = min(1.0, (time.time() - _transition_start) / _transition_duration) combine_frame = cv2.addWeighted(_last_speaking_frame, 1-alpha, target_frame, alpha, 0) else: combine_frame = target_frame # 缓存静音帧 _last_silent_frame = combine_frame.copy() else: combine_frame = target_frame else: self.speaking = True try: current_frame = self.paste_back_frame(res_frame,idx) except Exception as e: logger.warning(f"paste_back_frame error: {e}") continue if enable_transition: # 静音→说话过渡 if time.time() - _transition_start < _transition_duration and _last_silent_frame is not None: alpha = min(1.0, (time.time() - _transition_start) / _transition_duration) combine_frame = cv2.addWeighted(_last_silent_frame, 1-alpha, current_frame, alpha, 0) else: combine_frame = current_frame # 缓存说话帧 _last_speaking_frame = combine_frame.copy() else: combine_frame = current_frame #cv2.putText(combine_frame, "LiveTalking", (10, 20), cv2.FONT_HERSHEY_SIMPLEX, 0.3, (128,128,128), 1) if self.opt.transport=='virtualcam': if vircam==None: height, width,_= combine_frame.shape vircam = pyvirtualcam.Camera(width=width, height=height, fps=25, fmt=pyvirtualcam.PixelFormat.BGR,print_fps=True) vircam.send(combine_frame) elif video_track is not None and loop is not None: #webrtc image = combine_frame new_frame = VideoFrame.from_ndarray(image, format="bgr24") asyncio.run_coroutine_threadsafe(video_track._queue.put((new_frame,None)), loop) self.record_video_data(combine_frame) # RTMP 推流: 视频帧 if self._rtmp_push_url: if not _rtmp_started and not _rtmp_attempted: # 首次获得视频帧，启动 RTMP 推流 if self.width > 0 and self.height > 0: self.start_rtmp_push() _rtmp_attempted = True # 标记已尝试 if self._rtmp_pushing: # 启动成功 _rtmp_started = True if _rtmp_started: # 只有启动成功后才推流 self.rtmp_push_video_data(combine_frame) for audio_frame in audio_frames: frame,type,eventpoint = audio_frame frame = (frame * 32767).astype(np.int16) if self.opt.transport=='virtualcam': audio_tmp.put(frame.tobytes()) #TODO elif audio_track is not None and loop is not None: #webrtc new_frame = AudioFrame(format='s16', layout='mono', samples=frame.shape[0]) new_frame.planes[0].update(frame.tobytes()) new_frame.sample_rate=16000 asyncio.run_coroutine_threadsafe(audio_track._queue.put((new_frame,eventpoint)), loop) self.record_audio_data(frame) # RTMP 推流: 音频帧 if _rtmp_started: self.rtmp_push_audio_data(frame) if self.opt.transport=='virtualcam': vircam.sleep_until_next_frame() # 停止 RTMP 推流 self.stop_rtmp_push() if self.opt.transport=='virtualcam': audio_thread.join() vircam.close() logger.info('basereal process_frames thread stop') # def process_custom(self,audiotype:int,idx:int): # if self.curr_state!=audiotype: #从推理切到口播 # if idx in self.switch_pos: #在卡点位置可以切换 # self.curr_state=audiotype # self.custom_index=0 # else: # self.custom_index+=1