基于Python FastAPI集成NDI SDK实现PTZ摄像头直播的技术咨询

一、环境准备 #

• 安装NDI SDK：下载对应系统（Windows/macOS/Linux）的官方SDK并安装，确保系统能识别NDI设备。
• 安装Python依赖：

  pip install ndi-python fastapi uvicorn opencv-python numpy
  

  • ndi-python：官方维护的Python NDI绑定，提供核心NDI功能
  • fastapi/uvicorn：后端框架及ASGI服务器
  • opencv-python：处理视频帧编码与格式转换
  • numpy：辅助帧数据结构转换

二、核心NDI操作实现 #

1. 发现局域网内的NDI源（PTZ摄像头） #

扫描并列出网络中的NDI设备，找到目标PTZ摄像头：

import ndi

# 初始化NDI查找器
finder = ndi.find_create_v2()
if not finder:
    raise RuntimeError("无法初始化NDI查找器")

# 等待5秒查找NDI源
sources = ndi.find_wait_for_sources(finder, 5000)
print(f"找到NDI源数量: {len(sources)}")
for idx, source in enumerate(sources):
    print(f"{idx+1}. {source.ndi_name} | {source.url_address}")

# 释放资源
ndi.find_destroy(finder)

2. 接收NDI视频流 #

连接到指定NDI源并接收视频帧：

import ndi
import cv2
import numpy as np

# 配置NDI接收器格式（匹配OpenCV的BGR）
recv_settings = ndi.RecvCreateV3()
recv_settings.color_format = ndi.RECV_COLOR_FORMAT_BGRX
recv = ndi.recv_create_v3(recv_settings)
if not recv:
    raise RuntimeError("无法创建NDI接收器")

# 连接到目标PTZ摄像头（替换为实际NDI名称）
target_source = ndi.Source()
target_source.ndi_name = "你的PTZ摄像头NDI名称"
ndi.recv_connect(recv, target_source)

# 循环接收并显示帧
while True:
    frame_type, video_frame, _, _ = ndi.recv_capture_v2(recv, 100)
    
    if frame_type == ndi.FRAME_TYPE_VIDEO:
        # 转换NDI帧为OpenCV可用数组
        frame = np.array(video_frame.data, dtype=np.uint8).reshape(video_frame.yres, video_frame.xres, 4)[:, :, :3]
        
        # 此处可添加帧处理、PTZ控制逻辑
        cv2.imshow("NDI Live Feed", frame)
        
        # 释放帧资源
        ndi.recv_free_video_v2(recv, video_frame)
    
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

# 清理资源
ndi.recv_destroy(recv)
cv2.destroyAllWindows()

三、结合FastAPI实现HTTP直播 #

将NDI流转换为HTTP可访问的MJPEG直播流（低延迟场景适用）：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import ndi
import cv2
import numpy as np

app = FastAPI()
recv = None  # 全局NDI接收器实例

def init_ndi_receiver(source_name: str):
    """初始化NDI接收器"""
    global recv
    recv_settings = ndi.RecvCreateV3()
    recv_settings.color_format = ndi.RECV_COLOR_FORMAT_BGRX
    recv = ndi.recv_create_v3(recv_settings)
    target_source = ndi.Source()
    target_source.ndi_name = source_name
    ndi.recv_connect(recv, target_source)

def generate_frames():
    """生成MJPEG流帧数据"""
    while True:
        frame_type, video_frame, _, _ = ndi.recv_capture_v2(recv, 100)
        if frame_type == ndi.FRAME_TYPE_VIDEO:
            # 转换帧格式并编码为JPEG
            frame = np.array(video_frame.data, dtype=np.uint8).reshape(video_frame.yres, video_frame.xres, 4)[:, :, :3]
            ret, buffer = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 80])
            if ret:
                yield (b'--frame\r\n'
                       b'Content-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n')
            ndi.recv_free_video_v2(recv, video_frame)

@app.on_event("startup")
def startup_event():
    """服务启动时初始化NDI连接"""
    init_ndi_receiver("你的PTZ摄像头NDI名称")

@app.on_event("shutdown")
def shutdown_event():
    """服务关闭时清理资源"""
    if recv:
        ndi.recv_destroy(recv)

@app.get("/live")
def live_stream():
    """返回MJPEG直播流"""
    return StreamingResponse(generate_frames(), media_type="multipart/x-mixed-replace; boundary=frame")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

启动服务后，访问http://localhost:8000/live即可在浏览器或视频播放器中查看直播。

四、PTZ控制补充 #

若摄像头支持NDI控制，可通过元数据发送PTZ指令（具体格式需参考摄像头文档）：

# 示例PTZ控制命令
ptz_command = """<ndi_ptz>
    <pan>1.0</pan>   # 水平转动，范围-1.0到1.0
    <tilt>-0.5</tilt> # 垂直转动，范围-1.0到1.0
    <zoom>2.0</zoom>  # 变焦，根据设备支持范围调整
</ndi_ptz>"""

# 发送控制指令到NDI源
ndi.recv_send_metadata(recv, ptz_command)

若摄像头不支持NDI控制，可结合onvif-zeep包实现ONVIF协议控制，与NDI流接收逻辑并行即可。

五、注意事项 #

• 确保NDI设备与后端服务在同一局域网，或配置NDI网桥实现跨网络访问
• 高并发场景下，建议用多线程/异步处理帧捕获与编码，避免阻塞FastAPI事件循环
• 大带宽需求可切换到HLS/DASH协议，可使用ffmpeg-python进行转码

内容的提问来源于stack exchange，提问作者kanagavignesh psts-sg