图像处理篇---MJPEG视频流处理


文章目录

  • 前言
  • 一、MJPEG流基础概念
    • MJPEG流特点
      • 格式简单
      • 无压缩时序
      • HTTP协议传输
      • 边界标记
    • 常见应用场景
      • IP摄像头视频流
      • 嵌入式设备(如ESP32)视频输出
      • 简单视频监控系统
  • 二、基础处理方法
    • 方法1:使用OpenCV直接读取
      • 优点
      • 缺点
    • 方法2:手动解析HTTP流
      • 优点
      • 缺点
  • 三、高级处理方法
    • 方法3:异步IO处理(asyncio)
      • 优点
      • 缺点
    • 方法4:使用生成器管道处理
      • 优点
      • 缺点
  • 四、专业级处理方法
    • 方法5:使用FFmpeg作为后端
      • 优点
      • 缺点
    • 方法6:使用GStreamer管道
      • 优点
      • 缺点
  • 五、特殊场景处理
    • 处理需要认证的MJPEG流
  • 六、性能优化技巧
    • 降低分辨率
    • 跳过帧处理
    • 使用多线程
    • 硬件加速解码
  • 七、总结对比


前言

MJPEG(Motion JPEG)是一种简单的视频流格式,它将视频作为一系列JPEG图像传输。下面详细介绍Python处理MJPEG流的各种方法,从基础到高级实现。


一、MJPEG流基础概念

MJPEG流特点

格式简单

格式简单:由连续的JPEG图像组成

无压缩时序

无压缩时序:每帧独立压缩,无帧间压缩

HTTP协议传输

HTTP传输:通常通过HTTP协议传输

边界标记

边界标记:每帧以\xff\xd8开始,\xff\xd9结束

常见应用场景

IP摄像头视频流

嵌入式设备(如ESP32)视频输出

简单视频监控系统

二、基础处理方法

方法1:使用OpenCV直接读取

import cv2def opencv_reader(stream_url):cap = cv2.VideoCapture(stream_url)if not cap.isOpened():print("无法打开视频流")return# 设置缓冲区减少延迟cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)while True:ret, frame = cap.read()if not ret:print("视频流中断")breakcv2.imshow('OpenCV MJPEG', frame)if cv2.waitKey(1) & 0xFF == ord('q'):breakcap.release()cv2.destroyAllWindows()使用示例
opencv_reader("http://192.168.1.100/mjpeg/1")

优点

  1. 实现简单
  2. 自动处理JPEG解码

缺点

  1. 对某些MJPEG流兼容性不好
  2. 难以处理认证和特殊头部

方法2:手动解析HTTP流

import requests
import cv2
import numpy as np
from io import BytesIOdef manual_http_reader(stream_url):session = requests.Session()stream = session.get(stream_url, stream=True)buffer = b""try:for chunk in stream.iter_content(chunk_size=1024):buffer += chunkstart = buffer.find(b'\xff\xd8')end = buffer.find(b'\xff\xd9')if start != -1 and end != -1:jpeg_data = buffer[start:end+2]buffer = buffer[end+2:]# 转换为OpenCV图像img = cv2.imdecode(np.frombuffer(jpeg_data, np.uint8), cv2.IMREAD_COLOR)if img is not None:cv2.imshow('Manual MJPEG', img)if cv2.waitKey(1) & 0xFF == ord('q'):breakfinally:stream.close()session.close()cv2.destroyAllWindows()# 使用示例
manual_http_reader("http://192.168.1.100/mjpeg/1")

优点

  1. 完全控制流处理过程
  2. 可以处理认证和特殊HTTP头

缺点

  1. 实现较复杂
  2. 需要手动处理JPEG解码

三、高级处理方法

方法3:异步IO处理(asyncio)

import aiohttp
import asyncio
import cv2
import numpy as npasync def async_mjpeg_reader(url):async with aiohttp.ClientSession() as session:async with session.get(url) as resp:buffer = b""while True:chunk = await resp.content.read(1024)if not chunk:breakbuffer += chunkstart = buffer.find(b'\xff\xd8')end = buffer.find(b'\xff\xd9')if start != -1 and end != -1:jpeg = buffer[start:end+2]buffer = buffer[end+2:]img = cv2.imdecode(np.frombuffer(jpeg, np.uint8), cv2.IMREAD_COLOR)if img is not None:cv2.imshow('Async MJPEG', img)if cv2.waitKey(1) & 0xFF == ord('q'):break# 运行示例
#asyncio.run(async_mjpeg_reader("http://192.168.1.100/mjpeg/1"))

