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

相关文章

C语言实现两个变量值交换的三种方式

《C语言实现两个变量值交换的三种方式》两个变量值的交换是编程中最常见的问题之一,以下将介绍三种变量的交换方式,其中第一种方式是最常用也是最实用的,后两种方式一般只在特殊限制下使用,需要的朋友可以参考下... 目录1.使用临时变量(推荐)2.相加和相减的方式(值较大时可能丢失数据)3.按位异或运算1.使用临时

python中字符串拼接的几种方法及优缺点对比详解

《python中字符串拼接的几种方法及优缺点对比详解》在Python中,字符串拼接是常见的操作,Python提供了多种方法来拼接字符串,每种方法有其优缺点和适用场景,以下是几种常见的字符串拼接方法,需... 目录1. 使用 + 运算符示例:优缺点:2. 使用&nbsjsp;join() 方法示例:优缺点:3

java streamfilter list 过滤的实现

《javastreamfilterlist过滤的实现》JavaStreamAPI中的filter方法是过滤List集合中元素的一个强大工具,可以轻松地根据自定义条件筛选出符合要求的元素,本文就来... 目录1. 创建一个示例List2. 使用Stream的filter方法进行过滤3. 自定义过滤条件1. 定

使用C语言实现交换整数的奇数位和偶数位

《使用C语言实现交换整数的奇数位和偶数位》在C语言中,要交换一个整数的二进制位中的奇数位和偶数位,重点需要理解位操作,当我们谈论二进制位的奇数位和偶数位时,我们是指从右到左数的位置,本文给大家介绍了使... 目录一、问题描述二、解决思路三、函数实现四、宏实现五、总结一、问题描述使用C语言代码实现:将一个整

如何使用Python实现一个简单的window任务管理器

《如何使用Python实现一个简单的window任务管理器》这篇文章主要为大家详细介绍了如何使用Python实现一个简单的window任务管理器,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起... 任务管理器效果图完整代码import tkinter as tkfrom tkinter i

Python异步编程中asyncio.gather的并发控制详解

《Python异步编程中asyncio.gather的并发控制详解》在Python异步编程生态中,asyncio.gather是并发任务调度的核心工具,本文将通过实际场景和代码示例,展示如何结合信号量... 目录一、asyncio.gather的原始行为解析二、信号量控制法:给并发装上"节流阀"三、进阶控制

redis+lua实现分布式限流的示例

《redis+lua实现分布式限流的示例》本文主要介绍了redis+lua实现分布式限流的示例,可以实现复杂的限流逻辑,如滑动窗口限流,并且避免了多步操作导致的并发问题,具有一定的参考价值,感兴趣的可... 目录为什么使用Redis+Lua实现分布式限流使用ZSET也可以实现限流,为什么选择lua的方式实现

使用PyTorch实现手写数字识别功能

《使用PyTorch实现手写数字识别功能》在人工智能的世界里,计算机视觉是最具魅力的领域之一,通过PyTorch这一强大的深度学习框架,我们将在经典的MNIST数据集上,见证一个神经网络从零开始学会识... 目录当计算机学会“看”数字搭建开发环境MNIST数据集解析1. 认识手写数字数据库2. 数据预处理的

Redis中管道操作pipeline的实现

《Redis中管道操作pipeline的实现》RedisPipeline是一种优化客户端与服务器通信的技术,通过批量发送和接收命令减少网络往返次数,提高命令执行效率,本文就来介绍一下Redis中管道操... 目录什么是pipeline场景一:我要向Redis新增大批量的数据分批处理事务( MULTI/EXE

使用Python和python-pptx构建Markdown到PowerPoint转换器

《使用Python和python-pptx构建Markdown到PowerPoint转换器》在这篇博客中,我们将深入分析一个使用Python开发的应用程序,该程序可以将Markdown文件转换为Pow... 目录引言应用概述代码结构与分析1. 类定义与初始化2. 事件处理3. Markdown 处理4. 转