462 lines
18 KiB
Python
462 lines
18 KiB
Python
#! python
|
|
# Python Serial Port Extension for Win32, Linux, BSD, Jython
|
|
# serial driver for win32
|
|
# see __init__.py
|
|
#
|
|
# (C) 2001-2011 Chris Liechti <cliechti@gmx.net>
|
|
# this is distributed under a free software license, see license.txt
|
|
#
|
|
# Initial patch to use ctypes by Giovanni Bajo <rasky@develer.com>
|
|
|
|
import ctypes
|
|
from serial import win32
|
|
|
|
from serial.serialutil import *
|
|
|
|
|
|
def device(portnum):
|
|
"""Turn a port number into a device name"""
|
|
return 'COM%d' % (portnum+1) # numbers are transformed to a string
|
|
|
|
|
|
class Win32Serial(SerialBase):
|
|
"""Serial port implementation for Win32 based on ctypes."""
|
|
|
|
BAUDRATES = (50, 75, 110, 134, 150, 200, 300, 600, 1200, 1800, 2400, 4800,
|
|
9600, 19200, 38400, 57600, 115200)
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.hComPort = None
|
|
self._overlappedRead = None
|
|
self._overlappedWrite = None
|
|
self._rtsToggle = False
|
|
|
|
self._rtsState = win32.RTS_CONTROL_ENABLE
|
|
self._dtrState = win32.DTR_CONTROL_ENABLE
|
|
|
|
|
|
SerialBase.__init__(self, *args, **kwargs)
|
|
|
|
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 self._isOpen:
|
|
raise SerialException("Port is already open.")
|
|
# the "\\.\COMx" format is required for devices other than COM1-COM8
|
|
# not all versions of windows seem to support this properly
|
|
# so that the first few ports are used with the DOS device name
|
|
port = self.portstr
|
|
try:
|
|
if port.upper().startswith('COM') and int(port[3:]) > 8:
|
|
port = '\\\\.\\' + port
|
|
except ValueError:
|
|
# for like COMnotanumber
|
|
pass
|
|
self.hComPort = win32.CreateFile(port,
|
|
win32.GENERIC_READ | win32.GENERIC_WRITE,
|
|
0, # exclusive access
|
|
None, # no security
|
|
win32.OPEN_EXISTING,
|
|
win32.FILE_ATTRIBUTE_NORMAL | win32.FILE_FLAG_OVERLAPPED,
|
|
0)
|
|
if self.hComPort == win32.INVALID_HANDLE_VALUE:
|
|
self.hComPort = None # 'cause __del__ is called anyway
|
|
raise SerialException("could not open port %r: %r" % (self.portstr, ctypes.WinError()))
|
|
|
|
try:
|
|
self._overlappedRead = win32.OVERLAPPED()
|
|
self._overlappedRead.hEvent = win32.CreateEvent(None, 1, 0, None)
|
|
self._overlappedWrite = win32.OVERLAPPED()
|
|
#~ self._overlappedWrite.hEvent = win32.CreateEvent(None, 1, 0, None)
|
|
self._overlappedWrite.hEvent = win32.CreateEvent(None, 0, 0, None)
|
|
|
|
# Setup a 4k buffer
|
|
win32.SetupComm(self.hComPort, 4096, 4096)
|
|
|
|
# Save original timeout values:
|
|
self._orgTimeouts = win32.COMMTIMEOUTS()
|
|
win32.GetCommTimeouts(self.hComPort, ctypes.byref(self._orgTimeouts))
|
|
|
|
self._reconfigurePort()
|
|
|
|
# Clear buffers:
|
|
# Remove anything that was there
|
|
win32.PurgeComm(self.hComPort,
|
|
win32.PURGE_TXCLEAR | win32.PURGE_TXABORT |
|
|
win32.PURGE_RXCLEAR | win32.PURGE_RXABORT)
|
|
except:
|
|
try:
|
|
self._close()
|
|
except:
|
|
# ignore any exception when closing the port
|
|
# also to keep original exception that happened when setting up
|
|
pass
|
|
self.hComPort = None
|
|
raise
|
|
else:
|
|
self._isOpen = True
|
|
|
|
|
|
def _reconfigurePort(self):
|
|
"""Set communication 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 = (win32.MAXDWORD, 0, 0, 0, 0)
|
|
else:
|
|
timeouts = (0, 0, int(self._timeout*1000), 0, 0)
|
|
if self._timeout != 0 and self._interCharTimeout is not None:
|
|
timeouts = (int(self._interCharTimeout * 1000),) + timeouts[1:]
|
|
|
|
if self._writeTimeout is None:
|
|
pass
|
|
elif self._writeTimeout == 0:
|
|
timeouts = timeouts[:-2] + (0, win32.MAXDWORD)
|
|
else:
|
|
timeouts = timeouts[:-2] + (0, int(self._writeTimeout*1000))
|
|
win32.SetCommTimeouts(self.hComPort, ctypes.byref(win32.COMMTIMEOUTS(*timeouts)))
|
|
|
|
win32.SetCommMask(self.hComPort, win32.EV_ERR)
|
|
|
|
# Setup the connection info.
|
|
# Get state and modify it:
|
|
comDCB = win32.DCB()
|
|
win32.GetCommState(self.hComPort, ctypes.byref(comDCB))
|
|
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 = win32.NOPARITY
|
|
comDCB.fParity = 0 # Disable Parity Check
|
|
elif self._parity == PARITY_EVEN:
|
|
comDCB.Parity = win32.EVENPARITY
|
|
comDCB.fParity = 1 # Enable Parity Check
|
|
elif self._parity == PARITY_ODD:
|
|
comDCB.Parity = win32.ODDPARITY
|
|
comDCB.fParity = 1 # Enable Parity Check
|
|
elif self._parity == PARITY_MARK:
|
|
comDCB.Parity = win32.MARKPARITY
|
|
comDCB.fParity = 1 # Enable Parity Check
|
|
elif self._parity == PARITY_SPACE:
|
|
comDCB.Parity = win32.SPACEPARITY
|
|
comDCB.fParity = 1 # Enable Parity Check
|
|
else:
|
|
raise ValueError("Unsupported parity mode: %r" % self._parity)
|
|
|
|
if self._stopbits == STOPBITS_ONE:
|
|
comDCB.StopBits = win32.ONESTOPBIT
|
|
elif self._stopbits == STOPBITS_ONE_POINT_FIVE:
|
|
comDCB.StopBits = win32.ONE5STOPBITS
|
|
elif self._stopbits == STOPBITS_TWO:
|
|
comDCB.StopBits = win32.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 = win32.RTS_CONTROL_HANDSHAKE
|
|
elif self._rtsToggle:
|
|
comDCB.fRtsControl = win32.RTS_CONTROL_TOGGLE
|
|
else:
|
|
comDCB.fRtsControl = self._rtsState
|
|
if self._dsrdtr:
|
|
comDCB.fDtrControl = win32.DTR_CONTROL_HANDSHAKE
|
|
else:
|
|
comDCB.fDtrControl = self._dtrState
|
|
|
|
if self._rtsToggle:
|
|
comDCB.fOutxCtsFlow = 0
|
|
else:
|
|
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
|
|
|
|
if not win32.SetCommState(self.hComPort, ctypes.byref(comDCB)):
|
|
raise ValueError("Cannot configure port, some setting was wrong. Original message: %r" % ctypes.WinError())
|
|
|
|
#~ def __del__(self):
|
|
#~ self.close()
|
|
|
|
|
|
def _close(self):
|
|
"""internal close port helper"""
|
|
if self.hComPort:
|
|
# Restore original timeout values:
|
|
win32.SetCommTimeouts(self.hComPort, self._orgTimeouts)
|
|
# Close COM-Port:
|
|
win32.CloseHandle(self.hComPort)
|
|
if self._overlappedRead is not None:
|
|
win32.CloseHandle(self._overlappedRead.hEvent)
|
|
self._overlappedRead = None
|
|
if self._overlappedWrite is not None:
|
|
win32.CloseHandle(self._overlappedWrite.hEvent)
|
|
self._overlappedWrite = None
|
|
self.hComPort = None
|
|
|
|
def close(self):
|
|
"""Close port"""
|
|
if self._isOpen:
|
|
self._close()
|
|
self._isOpen = False
|
|
|
|
def makeDeviceName(self, port):
|
|
return device(port)
|
|
|
|
# - - - - - - - - - - - - - - - - - - - - - - - -
|
|
|
|
def inWaiting(self):
|
|
"""Return the number of characters currently in the input buffer."""
|
|
flags = win32.DWORD()
|
|
comstat = win32.COMSTAT()
|
|
if not win32.ClearCommError(self.hComPort, ctypes.byref(flags), ctypes.byref(comstat)):
|
|
raise SerialException('call to ClearCommError failed')
|
|
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:
|
|
win32.ResetEvent(self._overlappedRead.hEvent)
|
|
flags = win32.DWORD()
|
|
comstat = win32.COMSTAT()
|
|
if not win32.ClearCommError(self.hComPort, ctypes.byref(flags), ctypes.byref(comstat)):
|
|
raise SerialException('call to ClearCommError failed')
|
|
if self.timeout == 0:
|
|
n = min(comstat.cbInQue, size)
|
|
if n > 0:
|
|
buf = ctypes.create_string_buffer(n)
|
|
rc = win32.DWORD()
|
|
err = win32.ReadFile(self.hComPort, buf, n, ctypes.byref(rc), ctypes.byref(self._overlappedRead))
|
|
if not err and win32.GetLastError() != win32.ERROR_IO_PENDING:
|
|
raise SerialException("ReadFile failed (%r)" % ctypes.WinError())
|
|
err = win32.WaitForSingleObject(self._overlappedRead.hEvent, win32.INFINITE)
|
|
read = buf.raw[:rc.value]
|
|
else:
|
|
read = bytes()
|
|
else:
|
|
buf = ctypes.create_string_buffer(size)
|
|
rc = win32.DWORD()
|
|
err = win32.ReadFile(self.hComPort, buf, size, ctypes.byref(rc), ctypes.byref(self._overlappedRead))
|
|
if not err and win32.GetLastError() != win32.ERROR_IO_PENDING:
|
|
raise SerialException("ReadFile failed (%r)" % ctypes.WinError())
|
|
err = win32.GetOverlappedResult(self.hComPort, ctypes.byref(self._overlappedRead), ctypes.byref(rc), True)
|
|
read = buf.raw[:rc.value]
|
|
else:
|
|
read = bytes()
|
|
return bytes(read)
|
|
|
|
def write(self, data):
|
|
"""Output the given string over the serial port."""
|
|
if not self.hComPort: raise portNotOpenError
|
|
#~ if not isinstance(data, (bytes, bytearray)):
|
|
#~ raise TypeError('expected %s or bytearray, got %s' % (bytes, type(data)))
|
|
# convert data (needed in case of memoryview instance: Py 3.1 io lib), ctypes doesn't like memoryview
|
|
data = to_bytes(data)
|
|
if data:
|
|
#~ win32event.ResetEvent(self._overlappedWrite.hEvent)
|
|
n = win32.DWORD()
|
|
err = win32.WriteFile(self.hComPort, data, len(data), ctypes.byref(n), self._overlappedWrite)
|
|
if not err and win32.GetLastError() != win32.ERROR_IO_PENDING:
|
|
raise SerialException("WriteFile failed (%r)" % ctypes.WinError())
|
|
if self._writeTimeout != 0: # if blocking (None) or w/ write timeout (>0)
|
|
# Wait for the write to complete.
|
|
#~ win32.WaitForSingleObject(self._overlappedWrite.hEvent, win32.INFINITE)
|
|
err = win32.GetOverlappedResult(self.hComPort, self._overlappedWrite, ctypes.byref(n), True)
|
|
if n.value != len(data):
|
|
raise writeTimeoutError
|
|
return n.value
|
|
else:
|
|
return 0
|
|
|
|
def flush(self):
|
|
"""Flush of file like objects. In this case, wait until all data
|
|
is written."""
|
|
while self.outWaiting():
|
|
time.sleep(0.05)
|
|
# XXX could also use WaitCommEvent with mask EV_TXEMPTY, but it would
|
|
# require overlapped IO and its also only possible to set a single mask
|
|
# on the port---
|
|
|
|
def flushInput(self):
|
|
"""Clear input buffer, discarding all that is in the buffer."""
|
|
if not self.hComPort: raise portNotOpenError
|
|
win32.PurgeComm(self.hComPort, win32.PURGE_RXCLEAR | win32.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
|
|
win32.PurgeComm(self.hComPort, win32.PURGE_TXCLEAR | win32.PURGE_TXABORT)
|
|
|
|
def sendBreak(self, duration=0.25):
|
|
"""Send break condition. Timed, returns to idle state after given duration."""
|
|
if not self.hComPort: raise portNotOpenError
|
|
import time
|
|
win32.SetCommBreak(self.hComPort)
|
|
time.sleep(duration)
|
|
win32.ClearCommBreak(self.hComPort)
|
|
|
|
def setBreak(self, level=1):
|
|
"""Set break: Controls TXD. When active, to transmitting is possible."""
|
|
if not self.hComPort: raise portNotOpenError
|
|
if level:
|
|
win32.SetCommBreak(self.hComPort)
|
|
else:
|
|
win32.ClearCommBreak(self.hComPort)
|
|
|
|
def setRTS(self, level=1):
|
|
"""Set terminal status line: Request To Send"""
|
|
# remember level for reconfigure
|
|
if level:
|
|
self._rtsState = win32.RTS_CONTROL_ENABLE
|
|
else:
|
|
self._rtsState = win32.RTS_CONTROL_DISABLE
|
|
# also apply now if port is open
|
|
if self.hComPort:
|
|
if level:
|
|
win32.EscapeCommFunction(self.hComPort, win32.SETRTS)
|
|
else:
|
|
win32.EscapeCommFunction(self.hComPort, win32.CLRRTS)
|
|
|
|
def setDTR(self, level=1):
|
|
"""Set terminal status line: Data Terminal Ready"""
|
|
# remember level for reconfigure
|
|
if level:
|
|
self._dtrState = win32.DTR_CONTROL_ENABLE
|
|
else:
|
|
self._dtrState = win32.DTR_CONTROL_DISABLE
|
|
# also apply now if port is open
|
|
if self.hComPort:
|
|
if level:
|
|
win32.EscapeCommFunction(self.hComPort, win32.SETDTR)
|
|
else:
|
|
win32.EscapeCommFunction(self.hComPort, win32.CLRDTR)
|
|
|
|
def _GetCommModemStatus(self):
|
|
stat = win32.DWORD()
|
|
win32.GetCommModemStatus(self.hComPort, ctypes.byref(stat))
|
|
return stat.value
|
|
|
|
def getCTS(self):
|
|
"""Read terminal status line: Clear To Send"""
|
|
if not self.hComPort: raise portNotOpenError
|
|
return win32.MS_CTS_ON & self._GetCommModemStatus() != 0
|
|
|
|
def getDSR(self):
|
|
"""Read terminal status line: Data Set Ready"""
|
|
if not self.hComPort: raise portNotOpenError
|
|
return win32.MS_DSR_ON & self._GetCommModemStatus() != 0
|
|
|
|
def getRI(self):
|
|
"""Read terminal status line: Ring Indicator"""
|
|
if not self.hComPort: raise portNotOpenError
|
|
return win32.MS_RING_ON & self._GetCommModemStatus() != 0
|
|
|
|
def getCD(self):
|
|
"""Read terminal status line: Carrier Detect"""
|
|
if not self.hComPort: raise portNotOpenError
|
|
return win32.MS_RLSD_ON & self._GetCommModemStatus() != 0
|
|
|
|
# - - platform specific - - - -
|
|
|
|
def setBufferSize(self, rx_size=4096, tx_size=None):
|
|
"""\
|
|
Recommend a buffer size to the driver (device driver can ignore this
|
|
vlaue). Must be called before the port is opended.
|
|
"""
|
|
if tx_size is None: tx_size = rx_size
|
|
win32.SetupComm(self.hComPort, rx_size, tx_size)
|
|
|
|
def setXON(self, level=True):
|
|
"""\
|
|
Manually control flow - when software flow control is enabled.
|
|
This will send XON (true) and XOFF (false) to the other device.
|
|
WARNING: this function is not portable to different platforms!
|
|
"""
|
|
if not self.hComPort: raise portNotOpenError
|
|
if level:
|
|
win32.EscapeCommFunction(self.hComPort, win32.SETXON)
|
|
else:
|
|
win32.EscapeCommFunction(self.hComPort, win32.SETXOFF)
|
|
|
|
def outWaiting(self):
|
|
"""return how many characters the in the outgoing buffer"""
|
|
flags = win32.DWORD()
|
|
comstat = win32.COMSTAT()
|
|
if not win32.ClearCommError(self.hComPort, ctypes.byref(flags), ctypes.byref(comstat)):
|
|
raise SerialException('call to ClearCommError failed')
|
|
return comstat.cbOutQue
|
|
|
|
# functions useful for RS-485 adapters
|
|
def setRtsToggle(self, rtsToggle):
|
|
"""Change RTS toggle control setting."""
|
|
self._rtsToggle = rtsToggle
|
|
if self._isOpen: self._reconfigurePort()
|
|
|
|
def getRtsToggle(self):
|
|
"""Get the current RTS toggle control setting."""
|
|
return self._rtsToggle
|
|
|
|
rtsToggle = property(getRtsToggle, setRtsToggle, doc="RTS toggle control setting")
|
|
|
|
|
|
# assemble Serial class with the platform specific implementation and the base
|
|
# for file-like behavior. for Python 2.6 and newer, that provide the new I/O
|
|
# library, derive from io.RawIOBase
|
|
try:
|
|
import io
|
|
except ImportError:
|
|
# classic version with our own file-like emulation
|
|
class Serial(Win32Serial, FileLike):
|
|
pass
|
|
else:
|
|
# io library present
|
|
class Serial(Win32Serial, io.RawIOBase):
|
|
pass
|
|
|
|
|
|
# Nur Testfunktion!!
|
|
if __name__ == '__main__':
|
|
s = Serial(0)
|
|
sys.stdout.write("%s\n" % s)
|
|
|
|
s = Serial()
|
|
sys.stdout.write("%s\n" % s)
|
|
|
|
s.baudrate = 19200
|
|
s.databits = 7
|
|
s.close()
|
|
s.port = 0
|
|
s.open()
|
|
sys.stdout.write("%s\n" % s)
|
|
|