giomba/nes-proj
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Serial library for Python needed for some of the Tmote Sky tools

Browse Source
This commit is contained in:
adamdunkels 2007-11-13 13:23:02 +00:00
parent 2c1af74f37
commit d8717c264a
9 changed files with 965 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
tools/sky/serial/Makefile.am Normal file
View File

@ -0,0 +1,10 @@
AUTOMAKE_OPTIONS = foreign
tinyoslibdir=$(libdir)/tinyos
serialdir=$(tinyoslibdir)/serial
serial_DATA = __init__.py \
serialjava.py \
serialposix.py \
serialutil.py \
serialwin32.py

21
tools/sky/serial/__init__.py Normal file
View File

@ -0,0 +1,21 @@
#!/usr/bin/env python
#portable serial port access with python
#this is a wrapper module for different platform implementations
#
# (C)2001-2002 Chris Liechti <cliechti@gmx.net>
# 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 "Sorry no implementation for your platform available."
#no "mac" implementation. someone want's to write it? i have no access to a mac.

BIN
tools/sky/serial/__init__.pyc Normal file
View File

Binary file not shown.

197
tools/sky/serial/serialjava.py Normal file
View File

@ -0,0 +1,197 @@
#!jython
#module for serial IO for Jython and JavaComm
#see __init__.py
#
#(C) 2002 Chris Liechti <cliechti@gmx.net>
# this is distributed under a free software license, see license.txt
import sys, os, string, javax.comm
import serialutil
VERSION = string.split("$Revision: 1.1 $")[1] #extract CVS version
PARITY_NONE, PARITY_EVEN, PARITY_ODD, PARITY_MARK, PARITY_SPACE = (0,1,2,3,4)
STOPBITS_ONE, STOPBITS_TWO, STOPBITS_ONE_HALVE = (1, 2, 3)
FIVEBITS, SIXBITS, SEVENBITS, EIGHTBITS = (5,6,7,8)
portNotOpenError = ValueError('port not open')
def device(portnumber):
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]
class Serial(serialutil.FileLike):
def __init__(self,
port, #number of device, numbering starts at
#zero. if everything fails, the user
#can specify a device string, note
#that this isn't portable anymore
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 for waiting forever
xonxoff=0, #enable software flow control
rtscts=0, #enable RTS/CTS flow control
):
if type(port) == type(''): #strings are taken directly
portId = javax.comm.CommPortIdentifier.getPortIdentifier(port)
else:
portId = device(port) #numbers are transformed to a comportid obj
self.portstr = portId.getName()
try:
self.sPort = portId.open("python serial module", 10)
except Exception, msg:
self.sPort = None
raise serialutil.SerialException, "could not open port: %s" % msg
self.instream = self.sPort.getInputStream()
self.outstream = self.sPort.getOutputStream()
self.sPort.enableReceiveTimeout(30)
if bytesize == FIVEBITS:
self.databits = javax.comm.SerialPort.DATABITS_5
elif bytesize == SIXBITS:
self.databits = javax.comm.SerialPort.DATABITS_6
elif bytesize == SEVENBITS:
self.databits = javax.comm.SerialPort.DATABITS_7
elif bytesize == EIGHTBITS:
self.databits = javax.comm.SerialPort.DATABITS_8
else:
raise ValueError, "unsupported bytesize"
if stopbits == STOPBITS_ONE:
self.jstopbits = javax.comm.SerialPort.STOPBITS_1
elif stopbits == STOPBITS_ONE_HALVE:
self.jstopbits = javax.comm.SerialPort.STOPBITS_1_5
elif stopbits == STOPBITS_TWO:
self.jstopbits = javax.comm.SerialPort.STOPBITS_2
else:
raise ValueError, "unsupported number of stopbits"
if parity == PARITY_NONE:
self.jparity = javax.comm.SerialPort.PARITY_NONE
elif parity == PARITY_EVEN:
self.jparity = javax.comm.SerialPort.PARITY_EVEN
elif parity == PARITY_ODD:
self.jparity = javax.comm.SerialPort.PARITY_ODD
elif parity == PARITY_MARK:
self.jparity = javax.comm.SerialPort.PARITY_MARK
elif parity == PARITY_SPACE:
self.jparity = javax.comm.SerialPort.PARITY_SPACE
else:
raise ValueError, "unsupported parity type"
jflowin = jflowout = 0
if rtscts:
jflowin = jflowin | javax.comm.SerialPort.FLOWCONTROL_RTSCTS_IN
jflowout = jflowout | javax.comm.SerialPort.FLOWCONTROL_RTSCTS_OUT
if xonxoff:
jflowin = jflowin | javax.comm.SerialPort.FLOWCONTROL_XONXOFF_IN
jflowout = jflowout | javax.comm.SerialPort.FLOWCONTROL_XONXOFF_OUT
self.sPort.setSerialPortParams(baudrate, self.databits, self.jstopbits, self.jparity)
self.sPort.setFlowControlMode(jflowin | jflowout)
self.timeout = timeout
if timeout >= 0:
self.sPort.enableReceiveTimeout(timeout*1000)
else:
self.sPort.disableReceiveTimeout()
def close(self):
if self.sPort:
self.instream.close()
self.outstream.close()
self.sPort.close()
self.sPort = None
def setBaudrate(self, baudrate):
"""change baudrate after port is open"""
if not self.sPort: raise portNotOpenError
self.sPort.setSerialPortParams(baudrate, self.databits, self.jstopbits, self.jparity)
def inWaiting(self):
if not self.sPort: raise portNotOpenError
return self.instream.available()
def write(self, data):
if not self.sPort: raise portNotOpenError
self.outstream.write(data)
def read(self, size=1):
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 flushInput(self):
if not self.sPort: raise portNotOpenError
self.instream.skip(self.instream.available())
def flushOutput(self):
if not self.sPort: raise portNotOpenError
self.outstream.flush()
def sendBreak(self):
if not self.sPort: raise portNotOpenError
self.sPort.sendBreak()
def getDSR(self):
if not self.sPort: raise portNotOpenError
self.sPort.isDSR()
def getCD(self):
if not self.sPort: raise portNotOpenError
self.sPort.isCD()
def getRI(self):
if not self.sPort: raise portNotOpenError
self.sPort.isRI()
def getCTS(self):
if not self.sPort: raise portNotOpenError
self.sPort.isCTS()
def setDTR(self,on=1):
if not self.sPort: raise portNotOpenError
self.sPort.setDTR(on)
def setRTS(self,on=1):
if not self.sPort: raise portNotOpenError
self.sPort.setRTS(on)
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