优点

  1. 非阻塞IO,适合高性能应用
  2. 适合与其他异步任务集成

缺点

  1. 需要理解异步编程模型
  2. 与OpenCV的同步显示存在兼容问题

方法4:使用生成器管道处理

import requests
import cv2
import numpy as npdef mjpeg_stream_generator(url):session = requests.Session()stream = session.get(url, stream=True)buffer = b""try:for chunk in stream.iter_content(1024):buffer += chunkwhile True:start = buffer.find(b'\xff\xd8')end = buffer.find(b'\xff\xd9')if start == -1 or end == -1:breakjpeg = buffer[start:end+2]buffer = buffer[end+2:]yield jpegfinally:stream.close()session.close()def process_frames(generator):for jpeg in generator:img = cv2.imdecode(np.frombuffer(jpeg, np.uint8), cv2.IMREAD_COLOR)if img is not None:# 在这里添加自定义处理逻辑processed = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)cv2.imshow('Processed MJPEG', processed)if cv2.waitKey(1) & 0xFF == ord('q'):break# 使用示例
stream_gen = mjpeg_stream_generator("http://192.168.1.100/mjpeg/1")
process_frames(stream_gen)
cv2.destroyAllWindows()

优点

  1. 分离数据获取和处理逻辑
  2. 方便添加自定义处理管道
  3. 代码结构清晰

缺点

需要理解生成器概念

四、专业级处理方法

方法5:使用FFmpeg作为后端

import cv2
import subprocess
import numpy as npdef ffmpeg_reader(url, width=640, height=480):command = ['ffmpeg','-i', url,'-f', 'image2pipe','-pix_fmt', 'bgr24','-vcodec', 'rawvideo','-']pipe = subprocess.Popen(command, stdout=subprocess.PIPE, bufsize=10**8)try:while True:raw = pipe.stdout.read(width*height*3)if not raw:breakimg = np.frombuffer(raw, dtype='uint8').reshape((height, width, 3))cv2.imshow('FFmpeg MJPEG', img)if cv2.waitKey(1) & 0xFF == ord('q'):breakfinally:pipe.terminate()cv2.destroyAllWindows()# 使用示例
ffmpeg_reader("http://192.168.1.100/mjpeg/1")

优点

  1. 处理复杂流更可靠
  2. 支持更多视频格式和编码
  3. 可以添加各种FFmpeg滤镜

缺点

  1. 需要安装FFmpeg
  2. 系统资源占用较高

方法6:使用GStreamer管道

import cv2def gstreamer_reader(url):# GStreamer管道定义pipeline = (f'souphttpsrc location={url} ! ''jpegparse ! ''jpegdec ! ''videoconvert ! ''appsink emit-signals=true sync=false max-buffers=1 drop=true')cap = cv2.VideoCapture(pipeline, cv2.CAP_GSTREAMER)if not cap.isOpened():print("无法打开GStreamer管道")returnwhile True:ret, frame = cap.read()if not ret:print("读取帧失败")breakcv2.imshow('GStreamer MJPEG', frame)if cv2.waitKey(1) & 0xFF == ord('q'):breakcap.release()cv2.destroyAllWindows()# 使用示例
gstreamer_reader("http://192.168.1.100/mjpeg/1")

优点

  1. 低延迟
  2. 高度可配置的管道
  3. 良好的硬件加速支持

缺点

  1. 需要安装GStreamer
  2. 配置复杂

五、特殊场景处理

处理需要认证的MJPEG流

import requests
from requests.auth import HTTPBasicAuth
import cv2
import numpy as npdef auth_mjpeg_reader(url, username, password):session = requests.Session()stream = session.get(url, stream=True, auth=HTTPBasicAuth(username, password))buffer = b""try:for chunk in stream.iter_content(1024):buffer += chunkstart = buffer.find(b'\xff\xd8')end = buffer.find(b'\xff\xd9')if start != -1 and end != -1:jpeg = buffer[start:end+2]buffer = buffer[end+2:]img = cv2.imdecode(np.frombuffer(jpeg, np.uint8), cv2.IMREAD_COLOR)if img is not None:cv2.imshow('Auth MJPEG', img)if cv2.waitKey(1) & 0xFF == ord('q'):breakfinally:stream.close()session.close()cv2.destroyAllWindows()# 使用示例
auth_mjpeg_reader("http://192.168.1.100/mjpeg/1", "admin", "password")
  1. 处理不稳定的MJPEG流
