CVS: examples/libraries/monitor monitor.py, NONE, 1.1 main.c, 1.1, 1.2

Update of /cvsroot/mspgcc/examples/libraries/monitor
In directory sc8-pr-cvs16.sourceforge.net:/tmp/cvs-serv399

Modified Files:
	main.c 
Added Files:
	monitor.py 
Log Message:
- unify pin read answers
- add example code for machine mode


--- NEW FILE: monitor.py ---
#! /usr/bin/env python

"""\
MSP430 monitor example, using the machine mode.

This modlue can be used either as command line application for simple tests
or as extansion in other Python programs.

For the usage as command line tool, see the usage description down in this
file.

Usage as Python module:

>>> m = Monitor()
>>> m.open('/dev/ttyUSB0')
>>> m.erase(0x1000)
>>> m.read(0x1000, 32) == 32*'\xff'
True
>>> m.write(0x1000, "hello world")
>>> m.read(0x1000, 11)
'hello world'
>>> for n in range(3):
...     for i in range(8):
...         m.pin(5, None, 0)
...         m.pin(5, i, 1)
>>>

Of course it is also possible to use Monitor as base class.
"""

import time
import serial
import logging
import logging.config

MACHINE_START           = '\x1f'          # US
MACHINE_EXITCODE        = '\x1c'          # FS
MACHINE_END             = '\x04'          # EOT


class MonitorError(Exception): pass
class MonitorTimeout(MonitorError): pass

class Monitor:
    """This class provides access to the monitor"""
    
    def __init__(self):
        self.serial = None
        self.logger_box = logging.getLogger('Monitor')
        
    def open(self, port=None, baudrate=115200):
        self.logger_box.info('Opening serial port %r' % port)
        self.serial = serial.Serial(
            port,
            baudrate=baudrate,
            timeout=1,
        )
        self.reset()

    def __del__(self):
        self.close()
    
    def close(self):
        if self.serial is not None:
            self.serial.close()
            self.serial = None
    
    def command(self, command, quiet=False, timeout=4):
        """send a command and receive answer using the machine mode.
           quiet=True disables logging"""
        msg = '%s%s\n' % (MACHINE_START, command.encode('latin1'))
        self.serial.flushInput()
        #~ time.sleep(0.2)
        input = []
        line = 0
        self.logger_box.debug('command %r' % (msg, ))
        self.serial.write(msg) #in case unicode is passed
        while True:
            byte = self.serial.read(1)
            #~ self.logger_box.debug('byte %r' % (byte,))
            if not byte:
                # timeout disabled?
                if timeout is None: continue
                if timeout: timeout -= 1
                if timeout == 0:
                    if not quiet: self.logger_box.error('timeout %r' % ''.join(input))
                    raise MonitorTimeout('Timeout. No answer from Monitor.')
                else:
                    continue
            block = byte + self.serial.read(self.serial.inWaiting())
            for character in block:
                #~ self.logger_box.debug('character %r' % (character))
                if character == MACHINE_EXITCODE:
                    answer = ''.join(input)
                    del input[:]
                    stdout = None
                elif character == MACHINE_END:
                    try:
                        exitcode = int(''.join(input))
                    except ValueError:
                        raise MonitorError('garabge in exit code')
                    if not quiet: self.logger_box.info('answer %r' % (answer))
                    if exitcode:# != 0
                        if exitcode == -1:
                            raise MonitorError("no such command: %r" % command, exitcode)
                        elif exitcode == 1:
                            raise MonitorError("parameter error", exitcode)
                        else:
                            raise MonitorError("command failed with exit code %d" % exitcode, exitcode)
                    return answer
                else:
                    input.append(character)
    
        
    def abort(self):
        """\
        Abort a running command.
        """
        self.logger_box.info('abort')
        self.serial.write('\x03')
        time.sleep(0.1)

    def sync(self):
        """\
        Check connection to box. Send empty command and expect exit code 0.
        """
        self.logger_box.info('sync')
        self.command('')
    
    def reset(self):
        """\
        Reset box.
        """
        self.logger_box.info('reset')
        try:
            self.abort()
            self.serial.write('reset\n')
            time.sleep(0.5)
            while True:
                b = self.serial.read(1)
                if not b or b == '$': break
        except MonitorTimeout:
            pass
        self.sync()


    def adc(self, channel):
        """\
        Make an ADC measurement.
        """
        self.logger_box.info('adc')
        line = self.command('adc %s' % channel)
        return int(line)
    

    def erase(self, address):
        """\
        Erase memory
        """
        self.logger_box.info('erase 0x%04x' % (address, ))
        lines = self.command('erase %d' % (address, ))
    
    def write(self, address, data):
        """\
        Write memory
        """
        self.logger_box.info('write 0x%04x %r' % (address, data))
        lines = self.command('write %d %s' % (address, data.encode('hex')))
    
    def read(self, address, size):
        """\
        Read memory
        """
        self.logger_box.info('read 0x%04x %d' % (address, size))
        lines = self.command('hexdump %d %d' % (address, size))
        return ''.join([line[7:55].replace(' ', '').decode('hex') for line in lines.splitlines()])
    
    
    def pin(self, port, pin=None, value=None):
        """\
        Set pins
        """
        self.logger_box.info('pin %s %s %s' % (port, pin, value))
        if value is None:
            if pin is None:
                lines = self.command('P%d' % (port, ))
            else:
                lines = self.command('P%d.%d' % (port, pin))
            return int(lines)
        else:
            if pin is None:
                self.command('P%d=%d' % (port, value))
            else:
                self.command('P%d.%d=%d' % (port, pin, value))



