Python實(shí)現(xiàn)多路視頻多窗口播放功能

更新時(shí)間：2025年02月09日 11:43:53 作者：c+猿輔導(dǎo)

這篇文章主要為大家詳細(xì)介紹了Python實(shí)現(xiàn)多路視頻多窗口播放功能的相關(guān)知識(shí),文中的示例代碼講解詳細(xì),有需要的小伙伴可以跟隨小編一起學(xué)習(xí)一下

一、python實(shí)現(xiàn)多路視頻播放功能

服務(wù)端開(kāi)發(fā)后通常需要做功能測(cè)試、性能測(cè)試，通常postman、curl等作功能測(cè)試使用，長(zhǎng)跑服務(wù)性能postman并不太適合，如用c++實(shí)現(xiàn)播放器功能太慢，效率太低效，本文介紹一種用python來(lái)實(shí)現(xiàn)多路視頻播放的測(cè)試。

二、代碼實(shí)現(xiàn)

http申請(qǐng)視頻流地址并cv2播放功能

import json
import requests
import time
import threading
from threading import Thread
import signal
import base64
from io import BytesIO
import queue
import random
import sys
from openpyxl import load_workbook
import json
import cv2
import datetime
import os
import shutil
import numpy as np
requests.packages.urllib3.disable_warnings()


group_id_base = 31000000452158000001
default_group = 31000000452168000002
username = "admin"
password = "admin123"
sever_ip = "1237.0.0.1"
my_token = "D21DCD7B-9380-CC90-7DA4-673BDE3BF2CF"

def allocToken(sever_ip):
    url = '%s/micplatform/vuds/allocToken' % (sever_ip)
    print(url)
    headers = {
        "Content-Type": "application/json",
    }
    data = {
        "validateMethod": "name+password",
        "username": str(username),
        "userpassword": str(password),
        "refreshInterval": 3600,
        "requestid": "1"
    }
    response = requests.post(url, headers=headers, verify=False, data=json.dumps(data))
    print(response.text)
    resp = json.loads(response.text)
    print( "url "+ url +" token: " + str(resp["token"]))
    return resp["token"]
def AllocStream(deviceId):
    url = '%s/micplatform/vmd/realplayUrlAlloc' %(sever_ip)
    headers = {
        "Content-Type": "application/json",
    }
    data = {
        "requestid": "1",
        "token": my_token,
        "deviceid": deviceId,
        "protocol": "http+flv"
    }
    try:
        response = requests.post(url, headers=headers, verify=False, data=json.dumps(data))
        response.raise_for_status()  # Check for HTTP errors
        try:
            result = response.json()
            if "playurl" in result:
                print(f"{deviceId} 申請(qǐng)碼流成功 {result['playurl']}")
                return result['playurl']
            else:
                print(f"{deviceId} 申請(qǐng)碼流失敗 {response.json()}" + my_token)
                return ""
        except json.JSONDecodeError:
            print(f"{deviceId} 響應(yīng)解析失?。簾o(wú)法解析JSON")
            return ""
    except requests.RequestException as e:
        print(f"{deviceId} 請(qǐng)求失?。簕e}")
        return ""

def FreeStream(playurl):
    url = '%s/micplatform/vmd/realplayUrlRelease' %(sever_ip)
    headers = {
        "Content-Type": "application/json",
    }
    data = {
        "requestid": "1",
        "token": my_token,
        "playurl": playurl
    }
    try:
        response = requests.post(url, headers=headers, verify=False, data=json.dumps(data))
        response.raise_for_status()  # Check for HTTP errors
        try:
            result = response.json()
            if "resultDesc" in result and result['resultDesc'] == "成功":
                print(f"{playurl} 釋放碼流成功")
            else:
                print(f"{playurl} 申請(qǐng)碼流失敗")
        except json.JSONDecodeError:
            print(f"{playurl} 響應(yīng)解析失?。簾o(wú)法解析JSON" )
    except requests.RequestException as e:
        print(f"{playurl} 請(qǐng)求失?。簕e}")


