Hello,
I have recently purchased Digi Xbee3 zigbee mesh kit. I am using two of the Xbee devices(Sender and Receiver) with firmware: Digi Xbee3 802.15.4 (latest). Done all initial configuration and checked throughput/range test/sending-receiving API frames from console of XCTU(version 6.5.12) everything is working fine.
Now I want to do some test where I send API frames from Sender to Receiver for 1 minute and get throughput / RSSI / packet lost values. To do so I am trying to use a python script by using python digi xbee libraries.
First, I sent a frame from Sender by using Python, I have confirmed that the frame is correctly received and decoded from the XCTU console (the transfer/receive rate also seems to be fine). I am also able to receive by using Python script.but it is too slow when read_data() is used to read the data. The transmitter send all frames on time, but receiver is unable to cope up with the rate and does not display the output fast enough.
I am including the Python codes below for transmitter and receiver here. Please let me know if I am doing any mistake / a way to fix this.
Transmitter
import time
from digi.xbee.devices import XBeeDevice, RemoteXBeeDevice, XBee64BitAddress, XBeeMessage
SERIAL_PORT = '/dev/tty.usbserial-AB0OO9Y6'
BAUD_RATE = 9600
transmitter = XBeeDevice(SERIAL_PORT, BAUD_RATE)
# NEW_TIMEOUT_FOR_SYNC_OPERATIONS = 5 # sec
print("Current timeout: %d seconds" %transmitter.get_sync_ops_timeout())
try:
transmitter.open()
destination_address = RemoteXBeeDevice(transmitter, XBee64BitAddress.from_hex_string("13A2004201081B"))
payload_data = b"Hi ! Communicating over 802.15.4"
transmission_interval = 0.5 # sec
print("Transmitting data...")
start_time = time.time()
transmitted_packets = 0
while time.time() - start_time < 30:
transmitter.send_data(destination_address, payload_data)
print('Message is sent')
transmitted_packets += 1
time.sleep(transmission_interval)
print("Transmission completed.")
print("Total transmitted packets:", transmitted_packets)
except Exception as e:
print("Error:", str(e))
finally:
# Close the connection
transmitter.close()
Receiver
import time
from digi.xbee.devices import XBeeDevice, RemoteXBeeDevice, XBee64BitAddress
SERIAL_PORT = '/dev/tty.usbserial-AB0OPIU8'
BAUD_RATE = 9600
receiver = XBeeDevice(SERIAL_PORT, BAUD_RATE)
try:
receiver.open()
payload_size = len(b"Hi ! Communicating over 802.15.4")
print("Receiving data...")
start_time = time.time()
received_packets = 0
lost_packets = 0
total_bytes_received = 0
while time.time() - start_time < 60:
xbee_message = receiver.read_data(timeout=6)
# print(xbee_message)
if xbee_message:
received_packets += 1
payload_data = xbee_message.data
print(payload_data)
finally:
receiver.close()
Thank you
Ayaz