将Raspberry Pi与Grove Base Hat连接到ThingsBoard

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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
#
# Copyright © 2019-2024 The Thingsboard Authors
#
# Licensed under the Apache License, Version 2.0 (the"License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# HTTP://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an"AS IS"BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

import logging
import time
from tb_device_mqtt import TBDeviceMqttClient, TBPublishInfo
from grove.grove_mini_pir_motion_sensor import GroveMiniPIRMotionSensor
from grove.grove_ultrasonic_ranger import GroveUltrasonicRanger
from Seeed_Python_DHT.seeed_dht import DHT
from grove.grove_moisture_sensor import GroveMoistureSensor
from grove.button import Button
from grove.grove_ryb_led_button import GroveLedButton
from grove.grove_light_sensor_v1_2 import GroveLightSensor
from grove.grove_servo import GroveServo

# 日志配置，本例将输出到控制台
logging.basicConfig(level=logging.INFO,
 format='%(asctime)s - %(levelname)s - %(module)s - %(lineno)d - %(message)s',
 datefmt='%Y-%m-%d %H:%M:%S')

log = logging.getLogger(__name__)

THINGSBOARD_SERVER = 'THINGSBOARD_HOST'
ACCESS_TOKEN = 'ACCESS_TOKEN'


def main():

 # Grove - Servo connected to PWM端口
 servo = GroveServo(12)
 servo_angle = 90

 # Grove - mini PIR motion pir_sensor connected to端口D5
 pir_sensor = GroveMiniPIRMotionSensor(5)

 # Grove - Ultrasonic Ranger connected to端口D16
 ultrasonic_sensor = GroveUltrasonicRanger(16)

 # Grove - LED Button connected to端口D18
 button = GroveLedButton(18)

 # Grove - Moisture Sensor connected to端口A0
 moisture_sensor = GroveMoistureSensor(0)

 # Grove - Light Sensor connected to端口A2
 light_sensor = GroveLightSensor(2)
 light_state = False

 # Grove - Temperature&Humidity Sensor connected to端口D22
 dht_sensor = DHT('11', 22)

 # 服务器 RPC 请求回调（用于控制舵机与 LED）
 def on_server_side_rpc_request(client, request_id, request_body):
 log.info('received rpc: {}, {}'.format(request_id, request_body))
 if request_body['method'] == 'getLedState':
 client.send_rpc_reply(request_id, light_state)
 elif request_body['method'] == 'setLedState':
 light_state = request_body['params']
 button.led.light(light_state)
 elif request_body['method'] == 'setServoAngle':
 servo_angle = float(request_body['params'])
 servo.setAngle(servo_angle)
 elif request_body['method'] == 'getServoAngle':
 client.send_rpc_reply(request_id, servo_angle)

 # 连接 ThingsBoard
 client = TBDeviceMqttClient(THINGSBOARD_SERVER, username=ACCESS_TOKEN)
 client.set_server_side_rpc_request_handler(on_server_side_rpc_request)
 client.connect()

 # Callback on detect the motion from motion sensor
 def on_detect():
 log.info('motion detected')
 telemetry = {"motion": True}
 client.send_telemetry(telemetry)
 time.sleep(5)
 # 在仪表板中取消运动告警
 client.send_telemetry({"motion": False})
 log.info("Motion alert deactivated")

 # Callback from button if it was pressed or unpressed
 def on_event(index, event, tm):
 if button._GroveLedButton__btn.is_pressed():
 log.debug('button: single click')
 telemetry = {"button_press": True}
 client.send_telemetry(telemetry)
 log.info("Pressed")
 else:
 log.debug('button: single click')
 telemetry = {"button_press": False}
 client.send_telemetry(telemetry)
 log.info("Unpressed")
 if event & Button.EV_SINGLE_CLICK:
 button.led.light(True)
 elif event & Button.EV_DOUBLE_CLICK:
 button.led.blink()
 elif event & Button.EV_LONG_PRESS:
 button.led.light(False)

 # Adding the callback to the motion sensor
 pir_sensor.on_detect = on_detect
 # Adding the callback to the button
 button.on_event = on_event
 try:
 while True:
 distance = ultrasonic_sensor.get_distance()
 log.debug('distance: {} cm'.format(distance))

 humidity, temperature = dht_sensor.read()
 log.debug('temperature: {}C, humidity: {}%'.format(temperature, humidity))

 moisture = moisture_sensor.moisture
 log.debug('moisture: {}'.format(moisture))

 log.debug('light: {}'.format(light_sensor.light))

 # 格式化待发送至 ThingsBoard 的数据
 telemetry = {'distance': distance,
 'temperature': temperature,
 'humidity': humidity,
 'moisture': moisture,
 'light': light_sensor.light}

 # 发送数据
 client.send_telemetry(telemetry).get()

 time.sleep(.1)
 except Exception as e:
 raise e
 finally:
 client.disconnect()


if __name__ == '__main__':
 main()