def openVideo(streamtype,deviceid,stop_event):
    playurl = AllocStream(deviceid)
    if len(playurl) == 0:
        return

    if len(playurl) != 0:
        playurl_array.append(playurl)
    if (streamtype == 3 or streamtype == 2):
        cap = cv2.VideoCapture(playurl)
    else:
        cap = cv2.VideoCapture(0)
    while (not stop_event.is_set()):
        ret, frame = cap.read()  # get a frame
        if ret == True:
            # showdate = str(datetime.datetime.now())
            # font = cv2.FONT_HERSHEY_SIMPLEX
            # frame = cv2.putText(frame, showdate, (10, 100), font, 0.5, (0, 255, 255), 2, cv2.LINE_AA)
            cv2.imshow(deviceid, frame)  # show a frame
            if cv2.waitKey(1) & 0xFF == ord('q'):
                print("deviceid "+ deviceid + " receive the stop command")
                stop_event.set()  # 設(shè)置事件，通知其他線(xiàn)程停止
                break
        else:
            break
    cap.release()
    # 如果你的程序在退出時(shí)沒(méi)有正確關(guān)閉所有OpenCV窗口，那么可能是因?yàn)閏v2.destroyAllWindows()沒(méi)有在主線(xiàn)程中被調(diào)用。在所有線(xiàn)程結(jié)束后，確保在主線(xiàn)程中調(diào)用
    # cv2.destroyAllWindows()

device_datas = []
stop_event = threading.Event()  # 創(chuàng)建一個(gè)事件對(duì)象
def AllocStreamTaskByExcelFile(xlsfile):
    workbook = load_workbook(xlsfile)
    sheet = workbook.active
    for row in sheet.iter_rows(values_only=True):
        if len(row[0]) != 20 :
            continue
        if row[4] != "ON":
            continue
        print("insert davice_data id: ",row[0])
        device_datas.append(row[0])
    process_array_in_threads(device_datas)

playurl_array = []
def process_array_in_threads(device_datas):
    print(device_datas)
    start_time = time.time()
    threads = []
    for deviceid in device_datas:
        thread = threading.Thread(target=openVideo, args=(3,deviceid,stop_event))
        threads.append(thread)
        thread.start()
        print("start task stream "+ deviceid)
    for thread in threads:
        thread.join()

    cv2.destroyAllWindows()
    for playurl in playurl_array:
        FreeStream(playurl)
    playurl_array.clear()

if __name__ == '__main__':
    try:
        file_path = "config.ini"
        with open(file_path, 'r') as f:
            config = json.load(f)

        if 'serverUrl' in config:
            sever_ip = config['serverUrl']
        else:
            sever_ip = ""
            print("Please check serverUrl fielddata in config.ini")

        if 'user' in config:
            username = config['user']
        else:
            username = ""
            print("Please check username fielddata in config.ini")

        if 'password' in config:
            password = config['password']
        else:
            password = ""
            print("Please check password fielddata in config.ini")

    except FileNotFoundError:
        print(f"{file_path} does not exist")
        exit(0)
    except IOError:
        print(f"{file_path} exists but is not readable")
        exit(0)
    except json.JSONDecodeError:
        print(f"{file_path} is not a valid JSON file")
        exit(0)

    # 配置文件格式檢查
    if not sever_ip or not username or not password:
        print("Config file is missing required fields. Please check serverUrl, user, password.")
        exit(0)

    print("*************************************")
    print("get server url in config.ini: " + sever_ip)
    print("get user in config.ini: " + username)
    print("get password in config.ini: " + password)
    print("*************************************")
    my_token = allocToken(sever_ip)
    print(sever_ip + " get a token is  " + my_token)

    file = "name2id_vplatform.xlsx"
    AllocStreamTaskByExcelFile(file)

輸入文件 name2id_vplatform.xlsx

配置文件輸入?yún)?shù)：config.ini

三、打包代碼實(shí)現(xiàn)

基于pycharm軟件，安裝打包軟件

pip install pyinstaller
pyinstaller --onefile main.py
pyinstaller --onefile --distpath dist --out my_application.exe your_script.py

--onefile 表示創(chuàng)建一個(gè)獨(dú)立的文件。
--distpath dist 指定輸出目錄為 dist。
--out my_application 指定輸出的文件名為 my_application.exe。
your_script.py 是你想要打包的 Python 腳本。