From: Olivier Matz Date: Thu, 17 Jul 2014 17:56:46 +0000 (+0200) Subject: add a script for xbee tests X-Git-Url: http://git.droids-corp.org/?a=commitdiff_plain;h=0704ac13884035e9124507092664910f9936bb5d;p=protos%2Fxbee-avr.git add a script for xbee tests --- diff --git a/dump_stats.py b/dump_stats.py new file mode 100644 index 0000000..eab8833 --- /dev/null +++ b/dump_stats.py @@ -0,0 +1,210 @@ +#! /usr/bin/env python + +import os,sys,termios,atexit +import serial +from select import select +import cmd +#import pylab +from matplotlib import pylab +from math import * + +import struct +import numpy +import shlex +import time +import math +import warnings +warnings.filterwarnings("ignore","tempnam",RuntimeWarning, __name__) + +import logging +log = logging.getLogger("XbeeShell") +_handler = logging.StreamHandler() +_handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s")) +log.addHandler(_handler) +log.setLevel(1) + +class SerialLogger: + def __init__(self, ser, filein, fileout=None): + self.ser = ser + self.filein = filein + self.fin = open(filein, "a", 0) + if fileout: + self.fileout = fileout + self.fout = open(fileout, "a", 0) + else: + self.fileout = filein + self.fout = self.fin + def fileno(self): + return self.ser.fileno() + def read(self, *args): + res = self.ser.read(*args) + self.fin.write(res) + return res + def write(self, s): + self.fout.write(s) + self.ser.write(s) + def readline(self, *args): + res = self.ser.readline(*args) + self.fin.write(res) + return res + +class Interp(cmd.Cmd): + prompt = "Xbee> " + def __init__(self, tty, baudrate=57600): + cmd.Cmd.__init__(self) + self.ser = serial.Serial(tty,baudrate=baudrate,timeout=0.1) + self.escape = "\x01" # C-a + self.quitraw = "\x02" # C-b + self.serial_logging = False + self.default_in_log_file = "/tmp/xbee.in.log" + self.default_out_log_file = "/tmp/xbee.out.log" + + def do_log(self, args): + """Activate serial logs. + log logs input and output to + log logs input to and output to + log logs to /tmp/microb.log or the last used file""" + + if self.serial_logging: + log.error("Already logging to %s and %s" % (self.ser.filein, + self.ser.fileout)) + else: + self.serial_logging = True + files = [os.path.expanduser(x) for x in args.split()] + if len(files) == 0: + files = [self.default_in_log_file, self.default_out_log_file] + elif len(files) == 1: + self.default_in_log_file = files[0] + self.default_out_log_file = None + elif len(files) == 2: + self.default_in_log_file = files[0] + self.default_out_log_file = files[1] + else: + print "Can't parse arguments" + + self.ser = SerialLogger(self.ser, *files) + log.info("Starting serial logging to %s and %s" % (self.ser.filein, + self.ser.fileout)) + + + def do_unlog(self, args): + if self.serial_logging: + log.info("Stopping serial logging to %s and %s" % (self.ser.filein, + self.ser.fileout)) + self.ser = self.ser.ser + self.serial_logging = False + else: + log.error("No log to stop") + + def do_raw(self, args): + "Switch to RAW mode" + stdin = os.open("/dev/stdin",os.O_RDONLY) + stdout = os.open("/dev/stdout",os.O_WRONLY) + + stdin_termios = termios.tcgetattr(stdin) + raw_termios = stdin_termios[:] + + try: + log.info("Switching to RAW mode") + + # iflag + raw_termios[0] &= ~(termios.IGNBRK | termios.BRKINT | + termios.PARMRK | termios.ISTRIP | + termios.INLCR | termios.IGNCR | + termios.ICRNL | termios.IXON) + # oflag + raw_termios[1] &= ~termios.OPOST; + # cflag + raw_termios[2] &= ~(termios.CSIZE | termios.PARENB); + raw_termios[2] |= termios.CS8; + # lflag + raw_termios[3] &= ~(termios.ECHO | termios.ECHONL | + termios.ICANON | termios.ISIG | + termios.IEXTEN); + + termios.tcsetattr(stdin, termios.TCSADRAIN, raw_termios) + + mode = "normal" + while True: + ins,outs,errs=select([stdin,self.ser],[],[]) + for x in ins: + if x == stdin: + c = os.read(stdin,1) + if mode == "escape": + mode =="normal" + if c == self.escape: + self.ser.write(self.escape) + elif c == self.quitraw: + return + else: + self.ser.write(self.escape) + self.ser.write(c) + else: + if c == self.escape: + mode = "escape" + else: + self.ser.write(c) + elif x == self.ser: + os.write(stdout,self.ser.read()) + finally: + termios.tcsetattr(stdin, termios.TCSADRAIN, stdin_termios) + log.info("Back to normal mode") + + def do_dump_stats(self, args): + """send hello at power 4 and dump stats regulary""" + + self.ser.write("write power-level 4\n") + prev_reset = time.time() + try: + while True: + # send hellos during 1 s + self.ser.write("rc_proto_hello wing 30 30 tototiti\n") + time.sleep(1.5) + + # send a reset every 60 s + t = time.time() + if t - prev_reset > 60: + self.ser.write("write soft-reset\n") + prev_reset = t + + # display time & stats + self.ser.fin.write("\nTIME:%s\n"%(time.time())) + self.ser.write("rc_proto_stats show\n") + while True: + l = self.ser.readline() + sys.stdout.write(l) + if l == "": + break + except KeyboardInterrupt: + pass + + +if __name__ == "__main__": + try: + import readline,atexit + except ImportError: + pass + else: + histfile = os.path.join(os.environ["HOME"], ".xbee") + atexit.register(readline.write_history_file, histfile) + try: + readline.read_history_file(histfile) + except IOError: + pass + + device = "/dev/ttyS0" + if len(sys.argv) > 1: + device = sys.argv[1] + interp = Interp(device) + interp.do_log("") + while 1: + try: + interp.cmdloop() + except KeyboardInterrupt: + print + except Exception,e: + l = str(e).strip() + if l: + log.exception("%s" % l.splitlines()[-1]) + continue + break