基于OpenCv+Django的网络实时视频流传输（前后端分离）
生活随笔
收集整理的這篇文章主要介紹了
基于OpenCv+Django的网络实时视频流传输（前后端分离）
小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.
秋風閣——北溪入江流：https://focus-wind.com/
秋風閣——基于OpenCv+Django的網(wǎng)絡實時視頻流傳輸（前后端分離）
使用OpenCv捕獲攝像機畫面后，我們有時候需要將畫面顯示在界面上。本博客基于Django的前后端分離模式，將視頻流從后端讀取，傳送給前端顯示。
Django流傳輸實例：StreamingHttpResponse
在使用Django進行視頻流傳輸時，無法使用HttpResponse，JsonResponse等對象對內容直接傳輸，需要使用StreamingHttpResponse流式傳輸一個響應給瀏覽器。StreamingHttpResponse不是HttpResponse的子類，因此他們之間的API略有不同。StreamingHttpResponse與HttpResponse之間有以下顯著區(qū)別：
- 應該給StreamingHttpResponse一個迭代器，產(chǎn)生字節(jié)字符串作為內容。
- 不應該直接訪問StreamingHttpResponse的內容，除非通過迭代器響應對象本身。
- StreamingHttpResponse沒有content屬性。相反，他有一個streaming_content屬性。
- 無法使用類文件對象的tell()何write()方法。這樣會引起一個異常。
Django傳輸視頻流
因為使用Django的StreamingHttpResponse類進行流傳輸，所以我們首先需要生成一個視頻流的迭代器，在迭代器中，需要將從opencv中獲取到的numpy.ndarray三維數(shù)組轉換為字節(jié)類型的，然后傳輸?shù)角岸恕?/p>
傳輸視頻流：
在使用海康威視等分辨率較高的相機時，直接解碼，延遲過高，所以需要先對圖片進行壓縮，然后解碼。
經(jīng)測試，海康相機使用0.25的壓縮倍率顯示壓縮效率較好，當大于0.25時，延遲較高，小于0.25時，界面顯示較差
迭代器優(yōu)化：
def gen_display(camera):"""視頻流生成器功能。"""while True:# 讀取圖片ret, frame = camera.read()if ret:frame = cv2.resize(frame, (0, 0), fx=0.25, fy=0.25)# 將圖片進行解碼ret, frame = cv2.imencode('.jpeg', frame)if ret:# 轉換為byte類型的，存儲在迭代器中yield (b'--frame\r\n'b'Content-Type: image/jpeg\r\n\r\n' + frame.tobytes() + b'\r\n')前端顯示視頻流
在Django中配置路由后，在瀏覽器端直接訪問視頻url即可看到視頻顯示畫面。
在前端HTML5中，將視頻路由寫入img標簽的src屬性中，即可訪問視頻流界面。例如：<img src=‘https://ip:port/uri’
前端顯示視頻流：
顯示結果：
在前端顯示視頻流中，可以通過調整img標簽的屬性來調整界面顯示位置，顯示大小。所以在進行視頻流前后端傳輸中，在保證視頻顯示清晰度的情況下，建議使用前端來調整界面大小。
調整界面前端顯示視頻樣式：
<!DOCTYPE html> <html lang="en"> <head><meta charset="UTF-8"><meta http-equiv="X-UA-Compatible" content="IE=edge"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>基于OpenCv+Django的網(wǎng)絡實時視頻流傳輸（前后端分離）</title><style>#video {width: 500px;height: 500px;}</style> </head> <body><!-- 顯示視頻流 --><div align="center"><img src="http://127.0.0.1:8000/api/cv/display" id="video"></div> </body> </html>顯示結果：
視頻流傳輸優(yōu)化
在項目中，我們可能經(jīng)常需要對多個相機進行處理，而不是對一個相機進行操作，所以我們可以使用相機工廠來獲取相機。在實例化相機后，需要開啟一個線程，及時更新緩存隊列，確保OpenCv不會因為緩存過多而造成緩存區(qū)堵塞，界面延遲。
- 使用線程實時讀取OpenCv的內容到隊列中
- 使用相機工廠來獲取相機
在示例代碼中，camera_model為自定義model，其中代碼需要用到的數(shù)據(jù)有數(shù)據(jù)表記錄的唯一標識id，相機的訪問api：camera_api
相機類：
import queue import threadingimport cv2from apps.device.models import Cameraclass CameraException(Exception):message = None# 初始化異常def __init__(self, message: str):# 初始化異常，定位異常信息描述self.message = messagedef __str__(self):return self.messageclass BaseCamera:# 相機操作對象cam = None# 保存每一幀從rtsp流中讀取到的畫面，使用opencv讀取，為BGR圖片queue_image = queue.Queue(maxsize=10)# 后臺取幀線程thread = None# 相機Modelcamera_model = None# 相機基類def __init__(self, camera_model: Camera):"""使用rtsp流初始化相機參數(shù)rtsp格式：rtsp://[username]:[password]@[ip]:[port]/[codec]/[channel]/[subtype]/av_streamusername: 用戶名。例如admin。password: 密碼。例如12345。ip: 為設備IP。例如 192.0.0.64。port: 端口號默認為554，若為默認可不填寫。codec：有h264、MPEG-4、mpeg4這幾種。channel: 通道號，起始為1。例如通道1，則為ch1。subtype: 碼流類型，主碼流為main，輔碼流為sub。"""self.cam = cv2.VideoCapture(camera_model.camera_api)if self.cam.isOpened():# 相機打開成功,啟動線程讀取數(shù)據(jù)self.thread = threading.Thread(target=self._thread, daemon=True)self.thread.start()else:# 打開失敗，相機流錯誤raise CameraException("視頻流接口訪問失敗")def _thread(self):"""相機后臺進程，持續(xù)讀取相機opencv讀取時會將信息存儲到緩存區(qū)里，處理速度小于緩存區(qū)速度，會導致資源積累"""# 線程一直讀取視頻流，將最新的視頻流存在隊列中while self.cam.isOpened():ret, img = self.cam.read()if not ret or img is None:# 讀取相機失敗passelse:# 讀取內容成功，將數(shù)據(jù)存放在緩存區(qū)if self.queue_image.full():# 隊列滿，隊頭出隊self.queue_image.get()# 隊尾添加數(shù)據(jù)self.queue_image.put(img)else:# 隊尾添加數(shù)據(jù)self.queue_image.put(img)# 直接讀取圖片def read(self):"""直接讀取從rtsp流中獲取到的圖片，不進行額外加工可能為空，需做判空處理"""return self.queue_image.get()# 讀取視頻幀def get_frame(self):"""獲取加工后的圖片，可以直接返回給前端顯示"""img = self.queue_image.get()if img is None:return Noneelse:# 壓縮圖片，否則圖片過大，編碼效率慢，視頻延遲過高img = cv2.resize(img, (0, 0), fx=0.25, fy=0.25)# 對圖片進行編碼ret, jpeg = cv2.imencode('.jpeg', img)return jpeg.tobytes()class CameraFactory:"""相機工廠"""# 存儲實例化的所有相機cameras = {}@classmethoddef get_camera(cls, camera_id: int):# 通過相機id獲取相機camera = cls.cameras.get(camera_id)if camera is None:# 查看是否存在相機，存在訪問try:camera_model = Camera.objects.get(id=camera_id)base_camera = BaseCamera(camera_model=camera_model)if base_camera is not None:cls.cameras.setdefault(camera_id, base_camera)return cls.cameras.get(camera_id)else:return Noneexcept Camera.DoesNotExist:# 相機不存在return Noneexcept CameraException:# 相機實例失敗return Noneelse:# 存在相機，直接返回return cameraDjango views.py：
from django.http import StreamingHttpResponsefrom apps.device.Camera import CameraFactory, BaseCameradef gen_display(camera: BaseCamera):"""視頻流生成器功能。"""while True:# 讀取圖片frame = camera.get_frame()if frame is not None:yield (b'--frame\r\n'b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')def video(request):"""視頻流路由。將其放入img標記的src屬性中。例如：<img src='https://ip:port/uri' >"""# 視頻流相機對象camera_id = request.GET.get('camera_id')camera: BaseCamera = CameraFactory.get_camera(camera_id)# 使用流傳輸傳輸視頻流return StreamingHttpResponse(gen_display(camera), content_type='multipart/x-mixed-replace; boundary=frame')總結
以上是生活随笔為你收集整理的基于OpenCv+Django的网络实时视频流传输（前后端分离）的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 谷歌浏览器Chrome播放rtsp实时视
- 下一篇： vba 添加outlook 签名_如何在