test_digixbee.py

# install the digi-xbee package
import binascii
import json
import os
import pprint
import time

from digi.xbee.devices import XBeeDevice
from digi.xbee.models.address import XBee64BitAddress

tree = json.loads(os.environ["ACC_TREE_JSON"])
pfid = "0007-04-01-01"
addr64 = tree["enabledACS"][pfid]["xb_address"]


print("initializing zb")
device = XBeeDevice("/dev/ttyUSB1", 19_200)
device.open()
print("Hardware version: %s" % device.get_hardware_version().description)

# ------------- Configuring coordinator

device.set_parameter("NI", b"Xbee3 Coordinator")
device.reset()
print(device.get_parameter("NI"))
device.apply_changes()


# -------------  Discovering network
# Idea: scan and verify only and all ACC_TREE_JSON nodes are in network
xnet = device.get_network()

# Start the discovery process and wait for it to be over.
xnet.start_discovery_process()
while xnet.is_discovery_running():
    time.sleep(0.5)
print("network found")

print(xnet.get_devices())

# -------------  Configuring remote node
zb1 = xnet.get_device_by_64(XBee64BitAddress.from_hex_string(addr64))
print(zb1)
zb1.read_device_info()
print(zb1.get_hardware_version().description)
print(zb1.get_parameter("NI"))

zb1.set_parameter("NI", b"Xbee2 01")
zb1.apply_changes()
print(zb1.get_parameter("NI"))


# -------------  Receiving data
def my_data_received_callback(xbee_message):
    address = xbee_message.remote_device.get_64bit_addr()
    addr16 = xbee_message.remote_device.get_16bit_addr()
    data = xbee_message.data.decode("utf8")
    print("Received data from %s (%s): %s" % (address, addr16, data))


device.add_data_received_callback(my_data_received_callback)

while True:
    time.sleep(1)