對(duì)于一些特殊的行業(yè)（比如交警需要統(tǒng)計(jì)車(chē)輛信息，小區(qū)物業(yè)可能需要統(tǒng)計(jì)出入信息等），需要監(jiān)控視頻人數(shù)統(tǒng)計(jì)的功能，而這樣的功能一般由軟件層面進(jìn)行實(shí)現(xiàn)，接下來(lái)小編就帶來(lái)一個(gè)使用flask框架，調(diào)用yolov5庫(kù)進(jìn)行人數(shù)統(tǒng)計(jì)的方法，并提供出一個(gè)http API供其他應(yīng)用調(diào)用。
一、圖示
客戶(hù)端請(qǐng)求輸入一段視頻或者一個(gè)視頻流，輸出人數(shù)或其他目標(biāo)數(shù)量，上報(bào)給上層服務(wù)器端，即提供一個(gè)http API調(diào)用算法統(tǒng)計(jì)出人數(shù)，最終http上報(bào)總?cè)藬?shù)
二、準(zhǔn)備
相關(guān)技術(shù) python pytorch opencv http協(xié)議 post請(qǐng)求
Flask
Flask是一個(gè)Python實(shí)現(xiàn)web開(kāi)發(fā)的微框架，對(duì)于像我對(duì)web框架不熟悉的人來(lái)說(shuō)還是比較容易上手的。
Flask安裝
sudo pip install Flask
三、一個(gè)簡(jiǎn)單服務(wù)器應(yīng)用
為了稍微了解一下flask是如何使用的，先做一個(gè)簡(jiǎn)單的服務(wù)器例子。
第一個(gè)文件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')表示現(xiàn)在設(shè)定的ip為0.0.0.0，并且設(shè)定為0.0.0.0是非常方便的，如果你是在一臺(tái)遠(yuǎn)程電腦上設(shè)置服務(wù)器，并且那臺(tái)遠(yuǎn)程電腦的ip是172.1.1.1，那么在本地的電腦上可以設(shè)定ip為172.1.1.1來(lái)向服務(wù)器發(fā)起請(qǐng)求。
@app.route('/')表示發(fā)送request的地址是http://0.0.0.0:5000/，@app.route("/python")表示發(fā)送requests的地址為http://0.0.0.0:5000/python。
第二個(gè)文件是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)
四、向服務(wù)器發(fā)送圖片
服務(wù)器代碼
#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)
客戶(hù)端代碼
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
這種情況長(zhǎng)傳圖片是最快的，比用opencv先打開(kāi)后傳遞象素級(jí)的數(shù)字要快很多.
五、最終關(guān)鍵yolov5調(diào)用代碼：
#!/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/"
# 創(chuàng)建一個(gè)服務(wù)，把當(dāng)前這個(gè)python文件當(dāng)做一個(gè)服務(wù)
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))
# 讀取視頻時(shí)長(zhǎng)（幀總數(shù)）
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: # 檢測(cè)是否有數(shù)據(jù)
return ('fail..')
video_name= request.data.decode('utf-8')
# 獲取到POST過(guò)來(lái)的數(shù)據(jù)，因?yàn)槲疫@里傳過(guò)來(lái)的數(shù)據(jù)需要轉(zhuǎn)換一下編碼。根據(jù)晶具體情況而定
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)
# 把區(qū)獲取到的數(shù)據(jù)轉(zhuǎn)為JSON格式。
data_return={"code":200,"data":url_video,"message":"請(qǐng)求成功","sucsess":"true"}
return json.dumps(data_return)
else:
pass
# 返回JSON數(shù)據(jù)。
if __name__ == '__main__':
server.run(host='192.168.1.250', port=8888)
客戶(hù)端請(qǐng)求測(cè)試：
#!/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)小結(jié)
到此這篇關(guān)于用python實(shí)現(xiàn)監(jiān)控視頻人數(shù)統(tǒng)計(jì)的文章就介紹到這了,更多python學(xué)習(xí)資料請(qǐng)關(guān)注W3Cschool！
