如何用Python读取带时间戳元数据的Amazon IVS HLS流？

一、WebSocket连接错误的核心原因 #

• 你使用的STREAM_WS是HLS播放列表URL（.m3u8），这不是WebSocket服务端点，服务器返回HTTP 200是因为它是合法的HTTP资源，但不支持WebSocket协议，自然会拒绝连接。
• IVS的定时元数据推送需要使用播放器WebSocket URL，格式为wss://<ivs-domain>/live/<stream-id>，可通过两种方式获取：
  • 从IVS控制台的流详情页，在「播放设置」中找到「WebSocket播放URL」；
  • 用AWS SDK（如boto3）调用get_stream接口，获取playbackUrl后将https替换为wss，并移除末尾的.m3u8后缀。

二、正确读取定时元数据的Python代码 #

IVS的元数据订阅协议并非你使用的格式，正确的订阅消息需指定type和topics字段，以下是可运行的示例：

import asyncio
import websockets
import json

# 替换为你的IVS WebSocket播放URL（wss开头，无.m3u8后缀）
STREAM_WS = "wss://your-ivs-domain/live/your-stream-key"

async def fetch_ivs_metadata():
    try:
        async with websockets.connect(STREAM_WS) as websocket:
            # 发送订阅元数据的指令
            await websocket.send(json.dumps({"type": "subscribe", "topics": ["metadata"]}))
            print("已成功订阅元数据事件")

            while True:
                raw_msg = await websocket.recv()
                msg_data = json.loads(raw_msg)
                # 过滤出定时元数据事件
                if msg_data.get("type") == "message" and msg_data.get("topic") == "metadata":
                    metadata_payload = msg_data["payload"]
                    print(f"收到定时元数据: {metadata_payload}")
                    # 提取时间戳：metadata_payload.get("timestamp")
    except Exception as e:
        print(f"连接/读取错误: {str(e)}")

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

三、解决OpenCV读取流的高延迟问题 #

1. 启用IVS低延迟模式 #

在IVS控制台创建或修改流的播放配置，选择「低延迟」模式，可将延迟从标准HLS的20+秒降至2-5秒。

2. 优化OpenCV的缓存设置 #

直接用OpenCV读取流时，调整缓存大小减少延迟：

import cv2

cap = cv2.VideoCapture("YOUR_LOW_LATENCY_HLS_URL")
# 设置缓存帧数为1，减少缓存延迟
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)

while cap.isOpened():
    ret, frame = cap.read()
    if not ret:
        break
    # 处理帧
    cv2.imshow("IVS Stream", frame)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()

3. 用FFmpeg管道读取（更精细控制延迟） #

如果OpenCV的原生读取仍有延迟，可通过FFmpeg管道读取流，强制关闭缓存：

import cv2
import subprocess as sp
import numpy as np

# 替换为你的低延迟HLS URL
HLS_URL = "https://your-ivs-domain/live/your-stream-key/index.m3u8"
# 根据你的流分辨率调整宽高
WIDTH, HEIGHT = 1920, 1080

ffmpeg_cmd = [
    "ffmpeg",
    "-i", HLS_URL,
    "-fflags", "nobuffer",
    "-flags", "low_delay",
    "-framerate", "30",
    "-pix_fmt", "bgr24",
    "-vcodec", "rawvideo",
    "-an", "-f", "rawvideo", "pipe:"
]

pipe = sp.Popen(ffmpeg_cmd, stdout=sp.PIPE, bufsize=10**8)

while True:
    # 读取一帧原始数据
    raw_frame = pipe.stdout.read(WIDTH * HEIGHT * 3)
    if not raw_frame:
        break
    # 转换为OpenCV可用的帧格式
    frame = np.frombuffer(raw_frame, dtype=np.uint8).reshape((HEIGHT, WIDTH, 3))
    # 处理帧
    cv2.imshow("Low Latency Stream", frame)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

pipe.terminate()
cv2.destroyAllWindows()

四、额外注意事项 #

• 确保推流时已嵌入定时元数据：比如用FFmpeg推流时添加-metadata:s:v:0 timed_metadata="your-custom-data"参数，否则IVS不会推送元数据事件；
• 私有流需使用签名URL：如果你的IVS流是私有访问，WebSocket和HLS URL都需要生成带签名的临时URL，可通过AWS SDK生成。

内容的提问来源于stack exchange，提问作者JFCorleone