| #! python |
| # |
| # Base class and support functions used by various backends. |
| # |
| # This file is part of pySerial. https://github.com/pyserial/pyserial |
| # (C) 2001-2015 Chris Liechti <cliechti@gmx.net> |
| # |
| # SPDX-License-Identifier: BSD-3-Clause |
| |
| import io |
| import time |
| |
| # ``memoryview`` was introduced in Python 2.7 and ``bytes(some_memoryview)`` |
| # isn't returning the contents (very unfortunate). Therefore we need special |
| # cases and test for it. Ensure that there is a ``memoryview`` object for older |
| # Python versions. This is easier than making every test dependent on its |
| # existence. |
| try: |
| memoryview |
| except (NameError, AttributeError): |
| # implementation does not matter as we do not really use it. |
| # it just must not inherit from something else we might care for. |
| class memoryview(object): # pylint: disable=redefined-builtin,invalid-name |
| pass |
| |
| try: |
| unicode |
| except (NameError, AttributeError): |
| unicode = str # for Python 3, pylint: disable=redefined-builtin,invalid-name |
| |
| try: |
| basestring |
| except (NameError, AttributeError): |
| basestring = (str,) # for Python 3, pylint: disable=redefined-builtin,invalid-name |
| |
| |
| # "for byte in data" fails for python3 as it returns ints instead of bytes |
| def iterbytes(b): |
| """Iterate over bytes, returning bytes instead of ints (python3)""" |
| if isinstance(b, memoryview): |
| b = b.tobytes() |
| i = 0 |
| while True: |
| a = b[i:i + 1] |
| i += 1 |
| if a: |
| yield a |
| else: |
| break |
| |
| |
| # all Python versions prior 3.x convert ``str([17])`` to '[17]' instead of '\x11' |
| # so a simple ``bytes(sequence)`` doesn't work for all versions |
| def to_bytes(seq): |
| """convert a sequence to a bytes type""" |
| if isinstance(seq, bytes): |
| return seq |
| elif isinstance(seq, bytearray): |
| return bytes(seq) |
| elif isinstance(seq, memoryview): |
| return seq.tobytes() |
| elif isinstance(seq, unicode): |
| raise TypeError('unicode strings are not supported, please encode to bytes: {!r}'.format(seq)) |
| else: |
| b = bytearray() |
| for item in seq: |
| # this one handles int and bytes in Python 2.7 |
| # add conversion in case of Python 3.x |
| if isinstance(item, bytes): |
| item = ord(item) |
| b.append(item) |
| return bytes(b) |
| |
| # create control bytes |
| XON = to_bytes([17]) |
| XOFF = to_bytes([19]) |
| |
| CR = to_bytes([13]) |
| LF = to_bytes([10]) |
| |
| |
| PARITY_NONE, PARITY_EVEN, PARITY_ODD, PARITY_MARK, PARITY_SPACE = 'N', 'E', 'O', 'M', 'S' |
| STOPBITS_ONE, STOPBITS_ONE_POINT_FIVE, STOPBITS_TWO = (1, 1.5, 2) |
| FIVEBITS, SIXBITS, SEVENBITS, EIGHTBITS = (5, 6, 7, 8) |
| |
| PARITY_NAMES = { |
| PARITY_NONE: 'None', |
| PARITY_EVEN: 'Even', |
| PARITY_ODD: 'Odd', |
| PARITY_MARK: 'Mark', |
| PARITY_SPACE: 'Space', |
| } |
| |
| |
| class SerialException(IOError): |
| """Base class for serial port related exceptions.""" |
| |
| |
| class SerialTimeoutException(SerialException): |
| """Write timeouts give an exception""" |
| |
| |
| writeTimeoutError = SerialTimeoutException('Write timeout') |
| portNotOpenError = SerialException('Attempting to use a port that is not open') |
| |
| |
| class SerialBase(io.RawIOBase): |
| """\ |
| Serial port base class. Provides __init__ function and properties to |
| get/set port settings. |
| """ |
| |
| # default values, may be overridden 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, PARITY_MARK, PARITY_SPACE) |
| STOPBITS = (STOPBITS_ONE, STOPBITS_ONE_POINT_FIVE, STOPBITS_TWO) |
| |
| def __init__(self, |
| port=None, |
| baudrate=9600, |
| bytesize=EIGHTBITS, |
| parity=PARITY_NONE, |
| stopbits=STOPBITS_ONE, |
| timeout=None, |
| xonxoff=False, |
| rtscts=False, |
| write_timeout=None, |
| dsrdtr=False, |
| inter_byte_timeout=None, |
| **kwargs): |
| """\ |
| 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.is_open = False |
| self.portstr = None |
| self.name = None |
| # correct values are assigned below through properties |
| self._port = None |
| self._baudrate = None |
| self._bytesize = None |
| self._parity = None |
| self._stopbits = None |
| self._timeout = None |
| self._write_timeout = None |
| self._xonxoff = None |
| self._rtscts = None |
| self._dsrdtr = None |
| self._inter_byte_timeout = None |
| self._rs485_mode = None # disabled by default |
| self._rts_state = True |
| self._dtr_state = True |
| self._break_state = False |
| |
| # 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.write_timeout = write_timeout |
| self.xonxoff = xonxoff |
| self.rtscts = rtscts |
| self.dsrdtr = dsrdtr |
| self.inter_byte_timeout = inter_byte_timeout |
| # watch for backward compatible kwargs |
| if 'writeTimeout' in kwargs: |
| self.write_timeout = kwargs.pop('writeTimeout') |
| if 'interCharTimeout' in kwargs: |
| self.inter_byte_timeout = kwargs.pop('interCharTimeout') |
| if kwargs: |
| raise ValueError('unexpected keyword arguments: {!r}'.format(kwargs)) |
| |
| if port is not None: |
| self.open() |
| |
| # - - - - - - - - - - - - - - - - - - - - - - - - |
| |
| # to be implemented by subclasses: |
| # def open(self): |
| # def close(self): |
| |
| # - - - - - - - - - - - - - - - - - - - - - - - - |
| |
| @property |
| def port(self): |
| """\ |
| Get the current port setting. The value that was passed on init or using |
| setPort() is passed back. |
| """ |
| return self._port |
| |
| @port.setter |
| def port(self, port): |
| """\ |
| Change the port. |
| """ |
| if port is not None and not isinstance(port, basestring): |
| raise ValueError('"port" must be None or a string, not {}'.format(type(port))) |
| was_open = self.is_open |
| if was_open: |
| self.close() |
| self.portstr = port |
| self._port = port |
| self.name = self.portstr |
| if was_open: |
| self.open() |
| |
| @property |
| def baudrate(self): |
| """Get the current baud rate setting.""" |
| return self._baudrate |
| |
| @baudrate.setter |
| def baudrate(self, baudrate): |
| """\ |
| Change baud rate. It raises a ValueError if the port is open and the |
| baud rate is not possible. If the port is closed, then the value is |
| accepted and the exception is raised when the port is opened. |
| """ |
| try: |
| b = int(baudrate) |
| except TypeError: |
| raise ValueError("Not a valid baudrate: {!r}".format(baudrate)) |
| else: |
| if b < 0: |
| raise ValueError("Not a valid baudrate: {!r}".format(baudrate)) |
| self._baudrate = b |
| if self.is_open: |
| self._reconfigure_port() |
| |
| @property |
| def bytesize(self): |
| """Get the current byte size setting.""" |
| return self._bytesize |
| |
| @bytesize.setter |
| def bytesize(self, bytesize): |
| """Change byte size.""" |
| if bytesize not in self.BYTESIZES: |
| raise ValueError("Not a valid byte size: {!r}".format(bytesize)) |
| self._bytesize = bytesize |
| if self.is_open: |
| self._reconfigure_port() |
| |
| @property |
| def parity(self): |
| """Get the current parity setting.""" |
| return self._parity |
| |
| @parity.setter |
| def parity(self, parity): |
| """Change parity setting.""" |
| if parity not in self.PARITIES: |
| raise ValueError("Not a valid parity: {!r}".format(parity)) |
| self._parity = parity |
| if self.is_open: |
| self._reconfigure_port() |
| |
| @property |
| def stopbits(self): |
| """Get the current stop bits setting.""" |
| return self._stopbits |
| |
| @stopbits.setter |
| def stopbits(self, stopbits): |
| """Change stop bits size.""" |
| if stopbits not in self.STOPBITS: |
| raise ValueError("Not a valid stop bit size: {!r}".format(stopbits)) |
| self._stopbits = stopbits |
| if self.is_open: |
| self._reconfigure_port() |
| |
| @property |
| def timeout(self): |
| """Get the current timeout setting.""" |
| return self._timeout |
| |
| @timeout.setter |
| def timeout(self, timeout): |
| """Change timeout setting.""" |
| if timeout is not None: |
| try: |
| timeout + 1 # test if it's a number, will throw a TypeError if not... |
| except TypeError: |
| raise ValueError("Not a valid timeout: {!r}".format(timeout)) |
| if timeout < 0: |
| raise ValueError("Not a valid timeout: {!r}".format(timeout)) |
| self._timeout = timeout |
| if self.is_open: |
| self._reconfigure_port() |
| |
| @property |
| def write_timeout(self): |
| """Get the current timeout setting.""" |
| return self._write_timeout |
| |
| @write_timeout.setter |
| def write_timeout(self, timeout): |
| """Change timeout setting.""" |
| if timeout is not None: |
| if timeout < 0: |
| raise ValueError("Not a valid timeout: {!r}".format(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}".format(timeout)) |
| |
| self._write_timeout = timeout |
| if self.is_open: |
| self._reconfigure_port() |
| |
| @property |
| def inter_byte_timeout(self): |
| """Get the current inter-character timeout setting.""" |
| return self._inter_byte_timeout |
| |
| @inter_byte_timeout.setter |
| def inter_byte_timeout(self, ic_timeout): |
| """Change inter-byte timeout setting.""" |
| if ic_timeout is not None: |
| if ic_timeout < 0: |
| raise ValueError("Not a valid timeout: {!r}".format(ic_timeout)) |
| try: |
| ic_timeout + 1 # test if it's a number, will throw a TypeError if not... |
| except TypeError: |
| raise ValueError("Not a valid timeout: {!r}".format(ic_timeout)) |
| |
| self._inter_byte_timeout = ic_timeout |
| if self.is_open: |
| self._reconfigure_port() |
| |
| @property |
| def xonxoff(self): |
| """Get the current XON/XOFF setting.""" |
| return self._xonxoff |
| |
| @xonxoff.setter |
| def xonxoff(self, xonxoff): |
| """Change XON/XOFF setting.""" |
| self._xonxoff = xonxoff |
| if self.is_open: |
| self._reconfigure_port() |
| |
| @property |
| def rtscts(self): |
| """Get the current RTS/CTS flow control setting.""" |
| return self._rtscts |
| |
| @rtscts.setter |
| def rtscts(self, rtscts): |
| """Change RTS/CTS flow control setting.""" |
| self._rtscts = rtscts |
| if self.is_open: |
| self._reconfigure_port() |
| |
| @property |
| def dsrdtr(self): |
| """Get the current DSR/DTR flow control setting.""" |
| return self._dsrdtr |
| |
| @dsrdtr.setter |
| def dsrdtr(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.is_open: |
| self._reconfigure_port() |
| |
| @property |
| def rts(self): |
| return self._rts_state |
| |
| @rts.setter |
| def rts(self, value): |
| self._rts_state = value |
| if self.is_open: |
| self._update_rts_state() |
| |
| @property |
| def dtr(self): |
| return self._dtr_state |
| |
| @dtr.setter |
| def dtr(self, value): |
| self._dtr_state = value |
| if self.is_open: |
| self._update_dtr_state() |
| |
| @property |
| def break_condition(self): |
| return self._break_state |
| |
| @break_condition.setter |
| def break_condition(self, value): |
| self._break_state = value |
| if self.is_open: |
| self._update_break_state() |
| |
| # - - - - - - - - - - - - - - - - - - - - - - - - |
| # functions useful for RS-485 adapters |
| |
| @property |
| def rs485_mode(self): |
| """\ |
| Enable RS485 mode and apply new settings, set to None to disable. |
| See serial.rs485.RS485Settings for more info about the value. |
| """ |
| return self._rs485_mode |
| |
| @rs485_mode.setter |
| def rs485_mode(self, rs485_settings): |
| self._rs485_mode = rs485_settings |
| if self.is_open: |
| self._reconfigure_port() |
| |
| # - - - - - - - - - - - - - - - - - - - - - - - - |
| |
| _SAVED_SETTINGS = ('baudrate', 'bytesize', 'parity', 'stopbits', 'xonxoff', |
| 'dsrdtr', 'rtscts', 'timeout', 'write_timeout', |
| 'inter_byte_timeout') |
| |
| def get_settings(self): |
| """\ |
| Get current port settings as a dictionary. For use with |
| apply_settings(). |
| """ |
| return dict([(key, getattr(self, '_' + key)) for key in self._SAVED_SETTINGS]) |
| |
| def apply_settings(self, d): |
| """\ |
| Apply stored settings from a dictionary returned from |
| get_settings(). It's allowed to delete keys from the dictionary. These |
| values will simply left unchanged. |
| """ |
| for key in self._SAVED_SETTINGS: |
| if key in d and d[key] != getattr(self, '_' + key): # check against internal "_" value |
| setattr(self, key, d[key]) # set non "_" value to use properties write function |
| |
| # - - - - - - - - - - - - - - - - - - - - - - - - |
| |
| def __repr__(self): |
| """String representation of the current port settings and its state.""" |
| return '{name}<id=0x{id:x}, open={p.is_open}>(port={p.portstr!r}, ' \ |
| 'baudrate={p.baudrate!r}, bytesize={p.bytesize!r}, parity={p.parity!r}, ' \ |
| 'stopbits={p.stopbits!r}, timeout={p.timeout!r}, xonxoff={p.xonxoff!r}, ' \ |
| 'rtscts={p.rtscts!r}, dsrdtr={p.dsrdtr!r})'.format( |
| name=self.__class__.__name__, id=id(self), p=self) |
| |
| # - - - - - - - - - - - - - - - - - - - - - - - - |
| # compatibility with io library |
| # pylint: disable=invalid-name,missing-docstring |
| |
| def readable(self): |
| return True |
| |
| def writable(self): |
| return True |
| |
| def seekable(self): |
| return False |
| |
| def readinto(self, b): |
| data = self.read(len(b)) |
| n = len(data) |
| try: |
| b[:n] = data |
| except TypeError as err: |
| import array |
| if not isinstance(b, array.array): |
| raise err |
| b[:n] = array.array('b', data) |
| return n |
| |
| # - - - - - - - - - - - - - - - - - - - - - - - - |
| # context manager |
| |
| def __enter__(self): |
| return self |
| |
| def __exit__(self, *args, **kwargs): |
| self.close() |
| |
| # - - - - - - - - - - - - - - - - - - - - - - - - |
| |
| def send_break(self, duration=0.25): |
| """\ |
| Send break condition. Timed, returns to idle state after given |
| duration. |
| """ |
| if not self.is_open: |
| raise portNotOpenError |
| self.break_condition = True |
| time.sleep(duration) |
| self.break_condition = False |
| |
| # - - - - - - - - - - - - - - - - - - - - - - - - |
| # backwards compatibility / deprecated functions |
| |
| def flushInput(self): |
| self.reset_input_buffer() |
| |
| def flushOutput(self): |
| self.reset_output_buffer() |
| |
| def inWaiting(self): |
| return self.in_waiting |
| |
| def sendBreak(self, duration=0.25): |
| self.send_break(duration) |
| |
| def setRTS(self, value=1): |
| self.rts = value |
| |
| def setDTR(self, value=1): |
| self.dtr = value |
| |
| def getCTS(self): |
| return self.cts |
| |
| def getDSR(self): |
| return self.dsr |
| |
| def getRI(self): |
| return self.ri |
| |
| def getCD(self): |
| return self.cd |
| |
| def setPort(self, port): |
| self.port = port |
| |
| @property |
| def writeTimeout(self): |
| return self.write_timeout |
| |
| @writeTimeout.setter |
| def writeTimeout(self, timeout): |
| self.write_timeout = timeout |
| |
| @property |
| def interCharTimeout(self): |
| return self.inter_byte_timeout |
| |
| @interCharTimeout.setter |
| def interCharTimeout(self, interCharTimeout): |
| self.inter_byte_timeout = interCharTimeout |
| |
| def getSettingsDict(self): |
| return self.get_settings() |
| |
| def applySettingsDict(self, d): |
| self.apply_settings(d) |
| |
| def isOpen(self): |
| return self.is_open |
| |
| # - - - - - - - - - - - - - - - - - - - - - - - - |
| # additional functionality |
| |
| def read_all(self): |
| """\ |
| Read all bytes currently available in the buffer of the OS. |
| """ |
| return self.read(self.in_waiting) |
| |
| def read_until(self, terminator=LF, size=None): |
| """\ |
| Read until a termination sequence is found ('\n' by default), the size |
| is exceeded or until timeout occurs. |
| """ |
| lenterm = len(terminator) |
| line = bytearray() |
| while True: |
| c = self.read(1) |
| if c: |
| line += c |
| if line[-lenterm:] == terminator: |
| break |
| if size is not None and len(line) >= size: |
| break |
| else: |
| break |
| return bytes(line) |
| |
| def iread_until(self, *args, **kwargs): |
| """\ |
| Read lines, implemented as generator. It will raise StopIteration on |
| timeout (empty read). |
| """ |
| while True: |
| line = self.read_until(*args, **kwargs) |
| if not line: |
| break |
| yield line |
| |
| |
| # - - - - - - - - - - - - - - - - - - - - - - - - - |
| if __name__ == '__main__': |
| import sys |
| s = SerialBase() |
| sys.stdout.write('port name: {}\n'.format(s.name)) |
| sys.stdout.write('baud rates: {}\n'.format(s.BAUDRATES)) |
| sys.stdout.write('byte sizes: {}\n'.format(s.BYTESIZES)) |
| sys.stdout.write('parities: {}\n'.format(s.PARITIES)) |
| sys.stdout.write('stop bits: {}\n'.format(s.STOPBITS)) |
| sys.stdout.write('{}\n'.format(s)) |
