CVS: python/mspgcc/serial __init__.py,NONE,1.1 __init__.pyc,NONE,1.1 serialjava.py,NONE,1.1 serialposix.py,NONE,1.1 serialutil.py,NONE,1.1 serialutil.pyc,NONE,1.1 serialwin32.py,NONE,1.1 serialwin32.pyc,NONE,1.1
Chris Liechti <[email protected]>
| Newsgroups | gmane.comp.hardware.texas-instruments.msp430.gcc.cvs |
|---|---|
| Message-ID | <[email protected]> |
Update of /cvsroot/mspgcc/python/mspgcc/serial In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv24301/python/mspgcc/serial Added Files: __init__.py __init__.pyc serialjava.py serialposix.py serialutil.py serialutil.pyc serialwin32.py serialwin32.pyc Log Message: rename msp430 package to mspgcc (resolves naming/import conflicts with msp430.dll) --- NEW FILE: __init__.py --- #!/usr/bin/env python #portable serial port access with python #this is a wrapper module for different platform implementations # # (C)2001-2002 Chris Liechti <[email protected]> # this is distributed under a free software license, see license.txt import sys, os, string VERSION = string.split("$Revision: 1.1 $")[1] #extract CVS version #chose an implementation, depending on os if os.name == 'nt': #sys.platform == 'win32': from serialwin32 import * elif os.name == 'posix': from serialposix import * elif os.name == 'java': from serialjava import * else: raise Exception("Sorry: no implementation for your platform ('%s') available" % os.name) --- NEW FILE: __init__.pyc --- mò ¦\<Cc --- NEW FILE: serialjava.py --- #!jython #Python Serial Port Extension for Win32, Linux, BSD, Jython #module for serial IO for Jython and JavaComm #see __init__.py # #(C) 2002-2003 Chris Liechti <[email protected]> # this is distributed under a free software license, see license.txt import javax.comm from serialutil import * VERSION = "$Revision: 1.1 $".split()[1] #extract CVS version def device(portnumber): """Turn a port number into a device name""" enum = javax.comm.CommPortIdentifier.getPortIdentifiers() ports = [] while enum.hasMoreElements(): el = enum.nextElement() if el.getPortType() == javax.comm.CommPortIdentifier.PORT_SERIAL: ports.append(el) return ports[portnumber].getName() class Serial(SerialBase): """Serial port class, implemented with javax.comm and thus usable with jython and the appropriate java extension.""" def open(self): """Open port with current settings. This may throw a SerialException if the port cannot be opened.""" if self._port is None: raise SerialException("Port must be configured before it can be used.") if type(self._port) == type(''): #strings are taken directly portId = javax.comm.CommPortIdentifier.getPortIdentifier(self._port) else: portId = javax.comm.CommPortIdentifier.getPortIdentifier(device(self._port)) #numbers are transformed to a comportid obj try: self.sPort = portId.open("python serial module", 10) except Exception, msg: self.sPort = None raise SerialException("Could not open port: %s" % msg) self._reconfigurePort() self._instream = self.sPort.getInputStream() self._outstream = self.sPort.getOutputStream() self._isOpen = True def _reconfigurePort(self): """Set commuication parameters on opened port.""" if not self.sPort: raise SerialException("Can only operate on a valid port handle") self.sPort.enableReceiveTimeout(30) if self._bytesize == FIVEBITS: jdatabits = javax.comm.SerialPort.DATABITS_5 elif self._bytesize == SIXBITS: jdatabits = javax.comm.SerialPort.DATABITS_6 elif self._bytesize == SEVENBITS: jdatabits = javax.comm.SerialPort.DATABITS_7 elif self._bytesize == EIGHTBITS: jdatabits = javax.comm.SerialPort.DATABITS_8 else: raise ValueError("unsupported bytesize: %r" % self._bytesize) if self._stopbits == STOPBITS_ONE: jstopbits = javax.comm.SerialPort.STOPBITS_1 elif stopbits == STOPBITS_ONE_HALVE: self._jstopbits = javax.comm.SerialPort.STOPBITS_1_5 elif self._stopbits == STOPBITS_TWO: jstopbits = javax.comm.SerialPort.STOPBITS_2 else: raise ValueError("unsupported number of stopbits: %r" % self._stopbits) if self._parity == PARITY_NONE: jparity = javax.comm.SerialPort.PARITY_NONE elif self._parity == PARITY_EVEN: jparity = javax.comm.SerialPort.PARITY_EVEN elif self._parity == PARITY_ODD: jparity = javax.comm.SerialPort.PARITY_ODD #~ elif self._parity == PARITY_MARK: #~ jparity = javax.comm.SerialPort.PARITY_MARK #~ elif self._parity == PARITY_SPACE: #~ jparity = javax.comm.SerialPort.PARITY_SPACE else: raise ValueError("unsupported parity type: %r" % self._parity) jflowin = jflowout = 0 if self._rtscts: jflowin |= javax.comm.SerialPort.FLOWCONTROL_RTSCTS_IN jflowout |= javax.comm.SerialPort.FLOWCONTROL_RTSCTS_OUT if self._xonxoff: jflowin |= javax.comm.SerialPort.FLOWCONTROL_XONXOFF_IN jflowout |= javax.comm.SerialPort.FLOWCONTROL_XONXOFF_OUT self.sPort.setSerialPortParams(baudrate, jdatabits, jstopbits, jparity) self.sPort.setFlowControlMode(jflowin | jflowout) if self._timeout >= 0: self.sPort.enableReceiveTimeout(self._timeout*1000) else: self.sPort.disableReceiveTimeout() def close(self): """Close port""" if self._isOpen: if self.sPort: self._instream.close() self._outstream.close() self.sPort.close() self.sPort = None self._isOpen = False def makeDeviceName(self, port): return device(port) # - - - - - - - - - - - - - - - - - - - - - - - - def inWaiting(self): """Return the number of characters currently in the input buffer.""" if not self.sPort: raise portNotOpenError return self._instream.available() def read(self, size=1): """Read size bytes from the serial port. If a timeout is set it may return less characters as requested. With no timeout it will block until the requested number of bytes is read.""" if not self.sPort: raise portNotOpenError read = '' if size > 0: while len(read) < size: x = self._instream.read() if x == -1: if self.timeout >= 0: break else: read = read + chr(x) return read def write(self, data): """Output the given string over the serial port.""" if not self.sPort: raise portNotOpenError self._outstream.write(data) def flushInput(self): """Clear input buffer, discarding all that is in the buffer.""" if not self.sPort: raise portNotOpenError self._instream.skip(self._instream.available()) def flushOutput(self): """Clear output buffer, aborting the current output and discarding all that is in the buffer.""" if not self.sPort: raise portNotOpenError self._outstream.flush() def sendBreak(self): """Send break condition.""" if not self.sPort: raise portNotOpenError self.sPort.sendBreak() def setRTS(self,on=1): """Set terminal status line: Request To Send""" if not self.sPort: raise portNotOpenError self.sPort.setRTS(on) def setDTR(self,on=1): """Set terminal status line: Data Terminal Ready""" if not self.sPort: raise portNotOpenError self.sPort.setDTR(on) def getCTS(self): """Read terminal status line: Clear To Send""" if not self.sPort: raise portNotOpenError self.sPort.isCTS() def getDSR(self): """Read terminal status line: Data Set Ready""" if not self.sPort: raise portNotOpenError self.sPort.isDSR() def getRI(self): """Read terminal status line: Ring Indicator""" if not self.sPort: raise portNotOpenError self.sPort.isRI() def getCD(self): """Read terminal status line: Carrier Detect""" if not self.sPort: raise portNotOpenError self.sPort.isCD() if __name__ == '__main__': s = Serial(0, baudrate=19200, #baudrate bytesize=EIGHTBITS, #number of databits parity=PARITY_EVEN, #enable parity checking stopbits=STOPBITS_ONE, #number of stopbits timeout=3, #set a timeout value, None for waiting forever xonxoff=0, #enable software flow control rtscts=0, #enable RTS/CTS flow control ) s.setRTS(1) s.setDTR(1) s.flushInput() s.flushOutput() s.write('hello') print repr(s.read(5)) print s.inWaiting() del s --- NEW FILE: serialposix.py --- #!/usr/bin/env python #Python Serial Port Extension for Win32, Linux, BSD, Jython #module for serial IO for POSIX compatible systems, like Linux #see __init__.py # #(C) 2001-2003 Chris Liechti <[email protected]> # this is distributed under a free software license, see license.txt # #parts based on code from Grant B. Edwards <[email protected]>: # ftp://ftp.visi.com/users/grante/python/PosixSerial.py # references: http://www.easysw.com/~mike/serial/serial.html import sys, os, fcntl, termios, struct, select, errno from serialutil import * VERSION = "$Revision: 1.1 $".split()[1] #extract CVS version #Do check the Python version as some constants have moved. if (sys.hexversion < 0x020100f0): import TERMIOS else: TERMIOS = termios if (sys.hexversion < 0x020200f0): import FCNTL else: FCNTL = fcntl #try to detect the os so that a device can be selected... plat = sys.platform.lower() if plat[:5] == 'linux': #Linux (confirmed) def device(port): return '/dev/ttyS%d' % port elif plat == 'cygwin': #cywin/win32 (confirmed) def device(port): return '/dev/com%d' % (port + 1) elif plat == 'openbsd3': #BSD (confirmed) def device(port): return '/dev/ttyp%d' % port elif plat[:3] == 'bsd' or \ plat[:7] == 'freebsd' or \ plat[:7] == 'openbsd' or \ plat[:6] == 'darwin': #BSD (confirmed for freebsd4: cuaa%d) def device(port): return '/dev/cuaa%d' % port elif plat[:6] == 'netbsd': #NetBSD 1.6 testing by Erk def device(port): return '/dev/dty%02d' % port elif plat[:4] == 'irix': #IRIX (not tested) def device(port): return '/dev/ttyf%d' % port elif plat[:2] == 'hp': #HP-UX (not tested) def device(port): return '/dev/tty%dp0' % (port+1) elif plat[:5] == 'sunos': #Solaris/SunOS (confirmed) def device(port): return '/dev/tty%c' % (ord('a')+port) elif plat[:3] == 'aix': #aix def device(port): return '/dev/tty%d' % (port) else: #platform detection has failed... print """don't know how to number ttys on this system. ! Use an explicit path (eg /dev/ttyS1) or send this information to ! the author of this module: sys.platform = %r os.name = %r serialposix.py version = %s also add the device name of the serial port and where the counting starts for the first serial port. e.g. 'first serial port: /dev/ttyS0' and with a bit luck you can get this module running... """ % (sys.platform, os.name, VERSION) #no exception, just continue with a brave attempt to build a device name #even if the device name is not correct for the platform it has chances #to work using a string with the real device name as port paramter. def device(portum): return '/dev/ttyS%d' % portnum #~ raise Exception, "this module does not run on this platform, sorry." #whats up with "aix", "beos", .... #they should work, just need to know the device names. #load some constants for later use. #try to use values from TERMIOS, use defaults from linux otherwise TIOCMGET = hasattr(TERMIOS, 'TIOCMGET') and TERMIOS.TIOCMGET or 0x5415 TIOCMBIS = hasattr(TERMIOS, 'TIOCMBIS') and TERMIOS.TIOCMBIS or 0x5416 TIOCMBIC = hasattr(TERMIOS, 'TIOCMBIC') and TERMIOS.TIOCMBIC or 0x5417 TIOCMSET = hasattr(TERMIOS, 'TIOCMSET') and TERMIOS.TIOCMSET or 0x5418 #TIOCM_LE = hasattr(TERMIOS, 'TIOCM_LE') and TERMIOS.TIOCM_LE or 0x001 TIOCM_DTR = hasattr(TERMIOS, 'TIOCM_DTR') and TERMIOS.TIOCM_DTR or 0x002 TIOCM_RTS = hasattr(TERMIOS, 'TIOCM_RTS') and TERMIOS.TIOCM_RTS or 0x004 #TIOCM_ST = hasattr(TERMIOS, 'TIOCM_ST') and TERMIOS.TIOCM_ST or 0x008 #TIOCM_SR = hasattr(TERMIOS, 'TIOCM_SR') and TERMIOS.TIOCM_SR or 0x010 TIOCM_CTS = hasattr(TERMIOS, 'TIOCM_CTS') and TERMIOS.TIOCM_CTS or 0x020 TIOCM_CAR = hasattr(TERMIOS, 'TIOCM_CAR') and TERMIOS.TIOCM_CAR or 0x040 TIOCM_RNG = hasattr(TERMIOS, 'TIOCM_RNG') and TERMIOS.TIOCM_RNG or 0x080 TIOCM_DSR = hasattr(TERMIOS, 'TIOCM_DSR') and TERMIOS.TIOCM_DSR or 0x100 TIOCM_CD = hasattr(TERMIOS, 'TIOCM_CD') and TERMIOS.TIOCM_CD or TIOCM_CAR TIOCM_RI = hasattr(TERMIOS, 'TIOCM_RI') and TERMIOS.TIOCM_RI or TIOCM_RNG #TIOCM_OUT1 = hasattr(TERMIOS, 'TIOCM_OUT1') and TERMIOS.TIOCM_OUT1 or 0x2000 #TIOCM_OUT2 = hasattr(TERMIOS, 'TIOCM_OUT2') and TERMIOS.TIOCM_OUT2 or 0x4000 TIOCINQ = hasattr(TERMIOS, 'FIONREAD') and TERMIOS.FIONREAD or 0x541B TIOCM_zero_str = struct.pack('I', 0) TIOCM_RTS_str = struct.pack('I', TIOCM_RTS) TIOCM_DTR_str = struct.pack('I', TIOCM_DTR) class Serial(SerialBase): """Serial port class POSIX implementation. Serial port configuration is done with termios and fcntl. Runs on Linux and many other Un*x like systems.""" def open(self): """Open port with current settings. This may throw a SerialException if the port cannot be opened.""" if self._port is None: raise SerialException("Port must be configured before it can be used.") self.fd = None #open try: self.fd = os.open(self.portstr, os.O_RDWR|os.O_NOCTTY|os.O_NONBLOCK) except Exception, msg: self.fd = None raise SerialException("Could not open port: %s" % msg) #~ fcntl.fcntl(self.fd, FCNTL.F_SETFL, 0) #set blocking self._reconfigurePort() self._isOpen = True #~ self.flushInput() def _reconfigurePort(self): """Set commuication parameters on opened port.""" if self.fd is None: raise SerialException("Can only operate on a valid port handle") vmin = vtime = 0 #timeout is done via select try: iflag, oflag, cflag, lflag, ispeed, ospeed, cc = termios.tcgetattr(self.fd) except termios.error, msg: #if a port is nonexistent but has a /dev file, it'll fail here raise SerialException("Could not configure port: %s" % msg) #set up raw mode / no echo / binary cflag |= (TERMIOS.CLOCAL|TERMIOS.CREAD) lflag &= ~(TERMIOS.ICANON|TERMIOS.ECHO|TERMIOS.ECHOE|TERMIOS.ECHOK|TERMIOS.ECHONL| TERMIOS.ISIG|TERMIOS.IEXTEN) #|TERMIOS.ECHOPRT for flag in ('ECHOCTL', 'ECHOKE'): #netbsd workaround for Erk if hasattr(TERMIOS, flag): lflag &= ~getattr(TERMIOS, flag) oflag &= ~(TERMIOS.OPOST) iflag &= ~(TERMIOS.INLCR|TERMIOS.IGNCR|TERMIOS.ICRNL|TERMIOS.IGNBRK) if hasattr(TERMIOS, 'IUCLC'): iflag &= ~TERMIOS.IUCLC if hasattr(TERMIOS, 'PARMRK'): iflag &= ~TERMIOS.PARMRK #setup baudrate try: ispeed = ospeed = getattr(TERMIOS,'B%s' % (self._baudrate)) except AttributeError: raise ValueError('Invalid baud rate: %r' % self._baudrate) #setup char len cflag &= ~TERMIOS.CSIZE if self._bytesize == 8: cflag |= TERMIOS.CS8 elif self._bytesize == 7: cflag |= TERMIOS.CS7 elif self._bytesize == 6: cflag |= TERMIOS.CS6 elif self._bytesize == 5: cflag |= TERMIOS.CS5 else: raise ValueError('Invalid char len: %r' % self._bytesize) #setup stopbits if self._stopbits == STOPBITS_ONE: cflag &= ~(TERMIOS.CSTOPB) elif self._stopbits == STOPBITS_TWO: cflag |= (TERMIOS.CSTOPB) else: raise ValueError('Invalid stopit specification: %r' % self._stopbits) #setup parity iflag &= ~(TERMIOS.INPCK|TERMIOS.ISTRIP) if self._parity == PARITY_NONE: cflag &= ~(TERMIOS.PARENB|TERMIOS.PARODD) elif self._parity == PARITY_EVEN: cflag &= ~(TERMIOS.PARODD) cflag |= (TERMIOS.PARENB) elif self._parity == PARITY_ODD: cflag |= (TERMIOS.PARENB|TERMIOS.PARODD) else: raise ValueError('Invalid parity: %r' % self._parity) #setup flow control #xonxoff if hasattr(TERMIOS, 'IXANY'): if self._xonxoff: iflag |= (TERMIOS.IXON|TERMIOS.IXOFF) #|TERMIOS.IXANY) else: iflag &= ~(TERMIOS.IXON|TERMIOS.IXOFF|TERMIOS.IXANY) else: if self._xonxoff: iflag |= (TERMIOS.IXON|TERMIOS.IXOFF) else: iflag &= ~(TERMIOS.IXON|TERMIOS.IXOFF) #rtscts if hasattr(TERMIOS, 'CRTSCTS'): if self._rtscts: cflag |= (TERMIOS.CRTSCTS) else: cflag &= ~(TERMIOS.CRTSCTS) elif hasattr(TERMIOS, 'CNEW_RTSCTS'): #try it with alternate constant name if self._rtscts: cflag |= (TERMIOS.CNEW_RTSCTS) else: cflag &= ~(TERMIOS.CNEW_RTSCTS) #XXX should there be a warning if setting up rtscts (and xonxoff etc) fails?? #buffer #vmin "minimal number of characters to be read. = for non blocking" if vmin < 0 or vmin > 255: raise ValueError('Invalid vmin: %r ' % vmin) cc[TERMIOS.VMIN] = vmin #vtime if vtime < 0 or vtime > 255: raise ValueError('Invalid vtime: %r' % vtime) cc[TERMIOS.VTIME] = vtime #activate settings termios.tcsetattr(self.fd, TERMIOS.TCSANOW, [iflag, oflag, cflag, lflag, ispeed, ospeed, cc]) def close(self): """Close port""" if self._isOpen: if self.fd is not None: os.close(self.fd) self.fd = None self._isOpen = False def makeDeviceName(self, port): return device(port) # - - - - - - - - - - - - - - - - - - - - - - - - def inWaiting(self): """Return the number of characters currently in the input buffer.""" #~ s = fcntl.ioctl(self.fd, TERMIOS.FIONREAD, TIOCM_zero_str) s = fcntl.ioctl(self.fd, TIOCINQ, TIOCM_zero_str) return struct.unpack('I',s)[0] def read(self, size=1): """Read size bytes from the serial port. If a timeout is set it may return less characters as requested. With no timeout it will block until the requested number of bytes is read.""" if self.fd is None: raise portNotOpenError read = '' inp = None if size > 0: while len(read) < size: #print "\tread(): size",size, "have", len(read) #debug ready,_,_ = select.select([self.fd],[],[], self._timeout) if not ready: break #timeout buf = os.read(self.fd, size-len(read)) read = read + buf if self._timeout >= 0 and not buf: break #early abort on timeout return read def write(self, data): """Output the given string over the serial port.""" if self.fd is None: raise portNotOpenError t = len(data) d = data while t > 0: try: if self._writeTimeout is not None and self._writeTimeout > 0: _,ready,_ = select.select([],[self.fd],[], self._writeTimeout) if not ready: raise writeTimeoutError n = os.write(self.fd, d) if self._writeTimeout is not None and self._writeTimeout > 0: _,ready,_ = select.select([],[self.fd],[], self._writeTimeout) if not ready: raise writeTimeoutError d = d[n:] t = t - n except OSError,v: if v.errno != errno.EAGAIN: raise def flush(self): """Flush of file like objects. In this case, wait until all data is written.""" self.drainOutput() def flushInput(self): """Clear input buffer, discarding all that is in the buffer.""" if self.fd is None: raise portNotOpenError termios.tcflush(self.fd, TERMIOS.TCIFLUSH) def flushOutput(self): """Clear output buffer, aborting the current output and discarding all that is in the buffer.""" if self.fd is None: raise portNotOpenError termios.tcflush(self.fd, TERMIOS.TCOFLUSH) def sendBreak(self): """Send break condition.""" if self.fd is None: raise portNotOpenError termios.tcsendbreak(self.fd, 0) def setRTS(self,on=1): """Set terminal status line: Request To Send""" if self.fd is None: raise portNotOpenError if on: fcntl.ioctl(self.fd, TIOCMBIS, TIOCM_RTS_str) else: fcntl.ioctl(self.fd, TIOCMBIC, TIOCM_RTS_str) def setDTR(self,on=1): """Set terminal status line: Data Terminal Ready""" if self.fd is None: raise portNotOpenError if on: fcntl.ioctl(self.fd, TIOCMBIS, TIOCM_DTR_str) else: fcntl.ioctl(self.fd, TIOCMBIC, TIOCM_DTR_str) def getCTS(self): """Read terminal status line: Clear To Send""" if self.fd is None: raise portNotOpenError s = fcntl.ioctl(self.fd, TIOCMGET, TIOCM_zero_str) return struct.unpack('I',s)[0] & TIOCM_CTS != 0 def getDSR(self): """Read terminal status line: Data Set Ready""" if self.fd is None: raise portNotOpenError s = fcntl.ioctl(self.fd, TIOCMGET, TIOCM_zero_str) return struct.unpack('I',s)[0] & TIOCM_DSR != 0 def getRI(self): """Read terminal status line: Ring Indicator""" if self.fd is None: raise portNotOpenError s = fcntl.ioctl(self.fd, TIOCMGET, TIOCM_zero_str) return struct.unpack('I',s)[0] & TIOCM_RI != 0 def getCD(self): """Read terminal status line: Carrier Detect""" if self.fd is None: raise portNotOpenError s = fcntl.ioctl(self.fd, TIOCMGET, TIOCM_zero_str) return struct.unpack('I',s)[0] & TIOCM_CD != 0 # - - platform specific - - - - def drainOutput(self): """internal - not portable!""" if self.fd is None: raise portNotOpenError termios.tcdrain(self.fd) def nonblocking(self): """internal - not portable!""" if self.fd is None: raise portNotOpenError fcntl.fcntl(self.fd, FCNTL.F_SETFL, FCNTL.O_NONBLOCK) def fileno(self): """For easier of the serial port instance with select. WARNING: this function is not portable to different platforms!""" if self.fd is None: raise portNotOpenError return self.fd if __name__ == '__main__': s = Serial(0, baudrate=19200, #baudrate bytesize=EIGHTBITS, #number of databits parity=PARITY_EVEN, #enable parity checking stopbits=STOPBITS_ONE, #number of stopbits timeout=3, #set a timeout value, None for waiting forever xonxoff=0, #enable software flow control rtscts=0, #enable RTS/CTS flow control ) s.setRTS(1) s.setDTR(1) s.flushInput() s.flushOutput() s.write('hello') print repr(s.read(5)) print s.inWaiting() del s --- NEW FILE: serialutil.py --- #! python #Python Serial Port Extension for Win32, Linux, BSD, Jython #see __init__.py # #(C) 2001-2003 Chris Liechti <[email protected]> # this is distributed under a free software license, see license.txt PARITY_NONE, PARITY_EVEN, PARITY_ODD = 'N', 'E', 'O' STOPBITS_ONE, STOPBITS_TWO = (1, 2) FIVEBITS, SIXBITS, SEVENBITS, EIGHTBITS = (5,6,7,8) PARITY_NAMES = { PARITY_NONE: 'None', PARITY_EVEN: 'Even', PARITY_ODD: 'Odd', } XON = chr(17) XOFF = chr(19) #Python < 2.2.3 compatibility try: True except: True = 1 False = not True class SerialException(Exception): """Base class for serial port related exceptions.""" portNotOpenError = SerialException('Port not open') class SerialTimeoutException(SerialException): """Write timeouts give an exception""" writeTimeoutError = SerialTimeoutException("Write timeout") class FileLike(object): """An abstract file like class. This class implements readline and readlines based on read and writelines based on write. This class is used to provide the above functions for to Serial port objects. Note that when the serial port was opened with _NO_ timeout that readline blocks until it sees a newline (or the specified size is reached) and that readlines would never return and therefore refuses to work (it raises an exception in this case)! """ def read(self, size): raise NotImplementedError def write(self, s): raise NotImplementedError def readline(self, size=None, eol='\n'): """read a line which is terminated with end-of-line (eol) character ('\n' by default) or until timeout""" line = '' while 1: c = self.read(1) if c: line += c #not very efficient but lines are usually not that long if c == eol: break if size is not None and len(line) >= size: break else: break return line def readlines(self, sizehint=None, eol='\n'): """read a list of lines, until timeout sizehint is ignored""" if self.timeout is None: raise ValueError, "Serial port MUST have enabled timeout for this function!" lines = [] while 1: line = self.readline(eol=eol) if line: lines.append(line) if line[-1] != eol: #was the line received with a timeout? break else: break return lines def xreadlines(self, sizehint=None): """just call readlines - here for compatibility""" return self.readlines() def writelines(self, sequence): for line in sequence: self.write(line) def flush(self): """flush of file like objects""" pass class SerialBase(FileLike): """Serial port base class. Provides __init__ function and properties to get/set port settings.""" #default values, may be overriden in subclasses that do not support all values BAUDRATES = (50,75,110,134,150,200,300,600,1200,1800,2400,4800,9600, 19200,38400,57600,115200,230400,460800,500000,576000,921600, 1000000,1152000,1500000,2000000,2500000,3000000,3500000,4000000) BYTESIZES = (FIVEBITS, SIXBITS, SEVENBITS, EIGHTBITS) PARITIES = (PARITY_NONE, PARITY_EVEN, PARITY_ODD) STOPBITS = (STOPBITS_ONE, STOPBITS_TWO) def __init__(self, port = None, #number of device, numbering starts at #zero. if everything fails, the user #can specify a device string, note #that this isn't portable anymore #port will be opened if one is specified baudrate=9600, #baudrate bytesize=EIGHTBITS, #number of databits parity=PARITY_NONE, #enable parity checking stopbits=STOPBITS_ONE, #number of stopbits timeout=None, #set a timeout value, None to wait forever xonxoff=0, #enable software flow control rtscts=0, #enable RTS/CTS flow control writeTimeout=None, #set a timeout for writes dsrdtr=None, #None: use rtscts setting, dsrdtr override if true or false ): """Initialize comm port object. If a port is given, then the port will be opened immediately. Otherwise a Serial port object in closed state is returned.""" self._isOpen = False self._port = None #correct value is assigned below trough properties self._baudrate = None #correct value is assigned below trough properties self._bytesize = None #correct value is assigned below trough properties self._parity = None #correct value is assigned below trough properties self._stopbits = None #correct value is assigned below trough properties self._timeout = None #correct value is assigned below trough properties self._writeTimeout = None #correct value is assigned below trough properties self._xonxoff = None #correct value is assigned below trough properties self._rtscts = None #correct value is assigned below trough properties self._dsrdtr = None #correct value is assigned below trough properties #assign values using get/set methods using the properties feature self.port = port self.baudrate = baudrate self.bytesize = bytesize self.parity = parity self.stopbits = stopbits self.timeout = timeout self.writeTimeout = writeTimeout self.xonxoff = xonxoff self.rtscts = rtscts self.dsrdtr = dsrdtr if port is not None: self.open() def isOpen(self): """Check if the port is opened.""" return self._isOpen # - - - - - - - - - - - - - - - - - - - - - - - - #TODO: these are not realy needed as the is the BAUDRATES etc attribute... #maybe i remove them before the final release... def getSupportedBaudrates(self): return [(str(b), b) for b in self.BAUDRATES] def getSupportedByteSizes(self): return [(str(b), b) for b in self.BYTESIZES] def getSupportedStopbits(self): return [(str(b), b) for b in self.STOPBITS] def getSupportedParities(self): return [(PARITY_NAMES[b], b) for b in self.PARITIES] # - - - - - - - - - - - - - - - - - - - - - - - - def setPort(self, port): """Change the port. The attribute portstr is set to a string that contains the name of the port.""" was_open = self._isOpen if was_open: self.close() if port is not None: if type(port) in [type(''), type(u'')]: #strings are taken directly self.portstr = port else: self.portstr = self.makeDeviceName(port) else: self.portstr = None self._port = port if was_open: self.open() def getPort(self): """Get the current port setting. The value that was passed on init or using setPort() is passed back. See also the attribute portstr which contains the name of the port as a string.""" return self._port port = property(getPort, setPort, doc="Port setting") def setBaudrate(self, baudrate): """Change baudrate. It raises a ValueError if the port is open and the baudrate is not possible. If the port is closed, then tha value is accepted and the exception is raised when the port is opened.""" #~ if baudrate not in self.BAUDRATES: raise ValueError("Not a valid baudrate: %r" % baudrate) try: self._baudrate = int(baudrate) except TypeError: raise ValueError("Not a valid baudrate: %r" % baudrate) else: if self._isOpen: self._reconfigurePort() def getBaudrate(self): """Get the current baudrate setting.""" return self._baudrate baudrate = property(getBaudrate, setBaudrate, doc="Baudrate setting") def setByteSize(self, bytesize): """Change byte size.""" if bytesize not in self.BYTESIZES: raise ValueError("Not a valid byte size: %r" % bytesize) self._bytesize = bytesize if self._isOpen: self._reconfigurePort() def getByteSize(self): """Get the current byte size setting.""" return self._bytesize bytesize = property(getByteSize, setByteSize, doc="Byte size setting") def setParity(self, parity): """Change parity setting.""" if parity not in self.PARITIES: raise ValueError("Not a valid parity: %r" % parity) self._parity = parity if self._isOpen: self._reconfigurePort() def getParity(self): """Get the current parity setting.""" return self._parity parity = property(getParity, setParity, doc="Parity setting") def setStopbits(self, stopbits): """Change stopbits size.""" if stopbits not in self.STOPBITS: raise ValueError("Not a valid stopbit size: %r" % stopbits) self._stopbits = stopbits if self._isOpen: self._reconfigurePort() def getStopbits(self): """Get the current stopbits setting.""" return self._stopbits stopbits = property(getStopbits, setStopbits, doc="Stopbits setting") def setTimeout(self, timeout): """Change timeout setting.""" if timeout is not None: if timeout < 0: raise ValueError("Not a valid timeout: %r" % timeout) try: timeout + 1 #test if it's a number, will throw a TypeError if not... except TypeError: raise ValueError("Not a valid timeout: %r" % timeout) self._timeout = timeout if self._isOpen: self._reconfigurePort() def getTimeout(self): """Get the current timeout setting.""" return self._timeout timeout = property(getTimeout, setTimeout, doc="Timeout setting for read()") def setWriteTimeout(self, timeout): """Change timeout setting.""" if timeout is not None: if timeout < 0: raise ValueError("Not a valid timeout: %r" % timeout) try: timeout + 1 #test if it's a number, will throw a TypeError if not... except TypeError: raise ValueError("Not a valid timeout: %r" % timeout) self._writeTimeout = timeout if self._isOpen: self._reconfigurePort() def getWriteTimeout(self): """Get the current timeout setting.""" return self._writeTimeout writeTimeout = property(getWriteTimeout, setWriteTimeout, doc="Timeout setting for write()") def setXonXoff(self, xonxoff): """Change XonXoff setting.""" self._xonxoff = xonxoff if self._isOpen: self._reconfigurePort() def getXonXoff(self): """Get the current XonXoff setting.""" return self._xonxoff xonxoff = property(getXonXoff, setXonXoff, doc="Xon/Xoff setting") def setRtsCts(self, rtscts): """Change RtsCts flow control setting.""" self._rtscts = rtscts if self._isOpen: self._reconfigurePort() def getRtsCts(self): """Get the current RtsCts flow control setting.""" return self._rtscts rtscts = property(getRtsCts, setRtsCts, doc="RTS/CTS flow control setting") def setDsrDtr(self, dsrdtr=None): """Change DsrDtr flow control setting.""" if dsrdtr is None: #if not set, keep backwards compatibility and follow rtscts setting self._dsrdtr = self._rtscts else: #if defined independently, follow its value self._dsrdtr = dsrdtr if self._isOpen: self._reconfigurePort() def getDsrDtr(self): """Get the current DsrDtr flow control setting.""" return self._dsrdtr dsrdtr = property(getDsrDtr, setDsrDtr, "DSR/DTR flow control setting") # - - - - - - - - - - - - - - - - - - - - - - - - def __repr__(self): """String representation of the current port settings and its state.""" return "%s<id=0x%x, open=%s>(port=%r, baudrate=%r, bytesize=%r, parity=%r, stopbits=%r, timeout=%r, xonxoff=%r, rtscts=%r, dsrdtr=%r)" % ( self.__class__.__name__, id(self), self._isOpen, self.portstr, self.baudrate, self.bytesize, self.parity, self.stopbits, self.timeout, self.xonxoff, self.rtscts, self.dsrdtr, ) if __name__ == '__main__': s = SerialBase() print s.portstr print s.getSupportedBaudrates() print s.getSupportedByteSizes() print s.getSupportedParities() print s.getSupportedStopbits() print s --- NEW FILE: serialutil.pyc --- mò ¦\<Cc This class implements readline and readlines based on read and writelines based on write. This class is used to provide the above functions for to Serial port objects. Note that when the serial port was opened with _NO_ timeout that readline blocks until it sees a newline (or the specified size is reached) and that readlines would never return and therefore refuses to work (it raises an exception in this case)! c c (' ' by default) or until timeoutt c sizehint is ignoreds8 get/set port settings.i2 opened immediately. Otherwise a Serial port object in closed state is returned.N( contains the name of the port.R setPort() is passed back. See also the attribute portstr which contains the name of the port as a string.N( baudrate is not possible. If the port is closed, then tha value is accepted and the exception is raised when the port is opened.s t --- NEW FILE: serialwin32.py --- #! python #Python Serial Port Extension for Win32, Linux, BSD, Jython #serial driver for win32 #see __init__.py # #(C) 2001-2003 Chris Liechti <[email protected]> # this is distributed under a free software license, see license.txt import win32file # The base COM port and file IO functions. import win32event # We use events and the WaitFor[Single|Multiple]Objects functions. import win32con # constants. from serialutil import * VERSION = "$Revision: 1.1 $".split()[1] #extract CVS version #from winbase.h. these should realy be in win32con MS_CTS_ON = 16 MS_DSR_ON = 32 MS_RING_ON = 64 MS_RLSD_ON = 128 def device(portnum): """Turn a port number into a device name""" #the "//./COMx" format is required for devices >= 9 #not all versions of windows seem to support this propperly #so that the first few ports are used with the DOS device name if portnum < 9: return 'COM%d' % (portnum+1) #numbers are transformed to a string else: return r'\\.\COM%d' % (portnum+1) class Serial(SerialBase): """Serial port implemenation for Win32. This implemenatation requires a win32all installation.""" BAUDRATES = (50,75,110,134,150,200,300,600,1200,1800,2400,4800,9600, 19200,38400,57600,115200) def open(self): """Open port with current settings. This may throw a SerialException if the port cannot be opened.""" if self._port is None: raise SerialException("Port must be configured before it can be used.") self.hComPort = None try: self.hComPort = win32file.CreateFile(self.portstr, win32con.GENERIC_READ | win32con.GENERIC_WRITE, 0, # exclusive access None, # no security win32con.OPEN_EXISTING, win32con.FILE_ATTRIBUTE_NORMAL | win32con.FILE_FLAG_OVERLAPPED, None) except Exception, msg: self.hComPort = None #'cause __del__ is called anyway raise SerialException("could not open port: %s" % msg) # Setup a 4k buffer win32file.SetupComm(self.hComPort, 4096, 4096) #Save original timeout values: self._orgTimeouts = win32file.GetCommTimeouts(self.hComPort) self._rtsState = win32file.RTS_CONTROL_ENABLE self._dtrState = win32file.RTS_CONTROL_ENABLE self._reconfigurePort() # Clear buffers: # Remove anything that was there win32file.PurgeComm(self.hComPort, win32file.PURGE_TXCLEAR | win32file.PURGE_TXABORT | win32file.PURGE_RXCLEAR | win32file.PURGE_RXABORT) self._overlappedRead = win32file.OVERLAPPED() self._overlappedRead.hEvent = win32event.CreateEvent(None, 1, 0, None) self._overlappedWrite = win32file.OVERLAPPED() #~ self._overlappedWrite.hEvent = win32event.CreateEvent(None, 1, 0, None) self._overlappedWrite.hEvent = win32event.CreateEvent(None, 0, 0, None) self._isOpen = True def _reconfigurePort(self): """Set commuication parameters on opened port.""" if not self.hComPort: raise SerialException("Can only operate on a valid port handle") #Set Windows timeout values #timeouts is a tuple with the following items: #(ReadIntervalTimeout,ReadTotalTimeoutMultiplier, # ReadTotalTimeoutConstant,WriteTotalTimeoutMultiplier, # WriteTotalTimeoutConstant) if self._timeout is None: timeouts = (0, 0, 0, 0, 0) elif self._timeout == 0: timeouts = (win32con.MAXDWORD, 0, 0, 0, 0) else: timeouts = (0, 0, int(self._timeout*1000), 0, 0) if self._writeTimeout is None: pass elif self._writeTimeout == 0: timeouts = timeouts[:-2] + (0, win32con.MAXDWORD) else: timeouts = timeouts[:-2] + (0, int(self._writeTimeout*1000)) win32file.SetCommTimeouts(self.hComPort, timeouts) win32file.SetCommMask(self.hComPort, win32file.EV_ERR) # Setup the connection info. # Get state and modify it: comDCB = win32file.GetCommState(self.hComPort) comDCB.BaudRate = self._baudrate if self._bytesize == FIVEBITS: comDCB.ByteSize = 5 elif self._bytesize == SIXBITS: comDCB.ByteSize = 6 elif self._bytesize == SEVENBITS: comDCB.ByteSize = 7 elif self._bytesize == EIGHTBITS: comDCB.ByteSize = 8 else: raise ValueError("Unsupported number of data bits: %r" % self._bytesize) if self._parity == PARITY_NONE: comDCB.Parity = win32file.NOPARITY comDCB.fParity = 0 # Dis/Enable Parity Check elif self._parity == PARITY_EVEN: comDCB.Parity = win32file.EVENPARITY comDCB.fParity = 1 # Dis/Enable Parity Check elif self._parity == PARITY_ODD: comDCB.Parity = win32file.ODDPARITY comDCB.fParity = 1 # Dis/Enable Parity Check else: raise ValueError("Unsupported parity mode: %r" % self._parity) if self._stopbits == STOPBITS_ONE: comDCB.StopBits = win32file.ONESTOPBIT elif self._stopbits == STOPBITS_TWO: comDCB.StopBits = win32file.TWOSTOPBITS else: raise ValueError("Unsupported number of stop bits: %r" % self._stopbits) comDCB.fBinary = 1 # Enable Binary Transmission # Char. w/ Parity-Err are replaced with 0xff (if fErrorChar is set to TRUE) if self._rtscts: comDCB.fRtsControl = win32file.RTS_CONTROL_HANDSHAKE else: comDCB.fRtsControl = self._rtsState if self._dsrdtr: comDCB.fDtrControl = win32file.DTR_CONTROL_HANDSHAKE else: comDCB.fDtrControl = self._dtrState comDCB.fOutxCtsFlow = self._rtscts comDCB.fOutxDsrFlow = self._dsrdtr comDCB.fOutX = self._xonxoff comDCB.fInX = self._xonxoff comDCB.fNull = 0 comDCB.fErrorChar = 0 comDCB.fAbortOnError = 0 comDCB.XonChar = XON comDCB.XoffChar = XOFF try: win32file.SetCommState(self.hComPort, comDCB) except win32file.error, e: raise ValueError("Cannot configure port, some setting was wrong. Original message: %s" % e) #~ def __del__(self): #~ self.close() def close(self): """Close port""" if self._isOpen: if self.hComPort: #Restore original timeout values: win32file.SetCommTimeouts(self.hComPort, self._orgTimeouts) #Close COM-Port: win32file.CloseHandle(self.hComPort) self.hComPort = None self._isOpen = False def makeDeviceName(self, port): return device(port) # - - - - - - - - - - - - - - - - - - - - - - - - def inWaiting(self): """Return the number of characters currently in the input buffer.""" flags, comstat = win32file.ClearCommError(self.hComPort) return comstat.cbInQue def read(self, size=1): """Read size bytes from the serial port. If a timeout is set it may return less characters as requested. With no timeout it will block until the requested number of bytes is read.""" if not self.hComPort: raise portNotOpenError if size > 0: win32event.ResetEvent(self._overlappedRead.hEvent) flags, comstat = win32file.ClearCommError(self.hComPort) if self.timeout == 0: n = min(comstat.cbInQue, size) if n > 0: rc, buf = win32file.ReadFile(self.hComPort, win32file.AllocateReadBuffer(n), self._overlappedRead) win32event.WaitForSingleObject(self._overlappedRead.hEvent, win32event.INFINITE) read = str(buf) else: read = '' else: rc, buf = win32file.ReadFile(self.hComPort, win32file.AllocateReadBuffer(size), self._overlappedRead) n = win32file.GetOverlappedResult(self.hComPort, self._overlappedRead, 1) read = str(buf[:n]) else: read = '' return read def write(self, s): """Output the given string over the serial port.""" if not self.hComPort: raise portNotOpenError #print repr(s), if s: #~ win32event.ResetEvent(self._overlappedWrite.hEvent) err, n = win32file.WriteFile(self.hComPort, s, self._overlappedWrite) if err: #will be ERROR_IO_PENDING: # Wait for the write to complete. #~ win32event.WaitForSingleObject(self._overlappedWrite.hEvent, win32event.INFINITE) n = win32file.GetOverlappedResult(self.hComPort, self._overlappedWrite, 1) if n != len(s): raise writeTimeoutError def flushInput(self): """Clear input buffer, discarding all that is in the buffer.""" if not self.hComPort: raise portNotOpenError win32file.PurgeComm(self.hComPort, win32file.PURGE_RXCLEAR | win32file.PURGE_RXABORT) def flushOutput(self): """Clear output buffer, aborting the current output and discarding all that is in the buffer.""" if not self.hComPort: raise portNotOpenError win32file.PurgeComm(self.hComPort, win32file.PURGE_TXCLEAR | win32file.PURGE_TXABORT) def sendBreak(self): """Send break condition.""" if not self.hComPort: raise portNotOpenError import time win32file.SetCommBreak(self.hComPort) #TODO: how to set the correct duration?? time.sleep(0.020) win32file.ClearCommBreak(self.hComPort) def setRTS(self,level=1): """Set terminal status line: Request To Send""" if not self.hComPort: raise portNotOpenError if level: self._rtsState = win32file.RTS_CONTROL_ENABLE win32file.EscapeCommFunction(self.hComPort, win32file.SETRTS) else: self._rtsState = win32file.RTS_CONTROL_DISABLE win32file.EscapeCommFunction(self.hComPort, win32file.CLRRTS) def setDTR(self,level=1): """Set terminal status line: Data Terminal Ready""" if not self.hComPort: raise portNotOpenError if level: self._dtrState = win32file.DTR_CONTROL_ENABLE win32file.EscapeCommFunction(self.hComPort, win32file.SETDTR) else: self._dtrState = win32file.DTR_CONTROL_DISABLE win32file.EscapeCommFunction(self.hComPort, win32file.CLRDTR) def getCTS(self): """Read terminal status line: Clear To Send""" if not self.hComPort: raise portNotOpenError return MS_CTS_ON & win32file.GetCommModemStatus(self.hComPort) != 0 def getDSR(self): """Read terminal status line: Data Set Ready""" if not self.hComPort: raise portNotOpenError return MS_DSR_ON & win32file.GetCommModemStatus(self.hComPort) != 0 def getRI(self): """Read terminal status line: Ring Indicator""" if not self.hComPort: raise portNotOpenError return MS_RING_ON & win32file.GetCommModemStatus(self.hComPort) != 0 def getCD(self): """Read terminal status line: Carrier Detect""" if not self.hComPort: raise portNotOpenError return MS_RLSD_ON & win32file.GetCommModemStatus(self.hComPort) != 0 # - - platform specific - - - - def setXON(self, level=True): """Platform specific - set flow state.""" if not self.hComPort: raise portNotOpenError if level: win32file.EscapeCommFunction(self.hComPort, win32file.SETXON) else: win32file.EscapeCommFunction(self.hComPort, win32file.SETXOFF) #Nur Testfunktion!! if __name__ == '__main__': print __name__ s = Serial() print s s = Serial(0) print s s.baudrate = 19200 s.databits = 7 s.close() s.port = 3 s.open() print s --- NEW FILE: serialwin32.pyc --- mò ¦\<Cc win32all installation.i2 if the port cannot be opened.s. c "! c c return less characters as requested. With no timeout it will block until the requested number of bytes is read.i * *c !c c discarding all that is in the buffer.N( c c c c c c c c ( t ÿ ------------------------------------------------------- This SF.Net email is sponsored by xPML, a groundbreaking scripting language that extends applications into web and mobile media. Attend the live webcast and join the prime developer group breaking into this new coding territory! http://sel.as-us.falkag.net/sel?cmd=lnk&kid=110944&bid=241720&dat=121642