-
Notifications
You must be signed in to change notification settings - Fork 0
/
sds011.py
executable file
·123 lines (102 loc) · 2.97 KB
/
sds011.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
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
#!/usr/bin/python
# coding=utf-8
# "DATASHEET": http://cl.ly/ekot
# extended https://gist.github.com/kadamski/92653913a53baf9dd1a8
from __future__ import print_function
import serial, struct, sys, time
DEBUG = 1
CMD_MODE = 2
CMD_QUERY_DATA = 4
CMD_DEVICE_ID = 5
CMD_SLEEP = 6
CMD_FIRMWARE = 7
CMD_WORKING_PERIOD = 8
MODE_ACTIVE = 0
MODE_QUERY = 1
ser = serial.Serial()
ser.port = sys.argv[1]
ser.baudrate = 9600
ser.open()
ser.flushInput()
byte, data = 0, ""
def dump(d, prefix=''):
print(prefix + ' '.join(x.encode('hex') for x in d))
def construct_command(cmd, data=[]):
assert len(data) <= 12
data += [0,]*(12-len(data))
checksum = (sum(data)+cmd-2)%256
ret = "\xaa\xb4" + chr(cmd)
ret += ''.join(chr(x) for x in data)
ret += "\xff\xff" + chr(checksum) + "\xab"
if DEBUG:
dump(ret, '> ')
return ret
def process_data(d):
r = struct.unpack('<HHxxBB', d[2:])
pm25 = r[0]/10.0
pm10 = r[1]/10.0
checksum = sum(ord(v) for v in d[2:8])%256
print("PM 2.5: {} μg/m^3 PM 10: {} μg/m^3 CRC={}".format(pm25, pm10, "OK" if (checksum==r[2] and r[3]==0xab) else "NOK"))
def process_version(d):
r = struct.unpack('<BBBHBB', d[3:])
checksum = sum(ord(v) for v in d[2:8])%256
print("Y: {}, M: {}, D: {}, ID: {}, CRC={}".format(r[0], r[1], r[2], hex(r[3]), "OK" if (checksum==r[4] and r[5]==0xab) else "NOK"))
def read_response():
byte = 0
while byte != "\xaa":
byte = ser.read(size=1)
d = ser.read(size=9)
if DEBUG:
dump(d, '< ')
return byte + d
def cmd_set_mode(mode=MODE_QUERY):
ser.write(construct_command(CMD_MODE, [0x1, mode]))
read_response()
def cmd_query_data():
ser.write(construct_command(CMD_QUERY_DATA))
d = read_response()
if d[1] == "\xc0":
process_data(d)
def cmd_set_sleep(sleep=1):
mode = 0 if sleep else 1
ser.write(construct_command(CMD_SLEEP, [0x1, mode]))
read_response()
def cmd_set_working_period(period):
ser.write(construct_command(CMD_WORKING_PERIOD, [0x1, period]))
read_response()
def cmd_firmware_ver():
ser.write(construct_command(CMD_FIRMWARE))
d = read_response()
process_version(d)
def cmd_set_id(id):
id_h = (id>>8) % 256
id_l = id % 256
ser.write(construct_command(CMD_DEVICE_ID, [0]*10+[id_l, id_h]))
read_response()
if __name__ == "__main__":
command = ""
if len(sys.argv) > 2:
command = sys.argv[2]
if command == "sleep":
cmd_set_mode(0)
cmd_set_sleep(1)
elif command == "wakeup":
cmd_set_sleep(0)
elif command == "wakeup_wait":
cmd_set_sleep(0)
time.sleep(60)
elif command == "firmware_ver":
cmd_firmware_ver()
elif command == "active":
cmd_set_mode(0)
elif command == "query":
cmd_set_mode(1)
elif command == "read":
cmd_query_data()
else:
cmd_set_sleep(0)
cmd_set_mode(1)
time.sleep(3)
cmd_query_data()
cmd_set_mode(0)
cmd_set_sleep()