本项目使用 Digi Xbee3 无线 Zigbee 模块和 TE Connectivity MS8607 传感器来实现电池供电的 Zigbee 环境传感器。Xbee3 上的 ADC 用于测量系统电池电压。传感器数据和电池电压被发送到 Machinechat 的 JEDI One IoT 平台并进行监控。
项目描述
该项目的用例是监控冷冻机的温度和传感器系统的电池电压。MS8607 能够测量气压、温度和湿度,但本项目专注于温度和电压参数。MS8607 通过电缆连接到 Xbee 无线电,因此可以放置在冷冻机内以测量温度(还测量湿度以进一步了解冷冻机性能)。监控电池电压以确定电池何时退化到需要更换的程度。
Xbee3 模块上运行的 MicroPython 应用程序用于连接到 Zigbee 网络并读取传感器和 ADC。传感器数据和电池电压被发送到 Zigbee 协调器,然后通过 Machinechat 的 Zigbee 到 WiFi 传感器桥转发到 JEDI One IoT 平台(详见项目)。JEDI One 在具有以太网/WiFi 网络连接的 Raspberry Pi 4 上运行。
硬件
- RASPBERRY PI 4B/4GB
Raspberry PI 4 Model B 带 4GB SDRAM(用于运行 Machinechat JEDI One) - XB3-24Z8PT-J
Digi Xbee3 Zigbee PCB 天线 THT 模块 - DPP901G000
TE Connectivity MS8607 压力/温度/湿度传感器分线板 - TPS2042P
Texas Instruments TPS2042P 电源分配开关
软件
- JEDI One
JEDI One 是一个即用型 IoT 数据管理软件解决方案。功能包括:从传感器、设备和机器收集数据;构建直观的实时和历史数据及系统视图仪表板;创建规则以自动监控和响应数据条件;通过电子邮件和短信接收警报通知。 - MicroPython
MicroPython 是 Python 3 编程语言的精简高效实现,包含 Python 标准库的一小部分,并优化为在微控制器和受限环境中运行。
实施
传感器电路由两节 AA 碱性电池供电。主要组件包括 Digi Xbee 3 Zigbee 模块、Texas Instruments TPS2042 电源分配 IC 和 TE Connectivity MS8607 传感器板。TPS2042 用于控制 MS8607 和电池分压电路的电源。MicroPython 应用程序代码实现了 60 秒的睡眠周期,在读取和传输传感器数据之间以最小化电流消耗并延长电池寿命。下面的示意图展示了电路的实现方式。
Digi Xbee3 MicroPython MS8607 传感器项目详情
在MicroPython项目中,Xbee3模块被配置为Zigbee终端设备,其网络ID与Zigbee到WiFi桥接中使用的Xbee3协调器相同。MicroPython应用程序加入Zigbee网络,并启动一个无限循环,该循环启用TPS2042P开关,读取MS8607传感器,读取电池电压,禁用TPS2042P开关,构建传感器数据消息,发送到Zigbee协调器,休眠60秒并重复。
1 - 在Xbee3上设置MicroPython并刷入应用程序固件。参考Digi MicroPython编程指南
2 - 代码详解(PyCharm项目:sensor_ms8607THV,文件名:main.py)
初始设置并连接到Zigbee网络
# project: sensor_ms8607THV
# description: Xbee3 Zigbee Micropython MS8607 sensor project that sleeps x seconds, then wakes to send temperature,
# humidity, and battery voltage readings to coordinator
# Default template for Digi Xbee3 projects
import utime
import ustruct
import machine
import time
import xbee
from machine import Pin
x = xbee.XBee()
# list of commands in hex for MS8607 pressure sensor
c_reset = 0x1E # reset command
r_c1 = 0xA2 # read PROM C1 command
r_c2 = 0xA4 # read PROM C2 command
r_c3 = 0xA6 # read PROM C3 command
r_c4 = 0xA8 # read PROM C4 command
r_c5 = 0xAA # read PROM C5 command
r_c6 = 0xAC # read PROM C6 command
r_adc = 0x00 # read ADC command
r_d1 = 0x44 # convert D1 (OSR=1024)
r_d2 = 0x54 # convert D2 (OSR=1024)
p_address = 0x76 # pressure sensor i2c address
# list of commands in hex for MS8607 humidity sensor
h_address = 0x40 # humidty sensor i2c address
r_user = 0xE7 # read user register command
w_user = 0xE6 # write user register command
t_temp = 0xE3 # trigger temperature measurement, hold master
t_humi = 0xE5 # trigger humidity measurement, hold master
# set register format
REGISTER_FORMAT = '>h' # ">" big endian, "h" 2 bytes
REGISTER_SHIFT = 4 # rightshift 4 for 12 bit resolution
# set i2c clock to 100KHz
i2c = machine.I2C(1, freq=100000)
#Zigbee coordinator adddress in house
TARGET_64BIT_ADDR = b'\x00\x13\xA2\x00\x41\xAA\xCB\xBA'
# slave_addr = 0x76
# reset pressure sensor
def reset_ps():
slave_addr = p_address
data = bytearray([c_reset])
i2c.writeto(slave_addr, data)
return data
# scan i2c bus for active addresses
def scan_I2C():
devices = i2c.scan()
return devices
def read_c1(): # read PROM value C1
data = bytearray([r_c1])
i2c.writeto(slave_addr, data)
raw_c = i2c.readfrom(slave_addr, 2) # raw C is 2 bytes
value = int.from_bytes(raw_c, "big") # use builtin to convert to integer
return value
def read_c2(): # read PROM value C2
data = bytearray([r_c2])
i2c.writeto(slave_addr, data)
raw_c = i2c.readfrom(slave_addr, 2) # raw C is 2 bytes
value = int.from_bytes(raw_c, "big") # use builtin to convert to unsigned integer
return value
def read_c3(): # read PROM value C3
data = bytearray([r_c3])
i2c.writeto(slave_addr, data)
raw_c = i2c.readfrom(slave_addr, 2) # raw C is 2 bytes
value = int.from_bytes(raw_c, "big") # use builtin to convert to unsigned integer
return value
def read_c4(): # read PROM value C4
data = bytearray([r_c4])
i2c.writeto(slave_addr, data)
raw_c = i2c.readfrom(slave_addr, 2) # raw C is 2 bytes
value = int.from_bytes(raw_c, "big") # use builtin to convert to unsigned integer
return value
def read_c5(): # read PROM value C5
data = bytearray([r_c5])
i2c.writeto(slave_addr, data)
raw_c = i2c.readfrom(slave_addr, 2) # raw C is 2 bytes
value = int.from_bytes(raw_c, "big") # use builtin to convert to unsigned integer
return value
def read_c6(): # read PROM value C6
data = bytearray([r_c6])
i2c.writeto(slave_addr, data)
raw_c = i2c.readfrom(slave_addr, 2) # raw C is 2 bytes
value = int.from_bytes(raw_c, "big") # use builtin to convert to unsigned integer
return value
# start D1 conversion - pressure (24 bit unsigned)
def start_d1():
# print ('start D1 ')
data = bytearray([r_d1])
i2c.writeto(slave_addr, data)
# start D2 conversion - temperature (24 bit unsigned)
def start_d2():
# print ('start D2 ')
data = bytearray([r_d2])
i2c.writeto(slave_addr, data)
# read pressure sensor ADC
def read_adc(): # read ADC 24 bits unsigned
data = bytearray([r_adc])
i2c.writeto(slave_addr, data)
adc = i2c.readfrom(slave_addr, 3) # ADC is 3 bytes
value = int.from_bytes(adc, "big") # use builtin to convert to integer
return value
# read humidity sensor user register command 0xE7, default value = 0x02
# default resolution RH 12bit, T 14bit
def read_user():
data = bytearray([r_user])
i2c.writeto(slave_addr, data)
value = i2c.readfrom(slave_addr, 1)
return value
# read rh: send trigger rh command 0xE5
def read_rh():
data = bytearray([t_humi])
i2c.writeto(slave_addr, data)
raw_rh = i2c.readfrom(slave_addr, 2) # raw RH is 2 bytes
raw_value = int.from_bytes(raw_rh, "big") # use builtin to convert to integer
# rh_tc = (-0.15)*(25 - read_temp())# RH temp compensation
rh_value = (((raw_value / 65536) * 125) - 6) # calculate RH
return rh_value
# **************** Main Program *********************************
# set up variables d2, d3, d4 as pin outputs and set to 0
# use D4 to /EN TPS2042D power distribution switch, power MS8607 and voltage divider for ADC pin D0
d2 = Pin.board.D2
d2.mode(Pin.OUT)
d2.value(0)
d3 = Pin.board.D3
d3.mode(Pin.OUT)
d3.value(0)
d4 = Pin.board.D4
d4.mode(Pin.OUT)
d4.value(0) # set low to enable TPS2042P to enable i2c during startup
# set DIO9 and DIO10 to outputs with value 0
d9 = Pin.board.D9
d9.mode(Pin.OUT)
d9.value(0)
d10 = Pin.board.D10
d10.mode(Pin.OUT)
d10.value(0)
# set up ADC with 2.5V reference for pin D0 as analog input
#x = xbee.Xbee()
xbee.atcmd('AV', 1)
apin = machine.ADC('D0')
# delay for 30 seconds (long delay to connect to device if needed after reset)
#print("start 90 seconds delay")
utime.sleep(90)
#print("end 90 seconds delay")
try:
print('i2c scan addresses found: ', scan_I2C())
except:
print('i2c scan addresses not found')
try:
print('perform reset on pressure sensor, code = ', reset_ps())
except:
print('cannot reset pressure sensor')
# read and print humidity sensor user register
# slave_addr = h_address # set humidity sensor i2c address
# print('user register: ', read_user())
# read press sensor calibration PROM
slave_addr = p_address
try:
C1 = read_c1()
C2 = read_c2()
C3 = read_c3()
C4 = read_c4()
C5 = read_c5()
C6 = read_c6()
except:
print('cannot read pressure sensor calibration')
#print('PROM C1 = ', C1)
#print('PROM C2 = ', C2)
#print('PROM C3 = ', C3)
#print('PROM C4 = ', C4)
#print('PROM C5 = ', C5)
#print('PROM C6 = ', C6)
# check zigbee connection
while xbee.atcmd("AI") != 0:
print("#Trying to Connect...")
utime.sleep(1)
print("#Online...")
主循环
# Main loop
while True:
# set pin d4 low to enable TPS2042
# d2.value(1)
d4.value(0)
# delay for 2 seconds to stabilize
utime.sleep(2)
# take ADC reading on pin D0 (voltage divider across supply/battery)
raw_val = apin.read()
val_mv = int((raw_val * 2500)/4095 * 2)
print('supply voltage %d mV' % val_mv)
# start on D1 conversion for pressure sensor
try:
slave_addr = p_address # set i2c address to pressure sensor
start_d1() # start D1 conversion
utime.sleep(1) # short delay during conversion
raw_d1 = read_adc()
#print("D1= ", raw_d1)
except:
print("D1 conversion failed")
# start D2 conversion for temperature
try:
start_d2() # start D2 conversion
utime.sleep(1)
raw_d2 = read_adc()
#print("D2= ", raw_d2)
except:
print("D2 conversion failed")
# calulate pressure and temperature
try:
# difference between actual and ref P temp
dT = raw_d2 - (C5 * 256)
#print("dT= ", dT)
#
Temp = 2000 + (dT * (C6 / 8388608))
if Temp < 2000: # add 2nd order correction when temp < 20C
T2 = 3 * dT ** 2 / 8589934592
OFF2 = (61 * (Temp - 2000) ** 2) / 16
SENS2 = (29 * (Temp - 2000) ** 2) / 16
else:
T2 = 5 * dT ** 2 / 274877906944
OFF2 = 0
SENS2 = 0
if Temp < -1500: # add 2nd order correction when temp < -15C
OFF2 = OFF2 + (17 * (Temp + 1500) ** 2)
SENS2 = SENS2 + (9 * (Temp + 1500) ** 2)
# calculate corrected temp
Temp = (2000 - T2 + (dT * (C6 / 8388608))) / 100
fTemp = int(Temp * 9 / 5 + 32)
OFF = (C2 * 131072) + (C4 * dT / 64) - OFF2 # offset at actual P temperature
#print("OFF= ", OFF)
SENS = (C1 * 65536) + (C3 * dT / 128) - SENS2 # pressure offset at actual temperature
#print("SENS= ", SENS)
Pres = (raw_d1 * SENS / 2097152 - OFF) / 3276800 # barometric pressure
#print('P Temp = ', '%.1fC' % Temp)
#print('P Temp = ', '%.1fF' % fTemp)
#print('T2 = ', T2)
#print('OFF2 = ', OFF2)
#print('SENS2 = ', SENS2)
#print('Pressure = ', '%.1f ' % Pres)
utime.sleep(1)
except:
print("Temp and Pressure calculation failed")
# start on humidity sensor
try:
slave_addr = h_address
RH = int(read_rh() - 3.6 - (0.18 * Temp))
print('relative humidity: ', '%.1f percent' % RH) # temp compensated humidity
except:
print("humidity sensor conversion failed")
# turn off TPS2042
d4.value(1)
# build PTH sensor data payload for battery voltage, temp and humidity
try:
time_snapshot = str(utime.ticks_cpu())
# print_PTH = "PTH_sensor:" + time_snapshot + ":Press:" + str(Pres) + "mB:Temp:" + str(fTemp) + "F:Temp1:" + str(ftemp9) + "F:Humidity:" + str(RH) + "%:#"
# print_PTH = "PTH_sensor:" + time_snapshot + ":Press:" + str(Pres) + "mB:Temp:" + str(fTemp) + "F:Humidity:" + str(RH) + "%:#" #remove MCP9808
print_PTH = "PTH_sensor:" + time_snapshot + ":V_Ba:" + str(val_mv) + ":T_F:" + str(fTemp) + ":H_%:" + str(
RH) + ":#"
print(print_PTH)
except:
print("sensor payload build failed")
Pres = 99
fTemp = 99
RH = 99
# transmit PTH data over Zigbee to coordinator
d3.value(1)
try:
xbee.transmit(TARGET_64BIT_ADDR, print_PTH)
# xbee.transmit(ROUTER_64BIT_x1B2D, print_PTH)
#
except:
print("xbee coordinator transmit failed")
# set pin d4 high to turn off TPS2042P
#d4.value(1)
# print("set d4 high and delay for 1 seconds")
# utime.sleep(1)
# d2.value(0)
# d3.value(0)
print("sleeping for 60 seconds")
#sleep_ms = x.sleep_now(10000, True)
sleep_ms = x.sleep_now(60000, True)
print("woke up ")
该项目的最新源代码:sensor_ms8607THV在github上的链接如下:
设置 JEDI One
1 - 如果Raspberry Pi上尚未安装machinechat JEDI One,请参见以下内容:
- 获取Raspberry Pi版本的JEDI One DK-JEDIONE-RP
- 在Pi上安装,参见Raspberry Pi - 将JEDI One安装为服务
2 - 如果尚未实现,请设置Machinechat与Zigbee到WiFi传感器桥接(详见项目详情)。
3 - 设置JEDI One仪表板
在JEDI One中,选择“Dashboards”选项卡,然后选择“+”以添加新图表并进行配置。
命名图表,选择“Chart Type”,选择“Source”(PTH_xxx),选择“Property”(data2),输入“Units”并输入“Refresh Interval”。重复第二个图表,命名图表,选择“Source”(PTH_xxx),选择“Property”(data1),输入“Units”并输入“Refresh Interval”。完成后,仪表板应类似于下图。
结论
Digi的Xbee3 Zigbee模块、TE Connectivity的MS8607传感器、TI的TPS2042P电源IC和Machinechat的JEDI One物联网平台的结合,形成了一个功能强大的无线监控系统,用于监控冷冻室的温度和无线传感器的电池状态。该项目可以轻松修改以监控其他参数和传感器。JEDI One 可以轻松配置为在条件低于预设限制时提供电子邮件或短信警报。




