309 lines
10 KiB
Python
309 lines
10 KiB
Python
|
#! python
|
||
|
|
#
|
||
|
|
# This module implements a loop back connection receiving itself what it sent.
|
||
|
|
#
|
||
|
|
# The purpose of this module is.. well... You can run the unit tests with it.
|
||
|
|
# and it was so easy to implement ;-)
|
||
|
|
#
|
||
|
|
# This file is part of pySerial. https://github.com/pyserial/pyserial
|
||
|
|
# (C) 2001-2020 Chris Liechti <cliechti@gmx.net>
|
||
|
|
#
|
||
|
|
# SPDX-License-Identifier: BSD-3-Clause
|
||
|
|
#
|
||
|
|
# URL format: loop://[option[/option...]]
|
||
|
|
# options:
|
||
|
|
# - "debug" print diagnostic messages
|
||
|
|
from __future__ import absolute_import
|
||
|
|
|
||
|
|
import logging
|
||
|
|
import numbers
|
||
|
|
import time
|
||
|
|
try:
|
||
|
|
import urlparse
|
||
|
|
except ImportError:
|
||
|
|
import urllib.parse as urlparse
|
||
|
|
try:
|
||
|
|
import queue
|
||
|
|
except ImportError:
|
||
|
|
import Queue as queue
|
||
|
|
|
||
|
|
from serial.serialutil import SerialBase, SerialException, to_bytes, iterbytes, SerialTimeoutException, PortNotOpenError
|
||
|
|
|
||
|
|
# map log level names to constants. used in from_url()
|
||
|
|
LOGGER_LEVELS = {
|
||
|
|
'debug': logging.DEBUG,
|
||
|
|
'info': logging.INFO,
|
||
|
|
'warning': logging.WARNING,
|
||
|
|
'error': logging.ERROR,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
class Serial(SerialBase):
|
||
|
|
"""Serial port implementation that simulates a loop back connection in plain software."""
|
||
|
|
|
||
|
|
BAUDRATES = (50, 75, 110, 134, 150, 200, 300, 600, 1200, 1800, 2400, 4800,
|
||
|
|
9600, 19200, 38400, 57600, 115200)
|
||
|
|
|
||
|
|
def __init__(self, *args, **kwargs):
|
||
|
|
self.buffer_size = 4096
|
||
|
|
self.queue = None
|
||
|
|
self.logger = None
|
||
|
|
self._cancel_write = False
|
||
|
|
super(Serial, self).__init__(*args, **kwargs)
|
||
|
|
|
||
|
|
def open(self):
|
||
|
|
"""\
|
||
|
|
Open port with current settings. This may throw a SerialException
|
||
|
|
if the port cannot be opened.
|
||
|
|
"""
|
||
|
|
if self.is_open:
|
||
|
|
raise SerialException("Port is already open.")
|
||
|
|
self.logger = None
|
||
|
|
self.queue = queue.Queue(self.buffer_size)
|
||
|
|
|
||
|
|
if self._port is None:
|
||
|
|
raise SerialException("Port must be configured before it can be used.")
|
||
|
|
# not that there is anything to open, but the function applies the
|
||
|
|
# options found in the URL
|
||
|
|
self.from_url(self.port)
|
||
|
|
|
||
|
|
# not that there anything to configure...
|
||
|
|
self._reconfigure_port()
|
||
|
|
# all things set up get, now a clean start
|
||
|
|
self.is_open = True
|
||
|
|
if not self._dsrdtr:
|
||
|
|
self._update_dtr_state()
|
||
|
|
if not self._rtscts:
|
||
|
|
self._update_rts_state()
|
||
|
|
self.reset_input_buffer()
|
||
|
|
self.reset_output_buffer()
|
||
|
|
|
||
|
|
def close(self):
|
||
|
|
if self.is_open:
|
||
|
|
self.is_open = False
|
||
|
|
try:
|
||
|
|
self.queue.put_nowait(None)
|
||
|
|
except queue.Full:
|
||
|
|
pass
|
||
|
|
super(Serial, self).close()
|
||
|
|
|
||
|
|
def _reconfigure_port(self):
|
||
|
|
"""\
|
||
|
|
Set communication parameters on opened port. For the loop://
|
||
|
|
protocol all settings are ignored!
|
||
|
|
"""
|
||
|
|
# not that's it of any real use, but it helps in the unit tests
|
||
|
|
if not isinstance(self._baudrate, numbers.Integral) or not 0 < self._baudrate < 2 ** 32:
|
||
|
|
raise ValueError("invalid baudrate: {!r}".format(self._baudrate))
|
||
|
|
if self.logger:
|
||
|
|
self.logger.info('_reconfigure_port()')
|
||
|
|
|
||
|
|
def from_url(self, url):
|
||
|
|
"""extract host and port from an URL string"""
|
||
|
|
parts = urlparse.urlsplit(url)
|
||
|
|
if parts.scheme != "loop":
|
||
|
|
raise SerialException(
|
||
|
|
'expected a string in the form '
|
||
|
|
'"loop://[?logging={debug|info|warning|error}]": not starting '
|
||
|
|
'with loop:// ({!r})'.format(parts.scheme))
|
||
|
|
try:
|
||
|
|
# process options now, directly altering self
|
||
|
|
for option, values in urlparse.parse_qs(parts.query, True).items():
|
||
|
|
if option == 'logging':
|
||
|
|
logging.basicConfig() # XXX is that good to call it here?
|
||
|
|
self.logger = logging.getLogger('pySerial.loop')
|
||
|
|
self.logger.setLevel(LOGGER_LEVELS[values[0]])
|
||
|
|
self.logger.debug('enabled logging')
|
||
|
|
else:
|
||
|
|
raise ValueError('unknown option: {!r}'.format(option))
|
||
|
|
except ValueError as e:
|
||
|
|
raise SerialException(
|
||
|
|
'expected a string in the form '
|
||
|
|
'"loop://[?logging={debug|info|warning|error}]": {}'.format(e))
|
||
|
|
|
||
|
|
# - - - - - - - - - - - - - - - - - - - - - - - -
|
||
|
|
|
||
|
|
@property
|
||
|
|
def in_waiting(self):
|
||
|
|
"""Return the number of bytes currently in the input buffer."""
|
||
|
|
if not self.is_open:
|
||
|
|
raise PortNotOpenError()
|
||
|
|
if self.logger:
|
||
|
|
# attention the logged value can differ from return value in
|
||
|
|
# threaded environments...
|
||
|
|
self.logger.debug('in_waiting -> {:d}'.format(self.queue.qsize()))
|
||
|
|
return self.queue.qsize()
|
||
|
|
|
||
|
|
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.is_open:
|
||
|
|
raise PortNotOpenError()
|
||
|
|
if self._timeout is not None and self._timeout != 0:
|
||
|
|
timeout = time.time() + self._timeout
|
||
|
|
else:
|
||
|
|
timeout = None
|
||
|
|
data = bytearray()
|
||
|
|
while size > 0 and self.is_open:
|
||
|
|
try:
|
||
|
|
b = self.queue.get(timeout=self._timeout) # XXX inter char timeout
|
||
|
|
except queue.Empty:
|
||
|
|
if self._timeout == 0:
|
||
|
|
break
|
||
|
|
else:
|
||
|
|
if b is not None:
|
||
|
|
data += b
|
||
|
|
size -= 1
|
||
|
|
else:
|
||
|
|
break
|
||
|
|
# check for timeout now, after data has been read.
|
||
|
|
# useful for timeout = 0 (non blocking) read
|
||
|
|
if timeout and time.time() > timeout:
|
||
|
|
if self.logger:
|
||
|
|
self.logger.info('read timeout')
|
||
|
|
break
|
||
|
|
return bytes(data)
|
||
|
|
|
||
|
|
def cancel_read(self):
|
||
|
|
self.queue.put_nowait(None)
|
||
|
|
|
||
|
|
def cancel_write(self):
|
||
|
|
self._cancel_write = True
|
||
|
|
|
||
|
|
def write(self, data):
|
||
|
|
"""\
|
||
|
|
Output the given byte string over the serial port. Can block if the
|
||
|
|
connection is blocked. May raise SerialException if the connection is
|
||
|
|
closed.
|
||
|
|
"""
|
||
|
|
self._cancel_write = False
|
||
|
|
if not self.is_open:
|
||
|
|
raise PortNotOpenError()
|
||
|
|
data = to_bytes(data)
|
||
|
|
# calculate aprox time that would be used to send the data
|
||
|
|
time_used_to_send = 10.0 * len(data) / self._baudrate
|
||
|
|
# when a write timeout is configured check if we would be successful
|
||
|
|
# (not sending anything, not even the part that would have time)
|
||
|
|
if self._write_timeout is not None and time_used_to_send > self._write_timeout:
|
||
|
|
# must wait so that unit test succeeds
|
||
|
|
time_left = self._write_timeout
|
||
|
|
while time_left > 0 and not self._cancel_write:
|
||
|
|
time.sleep(min(time_left, 0.5))
|
||
|
|
time_left -= 0.5
|
||
|
|
if self._cancel_write:
|
||
|
|
return 0 # XXX
|
||
|
|
raise SerialTimeoutException('Write timeout')
|
||
|
|
for byte in iterbytes(data):
|
||
|
|
self.queue.put(byte, timeout=self._write_timeout)
|
||
|
|
return len(data)
|
||
|
|
|
||
|
|
def reset_input_buffer(self):
|
||
|
|
"""Clear input buffer, discarding all that is in the buffer."""
|
||
|
|
if not self.is_open:
|
||
|
|
raise PortNotOpenError()
|
||
|
|
if self.logger:
|
||
|
|
self.logger.info('reset_input_buffer()')
|
||
|
|
try:
|
||
|
|
while self.queue.qsize():
|
||
|
|
self.queue.get_nowait()
|
||
|
|
except queue.Empty:
|
||
|
|
pass
|
||
|
|
|
||
|
|
def reset_output_buffer(self):
|
||
|
|
"""\
|
||
|
|
Clear output buffer, aborting the current output and
|
||
|
|
discarding all that is in the buffer.
|
||
|
|
"""
|
||
|
|
if not self.is_open:
|
||
|
|
raise PortNotOpenError()
|
||
|
|
if self.logger:
|
||
|
|
self.logger.info('reset_output_buffer()')
|
||
|
|
try:
|
||
|
|
while self.queue.qsize():
|
||
|
|
self.queue.get_nowait()
|
||
|
|
except queue.Empty:
|
||
|
|
pass
|
||
|
|
|
||
|
|
@property
|
||
|
|
def out_waiting(self):
|
||
|
|
"""Return how many bytes the in the outgoing buffer"""
|
||
|
|
if not self.is_open:
|
||
|
|
raise PortNotOpenError()
|
||
|
|
if self.logger:
|
||
|
|
# attention the logged value can differ from return value in
|
||
|
|
# threaded environments...
|
||
|
|
self.logger.debug('out_waiting -> {:d}'.format(self.queue.qsize()))
|
||
|
|
return self.queue.qsize()
|
||
|
|
|
||
|
|
def _update_break_state(self):
|
||
|
|
"""\
|
||
|
|
Set break: Controls TXD. When active, to transmitting is
|
||
|
|
possible.
|
||
|
|
"""
|
||
|
|
if self.logger:
|
||
|
|
self.logger.info('_update_break_state({!r})'.format(self._break_state))
|
||
|
|
|
||
|
|
def _update_rts_state(self):
|
||
|
|
"""Set terminal status line: Request To Send"""
|
||
|
|
if self.logger:
|
||
|
|
self.logger.info('_update_rts_state({!r}) -> state of CTS'.format(self._rts_state))
|
||
|
|
|
||
|
|
def _update_dtr_state(self):
|
||
|
|
"""Set terminal status line: Data Terminal Ready"""
|
||
|
|
if self.logger:
|
||
|
|
self.logger.info('_update_dtr_state({!r}) -> state of DSR'.format(self._dtr_state))
|
||
|
|
|
||
|
|
@property
|
||
|
|
def cts(self):
|
||
|
|
"""Read terminal status line: Clear To Send"""
|
||
|
|
if not self.is_open:
|
||
|
|
raise PortNotOpenError()
|
||
|
|
if self.logger:
|
||
|
|
self.logger.info('CTS -> state of RTS ({!r})'.format(self._rts_state))
|
||
|
|
return self._rts_state
|
||
|
|
|
||
|
|
@property
|
||
|
|
def dsr(self):
|
||
|
|
"""Read terminal status line: Data Set Ready"""
|
||
|
|
if self.logger:
|
||
|
|
self.logger.info('DSR -> state of DTR ({!r})'.format(self._dtr_state))
|
||
|
|
return self._dtr_state
|
||
|
|
|
||
|
|
@property
|
||
|
|
def ri(self):
|
||
|
|
"""Read terminal status line: Ring Indicator"""
|
||
|
|
if not self.is_open:
|
||
|
|
raise PortNotOpenError()
|
||
|
|
if self.logger:
|
||
|
|
self.logger.info('returning dummy for RI')
|
||
|
|
return False
|
||
|
|
|
||
|
|
@property
|
||
|
|
def cd(self):
|
||
|
|
"""Read terminal status line: Carrier Detect"""
|
||
|
|
if not self.is_open:
|
||
|
|
raise PortNotOpenError()
|
||
|
|
if self.logger:
|
||
|
|
self.logger.info('returning dummy for CD')
|
||
|
|
return True
|
||
|
|
|
||
|
|
# - - - platform specific - - -
|
||
|
|
# None so far
|
||
|
|
|
||
|
|
|
||
|
|
# simple client test
|
||
|
|
if __name__ == '__main__':
|
||
|
|
import sys
|
||
|
|
s = Serial('loop://')
|
||
|
|
sys.stdout.write('{}\n'.format(s))
|
||
|
|
|
||
|
|
sys.stdout.write("write...\n")
|
||
|
|
s.write("hello\n")
|
||
|
|
s.flush()
|
||
|
|
sys.stdout.write("read: {!r}\n".format(s.read(5)))
|
||
|
|
|
||
|
|
s.close()
|