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.
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()
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()