xbee-recv.py

#!/usr/bin/env python3

import sys
import os
from serial import Serial

def read_frame(io):
  sbuf = io.read(1)
  if not sbuf[0] == 0x7E: return None

  sbuf = io.read(2)
  fbuf = bytearray(sbuf[0] << 8 | sbuf[1])

  for i in range(len(fbuf)):
    sbuf = io.read(1)
    if sbuf[0] != 0x7D:
      fbuf[i] = sbuf[0]
    else:
      sbuf = io.read(1)
      fbuf[i] = sbuf[0] ^ 0x20

  csum = 0xFF - (sum(fbuf) & 0xFF)
  sbuf = io.read(1)
  if sbuf[0] != csum: return None

  return fbuf

API_IOSAMPLE = 0x83

def decode_frame(fbuf):
  rec = None

  if fbuf[0] == API_IOSAMPLE:
    rec = {
      'type': fbuf[0],
      'sender': fbuf[1] << 8 | fbuf[2],
      'rssi': fbuf[3],
      'options': fbuf[4],
      'nr': fbuf[5],
      'mask': fbuf[6] << 8 | fbuf[7],
      'digital': fbuf[8] << 8 | fbuf[9],
    }

    rest = fbuf[10:]
    rec['analog'] = [(rest[i] << 8 | rest[i+1]) for i in range(0, len(rest), 2)]

  return rec

def main(args):
  dev = args[1]
  sio = Serial(dev, 9600)

  while True:
    fbuf = read_frame(sio)
    if fbuf:
      rec = decode_frame(fbuf)
      if rec:
        print(rec)

def usage_format():
  p = os.path.basename(sys.argv[0])
  return """
{p} - Dump XBee API frame received over a serial port
Usage: {p} <device>
Example:
  $ {p} /dev/ttyUSB2
  $ {p} /dev/xbee-adapter
Note:
- This code is incomplete and can only decode selected frame types.
""".lstrip().format(**locals())

def usage():
  sys.stderr.write(usage_format())
  sys.exit(0)

if __name__ == '__main__':
  if len(sys.argv) < 2:
    usage()
  main(sys.argv)