对于一些特殊的行业(比如交警需要统计车辆信息,小区物业可能需要统计出入信息等),需要监控视频人数统计的功能,而这样的功能一般由软件层面进行实现,接下来小编就带来一个使用flask框架,调用yolov5库进行人数统计的方法,并提供出一个http API供其他应用调用。
客户端请求输入一段视频或者一个视频流,输出人数或其他目标数量,上报给上层服务器端,即提供一个http API调用算法统计出人数,最终http上报总人数
相关技术 python pytorch opencv http协议 post请求
Flask
Flask是一个Python实现web开发的微框架,对于像我对web框架不熟悉的人来说还是比较容易上手的。
Flask安装
sudo pip install Flask
为了稍微了解一下flask是如何使用的,先做一个简单的服务器例子。
第一个文件hello.py。
from flask import Flask app = Flask(__name__) @app.route("/") def hello(): return 'hello world!' @app.route("/python") def hello_python(): return 'hello python!' if __name__ == '__main__': app.run(host='0.0.0.0')
app.run(host=‘0.0.0.0')表示现在设定的ip为0.0.0.0,并且设定为0.0.0.0是非常方便的,如果你是在一台远程电脑上设置服务器,并且那台远程电脑的ip是172.1.1.1,那么在本地的电脑上可以设定ip为172.1.1.1来向服务器发起请求。
@app.route('/')表示发送request的地址是http://0.0.0.0:5000/,@app.route("/python")表示发送requests的地址为http://0.0.0.0:5000/python。
第二个文件是request.py
import requests url = 'http://0.0.0.0:5000/' r = requests.get(url) print(r.status_code) print(r.text) url = 'http://0.0.0.0:5000/python' r = requests.get(url) print(r.status_code) print(r.text)
服务器代码
#coding:utf-8 from flask import request, Flask import os app = Flask(__name__) @app.route("/", methods=['POST']) def get_frame(): upload_file = request.files['file'] old_file_name = upload_file.filename file_path = os.path.join('/local/share/DeepLearning', 'new' + old_file_name) if upload_file: upload_file.save(file_path) print "success" return 'success' else: return 'failed' if __name__ == "__main__": app.run("0.0.0.0", port=5000)
客户端代码
import requests url = "http://0.0.0.0:5000" filepath='./t2.jpg' split_path = filepath.split('/') filename = split_path[-1] print(filename) file = open(filepath, 'rb') files = {'file':(filename, file, 'image/jpg')} r = requests.post(url,files = files) result = r.text print result
这种情况长传图片是最快的,比用opencv先打开后传递象素级的数字要快很多.
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2021/2/20 18:19 # @Author : xiaorun # @Site : # @File : yoloDetect.py # @Software: PyCharm import sys import threading from threading import Thread import time import os import cv2 from yolo import YOLO5 import json,jsonify import requests import flask from flask import request headers = {'Content-Type': 'application/json'} url_addr="http://123.206.106.55:8065/api/video/getPersonNum/" # 创建一个服务,把当前这个python文件当做一个服务 server = flask.Flask(__name__) server.debug = True def gen_detector(url_video): yolo = YOLO5() opt = parseData() yolo.set_config(opt.weights, opt.device, opt.img_size, opt.conf_thres, opt.iou_thres, True) yolo.load_model() camera = cv2.VideoCapture(url_video) # 读取视频的fps, 大小 fps = camera.get(cv2.CAP_PROP_FPS) size = (camera.get(cv2.CAP_PROP_FRAME_WIDTH), camera.get(cv2.CAP_PROP_FRAME_HEIGHT)) print("fps: {} size: {}".format(fps, size)) # 读取视频时长(帧总数) total = int(camera.get(cv2.CAP_PROP_FRAME_COUNT)) print("[INFO] {} total frames in video".format(total)) ret, frame = camera.read() if ret==False: video_parameter = {"accessKey": "1C7C48F44A3940EBBAQXTC736BF6530342", "code": "0000", "personNum": "video problem.."} response = requests.post(url=url_addr, headers=headers, data=json.dumps(video_parameter)) print(response.json()) max_person=0 while total>0: total=total-1 ret,frame=camera.read() if ret == True: objs = yolo.obj_detect(frame) if max_person<=len(objs): max_person=len(objs) for obj in objs: cls = obj["class"] cor = obj["color"] conf = '%.2f' % obj["confidence"] label = cls + " " x, y, w, h = obj["x"], obj["y"], obj["w"], obj["h"] cv2.rectangle(frame, (int(x), int(y)), (int(x + w), int(y + h)), tuple(cor)) cv2.putText(frame, label, (int(x), int(y)), cv2.FONT_HERSHEY_SIMPLEX, 1, cor, thickness=2) person = "there are {} person ".format(len(objs)) cv2.putText(frame, person, (20, 20), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), thickness=3) video_parameter = {"accessKey": "1C7C48F44A3940EBBAQXTC736BF6530342", "code": "0000", "personNum": str(max_person)} if total==0: response = requests.post(url=url_addr, headers=headers, data=json.dumps(video_parameter)) print(response.json()) cv2.imshow("test",frame) if cv2.waitKey(1)==ord("q"): break @server.route('/video', methods=['post']) def get_video(): if not request.data: # 检测是否有数据 return ('fail..') video_name= request.data.decode('utf-8') # 获取到POST过来的数据,因为我这里传过来的数据需要转换一下编码。根据晶具体情况而定 video_json = json.loads(video_name) print(video_json) accessKey=video_json["accessKey"] if accessKey=="1C7C48F44A3940EBBAQXTC736BF6530342": code=video_json["code"] url_video=video_json["url"] print(url_video) gen_detector(url_video) # 把区获取到的数据转为JSON格式。 data_return={"code":200,"data":url_video,"message":"请求成功","sucsess":"true"} return json.dumps(data_return) else: pass # 返回JSON数据。 if __name__ == '__main__': server.run(host='192.168.1.250', port=8888)
客户端请求测试:
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2021/5/12 15:12 # @Author : xiaorun # @Site : # @File : test_post.py # @Software: PyCharm import requests,json headers = {'Content-Type': 'application/json'} user_info = {"accessKey":"1C7C48F44A3940EBBAQXTC736BF6530342", "code":"N000001", "url":"http:xxxx/video/xxxx.mp4" } r = requests.post("http://8.8.9.76:8888/video",headers=headers, data=json.dumps(user_info)) print (r.text)