Machinechat 与电池供电的 Xbee3 无线 Zigbee MS8607 环境传感器

本项目使用 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 秒的睡眠周期,在读取和传输传感器数据之间以最小化电流消耗并延长电池寿命。下面的示意图展示了电路的实现方式。


04_00

Zigbee传感器Scheme-it原理图和BOM链接

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,请参见以下内容:

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 可以轻松配置为在条件低于预设限制时提供电子邮件或短信警报。

参考文献