Windows 7 (64 бита), Python 2.7.18 (32 бита)
У меня реализован следующий класс для обеспечения последовательной связи в рамках более крупной программы:
## serial communications thread
class Communication(threading.Thread):
"""
Serial communications class. Runs in own thread. This class handles
everything related to interfacing with the serial connection.
"""
## thread initialisation
def __init__(self, port, gui):
"""
Thread initialisation. Sets up the thread, data queue, state
variables and connection.
Arguments:
port (str): The serial port to use.
gui (ui): Instance of the ui() class which is controlling the
GUI.
No returns.
"""
global baud
## initialise self as a thread
threading.Thread.__init__(self)
## set up local reference to variables from other threads
self.gui = gui
## set up a FIFO queue to handle received serial data
self.rx = Queue.Queue()
## set up a state variable
self.terminate = False
## actually open the serial connection
self.lib_serial = gui.serial ## this is 'serial' from the pyserial library. we get this
## from the gui because it is handling the case
## where the user forgot to install pyserial
self.connection = self.lib_serial.Serial(port, baud, timeout = 0.1, write_timeout = 1)
## what to do when start() is called on this thread class
def run (self):
"""
Main thread loop (runs until terminated). Gets data from the serial
port, cleans it up and adds it to the receive buffer queue and
communication log.
No arguments.
No returns.
"""
## run forever (in a seperate thread)
while self.terminate == False:
## read a line of data, with error handling
try:
line = self.readline()
## strip the endline characters from it
cleanLine = line.decode("utf-8").rstrip('\r\n')
## only proceed if the data isn't a blank line
if cleanLine != '':
## print out the line
self.gui.add_communication(cleanLine, 'rx')
self.rx.put_nowait(cleanLine)
## if the read didn't succeed
except:
print('Error reading from serial connection')
## reimplementation of the pyserial readline() function to accept CR or
## LF as newline characters. also dumps received data after ten seconds
def readline(self):
"""
Reads data until a newline, carrige return or the passage of ten
seconds of time is seen (whichever is first) and returns it.
No arguments.
Returns:
(bytes) Data received.
"""
## set up a byte array to receive characters
line = bytearray()
## note the start time for timeouts
start_time = time.time()
while True:
## try to get a byte of data
c = self.connection.read(1)
## if there is a byte of data..
if c:
## add it to the line buffer
line += c
## if there is a carrige return (\r) or line-feed (\n)..
if line[1:] == b'\r' or line[1:] == b'\n':
## break the loop
break
## if there is no more data available..
else:
break
## if this read has been looping for more than ten seconds, timeout
## (but don't raise an error)
if (time.time() - start_time) > 10:
break
## return whatever data we have
return bytes(line)
## write to the serial port one character at a time (to avoid MVME overrun
## errors)
def write(self, data, eol = '\r', delay = 0.001):
"""
Sends data to the serial port one character at a time to avoid MVME
buffer overrun issues. Wraps the actual write function.
Arguments:
data (str): The data to send.
eol (str): The end of line character to terminate the data with.
delay (float): How long (in seconds) to wait between sending
successive characters.
No returns.
"""
## add the line end character to the string
data = data + eol
## iterate through the string character by character
for letter in list(data):
## send the character
self.write_raw(letter)
## wait
time.sleep(delay)
## log the string to the GUI
self.gui.add_communication(str(data), 'tx')
## write to the serial port. convert to bytes if not already
def write_raw(self, data, is_bytes = False):
"""
Converts data to bytes and writes it to the serial port.
Arguments:
data (str or bytes): Data to write.
is_bytes (bool): True if the data is bytes, False otherwise.
No returns.
"""
## convert to bytes
if is_bytes == False and sys.version_info >= (3, 0):
data = bytes(data, encoding = 'utf-8', errors = 'ignore')
try:
## write to the serial port
self.connection.write(data)
## flush serial buffers
self.connection.flush()
except:
## complain about errors
print('Serial write error')
self.gui.add_communication('Serial write error!', 'error')
## read data from the receive queue
def read(self):
"""
Reads data from the receive queue.
No arguments.
Returns:
(str) New data from the receive queue.
"""
## get data from the receive queue without blocking. if there is none,
## just return an empty string
try:
new_data = self.rx.get_nowait()
except Queue.Empty:
new_data = ''
return new_data
## shut down the serial connection
def kill(self):
"""
Terminates the serial connection and serial thread (this thread).
No arguments.
No returns.
"""
## terminate the thread main loop
self.terminate = True
## close the serial connection
self.connection.close()
К сожалению, для цели P C этот код вызовет «ошибку времени выполнения Visual C ++», сообщающую, что программа запросила завершение необычным способом. Это происходит, когда первая строка данных принимается от последовательного устройства или иногда, когда первая строка данных отправляется на последовательное устройство.
Используемый последовательный порт аппаратный (физический последовательный порт на машина). Линии управления потоком не используются.
Это оставляет мне несколько вопросов:
- Почему возникает эта ошибка или при каких условиях эта ошибка может возникать для других программ? Можете ли вы привести примеры или ссылки?
- Это связано с вызовом self. gui .add_communication () для обновления GUI, запущенного в главном потоке? Если да, то какой механизм я могу использовать для передачи этих обновлений в GUI, учитывая, что основной поток заблокирован mainl oop (), работающим на Tk root?
- Может ли запись в pyserial () вызов принимает строку в качестве данных в этой версии Python, вместо необходимости преобразования в байты, как в Python 3?
- Почему Python не возникает исключение до появления ошибки времени выполнения?