kind = "audio" How to use queue output for Audio? Audio 如何用队列输出？

[gg22mm]

The sound problem generated when creating virtual digital humans/在做虚拟数字人时 时时生成的声音问题
How to use queue output for Audio? Audio 如何用队列输出？

# All code
import asyncio
import json
import os
import uuid
import numpy as np
import math
import numpy as numpy
import cv2
from av import AudioFrame, VideoFrame #单独处理 音频 与 视频
import queue
import threading
from pydub import AudioSegment
# 
from aiortc import MediaStreamTrack, RTCPeerConnection, RTCSessionDescription, VideoStreamTrack
from aiortc.contrib.media import MediaPlayer, MediaRelay
#
import uvicorn 
from fastapi import FastAPI, Request
# 
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

app = FastAPI()

# 创建两个队列，一个用于视频帧，一个用于音频样本
video_queue = queue.Queue(maxsize=50000)
audio_queue = queue.Queue(maxsize=50000)

# 模拟音频帧生成
def load_audio(audio_path):
    
    # 加载音频文件
    audio = AudioSegment.from_file(audio_path)#("yongen_diy.wav", format="wav") #("1.mp3", format="mp3")
   
    # 转换为 numpy 数组
    samples = np.array(audio.get_array_of_samples())
    frame_rate = audio.frame_rate  # 44100 意思是：每秒采样44100次
    frame_size=frame_rate//25 #一秒声音分担到25张图片上。 # 44100 意思是：每秒采样44100次    
    
    # 分割音频为帧
    for i in range(0, len(samples), frame_size): #输出0-len(samples)的数据,间隔步长为：frame_size 
        frame = samples[i:i + frame_size]
        # print( str(i)+':'+ str(i+frame_size) ) # 0:600 600:1200 1200:1800 1800:2400
       
        if len(frame) == frame_size:  # 确保每个帧的长度一致 600 600
            audio_queue.put(frame)
            # time.sleep(0.02)  # 模拟音频帧率 50fps

# 模拟视频帧生成 - 每秒25帧
def load_video(video_path):
    cap = cv2.VideoCapture(video_path)
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        if not video_queue.full():
            video_queue.put(frame)  # 将视频帧放入队列
            # time.sleep(0.04)  # 模拟音频帧率 25fps 


# 假设我们有一个视频文件和一个音频文件
audio_path = 'yongen_diy.wav'
video_path = '1.mp4'

# 创建线程来加载数据
video_thread = threading.Thread(target=load_video, args=(video_path,))
audio_thread = threading.Thread(target=load_audio, args=(audio_path,))    
    
video_thread.start()
audio_thread.start()

# 自定义视频 video 配合队列 - is ok
class diyKindVideo(VideoStreamTrack):   

    kind = "video"

    def __init__(self):
        super().__init__()   # 必须初始化 VideoStreamTrack    
        self.cap = video_queue  #cv2.VideoCapture(0, cv2.CAP_DSHOW) #本地摄像头    

    async def recv(self):
        # video_frame = self.cap.get(timeout=1) #.read()主要是为了读取上面打开的视频的，以后一行可以用队列实现
        video_frame = video_queue.get(timeout=1)  # 从视频队列中取帧
        print(f"------视频帧 Video frame shape: {video_frame.shape}")  

        # 1、推流视频
        # # edges 图像识别修改 - 哈哈，视频二次推流搞定了，哈哈..~~
        # frame2 = cv2.cvtColor(cv2.Canny(video_frame, 100, 200), cv2.COLOR_GRAY2BGR)

        new_frame = VideoFrame.from_ndarray(video_frame, format="bgr24")
        pts, time_base = await self.next_timestamp()
        new_frame.pts = pts #int(self._index * self._time_base / self._frame_rate) 
        new_frame.time_base = time_base #'1/' + str(time_base)
        return new_frame

# 自定义声音 audio 配合队列 - ---------------------------------- not ok , why?????????????????????????????????????????
class diyKindAudio(MediaStreamTrack):   

    kind = "audio"

    def __init__(self):
        super().__init__() #必须初始化 VideoStreamTrack   
        self.cap = audio_queue  #cv2.VideoCapture(0, cv2.CAP_DSHOW) #本地摄像头        

    async def recv(self):
        audio_frame = audio_queue.get(timeout=1)  # 从音频队列中取样本
        print(f"------音频帧 Audio data shape: {audio_frame.shape}")       
        
        # 2、推流声音  
        frame = audio_frame.astype(np.int16)         
        new_frame = AudioFrame(format='s16', layout='mono', samples=frame.shape[0]) # 创建一个单声道，采样率为44100Hz的AudioFrame  
        new_frame.planes[0].update(frame.tobytes()) # __video_frame_buffer=frame.tobytes()
        new_frame.sample_rate=24000 #原来：16000 
        return new_frame  

#####################################################################################

# 异步回调处理 - 从 JSON 请求中提取参数、创建 RTCPeerConnection 、生成唯一 ID
@app.post("/offer")
async def offer(request: Request):    

    fid = RTCPeerConnection()  #浏览器生成 fid 连接  
    fids.add(fid)  #原生python 向集合 fids 中添加新的fid连接 
    print('-----------现在一共有连接数fids----------：',fids)

    # RTC心跳
    @fid.on("connectionstatechange")
    async def on_connectionstatechange():
        print("Connection state is %s" % fid.connectionState)
        if fid.connectionState == "failed":
            await fid.close()
            fids.discard(fid)

  
    fid.addTrack(diyKindVideo()) #视频Ok  
    fid.addTrack(diyKindAudio()) #音频Ok 
   
    
    # ----- 继续 -------    
    params = await request.json()

    # 假设我们已经收到了远程描述
    _desc = RTCSessionDescription(sdp=params["sdp"], type=params["type"])     
    await fid.setRemoteDescription(_desc)

    # 创建一个 answer
    answer = await fid.createAnswer()
    await fid.setLocalDescription(answer)   

    return {"sdp": fid.localDescription.sdp, "type": fid.localDescription.type}



# 方式一、添加不分离模版

# 渲染 HTML 模板的路径 - 不知道为什么，只有用python渲染的html才可以访问， 直了前后端分离方式就不行
@app.get("/")
def index(request: Request):
    return templates.TemplateResponse("index_fastapi_dev2.html", {"request": request, "name": "World"})    

static_dir = os.path.join(os.path.dirname(__file__), "static")
app.mount("/",StaticFiles(directory=static_dir, html=True),name="static")   

# 方式二、分离方式

if __name__ == "__main__":    
    uvicorn.run(app, host="0.0.0.0", port=8080,log_level='info',loop="asyncio")