使用百度AI实现视频的人流量统计（静态+动态）

百度AI提供了视频监测的诸多接口，下面以人体分析-人流量统计为例介绍。

目录

• 使用百度AI实现视频的人流量统计（静态+动态）
• • 必备条件
  • 静态人流量统计
  • • 输入输出及参数说明
    • 代码
    • 示例
  • 动态人流量统计
  • • 输入输出及参数说明
    • 代码
    • 示例

必备条件

参考《官方文档》，一切改动以官方文档为准。

1. 创建账号，开启项目

在开始之前，需要自行注册百度AI账号，并在主界面创建项目，如下图所示。

2. 获取Access Token
   

进入项目，通过提供的API key和Secret key向服务器获取token，详见网站。

# encoding:utf-8
import requests 

# client_id 为官网获取的AK， client_secret 为官网获取的SK
host = 'https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id=【官网获取的AK】&client_secret=【官网获取的SK】'
response = requests.get(host)
if response:
    print(response.json())

获取token，用于与远程通讯的唯一凭证。

静态人流量统计

接口描述：对于输入的一张图片识别和统计图像当中的人体个数。注⚠️：只统计了人头，不支持追踪和去重。

适用于3米以上的中远距离俯拍，以头部为主要识别目标统计人数，适应各类人流密集场景（如：机场、车展、景区、广场等）；默认识别整图中的人数，支持指定不规则区域的人数统计，同时可输出渲染图片。注⚠️：实际测试中发现对任务场景的要求挺高，详见测试图例。

输入输出及参数说明

输入：path = 本地图片文件的地址，access_token = 上一步获取的token（一周内有效，过期需要重新获取），area非必选，框定了范围，默认全局。
输出：person_num = 人数，image = 默认False，为True时返回人头检测的渲染图。
注⚠️：服务器返回格式为json，需要.json().get()

person_num = response.json().get('person_num')
img_str = response.json().get('image')

代码

读取本地图像文件，转化为base64格式，并发送至request_url

import requests
import base64
import json

'''
人流量统计
'''


def save_base_image(img_str,filename):

    img_data = base64.b64decode(img_str)
    with open(filename, 'wb') as f:
        f.write(img_data)
    return


request_url = "https://aip.baidubce.com/rest/2.0/image-classify/v1/body_num"
# 二进制方式打开图片文件
f = open('[图片地址]', 'rb')
img = base64.b64encode(f.read())
img_name = 'result.png'

params = {"image": img,
          "show": True}
access_token = '[自己搞]'
request_url = request_url + "?access_token=" + access_token
headers = {'content-type': 'application/x-www-form-urlencoded'}
response = requests.post(request_url, data=params, headers=headers)
if response:

    # response = response.decode('utf-8')
    # data = json.loads(response)
    person_num = response.json().get('person_num')
    print('person_num', person_num)
    img_str = response.json().get('image')
    save_base_image(img_str, img_name)

示例

可以看到该算法基本实现人体（头）检测，并实现了计数。但是在密集情况（我国真实的车站场景）下，统计结果并不是特别准确，尤其是在目标被遮挡的不良条件下表现欠佳：图中包括了口罩遮挡和人物之间相互遮挡。

动态人流量统计

接口描述：传入监控视频抓拍图片序列，进行人体追踪，返回每个人体框的坐标和所属ID；并根据目标轨迹判断进出区域行为，进行动态人数统计，返回区域进出人数。同时可输出渲染结果图（含统计值和跟踪框渲染）。
（注⚠️：官方特别说明建议抽帧频率为5fps，which·我看到这里才明白·means这根本不是视频处理，需要前期抽帧之后以图像格式完成统计和追踪。。。害）

输入输出及参数说明

输入：path = 本地序列化图像的文件夹，access_token = 上上一步获取的token，area非必选，框定了范围，默认全局。
输出：person_num = 人数，image = 默认False，为True时返回人头检测的渲染图。

代码

抽帧

import cv2
import os
import shutil


