如何用Python asyncio实现声音依次播放且视频持续播放？

现有以下代码，尝试用asyncio实现视频播放与声音的异步处理，但当前两个声音会同时播放。需要修改代码实现声音依次播放，同时保持视频正常播放。考虑使用asyncio.Queue()但不熟悉其用法，寻求帮助。

原代码：

import cv2
import os
from playsound import playsound as ps
import pygame
import time
import asyncio

video_path = os.path.join('videos_in', 'video.mp4')
sidewalk_path = os.path.join('voices', 'audio1.mp3')
driveway_path = os.path.join('voices', 'audio2.mp3')
cap = cv2.VideoCapture(video_path)
# Get FPS of original video
original_fps = int(cap.get(cv2.CAP_PROP_FPS))

pygame.init()
pygame.mixer.init()

async def sidewalk_sound():
    sc(sidewalk_path)

async def driveway_sound():
    sc(driveway_path)

async def playgo(tasks):
    for task in tasks:
        await asyncio.create_task(task)

class SoundClass:
    def __call__(self, path):
        sd = pygame.mixer.Sound(path)
        sd.play()
        return None

sc = SoundClass()
piv = 1

while True:
    task = []
    ret, frame = cap.read()
    if piv:
        task.append(driveway_sound())
        task.append(sidewalk_sound())
        piv = 0
    asyncio.run(playgo(task))
    cv2.imshow('video', frame)

    delay = int(1000 / original_fps)
    if cv2.waitKey(delay) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()

要实现声音依次播放且不阻塞视频，核心是让异步任务能等待单条声音播放完成，同时用队列管理待播放的声音，让视频循环和声音处理在同一个异步事件循环中运行。具体修改如下：

关键修改点 #

• 让声音播放函数等待声音播放完毕：通过pygame.mixer.Sound.get_length()获取音频时长，用asyncio.sleep()等待对应时间（避免阻塞事件循环）。
• 使用asyncio.Queue()存储待播放的音频路径，单独启动一个异步消费者任务，依次取出路径播放。
• 视频播放逻辑封装为异步函数，和声音消费任务一起在同一个事件循环中运行，避免每次循环重建事件循环。

修改后的代码 #

import cv2
import os
import pygame
import asyncio

video_path = os.path.join('videos_in', 'video.mp4')
sidewalk_path = os.path.join('voices', 'audio1.mp3')
driveway_path = os.path.join('voices', 'audio2.mp3')
cap = cv2.VideoCapture(video_path)
original_fps = int(cap.get(cv2.CAP_PROP_FPS))

pygame.init()
pygame.mixer.init()

# 创建声音播放队列
sound_queue = asyncio.Queue()

async def play_sound(path):
    """播放单条音频并等待播放完成"""
    sd = pygame.mixer.Sound(path)
    sd.play()
    # 获取音频时长，异步等待播放完成
    await asyncio.sleep(sd.get_length())

async def sound_consumer():
    """队列消费者：依次播放队列中的音频"""
    while True:
        path = await sound_queue.get()
        if path is None:
            break
        await play_sound(path)
        sound_queue.task_done()

async def video_player():
    """异步播放视频"""
    piv = 1
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        
        # 第一次循环时将两个音频加入队列
        if piv:
            await sound_queue.put(driveway_path)
            await sound_queue.put(sidewalk_path)
            piv = 0
        
        # 显示视频帧
        cv2.imshow('video', frame)
        
        # 按原视频帧率控制延迟，同时处理事件循环的回调
        delay = int(1000 / original_fps)
        if cv2.waitKey(delay) & 0xFF == ord('q'):
            break
    
    cap.release()
    cv2.destroyAllWindows()
    # 发送停止信号给消费者
    await sound_queue.put(None)

async def main():
    # 启动声音消费者任务
    consumer_task = asyncio.create_task(sound_consumer())
    # 运行视频播放器
    await video_player()
    # 等待队列中所有任务完成
    await sound_queue.join()
    # 取消消费者任务
    consumer_task.cancel()
    try:
        await consumer_task
    except asyncio.CancelledError:
        pass

if __name__ == "__main__":
    asyncio.run(main())

代码说明 #

• 队列管理：sound_queue用来存储待播放的音频路径，sound_consumer作为异步任务不断从队列取路径，播放完成后再取下一个，实现依次播放。
• 异步等待音频：play_sound中用asyncio.sleep()等待音频时长，不会阻塞事件循环，保证视频能正常刷新。
• 共享事件循环：视频播放和声音消费都在同一个事件循环中运行，避免了原代码中每次循环重建事件循环的问题，效率更高。
• 退出处理：按下q退出时，会释放视频资源，发送停止信号给消费者任务，确保程序正常退出。

内容的提问来源于stack exchange，提问作者2 weeks later