ffmedia的拉流+推流测试
#!/usr/bin/env python3import ff_pymedia as m
def align(x, a):
return (x + a - 1) & ~(a - 1)
def test_ffmedia(rtsp_urls, rtmp_urls):
input_source = None
try:
# 视频流拉流，设置参数
# rtsp流的地址
# rtsp流的传输协议，默认为UDP流-0，TCP流-1
# 是否拉取视频流，默认为真
# 是否拉取音频流，默认为假
input_source = m.ModuleRtspClient(rtsp_urls, m.RTSP_STREAM_TYPE(0), True, False)
print(f"input_source.videoFPS: {input_source.videoFPS}")
ret = input_source.init()
last_module = input_source
if ret < 0:
print("input_source init failed")
exit(-1)
# 视频流解码
input_para = last_module.getOutputImagePara()
# print(f"last_module.videoFPS: {last_module.videoFPS()}")
print(
f"input_para.v4l2Fmt: {input_para.v4l2Fmt}, {m.v4l2GetFmtByName('H264')}, {m.v4l2GetFmtByName('BGR24')}, {m.v4l2GetFmtByName('NV12')}")
print(f"input_para.width: {input_para.width}")
print(f"input_para.height: {input_para.height}")
if input_para.v4l2Fmt == m.v4l2GetFmtByName("H264") or \
input_para.v4l2Fmt == m.v4l2GetFmtByName("MJPEG") or \
input_para.v4l2Fmt == m.v4l2GetFmtByName("H265"):
print("加入解码...")
dec = m.ModuleMppDec()
dec.setProductor(last_module)
ret = dec.init()
if ret < 0:
print("dec init failed")
exit(-1)
input_para = dec.getOutputImagePara()
dec.setOutputImagePara(
m.ImagePara(input_para.width, input_para.height, input_para.hstride, input_para.vstride,
m.v4l2GetFmtByName('BGR24')))
last_module = dec
# # 输出内容正常
# input_para = last_module.getOutputImagePara()
# drm_display = m.ModuleDrmDisplay()
# drm_display.setPlanePara(m.v4l2GetFmtByName("BGR24"), 0,
# m.PLANE_TYPE.PLANE_TYPE_OVERLAY_OR_PRIMARY, 0xff, 1, 0)
# drm_display.setProductor(last_module)
# ret = drm_display.init()
# if ret < 0:
# print("drm display init failed")
# return 1
# else:
# t_h = drm_display.getDisplayPlaneH()
# t_w = drm_display.getDisplayPlaneW()
# w = min(t_w, input_para.width)
# h = min(t_h, input_para.height)
# x = (t_w - w) // 2
# y = (t_h - h) // 2
# drm_display.setWindowSize(x, y, w, h)
# 编码参数重定
print("成功加入解码")
# 视频流解码
print("加入编码")
enc = m.ModuleMppEnc(m.EncodeType.ENCODE_TYPE_H264)
enc.setProductor(last_module)
enc.setBufferCount(16)
enc.setDuration(0)# Use the input source timestamp
ret = enc.init()
if ret < 0:
print("ModuleMppEnc init failed")
exit(-1)
last_module = enc
print("成功加入编码")
# 解码后输出的内容就坏掉了！！！
# 视频流推流
push_c = m.ModuleRtmpClient(rtmp_urls, m.ImagePara(), 0)
push_c.setProductor(last_module)
ret = push_c.init()
if ret < 0:
print("Fail to init rtmp client push")
exit(-1)
# 启动拉流
input_source.start()
# 管道日志
input_source.dumpPipe()
# 等待用户输入
text = input("wait...")
# 打印管道日志
input_source.dumpPipeSummary()
# 停止拉流
input_source.stop()
except Exception as e:
print(f"异常: {e}")
finally:
pass
# 使用示例
if __name__ == "__main__":
# 两路RTSP流地址
rtsp_urls = [
"rtsp://xxxx:xxxx@192.168.8.83:554/h264/ch1/main/av_stream",
"rtsp://xxxx:xxxx@192.168.8.211:554/h264/ch1/main/av_stream"
]
# 两路RTMP推流地址
rtmp_urls = [
"rtmp://172.16.110.202:1935/live/origin_185384157d27690c9bb5ae5c2d65d753",
"rtmp://172.16.110.202:1935/live/origin_c568b8a53373a02905ce5d1fb1152b16"
]
test_ffmedia(rtsp_urls, rtmp_urls)
未报错~个人觉得是编码有问题~~~demo.py上面抄下来自己写的demo。
上述是测试案例，发现经过ModuleMppEnc编码后 流数据显示就不正常了，麻烦测试~已尝试各种办法 无果，请指教
发一下运行log看看 dengkx 发表于 2025-9-16 11:50
发一下运行log看看
请查看附件~
页:
[1]