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()
不幸的是，在目标PC上，此代码将导致“ Visual C ++运行时错误”，表明程序已以异常方式请求终止。当从串行设备接收到第一行数据时，或者偶尔将第一行数据发送到串行设备时，会发生这种情况。
正在使用的串行端口位于硬件中（机器上的物理串行端口）。不使用流量控制线。
这给我一些问题：
Windows 7（64位），Python 2.7.18（32位），我实现了以下类，以作为更大程序的一部分提供串行通信：##串行通信线程类Communication（...