用python实现监控视频人数统计

目录

• 一、图示

• 二、准备

• 三、一个简单服务器应用

• 四、向服务器发送图片

• 五、最终关键yolov5调用代码：

一、图示

客户端请求输入一段视频或者一个视频流，输出人数或其他目标数量，上报给上层服务器端，即提供一个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先打开后传递象素级的数字要快很多.

五、最终关键yolov5调用代码：

#!/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: {}\nsize: {}".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)

来源：https://blog.csdn.net/xiao__run/article/details/117026393