数据是通过套接字接收的，没有更多的外壳，它们是纯IP B帧，以NAL标头（类似于00 00 00 01）开头。我现在使用 pyav 来解码帧，但我只能在收到第二个 pps 信息（在关键帧中）后解码数据（因此我发送到解码线程的数据块可以以 pps 和 sps 开头），否则解码（）或demux（）将返回错误“不存在的PPS 0引用的decode_slice_header错误”。
我想将数据馈送到可以记住前一个P帧的支持解码器，因此在馈送一个B帧后，解码器返回解码后的视频帧。或者某种形式的 IO，可以作为容器打开，并通过另一个线程不断向其中写入数据。
这是我的关键代码：
#read thread... read until get a key frame, then make a new io.BytesIO() to store the new data.
rawFrames = io.BytesIO()
while flag_get_keyFrame:()
....
content= socket.recv(2048)
rawFrames.write(content)
....
#decode thread... decode content between two key frames
....
rawFrames.seek(0)
container = av.open(rawFrames)
for packet in container.demux():
for frame in packet.decode():
self.frames.append(frame)
....
我的代码将播放视频，但有 3~4 秒的延迟。所以我不会把所有的东西都放在这里，因为我知道它实际上并没有达到我想要实现的目标。 我想在收到第一个关键帧后播放视频，并在收到后续帧后立即解码它们。 Pyav opencv ffmpeg 或其他东西，我怎样才能实现我的目标？
经过几个小时也找到了这个问题的答案。我自己弄清楚了。
对于单线程，可以执行以下操作：
rawData = io.BytesIO()
container = av.open(rawData, format="h264", mode='r')
cur_pos = 0
while True:
data = await websocket.recv()
rawData.write(data)
rawData.seek(cur_pos)
for packet in container.demux():
if packet.size == 0:
continue
cur_pos += packet.size
for frame in packet.decode():
self.frames.append(frame)
这就是基本思想。我已经制定了一个通用版本，它将接收线程和解码线程分开。如果 CPU 跟不上解码速度，代码也会跳帧，并从下一个关键帧开始解码（这样就不会出现撕裂的绿屏效果）。这是代码的完整版本：
import asyncio
import av
import cv2
import io
from multiprocessing import Process, Queue, Event
import time
import websockets
def display_frame(frame, start_time, pts_offset, frame_rate):
if frame.pts is not None:
play_time = (frame.pts - pts_offset) * frame.time_base.numerator / frame.time_base.denominator
if start_time is not None:
current_time = time.time() - start_time
time_diff = play_time - current_time
if time_diff > 1 / frame_rate:
return False
if time_diff > 0:
time.sleep(time_diff)
img = frame.to_ndarray(format='bgr24')
cv2.imshow('Video', img)
return True
def get_pts(frame):
return frame.pts
def render(terminated, data_queue):
rawData = io.BytesIO()
cur_pos = 0
frames_buffer = []
start_time = None
pts_offset = None
got_key_frame = False
while not terminated.is_set():
try:
data = data_queue.get_nowait()
except:
time.sleep(0.01)
continue
rawData.write(data)
rawData.seek(cur_pos)
if cur_pos == 0:
container = av.open(rawData, mode='r')
original_codec_ctx = container.streams.video[0].codec_context
codec = av.codec.CodecContext.create(original_codec_ctx.name, 'r')
cur_pos += len(data)
dts = None
for packet in container.demux():
if packet.size == 0:
continue
dts = packet.dts
if pts_offset is None:
pts_offset = packet.pts
if not got_key_frame and packet.is_keyframe:
got_key_frame = True
if data_queue.qsize() > 8 and not packet.is_keyframe:
got_key_frame = False
continue
if not got_key_frame:
continue
frames = codec.decode(packet)
if start_time is None:
start_time = time.time()
frames_buffer += frames
frames_buffer.sort(key=get_pts)
for frame in frames_buffer:
if display_frame(frame, start_time, pts_offset, codec.framerate):
frames_buffer.remove(frame)
if cv2.waitKey(1) & 0xFF == ord('q'):
break
if dts is not None:
container.seek(25000)
rawData.seek(cur_pos)
if cv2.waitKey(1) & 0xFF == ord('q'):
break
terminated.set()
cv2.destroyAllWindows()
async def receive_encoded_video(websocket, path):
data_queue = Queue()
terminated = Event()
p = Process(
target=render,
args=(terminated, data_queue)
)
p.start()
while not terminated.is_set():
try:
data = await websocket.recv()
except:
break
data_queue.put(data)
terminated.set()
正常情况下会有 3~4 秒的延迟，因为您正在读取编码数据并通过 CPU 进行解码需要时间。
任何人都可以帮助我如何组织这种方法的循环缓冲区吗？ 谢谢！