import requests
import time
import cv2
import numpy as npdef robust_mjpeg_reader(url, max_retries=5, retry_delay=1):retry_count = 0while retry_count < max_retries:try:session = requests.Session()stream = session.get(url, stream=True, timeout=5)buffer = b""for chunk in stream.iter_content(1024):buffer += chunkstart = buffer.find(b'\xff\xd8')end = buffer.find(b'\xff\xd9')if start != -1 and end != -1:jpeg = buffer[start:end+2]buffer = buffer[end+2:]img = cv2.imdecode(np.frombuffer(jpeg, np.uint8), cv2.IMREAD_COLOR)if img is not None:cv2.imshow('Robust MJPEG', img)if cv2.waitKey(1) & 0xFF == ord('q'):stream.close()session.close()cv2.destroyAllWindows()returnretry_count = 0  # 重置重试计数except Exception as e:print(f"发生错误: {e}, 尝试重新连接...")retry_count += 1time.sleep(retry_delay)finally:if 'stream' in locals():stream.close()if 'session' in locals():session.close()print("达到最大重试次数,退出")cv2.destroyAllWindows()# 使用示例
robust_mjpeg_reader("http://192.168.1.100/mjpeg/1")

六、性能优化技巧

降低分辨率

# 对于手动解析的方法
img = cv2.imdecode(..., cv2.IMREAD_COLOR)
img = cv2.resize(img, (320, 240))

跳过帧处理

frame_counter = 0
frame_skip = 2  # 每3帧处理1帧for jpeg in generator:frame_counter += 1if frame_counter % (frame_skip + 1) != 0:continue# 处理帧...

使用多线程

from threading import Thread
from queue import Queueclass MJPEGBackgroundReader:def __init__(self, url, max_queue=5):self.url = urlself.queue = Queue(maxsize=max_queue)self.stop_event = Falseself.thread = Thread(target=self._reader_thread)self.thread.daemon = Truedef start(self):self.thread.start()return selfdef _reader_thread(self):# 实现读取逻辑填充队列passdef read(self):return self.queue.get()def stop(self):self.stop_event = Trueself.thread.join()

硬件加速解码

# 使用OpenCV的CUDA加速
img = cv2.imdecode(..., cv2.IMREAD_COLOR)
gpu_img = cv2.cuda_GpuMat()
gpu_img.upload(img)
# 在GPU上处理图像...

七、总结对比

方法 适用场景 优点 缺点
OpenCV直接读取 快速原型开发 简单易用 兼容性问题
手动HTTP解析 需要完全控制 灵活可靠 实现复杂
异步IO 高并发应用 非阻塞高效 异步编程难度
生成器管道 复杂处理流程 结构清晰 需要理解生成器
FFmpeg后端 专业级应用 强大可靠 依赖FFmpeg
GStreamer 低延迟需求 高性能 配置复杂

选择合适的方法取决于你的具体需求:
快速测试:OpenCV直接读取
生产环境:手动HTTP解析或FFmpeg
高性能需求:GStreamer或异步IO
复杂处理:生成器管道
所有方法都可以根据需要进行组合和扩展,构建适合自己项目的MJPEG处理解决方案。


本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mfbz.cn/a/240.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

【算法-哈希表】常见算法题的哈希表套路拆解

算法相关知识点可以通过点击以下链接进行学习一起加油&#xff01;双指针滑动窗口二分查找前缀和位运算模拟链表 在刷题的过程中&#xff0c;我们会频繁遇到一些“高频套路”——而哈希表正是其中最常用也最高效的工具之一。它能帮助我们在 O(1) 的时间复杂度内完成查找、插入与…

异地多活单元化架构下的微服务体系

治理服务间的跨IDC调用&#xff0c;而数据库层面还是要跨IDC 服务注册中心拆开、 金融要求&#xff0c;距离太远&#xff0c;异地备库&#xff0c;如果延迟没读到数据就可能有资损&#xff0c;IDC3平时不能用&#xff0c;IDC1挂了还是有数据同步问题&#xff0c;IDC3日常维护…

