Python实现多路视频多窗口播放功能

2025-02-10 04:50

本文主要是介绍Python实现多路视频多窗口播放功能,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

《Python实现多路视频多窗口播放功能》这篇文章主要为大家详细介绍了Python实现多路视频多窗口播放功能的相关知识,文中的示例代码讲解详细,有需要的小伙伴可以跟随小编一起学习一下...

一、python实现多路视频播放功能

服务端开发后通常需要做功能测试、性能测试,通常postman、curl等作功能测试使用,长跑服务性能postman并不太适合,如用c++实现播放器功能太慢,效率太低效,本文介绍一种用python来实现多路视频播放的测试。

二、代码实现

http申请视频流地址并cv2播放功能

import json
import requests
import time
import threading
from threading import Thread
import signal
import bphpase64
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} 申请码流成功 {result['playurl']}")
                return result['playurl']
            else:
                print(f"{deviceId} 申请码流失败 {response.json()}" + my_token)
                return ""
        except json.JSONDecodeError:
            print(f"{deviceId} 响应解析失败:无法解析JSON")
            return ""
    except requests.RequestException as e:
        print(f"{deviceId} 请求失败:{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} 申请码流失败")
        except json.JSONDecodeError:
            print(f"{playurl} 响应解析失败:无法解析JSON" )
    except requests.RequestException as e:
        print(f"{playurl} 请求失败:{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) &ahttp://www.chinasem.cnmp; 0xFF == ord('q'):
                print("deviceid "+ deviceid + " receive the stop command")
                stop_event.set()  # 设置事件,通知其他线程停止
                break
        else:
            break
    cap.release()
    # 如果你的程序在退出时没有正确关闭所有OpenCV窗口,那么可能是因为cv2.destroyAllWindows()没有在主线程中被调用。在所有线程结束后,确保在主线程中调用
    # cv2.destroyAllWindows()

device_datas = []
stop_event = threading.Event()  # 创建一个事件对象
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)

  javascript      if 'serverUrl' in config:
            sever_ip = config['serverUrl']
        else:
            sever_ip = ""
       js     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 IOErrorandroid:
        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

Python实现多路视频多窗口播放功能

配置文件输入参数:config.ini

Python实现多路视频多窗口播放功能

三、打包代码实现

基于pycharm软件安装打包软件

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

--onefile 表示创建一个独立的文件。
--distpath dist 指定输出目录为 dist。
--out my_application 指定输出的文件名为 my_application.exe。
your_script.py 是你想要打包的 Python 脚本

Python实现多路视频多窗口播放功能

总结

本文实现了最简单最快的方式实现播放器功能,python实现视频播放多路实时流的视频。

到此这篇关于Python实现多路视频多窗口播放功能的文章就介绍到这了,更多相关Python多路视频播放内容请搜索China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程China编程(www.chinasem.cn)!

这篇关于Python实现多路视频多窗口播放功能的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1153373

相关文章

Python中pywin32 常用窗口操作的实现

《Python中pywin32常用窗口操作的实现》本文主要介绍了Python中pywin32常用窗口操作的实现,pywin32主要的作用是供Python开发者快速调用WindowsAPI的一个... 目录获取窗口句柄获取最前端窗口句柄获取指定坐标处的窗口根据窗口的完整标题匹配获取句柄根据窗口的类别匹配获取句

利用Python打造一个Excel记账模板

《利用Python打造一个Excel记账模板》这篇文章主要为大家详细介绍了如何使用Python打造一个超实用的Excel记账模板,可以帮助大家高效管理财务,迈向财富自由之路,感兴趣的小伙伴快跟随小编一... 目录设置预算百分比超支标红预警记账模板功能介绍基础记账预算管理可视化分析摸鱼时间理财法碎片时间利用财

在 Spring Boot 中实现异常处理最佳实践

《在SpringBoot中实现异常处理最佳实践》本文介绍如何在SpringBoot中实现异常处理,涵盖核心概念、实现方法、与先前查询的集成、性能分析、常见问题和最佳实践,感兴趣的朋友一起看看吧... 目录一、Spring Boot 异常处理的背景与核心概念1.1 为什么需要异常处理?1.2 Spring B

Python中的Walrus运算符分析示例详解

《Python中的Walrus运算符分析示例详解》Python中的Walrus运算符(:=)是Python3.8引入的一个新特性,允许在表达式中同时赋值和返回值,它的核心作用是减少重复计算,提升代码简... 目录1. 在循环中避免重复计算2. 在条件判断中同时赋值变量3. 在列表推导式或字典推导式中简化逻辑

python处理带有时区的日期和时间数据

《python处理带有时区的日期和时间数据》这篇文章主要为大家详细介绍了如何在Python中使用pytz库处理时区信息,包括获取当前UTC时间,转换为特定时区等,有需要的小伙伴可以参考一下... 目录时区基本信息python datetime使用timezonepandas处理时区数据知识延展时区基本信息

Python位移操作和位运算的实现示例

《Python位移操作和位运算的实现示例》本文主要介绍了Python位移操作和位运算的实现示例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录1. 位移操作1.1 左移操作 (<<)1.2 右移操作 (>>)注意事项:2. 位运算2.1

如何在 Spring Boot 中实现 FreeMarker 模板

《如何在SpringBoot中实现FreeMarker模板》FreeMarker是一种功能强大、轻量级的模板引擎,用于在Java应用中生成动态文本输出(如HTML、XML、邮件内容等),本文... 目录什么是 FreeMarker 模板?在 Spring Boot 中实现 FreeMarker 模板1. 环

Qt实现网络数据解析的方法总结

《Qt实现网络数据解析的方法总结》在Qt中解析网络数据通常涉及接收原始字节流,并将其转换为有意义的应用层数据,这篇文章为大家介绍了详细步骤和示例,感兴趣的小伙伴可以了解下... 目录1. 网络数据接收2. 缓冲区管理(处理粘包/拆包)3. 常见数据格式解析3.1 jsON解析3.2 XML解析3.3 自定义

使用Python和Pyecharts创建交互式地图

《使用Python和Pyecharts创建交互式地图》在数据可视化领域,创建交互式地图是一种强大的方式,可以使受众能够以引人入胜且信息丰富的方式探索地理数据,下面我们看看如何使用Python和Pyec... 目录简介Pyecharts 简介创建上海地图代码说明运行结果总结简介在数据可视化领域,创建交互式地

SpringMVC 通过ajax 前后端数据交互的实现方法

《SpringMVC通过ajax前后端数据交互的实现方法》:本文主要介绍SpringMVC通过ajax前后端数据交互的实现方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价... 在前端的开发过程中,经常在html页面通过AJAX进行前后端数据的交互,SpringMVC的controll