392
tools/sky/serial/serialposix.py Normal file
View File

@ -0,0 +1,392 @@
#!/usr/bin/env python
#module for serial IO for POSIX compatible systems, like Linux
#see __init__.py
#
#(C) 2001-2002 Chris Liechti <cliechti@gmx.net>
# this is distributed under a free software license, see license.txt
#
#parts based on code from Grant B. Edwards <grante@visi.com>:
# ftp://ftp.visi.com/users/grante/python/PosixSerial.py
# references: http://www.easysw.com/~mike/serial/serial.html
import sys, os, fcntl, termios, struct, string, select
import serialutil
VERSION = string.split("$Revision: 1.1 $")[1] #extract CVS version
PARITY_NONE, PARITY_EVEN, PARITY_ODD = range(3)
STOPBITS_ONE, STOPBITS_TWO = (1, 2)
FIVEBITS, SIXBITS, SEVENBITS, EIGHTBITS = (5,6,7,8)
#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 = string.lower(sys.platform)
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[:6] == 'netbsd' 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[: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] == 'dgux': #Digital UNIX (not tested)
def device(port):
return '/dev/tty0%d' % (port+1)
else:
#platform detection has failed...
info = "sys.platform = %r\nos.name = %r\nserialposix.py version = %s" % (sys.platform, os.name, VERSION)
print """send this information to the author of this module:
%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...
"""
raise Exception, "this module does not run on this platform, sorry."
#whats up with "aix", "beos", "sco", ....
#they should work, just need to know the device names.
# construct dictionaries for baud rate lookups
baudEnumToInt = {}
baudIntToEnum = {}
for rate in (0,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
):
try:
i = eval('TERMIOS.B'+str(rate))
baudEnumToInt[i]=rate
baudIntToEnum[rate] = i
except:
pass
#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)
portNotOpenError = ValueError('port not open')
class Serial(serialutil.FileLike):
def __init__(self,
port, #number of device, numbering starts at
#zero. if everything fails, the user
#can specify a device string, note
#that this isn't portable anymore
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 for waiting forever
xonxoff=0, #enable software flow control
rtscts=0, #enable RTS/CTS flow control
):
"""init comm port"""
self.fd = None
self.timeout = timeout
vmin = vtime = 0 #timeout is done via select
#open
if type(port) == type(''): #strings are taken directly
self.portstr = port
else:
self.portstr = device(port) #numbers are transformed to a os dependant string
try:
self.fd = os.open(self.portstr, os.O_RDWR|os.O_NOCTTY|os.O_NONBLOCK)
except Exception, msg:
self.fd = None
raise serialutil.SerialException, "could not open port: %s" % msg
fcntl.fcntl(self.fd, FCNTL.F_SETFL, 0) #set blocking
try:
self.__tcgetattr() #read current settings
except termios.error, msg: #if a port is nonexistent but has a /dev file, it'll fail here
raise serialutil.SerialException, "could not open port: %s" % msg
#set up raw mode / no echo / binary
self.cflag = self.cflag | (TERMIOS.CLOCAL|TERMIOS.CREAD)
self.lflag = self.lflag & ~(TERMIOS.ICANON|TERMIOS.ECHO|TERMIOS.ECHOE|TERMIOS.ECHOK|TERMIOS.ECHONL|
TERMIOS.ECHOCTL|TERMIOS.ECHOKE|TERMIOS.ISIG|TERMIOS.IEXTEN) #|TERMIOS.ECHOPRT
self.oflag = self.oflag & ~(TERMIOS.OPOST)
if hasattr(TERMIOS, 'IUCLC'):
self.iflag = self.iflag & ~(TERMIOS.INLCR|TERMIOS.IGNCR|TERMIOS.ICRNL|TERMIOS.IUCLC|TERMIOS.IGNBRK)
else:
self.iflag = self.iflag & ~(TERMIOS.INLCR|TERMIOS.IGNCR|TERMIOS.ICRNL|TERMIOS.IGNBRK)
#setup baudrate
try:
self.ispeed = self.ospeed = baudIntToEnum[baudrate]
except:
raise ValueError,'invalid baud rate: %s' % baudrate
#setup char len
self.cflag = self.cflag & ~TERMIOS.CSIZE
if bytesize == 8:
self.cflag = self.cflag | TERMIOS.CS8
elif bytesize == 7:
self.cflag = self.cflag | TERMIOS.CS7
elif bytesize == 6:
self.cflag = self.cflag | TERMIOS.CS6
elif bytesize == 5:
self.cflag = self.cflag | TERMIOS.CS5
else:
raise ValueError,'invalid char len: '+str(clen)
#setup stopbits
if stopbits == STOPBITS_ONE:
self.cflag = self.cflag & ~(TERMIOS.CSTOPB)
elif stopbits == STOPBITS_TWO:
self.cflag = self.cflag | (TERMIOS.CSTOPB)
else:
raise ValueError,'invalid stopit specification:'+str(stopbits)
#setup parity
self.iflag = self.iflag & ~(TERMIOS.INPCK|TERMIOS.ISTRIP)
if parity == PARITY_NONE:
self.cflag = self.cflag & ~(TERMIOS.PARENB|TERMIOS.PARODD)
elif parity == PARITY_EVEN:
self.cflag = self.cflag & ~(TERMIOS.PARODD)
self.cflag = self.cflag | (TERMIOS.PARENB)
elif parity == PARITY_ODD:
self.cflag = self.cflag | (TERMIOS.PARENB|TERMIOS.PARODD)
else:
raise ValueError,'invalid parity: '+str(par)
#setup flow control
#xonxoff
if hasattr(TERMIOS, 'IXANY'):
if xonxoff:
self.iflag = self.iflag | (TERMIOS.IXON|TERMIOS.IXOFF|TERMIOS.IXANY)
else:
self.iflag = self.iflag & ~(TERMIOS.IXON|TERMIOS.IXOFF|TERMIOS.IXANY)
else:
if xonxoff:
self.iflag = self.iflag | (TERMIOS.IXON|TERMIOS.IXOFF)
else:
self.iflag = self.iflag & ~(TERMIOS.IXON|TERMIOS.IXOFF)
#rtscts
if hasattr(TERMIOS, 'CRTSCTS'):
if rtscts:
self.cflag = self.cflag | (TERMIOS.CRTSCTS)
else:
self.cflag = self.cflag & ~(TERMIOS.CRTSCTS)
elif hasattr(TERMIOS, 'CNEW_RTSCTS'): #try it with alternate constant name
if rtscts:
self.cflag = self.cflag | (TERMIOS.CNEW_RTSCTS)
else:
self.cflag = self.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: '+str(vmin)
self.cc[TERMIOS.VMIN] = vmin
#vtime
if vtime<0 or vtime>255:
raise ValueError,'invalid vtime: '+str(vtime)
self.cc[TERMIOS.VTIME] = vtime
#activate settings
self.__tcsetattr()
def __tcsetattr(self):
"""internal function to set port attributes"""
termios.tcsetattr(self.fd, TERMIOS.TCSANOW, [self.iflag,self.oflag,self.cflag,self.lflag,self.ispeed,self.ospeed,self.cc])
def __tcgetattr(self):
"""internal function to get port attributes"""
self.iflag,self.oflag,self.cflag,self.lflag,self.ispeed,self.ospeed,self.cc = termios.tcgetattr(self.fd)
def close(self):
"""close port"""
if self.fd:
os.close(self.fd)
self.fd = None
def setBaudrate(self, baudrate):
"""change baudrate after port is open"""
if not self.fd: raise portNotOpenError
self.__tcgetattr() #read current settings
#setup baudrate
try:
self.ispeed = self.ospeed = baudIntToEnum[baudrate]
except:
raise ValueError,'invalid baud rate: %s' % baudrate
self.__tcsetattr()
def inWaiting(self):
"""how many character are in the input queue"""
#~ 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 write(self, data):
"""write a string to the port"""
if not self.fd: raise portNotOpenError
t = len(data)
d = data
while t>0:
n = os.write(self.fd, d)
d = d[n:]
t = t - n
def read(self, size=1):
"""read a number of bytes from the port.
the default is one (unlike files)"""
if not self.fd: 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 flushInput(self):
"""clear input queue"""
if not self.fd:
raise portNotOpenError
termios.tcflush(self.fd, TERMIOS.TCIFLUSH)
def flushOutput(self):
"""flush output"""
if not self.fd:
raise portNotOpenError
termios.tcflush(self.fd, TERMIOS.TCOFLUSH)
def sendBreak(self):
"""send break signal"""
if not self.fd:
raise portNotOpenError
termios.tcsendbreak(self.fd, 0)
def drainOutput(self):
"""internal - not portable!"""
if not self.fd: raise portNotOpenError
termios.tcdrain(self.fd)
def nonblocking(self):
"""internal - not portable!"""
if not self.fd:
raise portNotOpenError
fcntl.fcntl(self.fd, FCNTL.F_SETFL, FCNTL.O_NONBLOCK)
def getDSR(self):
"""read terminal status line"""
if not self.fd: raise portNotOpenError
s = fcntl.ioctl(self.fd, TIOCMGET, TIOCM_zero_str)
return struct.unpack('I',s)[0] & TIOCM_DSR
def getCD(self):
"""read terminal status line"""
if not self.fd: raise portNotOpenError
s = fcntl.ioctl(self.fd, TIOCMGET, TIOCM_zero_str)
return struct.unpack('I',s)[0] & TIOCM_CD
def getRI(self):
"""read terminal status line"""
if not self.fd: raise portNotOpenError
s = fcntl.ioctl(self.fd, TIOCMGET, TIOCM_zero_str)
return struct.unpack('I',s)[0] & TIOCM_RI
def getCTS(self):
"""read terminal status line"""
if not self.fd: raise portNotOpenError
s = fcntl.ioctl(self.fd, TIOCMGET, TIOCM_zero_str)
return struct.unpack('I',s)[0] & TIOCM_CTS
def setDTR(self,on=1):
"""set terminal status line"""
if not self.fd: raise portNotOpenError
if on:
fcntl.ioctl(self.fd, TIOCMBIS, TIOCM_DTR_str)
else:
fcntl.ioctl(self.fd, TIOCMBIC, TIOCM_DTR_str)
def setRTS(self,on=1):
"""set terminal status line"""
if not self.fd: raise portNotOpenError
if on:
fcntl.ioctl(self.fd, TIOCMBIS, TIOCM_RTS_str)
else:
fcntl.ioctl(self.fd, TIOCMBIC, TIOCM_RTS_str)
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

