目录
一、python实现多路视频播放功能二、代码实现三、打包代码实现总结一、python实现多路视频播放功能
服务端开发后通常需要做功能测试、性能测试，通常postman、curl等作功能测试使用，长跑服务性能postman并不太适合，如用c++实现播放器功能太慢，效率太低效，本文介绍一种用python来实现多路视频播放的测试。
二、代码实现
http申请视频流地址并cv2播放功能
import jsonimport requestsimport timeimport threadingfrom threading import Threadimport signalimport base64from io import BytesIOimport queueimport randomimport sysfrom openpyxl import load_workbookimport jsonimport cv2import datetimeimport osimport shutilimport numpy as nprequests.packages.urllib3.disable_warnings()group_id_base = 31000000452158000001default_group = 31000000452168000002username = "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) & 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) 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
配置文件输入参数：config.ini
三、打包代码实现
基于pycharm软件，安装打包软件
pip install pyinstallerpyinstaller --onefile main.pypyinstaller --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实现视频播放多路实时流的视频。
