serial_spy.py

#!/usr/bin/python

# Thanks to "Very simple serial terminal" by Chris Liechti <[email protected]>
import sys, re, serial, threading, datetime, time
try:
    from serial.tools.list_ports import comports
except ImportError:
    comports = None

#WHEN_TO_PRINT = ''         # NOW
WHEN_TO_PRINT = chr(0x0A)   # LineFeed

thread_list = []

def hexdump(port, src, length=16, sep='.'):
    stamp = datetime.datetime.now()
    FILTER = ''.join([(len(repr(chr(x))) == 3) and chr(x) or sep for x in range(256)])
    lines = []
    for c in xrange(0, len(src), length):
        chars = src[c:c+length]
        hex = ' '.join(["%02x" % ord(x) for x in chars])
        if len(hex) > 24:
            hex = "%s %s" % (hex[:24], hex[24:])
        printable = ''.join(["%s" % ((ord(x) <= 127 and FILTER[ord(x)]) or sep) for x in chars])
        lines.append("%08x:  %-*s  |%s|\n" % (c, length*3, hex, printable))
    sys.stdout.write("\n%s\t%s\n%s" % (stamp, port, ''.join(lines)))
    sys.stdout.flush()

def dump_port_list():
    re_usb = re.compile('.*ttyUSB\d+')
    ports = []
    if comports:
        sys.stderr.write('\n--- Opening ports:\n')
        for port, desc, hwid in sorted(comports()):
            #~ sys.stderr.write('--- %-20s %s [%s]\n' % (port, desc, hwid))
            if re_usb.match(port):
                sys.stderr.write('--- %-20s %s\n' % (port, desc))
                ports.append(port)
    return ports

class serial_spy(object):
    def __init__(self, port, terminator='', baudrate=115202, parity=serial.PARITY_NONE, rtscts=False, xonxoff=False, echo=False):
        try:
            self.serial = serial.serial_for_url(port, baudrate, parity=parity, rtscts=rtscts, xonxoff=xonxoff, timeout=1)
        except AttributeError:
            # happens when the installed pyserial is older than 2.5. use the
            # Serial class directly then.
            self.serial = serial.Serial(port, baudrate, parity=parity, rtscts=rtscts, xonxoff=xonxoff, timeout=1)
        self.echo = echo
        self.dtr_state = True
        self.rts_state = True
        self.break_state = False
        self.terminator = terminator
        self.buf = ''


    def start(self):
        """Start reader thread"""
        self.alive = True
        # start serial->console thread
        self.receiver_thread = threading.Thread(target=self.reader)
        self.receiver_thread.daemon = True
        self.receiver_thread.start()

    def reader(self):
        """loop and copy serial->console"""
        try:
            while self.alive:
                #print "%s is alive!" % self.serial.port
                while self.serial.inWaiting():
                    data = self.serial.read(self.serial.inWaiting())
                    if self.terminator == '':
                        print "terminator NOW"
                        hexdump(self.serial.port, data)
                    else:
                        self.buf += data
                        a = self.buf.rpartition(self.terminator)
                        self.buf = a[2]
                        string = ''.join(a[0:2])
                        if string:
                            hexdump(self.serial.port, string)
        except serial.SerialException, e:
            self.alive = False
            # would be nice if the console reader could be interruptted at this
            # point...
            raise

def main():
    for port in dump_port_list():
        try:
            this = serial_spy(port, WHEN_TO_PRINT, 115200)
        except serial.SerialException, e:
            sys.stderr.write("could not open port %r: %s\n" % (port, e))
            sys.exit(1)

        this.start()
        thread_list.append(this)

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print "Keyboard interrupt"
        exit()

# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
if __name__ == '__main__':
    main()