一.API Key和Secret Key准备工作
1.1在百度智能云平台上注册账号，
1.2进入百度智能云管理中心：
1.3.汇集信息
驾驶行为分析 API Key: 【官网获取的AK】 Secret Key: 【官网获取的SK】 host = 'https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id=【官网获取的AK】&client_secret=【官网获取的SK】'
二.检测图像中的驾驶行为
2.1根据官网给的样例，更改自己的ak和sk,如下所示
def detect_image(frame): request_url = "https://aip.baidubce.com/rest/2.0/image-classify/v1/driver_behavior" image = cv2.imencode('.jpg', frame)[1] img = str(base64.b64encode(image))[2:-1] params = {"image": img} host = 'https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id=【官网获取的AK】&client_secret=【官网获取的SK】' response_host = requests.get(host) access_token = [] useriddatas = json.loads(response_host.text) access_token.append(str(useriddatas["access_token"])) access = (''.join(access_token)) access_token = access 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) data = response.json() datas = str(data) if len(datas) > 200: #防止检测过快无法触及并发，这里我是使用的免费的次数 find_behavior(data) else: pass
2.2从response.json()中解析行为：
def find_behavior(data): ''' smoke //吸烟， cellphone //打手机 ， not_buckling_up // 未系安全带， not_facing_front // 视角未看前方， yawning // 打哈欠， eyes_closed // 闭眼， head_lowered // 低头 ''' # print(data) data_smoke = data['person_info'][0]['attributes']['smoke']['score'] data_cellphone = data['person_info'][0]['attributes']['cellphone']['score'] data_not_buckling_up = data['person_info'][0]['attributes']['not_buckling_up']['score'] data_not_facing_front = data['person_info'][0]['attributes']['not_facing_front']['score'] data_yawning = data['person_info'][0]['attributes']['yawning']['score'] data_eyes_closed = data['person_info'][0]['attributes']['eyes_closed']['score'] data_head_lowered = data['person_info'][0]['attributes']['head_lowered']['score'] All_behavior = [data_smoke, data_cellphone, data_not_buckling_up, data_not_facing_front, data_yawning, data_eyes_closed, data_head_lowered] All_behavior = np.array(All_behavior) # 转格式 Danger_behavior = np.sum(All_behavior >= 0.4) # 存在危险驾驶行为（该行为概率大于0.4）的个数 if Danger_behavior > 0: """ from pygame import mixer #控制音频流模块 mixer.init() #初始化混音器模块 mixer.music.load('./存在危险驾驶行为.mp3') #载入待播放音乐文件 假定为2秒 mixer.music.play() #开始播放音乐流 sleep(2) #休眠2秒供播放使用 mixer.music.stop() # 停止播放 """ print("存在危险驾驶行为") else: print("安全驾驶中")
2.3.将图像检测怼到视频中
if __name__ == "__main__": capture = cv2.VideoCapture(0) fps = 0.0 while (True): t1 = time.time() # 读取某一帧 ref, frame = capture.read() # 格式转变，BGRtoRGB frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) # 转变成Image # 进行检测 detect_image(frame) # time.sleep(0.2) frame = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR) # fps = int(round(capture.get(cv2.CAP_PROP_FPS))) fps = (fps + (1. / (time.time() - t1))) / 2 frame = cv2.putText(frame, "fps= %.2f" % (fps), (0, 40), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2) cv2.imshow("video", frame) c = cv2.waitKey(1) & 0xff if c == 27: capture.release() break capture.release() cv2.destroyAllWindows()
综上所述：完整代码demo:
# encoding:utf-8 import json import time import numpy as np import requests import base64 import cv2 ''' 驾驶行为分析 API Key: USIybtKpK5CB2B1o6RSyYYz1 Secret Key: CFSlR51sTQnocifu8mmrn8sv7XNb5vwS host = 'https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id=【官网获取的AK】&client_secret=【官网获取的SK】' ''' def find_behavior(data): ''' smoke //吸烟， cellphone //打手机 ， not_buckling_up // 未系安全带， not_facing_front // 视角未看前方， yawning // 打哈欠， eyes_closed // 闭眼， head_lowered // 低头 ''' # print(data) data_smoke = data['person_info'][0]['attributes']['smoke']['score'] data_cellphone = data['person_info'][0]['attributes']['cellphone']['score'] data_not_buckling_up = data['person_info'][0]['attributes']['not_buckling_up']['score'] data_not_facing_front = data['person_info'][0]['attributes']['not_facing_front']['score'] data_yawning = data['person_info'][0]['attributes']['yawning']['score'] data_eyes_closed = data['person_info'][0]['attributes']['eyes_closed']['score'] data_head_lowered = data['person_info'][0]['attributes']['head_lowered']['score'] All_behavior = [data_smoke, data_cellphone, data_not_buckling_up, data_not_facing_front, data_yawning, data_eyes_closed, data_head_lowered] All_behavior = np.array(All_behavior) # 转格式 Danger_behavior = np.sum(All_behavior >= 0.4) # 存在危险驾驶行为（该行为概率大于0.4）的个数 if Danger_behavior > 0: """ from pygame import mixer #控制音频流模块 mixer.init() #初始化混音器模块 mixer.music.load('./存在危险驾驶行为.mp3') #载入待播放音乐文件 假定为2秒 mixer.music.play() #开始播放音乐流 sleep(2) #休眠2秒供播放使用 mixer.music.stop() # 停止播放 """ print("存在危险驾驶行为") else: print("安全驾驶中") def detect_image(frame): request_url = "https://aip.baidubce.com/rest/2.0/image-classify/v1/driver_behavior" image = cv2.imencode('.jpg', frame)[1] img = str(base64.b64encode(image))[2:-1] params = {"image": img} host ='https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id=【官网获取的AK】&client_secret=【官网获取的SK】' response_host = requests.get(host) access_token = [] useriddatas = json.loads(response_host.text) access_token.append(str(useriddatas["access_token"])) access = (''.join(access_token)) access_token = access 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) data = response.json() datas = str(data) if len(datas) > 200: find_behavior(data) else: pass if __name__ == "__main__": capture = cv2.VideoCapture(0) fps = 0.0 while (True): t1 = time.time() # 读取某一帧 ref, frame = capture.read() # 格式转变，BGRtoRGB frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) # 转变成Image # 进行检测 detect_image(frame) # time.sleep(0.2) frame = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR) # fps = int(round(capture.get(cv2.CAP_PROP_FPS))) fps = (fps + (1. / (time.time() - t1))) / 2 frame = cv2.putText(frame, "fps= %.2f" % (fps), (0, 40), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2) cv2.imshow("video", frame) c = cv2.waitKey(1) & 0xff if c == 27: capture.release() break capture.release() cv2.destroyAllWindows()