if __name__ == '__main__':
    import sys
    import optparse

    parser = optparse.OptionParser(
        usage="""\
%prog [options] command [command ...]

"command" is executed on the target. When multiple commands are given, one
after the other is executed. Parameters to commands are passed within quotes:

    monitor.py -p /dev/ttyUSB0 "adc 1"

primitive performance test:
    monitor.py -p /dev/ttyUSB0 --stats --repeat --quiet "P5.1=1"

run the built-in doctests:
    monitor.py -p /dev/ttyUSB0 __test__
""")
    
    parser.add_option("-p", "--port", dest="port",
        help="specify seriall port", default=None)
        
    parser.add_option("", "--repeat", dest="repeat", action="store_true",
        help="repeat the given commands forever in a loop", default=False)
    
    parser.add_option("-q", "--quiet", dest="quiet", action="store_true",
        help="don't produce any outputs", default=False)
        
    parser.add_option("", "--stats", dest="statistics", action="store_true",
        help="print statistics", default=False)
    
    parser.add_option("-D", "--debug", dest="debug", action="store_true",
        help="enable debug outputs", default=False)

    parser.add_option("-b", "--baudrate", dest="baudrate", action="store", type='int',
        help="set baudrate, default=115200", default=115200)

    (options, args) = parser.parse_args()
    
    logging.basicConfig()
    logger = logging.getLogger('Monitor')
    
    logger.setLevel(logging.WARNING)

    if options.debug:
        logger.setLevel(logging.NOTSET)
        logging.getLogger().setLevel(logging.NOTSET)
    
    if not args:
        parser.error('missing command')
    elif '__test__' in args:
        import doctest
        doctest.testmod()
        sys.exit(0)
    
    # instantiate BSL communication object
    box = Monitor()
    box.open(
        options.port,
        baudrate=options.baudrate,
    )
    
    start_time = time.time()
    try:
        iteration = 1
        try:
            while True:
                try:
                    for arg in args:
                        answer = box.command(arg)
                        if answer and not options.quiet: print answer
                    if not options.repeat: break
                    if options.statistics: # and (iteration & 0xf) == 0:
                        sys.stdout.write('iteration %d\r' % iteration)
                    iteration += 1
                except MonitorError:
                    if options.repeat:
                        sys.stderr.write("failure at iteration %d\n" % iteration)
                    raise
        finally:
            end_time = time.time()
            if options.statistics:
                sys.stdout.write('Run time: %.1f s for %d iterations (%.2f iter/s)\n' % (
                    end_time - start_time,
                    iteration,
                    iteration / (end_time - start_time)
                ))
    except SystemExit:
        raise
    except KeyboardInterrupt, e:
        sys.stderr.write('User abort\n')
        sys.exit(1)
    except Exception, e:
        if options.debug:
            raise
        sys.stderr.write("ERROR: %s\n" % e)
        sys.exit(2)

Index: main.c
===================================================================
RCS file: /cvsroot/mspgcc/examples/libraries/monitor/main.c,v
retrieving revision 1.1
retrieving revision 1.2
diff -u -w -d -r1.1 -r1.2
--- main.c	24 Mar 2008 22:07:46 -0000	1.1
+++ main.c	25 Mar 2008 00:45:46 -0000	1.2
@@ -292,7 +292,7 @@
             if (input[4] == '\0') {
                 // read
                 *PORT_DIR[port] &= ~mask;
-                printf("%s=%c\n", input, (*PORT_IN[port] & mask) ? '1' : '0');
+                printf("%c\n", (*PORT_IN[port] & mask) ? '1' : '0');
                 return SUCCESS;
             } else if (input[4] == '=') {
                 //write
@@ -312,7 +312,7 @@
             return SUCCESS;
         } else if (input[2] == '\0') {
             *PORT_DIR[port] = 0;
-            printf("0x%02x\n", *PORT_IN[port]);
+            printf("%d\n", *PORT_IN[port]);
             return SUCCESS;
         } else if (input[2] == '=') {
             // port assignment


-------------------------------------------------------------------------
This SF.net email is sponsored by: Microsoft
Defy all challenges. Microsoft(R) Visual Studio 2008.
http://clk.atdmt.com/MRT/go/vse0120000070mrt/direct/01/