def get_frame_from_video(video_name, interval):
    """
    Args:
        video_name:输入视频名字
        interval: 保存图片的帧率间隔
    Returns:
    """

    # 保存图片的路径
    save_path = video_name.split('.mp4')[0] + '/'
    is_exists = os.path.exists(save_path)
    if not is_exists:
        os.makedirs(save_path)
        print('path of %s is build' % save_path)
    else:
        shutil.rmtree(save_path)
        os.makedirs(save_path)
        print('path of %s already exist and rebuild' % save_path)

    # 开始读视频
    video_capture = cv2.VideoCapture(video_name)
    i = 0
    j = 0

    while True:
        success, frame = video_capture.read()
        i += 1
        if i % interval == 0:
            # 保存图片
            j += 1
            save_name = save_path + str(j) + '_' + str(i) + '.jpg'
            cv2.imwrite(save_name, frame)
            # cv2.imwrite('/Users/panyining/PycharmProjects/video_processing/test/'+save_name, frame)

            print('image of %s is saved' % save_name)
        if not success:
            print('video is all read')
            break


if __name__ == '__main__':
    # 视频文件名字
    video_name = '[你的视频].mp4'
    interval = 5 # fps
    get_frame_from_video(video_name, interval)

（动态）人流统计及追踪

import os
import urllib
from urllib.request import urlopen
import re
import base64
import numpy as np
import cv2

img_path = "[文件夹地址]"

# 读文件夹下的jpg
images = os.listdir(img_path)
images = sorted(images) # shape: 1920*888
request_url = "https://aip.baidubce.com/rest/2.0/image-classify/v1/body_tracking"

# 便利图片
for i in images:
    f = open(img_path + '/' + i, 'rb')
    try:
        img = base64.b64encode(f.read())
        params = {"area": "1, 1, 500, 1, 888, 1000, 1, 888", "case_id": 1, "case_init": "false", "dynamic": "true",
                  "image": img, "show": "true"}
        params = urllib.parse.urlencode(params).encode("utf-8")
        access_token = '[自己搞]'
        request_url = request_url + "?access_token=" + access_token
        request = urllib.request.Request(url=request_url, data=params)
        request.add_header('Content-Type', 'application/x-www-form-urlencoded')
        response = urlopen(request)
        content = response.read().decode('utf-8')

        if content:
            print(i)
            print(content)
            relink = '"image": "(.*)",'
            info = re.findall(relink, content)
            str_to_bytes = bytes(info[0], encoding="utf8")
            img_b64decode = base64.b64decode(str_to_bytes)  # base64解码

            img_array = np.fromstring(img_b64decode, np.uint8)  # 转换np序列
            img = cv2.imdecode(img_array, cv2.COLOR_BGR2RGB)  # 转换Opencv格式
            cv2.imwrite('[渲染结果保存的文件夹]' + i, img)

            # cv2.imshow("img", img)
            # cv2.waitKey(100)

            request_url = "https://aip.baidubce.com/rest/2.0/image-classify/v1/body_tracking"

    except:
        print('find .DS_Store!!!') # macOS会出现隐藏文件
        continue

实现人数预警

#filename:原图片名（本地存储包括路径）；resultfilename:处理后的文件保存名称(每个人打标)

#warningnum告警人数

def crowd_num_warning(filename,resultfilename,warningnum):

    person_num=body_num(filename,resultfilename)  # body_num统计area内人数

    if person_num>warningnum:

        warningmessage="警告：人数过于拥挤，最大人数"+str(warningnum)+"，当前人数:"+str(person_num)

        print(warningmessage)

        # 增加其他预警处理代码，比如将信息通过短信，APP发布给相关人

crowd_num_warning('crowd2.jpg','crowd2_num.jpg',30)

示例

百度AI的人流量统计测试完毕。由于前一步的抽帧，整的工作做的毫无意义，目前看来并不能为本人的项目提供实用性的帮助。附上近期看到大佬做的关于目标追踪的算法介绍，算是该工作的理论深化叭。

以上。

标签：人流量,演示,img,AI,request,num,import,path,response
来源： https://blog.csdn.net/weixin_41919571/article/details/110449977