#! python #serial driver for win32 #see __init__.py # #(C) 2001-2002 Chris Liechti # 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.4 $")[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 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)
