Raspberry Pi and XBee Asynchronous Operation

In my last post <link> I whined a bit about using the Python XBee library on the Pi; it's complicated, but I made it work.  This is not actually too hard to understand, but it is more complicated that just checking to see if something is out there and then using it.

What I did was use the XBee library's asynchronous call, which creates another thread, to receive the message, then put the message on a queue.  In the original thread, I check the queue and pull off the message, dismantle it and use it.  Actually, this isn't a terrible way to do it, just very different from what I've done before.

Then, I realized that waiting on the queue to have something in it was silly.  I simply checked the queue to see if something was there and if not continued.  If there is something there, I call a routine to handle it.  To test it, I added the Python scheduler to the code and set up a timer so that, every 30 seconds, I send a status request message out to my network, and see if the answer comes back.  This is a feature of my network, not a general thing.  What I did was enable the current Arduino house controller to respond to a very simple message by sending the status of a few devices as a broadcast.  This allows me to have devices that send the query and look at the response to see what's going on.  It also helps by being a source of messages that I could look for.

The code got pretty complex, but I tried to comment the heck out of it so you can see what is going on:

The Raspberry Pi Script

#! /usr/bin/python
# This is an example of asyncronous receive
# What it actually does is fork off a new process
# to do the XBee receive.  This way, the main
# code can go do somthing else and hand waiting
# for the XBee messages to come in to another
# process.

from xbee import ZigBee
from apscheduler.scheduler import Scheduler
import time
import serial
import Queue

# on the Raspberry Pi the serial port is ttyAMA0
PORT = '/dev/ttyAMA0'
BAUD_RATE = 9600

# The XBee addresses I'm dealing with
BROADCAST = '\x00\x00\x00\x00\x00\x00\xff\xff'
UNKNOWN = '\xff\xfe' # This is the 'I don't know' 16 bit address


packets = Queue.Queue()

# Open serial port
ser = serial.Serial(PORT, BAUD_RATE)

# this is a call back function.  When a message
# comes in this function will get the data
def message_received(data):
        packets.put(data, block=False)
        print 'gotta packet'

def sendPacket(where, what):
        # I'm only going to send the absolute minimum.
        zb.send('tx',
                dest_addr_long = where,
                # I always use the 'unknown' value for this
                # it's too much trouble to keep track of two
                # addresses for the device
                dest_addr = UNKNOWN,
                data = what)

# In my house network sending a '?\r' (question mark, carriage
# return) causes the controller to send a packet with some status
# information in it as a broadcast.  As a test, I'll send it and
# the receive above should catch the response.
def sendQueryPacket():
        # I'm broadcasting this message only
        # because it makes it easier for a monitoring
        # XBee to see the packet.  This is a test
        # module, remember?
        print 'sending query packet'
        sendPacket(BROADCAST, '?\r')

# OK, another thread has caught the packet from the XBee network,
# put it on a queue, this process has taken it off the queue and
# passed it to this routine, now we can take it apart and see
# what is going on ... whew!
def handlePacket(data):
        print 'In handlePacket: ',
        print data['id'],
        if data['id'] == 'tx_status':
                print data['deliver_status'].encode('hex')
        elif data['id'] == 'rx':
                print data['rf_data']
        else:
                print 'Unimplemented frame type'


# Create XBee library API object, which spawns a new thread
zb = ZigBee(ser, callback=message_received)

sendsched = Scheduler()
sendsched.start()

# every 30 seconds send a house query packet to the XBee network
sendsched.add_interval_job(sendQueryPacket, seconds=30)

# Do other stuff in the main thread
while True:
        try:
                time.sleep(0.1)
                if packets.qsize() > 0:
                        # got a packet from recv thread
                        # See, the receive thread gets them
                        # puts them on a queue and here is
                        # where I pick them off to use
                        newPacket = packets.get_nowait()
                        # now go dismantle the packet
                        # and use it.
                        handlePacket(newPacket)
        except KeyboardInterrupt:
                break

# halt() must be called before closing the serial
# port in order to ensure proper thread shutdown
zb.halt()
ser.close()

Yes, this also means I can both send and receive XBee messages.  So, I have a scheduled XBee message that can get the response, a separate thread to receive XBee messages and not load down the main thread, the ability to get the data out of the message and use it.  Now, I need to combine that with a web server that can display the results.

One thing that needs to be understood about the Python XBee library:  it only returns good packets.  If there is a collision and the checksum doesn't work, you don't get the message.  If the message is fragmented by noise, you don't get the message.  If anything goes wrong, you don't get the message.  That makes debugging a network problem almost impossible since you can't see anything when things go bad.  So, keep an arduino around to look at stuff or build a sniffer like I did <link> to follow the traffic.  XCTU can help, but remember, if the message is sent to a specific address, it's invisible to the output of a different XBee and XBees don't have a promiscuous mode like Ethernet chips.

I'm going to look at web servers now.

Edit: It took exactly 5 minutes to get a web server running.  I'm starting to like this board.