python打卡day20

特征降维------特征组合&#xff08;以SVD为例&#xff09; 知识点回顾&#xff1a; 奇异值的应用&#xff1a; 特征降维&#xff1a;对高维数据减小计算量、可视化数据重构&#xff1a;比如重构信号、重构图像&#xff08;可以实现有损压缩&#xff0c;k 越小压缩率越高&#…

【C/C++】范围for循环

&#x1f4d8; C 范围 for 循环详解&#xff08;Range-based for loop&#xff09; 一、什么是范围 for 循环&#xff1f; 范围 for 循环&#xff08;Range-based for loop&#xff09; 是 C11 引入的一种简化容器/数组遍历的方式。它通过自动调用容器的 begin() 和 end() 方法…

MySQL 与 Elasticsearch 数据一致性方案

MySQL 与 Elasticsearch 数据一致性方案 前言一、同步双写&#xff08;Synchronous Dual Write&#xff09;&#x1f504;二、异步双写&#xff08;Asynchronous Dual Write&#xff09;&#x1f4e4;三、定时同步&#xff08;Scheduled Synchronization&#xff09;&#x1f5…

【相机标定】OpenCV 相机标定中的重投影误差与角点三维坐标计算详解

摘要&#xff1a; 本文将从以下几个方面展开&#xff0c;结合典型代码深入解析 OpenCV 中的相机标定过程&#xff0c;重点阐述重投影误差的计算方法与实际意义&#xff0c;并通过一个 calcBoardCornerPositions() 函数详细讲解棋盘格角点三维坐标的构建逻辑。 在计算机视觉领域…

【MySQL】联合查询

个人主页&#xff1a;♡喜欢做梦 欢迎 &#x1f44d;点赞 ➕关注 ❤️收藏 &#x1f4ac;评论 目录 一、什么是联合查询 1.概念 2.语法要求 3.示例 4.为什么要使用联合查询 内连接 1.概念 2.语法 3.步骤&#xff1a; 外连接 1.概念 2.分类&#xff1a; 左外连…

如何从极狐GitLab 容器镜像库中删除容器镜像?

极狐GitLab 是 GitLab 在中国的发行版&#xff0c;关于中文参考文档和资料有&#xff1a; 极狐GitLab 中文文档极狐GitLab 中文论坛极狐GitLab 官网 从容器镜像库中删除容器镜像 (BASIC ALL) 您可以从您的容器镜像库中删除容器镜像。 要基于特定标准自动删除容器镜像&#x…

【Git】【commit】查看未推送的提交查看指定commit的修改内容合并不连续的commit

文章目录 1. 查看未推送的提交方法一 &#xff1a;git status方法二&#xff1a;git log方法三&#xff1a;git cherry方法四&#xff1a;git rev-list 2. 查看指定commit的修改方法一&#xff1a;git show方法二&#xff1a;git log方法三&#xff1a;git diff 3. 合并不连续的…

神经网络—感知器、多层感知器

文章目录 前言一、生物神经元与感知器的类比二、感知器1、简单感知器2、多层感知器&#xff08;1&#xff09;多层感知机结构 3、神经网络结构 总结1、感知器的局限性如何突破感知器的局限性&#xff1f; 2、感知器的应用 前言 感知器&#xff08;Perceptron&#xff09;是神经…

C++:扫雷游戏

一.扫雷游戏项目设计 1.文件结构设计 首先我们要先定义三个文件 ①test.c //文件中写游戏的测试逻辑 ②game.c //文件中写游戏中函数的实现等 ③game.h //文件中写游戏需要的数据类型和函数声明等 2.扫雷游戏的主体结构 使⽤控制台实现经典的扫雷游戏 •游戏可以通过菜单…

k8s的pod挂载共享内存

k8s的pod挂载共享内存&#xff0c;限制不生效问题&#xff1a; 注&#xff1a;/dev/shm 是 Linux 系统中用于共享内存的特殊路径。通过将 emptyDir 的 medium 设置为 Memory&#xff0c;可以确保 /dev/shm 正确地挂载到一个基于内存的文件系统&#xff0c;从而实现高效的共享内…