M5Stack UnitV2口罩识别并上传阿里云

内容介绍

AI摄像头 ––UnitV2 识别宿舍戴口罩情况

PART1硬件介绍

UnitV2是M5Stack推出的一款高效率的AI识别模块, 采用Sigmstar SSD202D(集成双核Cortex-A7 1.2Ghz处理器)控制核心，集成128MB-DDR3内存，512MB NAND Flash, 1080P摄像头。内嵌Linux操作系统，集成丰富的软硬件资源与开发工具，致力带给用户开箱即用的体验。

任务分析

识别办公室/宿舍出入人员是否有正常佩戴口罩，发现异常立即报警，并将每日的数据上传到PC端，生成报表。作为扩展，以上四个题目也可以选择上传到云端，生成报表。

PART2 实现思路

任务拆解

1.首先需要识别人是否出现，然后识别人是否带有口罩。

2.利用python爬虫爬取http://10.254.239.1/或者http://unitv2.py/输出的json信息，通过ai摄像头输出的物体数量来判断是否带有口罩

3.利用本地pc的python-mqtt模拟设备上传数据到阿里云，并实时查看当前戴口罩的人数和查看线上记录

4.利用python调用钉钉机器人来给钉钉群聊发送当前是否带有口罩的信息。

具体步骤

0.安装驱动

在m5stack官网下载安装驱动m5-docs

将驱动压缩包解压至桌面路径->进入设备管理器中选中当前未识别的设备->右键选择自定义更新->选中压缩包解压的路径->点击确认,等待更新完成。

1.训练模型

使用V-Training进行人脸识别训练和戴口罩识别训练

1.1

注册V-Training并登录，点击Start->新建项目->导入图片->NEXT->Object Detection

利用ai模型辅助训练

1.2

选择图片中你想要识别的物体创建标签，选择多张照片进行分析

1.3

点击Refresh可以刷新查看任务最新状态,训练完成后,将获得模型Download下载链接,以及损失曲线。

若损失曲线显示一张下降的曲线，则说明模型训练成功

1.4模型验证

通过浏览器访问域名unitv2.py或IP：10.254.239.1，切换功能至Object Recognition,点击add按键上传模型。

查看模型输出json文件是否符合预期效果

2 python爬虫爬取网页输出json文件

安装webdriver使用selenium，首先使用webdriver打开http://10.254.239.1/，使用模拟按键点击打开Object Recognition，并点击run

在执行完js程序后查找json输出的id标签，导出json文件，查找识别到的物品数量

（以下为单独的python爬虫程序，附录程序为完整版）

# 爬虫代码
# 指定chrome的驱动
# 执行到这里的时候Selenium会到指定的路径将chrome driver程序运行起来
driver = webdriver.Chrome('C:\Program Files\Google\Chrome\Application\chromedriver.exe')
# get 方法 打开指定网址
driver.get('http://10.254.239.1/')
html = driver.page_source
time.sleep(8)

driver.find_element_by_id("object_recognition")  # 模拟按键输入
driver.find_element_by_id("object_recognition").click()  # 单机元素
time.sleep(10)
driver.find_element_by_id("yolo_run")  # 模拟按键输入
driver.find_element_by_id("yolo_run").click()  # 单机元素
time.sleep(10)

def publish_loop():
    global mark_person, nomark_person
    mark_person = 0
    # dingding = DingDingWebHook(secret=my_secret, url=my_url)

    while 1:
        element_keyword = driver.find_element_by_id('func-result-pre')
        # print(element_keyword.text)
        json_date = element_keyword.text
        json_jiexi = json.loads(json_date)
        try:
            person_num: object = json_jiexi['num']
            assert isinstance(person_num, object)
            print(person_num)
            if person_num == 1:
                nomark_person += 1
                # 钉钉预警
                print('1111')
                dingdingpush1()
            elif person_num == 2:
                mark_person += 1
                # 钉钉预警
                print('2222')
                dingdingpush2()
            time.sleep(5)
            switchPost = json_switch_set('1', mark_person)
            client.publish(topic_post, payload=switchPost, qos=0)
        except:
            time.sleep(1)

3.python模拟设备数据上传阿里云

代码较长，已放附录

4.通过钉钉自定义机器人给宿舍群发送警报信息和祝福信息

class DingDingWebHook(object):
    def __init__(self, secret=None, url=None):
        """
        :param secret: 安全设置的加签秘钥
        :param url: 机器人没有加签的WebHook_url
        """
        if secret is not None:
            secret = secret
        else:
            secret = ''  # 加签秘钥
        if url is not None:
            url = url
        else:
            url = "https://oapi.dingtalk.com/robot/send?access_token=复制钉钉自定义机器人中的码"  # 无加密的url

        timestamp = round(time.time() * 1000)  # 时间戳
        secret_enc = secret.encode('utf-8')
        string_to_sign = '{}\n{}'.format(timestamp, secret)
        string_to_sign_enc = string_to_sign.encode('utf-8')
        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))  # 最终签名

        self.webhook_url = url + '&timestamp={}&sign={}'.format(timestamp, sign)  # 最终url，url+时间戳+签名

    def send_meassage(self, data):
        """
        发送消息至机器人对应的群
        :rtype: object
        :param data: 发送的内容
        :return:
        """
        header = {
            "Content-Type": "application/json",
            "Charset": "UTF-8"
        }
        send_data = json.dumps(data)  # 将字典类型数据转化为json格式
        send_data = send_data.encode("utf-8")  # 编码为UTF-8格式
        request = urllib.request.Request(url=self.webhook_url, data=send_data, headers=header)  # 发送请求

        opener = urllib.request.urlopen(request)  # 将请求发回的数据构建成为文件格式
        print(opener.read())  # 打印返回的结果


def dingdingpush1():
    my_secret = ''
    my_url = 'https://oapi.dingtalk.com/robot/send?access_token=复制钉钉自定义机器人中的码'
    my_data = {
        "msgtype": "markdown",
        "markdown": {"title": "unitv2",
                     "text": "# **350宿舍戴口罩情况**\n"
                             "\n检测到未佩戴口罩，请立即佩戴口罩 "
                     },
        "at": {
            "atMobiles": [],
            "isAtAll": True}  # 是否@所有人
    }
    dingding = DingDingWebHook(secret=my_secret, url=my_url)
    dingding.send_meassage(my_data)


def dingdingpush2():
    my_secret = ''
    my_url = 'https://oapi.dingtalk.com/robot/send?access_token=复制钉钉自定义机器人中的码'
    my_data2 = {
        "msgtype": "markdown",
        "markdown": {"title": "unitv2",
                     "text": "# **350宿舍戴口罩情况**\n""\n已佩戴口罩，注意环境安全 "},
        "at": {
            "atMobiles": [],
            "isAtAll": True}  # 是否@所有人
    }
    dingding = DingDingWebHook(secret=my_secret, url=my_url)
    dingding.send_meassage(my_data2)

PART3 功能展示

未佩戴口罩时