我想获取互联网音频/广播流(特别是Longplayer,单击以获取直接流URL)并用python播放。
最好是在后台运行,以使脚本能够继续运行其主循环。 (例如,作为游戏背景音乐或其他东西,尽管Pyglet,PyGame等可能会为此提供自己的工具。)
我已经看到一些使用requests
录制网络广播并将其转储到文件中的可能过时的示例,但这不是我想要的,答案的注释似乎包含有关requests
的争论,尤其是其他问题? (请参阅here)
我愿意使用可以pip
的任何软件包,只要它可以在Python 3.X中使用。 (当前使用3.6纯粹是因为我还没有花大力气安装3.7)
重申一下,我不想保存该流,只需立即将其播放(或在需要时进行缓冲?)再返回给用户即可。最好是在不阻止脚本的情况下进行脚本,我想这需要多线程/多处理,但这仅次于播放。
最佳答案
这些似乎很简单的问题似乎总是如此,细节在于魔鬼。我最终写了一些代码来解决这个问题。可以使用python3 -m pip install ffmpeg-python PyOpenAL
安装pip依赖项。代码的工作流程可以分为两个步骤:
这是该代码的样子(带有一些注释)。如果您对代码或此答案的任何其他部分有任何疑问,请告诉我。
import ctypes
import ffmpeg
import numpy as np
from openal.al import *
from openal.alc import *
from queue import Queue, Empty
from threading import Thread
import time
from urllib.request import urlopen
def init_audio():
#Create an OpenAL device and context.
device_name = alcGetString(None, ALC_DEFAULT_DEVICE_SPECIFIER)
device = alcOpenDevice(device_name)
context = alcCreateContext(device, None)
alcMakeContextCurrent(context)
return (device, context)
def create_audio_source():
#Create an OpenAL source.
source = ctypes.c_uint()
alGenSources(1, ctypes.pointer(source))
return source
def create_audio_buffers(num_buffers):
#Create a ctypes array of OpenAL buffers.
buffers = (ctypes.c_uint * num_buffers)()
buffers_ptr = ctypes.cast(
ctypes.pointer(buffers),
ctypes.POINTER(ctypes.c_uint),
)
alGenBuffers(num_buffers, buffers_ptr)
return buffers_ptr
def fill_audio_buffer(buffer_id, chunk):
#Fill an OpenAL buffer with a chunk of PCM data.
alBufferData(buffer_id, AL_FORMAT_STEREO16, chunk, len(chunk), 44100)
def get_audio_chunk(process, chunk_size):
#Fetch a chunk of PCM data from the FFMPEG process.
return process.stdout.read(chunk_size)
def play_audio(process):
#Queues up PCM chunks for playing through OpenAL
num_buffers = 4
chunk_size = 8192
device, context = init_audio()
source = create_audio_source()
buffers = create_audio_buffers(num_buffers)
#Initialize the OpenAL buffers with some chunks
for i in range(num_buffers):
buffer_id = ctypes.c_uint(buffers[i])
chunk = get_audio_chunk(process, chunk_size)
fill_audio_buffer(buffer_id, chunk)
#Queue the OpenAL buffers into the OpenAL source and start playing sound!
alSourceQueueBuffers(source, num_buffers, buffers)
alSourcePlay(source)
num_used_buffers = ctypes.pointer(ctypes.c_int())
while True:
#Check if any buffers are used up/processed and refill them with data.
alGetSourcei(source, AL_BUFFERS_PROCESSED, num_used_buffers)
if num_used_buffers.contents.value != 0:
used_buffer_id = ctypes.c_uint()
used_buffer_ptr = ctypes.pointer(used_buffer_id)
alSourceUnqueueBuffers(source, 1, used_buffer_ptr)
chunk = get_audio_chunk(process, chunk_size)
fill_audio_buffer(used_buffer_id, chunk)
alSourceQueueBuffers(source, 1, used_buffer_ptr)
if __name__ == "__main__":
url = "http://icecast.spc.org:8000/longplayer"
#Run FFMPEG in a separate process using subprocess, so it is non-blocking
process = (
ffmpeg
.input(url)
.output("pipe:", format='s16le', acodec='pcm_s16le', ac=2, ar=44100, loglevel="quiet")
.run_async(pipe_stdout=True)
)
#Run audio playing OpenAL code in a separate thread
thread = Thread(target=play_audio, args=(process,), daemon=True)
thread.start()
#Some example code to show that this is not being blocked by the audio.
start = time.time()
while True:
print(time.time() - start)
关于python - 如何接收 “icecast”网络广播流以使用Python立即播放?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/57438530/