基于阿里云物联网平台实现的简易出入监控

1
2
3
{
    'photo': 'xxxxxxx.jpg'
}

1
SELECT deviceName() deviceName, photo FROM "/a1O4b4XcICc/+/user/test"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
# -*- coding: utf-8 -*-

import logging
import json
import requests

def post(data):
    webhook_url='https://oapi.dingtalk.com/robot/send?access_token=${Token}' #钉钉群机器人的webhook的URL
    headers = {'Content-Type': 'application/json; charset=utf-8'}
    post_data = json.dumps(data)
    try:
        response = requests.post(webhook_url, headers=headers, data=post_data)
        logging.info('Send success')
    except requests.exceptions.HTTPError as exc:
        logging.error("Send Error,HTTP error: %d, reason: %s" % (exc.response.status_code, exc.response.reason))
        raise
    except requests.exceptions.ConnectionError:
        logging.error("Send Error,HTTP connection error!")
        raise
    else:
        result = response.json()
        logging.info('Send Error:%s' % result)
        if result['errcode']:
            error_data = {"msgtype": "text", "text": {"content": "Send Error, reason:%s" % result['errmsg']}, "at": {"isAtAll": True}}
            logging.error("Send Error:%s" % error_data)
            requests.post(webhook_url, headers=headers, data=json.dumps(error_data))
        return result

def post_makedown(title,text):
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": text
        },
        "at": {
            "atMobiles": [],
            "isAtAll": False
        }
    }
    post(data)

def handler(event, context):
    logger = logging.getLogger()
    evt = json.loads(event)
    #OSS endpoint url
    post_makedown('告警','![screenshot](https://${bucket}.oss-cn-hangzhou.aliyuncs.com/photo/%s)' % evt.get('photo',''))
    logger.info('photo name is %s', evt.get('photo',''))
    return 'OK'

1
2
3
4
5
6
1. pip install aliyun-python-sdk-iot-client
2. pip install oss2
3. mkdir py-demo (项目程序文件夹)
4. cd py-demo
5. mkdir photo (本地照片临时目录)
6. vim monitor.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
# -*- coding: utf-8 -*-

import json
import uuid
import logging
from time import sleep
from picamera import PiCamera
import RPi.GPIO as GPIO
import oss2
import aliyunsdkiotclient.AliyunIotMqttClient as iot

# 参数定义
options = {
    'productKey': '${productKey}',  # 设备标识三元组
    'deviceName': '${deviceName}',  # 设备标识三元组
    'deviceSecret': '${deviceSecret}',  # 设备标识三元组
    'port': 1883,  # iot mqtt port
    'host': 'iot-as-mqtt.cn-shanghai.aliyuncs.com',  # iot mqtt endpoint
    'endpoint': 'http://oss-cn-hangzhou.aliyuncs.com',  # oss endpoint
    'ak': '${ak}',
    'sk': '${sk}',
    'bucket': '${bucket}',  # oss bucket
    'ir_pin': 24  # 人体红外感应器设置读取针脚标号
}

topic = '/' + options['productKey'] + '/' + options['deviceName'] + '/user/test'

# 拍照存oss，并发送通知
def takephoto2oss(client):

    #拍照
    photo_filename = str(uuid.uuid1()) + ".jpg"
    print('take photo :' + photo_filename)
    camera.capture('photo/' + photo_filename)

    #存OSS
    print('save photo to oss :' + photo_filename)
    bucket.put_object_from_file(
        'photo/' + photo_filename, 'photo/' + photo_filename)

    #消息上送
    payload_json = {
        'photo': photo_filename
    }
    print('send data to iot server: ' + str(payload_json))
    client.publish(topic, payload = str(payload_json))


def on_connect(client, userdata, flags_dict, rc):
    print("Connected with result code " + str(rc))


def on_disconnect(client, userdata, flags_dict, rc):
    print("Disconnected.")


if __name__ == '__main__':
    # GPIO 初始化
    GPIO.setmode(GPIO.BCM)
    GPIO.setwarnings(False)
    GPIO.setup(options['ir_pin'], GPIO.IN)

    # 摄像头 初始化
    camera = PiCamera()
    camera.resolution = (640, 480)
    camera.vflip = True
    camera.hflip = True

    # OSS 初始化
    auth = oss2.Auth(options['ak'], options['sk'])
    bucket = oss2.Bucket(auth, options['endpoint'], options['bucket'])

    # IOT Mqtt 初始化
    client = iot.getAliyunIotMqttClient(options['productKey'], options['deviceName'], options['deviceSecret'], secure_mode = 3)
    client.on_connect = on_connect
    client.connect(host=options['productKey'] + '.' + options['host'], port=options['port'], keepalive = 60)

    while True:
        # 当高电平信号输入时报警
        if GPIO.input(options['ir_pin']) == True:
            print " Someone is coming!"
            takephoto2oss(client)
        else:
            continue
        sleep(3)

1
python monitor.py