BIN
tools/sky/serial/serialposix.pyc Normal file
View File

Binary file not shown.

65
tools/sky/serial/serialutil.py Normal file
View File

@ -0,0 +1,65 @@
class SerialException(Exception):
pass
class FileLike:
"""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

BIN
tools/sky/serial/serialutil.pyc Normal file
View File

Binary file not shown.

280
tools/sky/serial/serialwin32.py Normal file
View File

@ -0,0 +1,280 @@
#! python
#serial driver for win32
#see __init__.py
#
#(C) 2001-2002 Chris Liechti <cliechti@gmx.net>
# 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.
import sys, string
import serialutil
VERSION = string.split("$Revision: 1.1 $")[1] #extract CVS version
PARITY_NONE, PARITY_EVEN, PARITY_ODD = range(3)
STOPBITS_ONE, STOPBITS_TWO = (1, 2)
FIVEBITS, SIXBITS, SEVENBITS, EIGHTBITS = (5,6,7,8)
portNotOpenError = ValueError('port not open')
#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
class Serial(serialutil.FileLike):
def __init__(self,
port, #number of device, numbering starts at
#zero. if everything fails, the user
#can specify a device string, note
#that this isn't portable anymore
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 for waiting forever
xonxoff=0, #enable software flow control
rtscts=0, #enable RTS/CTS flow control
):
"""initialize comm port"""
self.timeout = timeout
if type(port) == type(''): #strings are taken directly
self.portstr = port
else:
# CSS 20040528 - open wasn't working for COM10 and greater, but by
# chance the '\\.\COM10' format seems to work, yay! But, only use
# if for COM10 and greater in case it introduces some other
# incompatibility.
if port < 9:
self.portstr = 'COM%d' % (port+1) #numbers are transformed to a string
else:
self.portstr = '\\\\.\\COM%d' % (port+1) #WIN NT format??
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 serialutil.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)
#Set Windows timeout values
#timeouts is a tuple with the following items:
#(ReadIntervalTimeout,ReadTotalTimeoutMultiplier,
# ReadTotalTimeoutConstant,WriteTotalTimeoutMultiplier,
# WriteTotalTimeoutConstant)
if timeout is None:
timeouts = (0, 0, 0, 0, 0)
elif timeout == 0:
timeouts = (win32con.MAXDWORD, 0, 0, 0, 0)
else:
#timeouts = (0, 0, 0, 0, 0) #timeouts are done with WaitForSingleObject
#timeouts = (win32con.MAXDWORD, 0, 0, 0, 1000) #doesn't works
#timeouts = (timeout*1000, 0, timeout*1000, 0, 0)
timeouts = (0, 0, timeout*1000, 0, timeout*1000)
win32file.SetCommTimeouts(self.hComPort, timeouts)
#win32file.SetCommMask(self.hComPort, win32file.EV_RXCHAR | win32file.EV_TXEMPTY |
# win32file.EV_RXFLAG | win32file.EV_ERR)
win32file.SetCommMask(self.hComPort,
win32file.EV_RXCHAR | win32file.EV_RXFLAG | win32file.EV_ERR)
#win32file.SetCommMask(self.hComPort, win32file.EV_ERR)
# Setup the connection info.
# Get state and modify it:
comDCB = win32file.GetCommState(self.hComPort)
comDCB.BaudRate = baudrate
if bytesize == FIVEBITS:
comDCB.ByteSize = 5
elif bytesize == SIXBITS:
comDCB.ByteSize = 6
elif bytesize == SEVENBITS:
comDCB.ByteSize = 7
elif bytesize == EIGHTBITS:
comDCB.ByteSize = 8
if parity == PARITY_NONE:
comDCB.Parity = win32file.NOPARITY
comDCB.fParity = 0 # Dis/Enable Parity Check
elif parity == PARITY_EVEN:
comDCB.Parity = win32file.EVENPARITY
comDCB.fParity = 1 # Dis/Enable Parity Check
elif parity == PARITY_ODD:
comDCB.Parity = win32file.ODDPARITY
comDCB.fParity = 1 # Dis/Enable Parity Check
if stopbits == STOPBITS_ONE:
comDCB.StopBits = win32file.ONESTOPBIT
elif stopbits == STOPBITS_TWO:
comDCB.StopBits = win32file.TWOSTOPBITS
comDCB.fBinary = 1 # Enable Binary Transmission
# Char. w/ Parity-Err are replaced with 0xff (if fErrorChar is set to TRUE)
if rtscts:
comDCB.fRtsControl = win32file.RTS_CONTROL_HANDSHAKE
comDCB.fDtrControl = win32file.DTR_CONTROL_HANDSHAKE
else:
comDCB.fRtsControl = win32file.RTS_CONTROL_ENABLE
comDCB.fDtrControl = win32file.DTR_CONTROL_ENABLE
comDCB.fOutxCtsFlow = rtscts
comDCB.fOutxDsrFlow = rtscts
comDCB.fOutX = xonxoff
comDCB.fInX = xonxoff
comDCB.fNull = 0
comDCB.fErrorChar = 0
comDCB.fAbortOnError = 0
win32file.SetCommState(self.hComPort, comDCB)
# Clear buffers:
# Remove anything that was there
win32file.PurgeComm(self.hComPort,
win32file.PURGE_TXCLEAR | win32file.PURGE_TXABORT |
win32file.PURGE_RXCLEAR | win32file.PURGE_RXABORT)
#print win32file.ClearCommError(self.hComPort) #flags, comState =
#self.overlapped = win32file.OVERLAPPED()
#self.overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
def __del__(self):
self.close()
def close(self):
"""close port"""
if self.hComPort:
#Wait until data is transmitted, but not too long... (Timeout-Time)
#while 1:
# flags, comState = win32file.ClearCommError(hComPort)
# if comState.cbOutQue <= 0 or calcTimeout(startTime) > timeout:
# break
self.setRTS(0)
self.setDTR(0)
#Clear buffers:
win32file.PurgeComm(self.hComPort,
win32file.PURGE_TXCLEAR | win32file.PURGE_TXABORT |
win32file.PURGE_RXCLEAR | win32file.PURGE_RXABORT)
#Restore original timeout values:
win32file.SetCommTimeouts(self.hComPort, self.orgTimeouts)
#Close COM-Port:
win32file.CloseHandle(self.hComPort)
self.hComPort = None
def setBaudrate(self, baudrate):
"""change baudrate after port is open"""
if not self.hComPort: raise portNotOpenError
# Setup the connection info.
# Get state and modify it:
comDCB = win32file.GetCommState(self.hComPort)
comDCB.BaudRate = baudrate
win32file.SetCommState(self.hComPort, comDCB)
def inWaiting(self):
"""returns the number of bytes waiting to be read"""
flags, comstat = win32file.ClearCommError(self.hComPort)
return comstat.cbInQue
def read(self, size=1):
"read num bytes from serial port"
if not self.hComPort: raise portNotOpenError
read = ''
if size > 0:
overlapped = win32file.OVERLAPPED()
overlapped.hEvent = win32event.CreateEvent(None, 1, 0, None)
if self.timeout == 0:
flags, comstat = win32file.ClearCommError(self.hComPort)
n = min(comstat.cbInQue, size)
if n > 0:
rc, buf = win32file.ReadFile(self.hComPort, win32file.AllocateReadBuffer(n), overlapped)
win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE)
read = str(buf)
else:
flags, comstat = win32file.ClearCommError(self.hComPort)
rc, buf = win32file.ReadFile(self.hComPort, win32file.AllocateReadBuffer(size), overlapped)
n = win32file.GetOverlappedResult(self.hComPort, overlapped, 1)
read = str(buf[:n])
return read
def write(self, s):
"write string to serial port"
if not self.hComPort: raise portNotOpenError
#print repr(s),
overlapped = win32file.OVERLAPPED()
overlapped.hEvent = win32event.CreateEvent(None, 1, 0, None)
err, n = win32file.WriteFile(self.hComPort, s, overlapped)
if err: #will be ERROR_IO_PENDING:
# Wait for the write to complete.
win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE)
def flushInput(self):
if not self.hComPort: raise portNotOpenError
win32file.PurgeComm(self.hComPort, win32file.PURGE_RXCLEAR | win32file.PURGE_RXABORT)
def flushOutput(self):
if not self.hComPort: raise portNotOpenError
win32file.PurgeComm(self.hComPort, win32file.PURGE_TXCLEAR | win32file.PURGE_TXABORT)
def sendBreak(self):
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"""
if not self.hComPort: raise portNotOpenError
if level:
win32file.EscapeCommFunction(self.hComPort, win32file.SETRTS)
else:
win32file.EscapeCommFunction(self.hComPort, win32file.CLRRTS)
def setDTR(self,level=1):
"""set terminal status line"""
if not self.hComPort: raise portNotOpenError
if level:
win32file.EscapeCommFunction(self.hComPort, win32file.SETDTR)
else:
win32file.EscapeCommFunction(self.hComPort, win32file.CLRDTR)
def getCTS(self):
"""read terminal status line"""
if not self.hComPort: raise portNotOpenError
return MS_CTS_ON & win32file.GetCommModemStatus(self.hComPort) != 0
def getDSR(self):
"""read terminal status line"""
if not self.hComPort: raise portNotOpenError
return MS_DSR_ON & win32file.GetCommModemStatus(self.hComPort) != 0
def getRI(self):
"""read terminal status line"""
if not self.hComPort: raise portNotOpenError
return MS_RING_ON & win32file.GetCommModemStatus(self.hComPort) != 0
def getCD(self):
"""read terminal status line"""
if not self.hComPort: raise portNotOpenError
return MS_RLSD_ON & win32file.GetCommModemStatus(self.hComPort) != 0
#Nur Testfunktion!!
if __name__ == '__main__':
print __name__
s = Serial(0)