pyserial和wxpython matplotlib线程中读取方法失败

2024-03-29 02:38:00 发布

您现在位置:Python中文网/ 问答频道 /正文

我正在尝试构建一个gui,从串行端口接收数据并显示在绘图中(使用matplotlib)。但是当我打开端口时,read()失败了。我就是不明白为什么。谁能给我一些建议吗?那会很感激的! 以下是我的部分代码: `在

class PlotFigure(wx.Frame):
    """Matplotlib wxFrame with animation effect"""
    def __init__(self):
        wx.Frame.__init__(self, None, wx.ID_ANY, title="Figure for figures", size=(1200, 1200))
        # Matplotlib Figure
        self.fig = Figure((6, 4), 100)
        # bind the Figure to the backend specific canvas
        self.canvas = FigureCanvas(self, wx.ID_ANY, self.fig)
        # add a subplot
        self.ax = self.fig.add_subplot(111)
        # limit the X and Y axes dimensions
        self.ax.set_ylim([-180, 180])
        self.ax.set_xlim([0, POINTS])

        self.ax.set_autoscale_on(False)
        self.ax.set_xticks([])
        # we want a tick every 10 point on Y (101 is to have 10
        self.ax.set_yticks(range(-180, 180, 50))
        # disable autoscale, since we don't want the Axes to ad
        # draw a grid (it will be only for Y)
        self.ax.grid(True)
        # generates first "empty" plots
        self.user1=self.user2=self.user3 = [None] * POINTS
        self.l_user1,=self.ax.plot(range(POINTS),self.user1,label='data1')
    self.l_user2,=self.ax.plot(range(POINTS),self.user2,label='data2')
    self.l_user3,=self.ax.plot(range(POINTS),self.user3,label='data3')


        # add the legend
        self.ax.legend(loc='upper center',
                           ncol=4,
                           prop=font_manager.FontProperties(size=10))
        # force a draw on the canvas()
        # trick to show the grid and the legend
        self.canvas.draw()
        # save the clean background - everything but the line
        # is drawn and saved in the pixel buffer background
        self.bg = self.canvas.copy_from_bbox(self.ax.bbox)
        # bind events coming from timer with id = TIMER_ID
        # to the onTimer callback function
        wx.EVT_TIMER(self, TIMER_ID, self.onTimer)

    #for serial
    self.ser=serial.Serial('COM1', 115200)
    #print self.ser

    def onTimer(self, evt):
    print 'onTimer....'
        """callback function for timer events"""
        # restore the clean background, saved at the beginning
        self.canvas.restore_region(self.bg)
        # update the data
    time.sleep(1)
        self.ser.flushInput()
        print 'before the read method....'
    data=self.ser.read(12)
        print 'after the read method... just cant reach here...'
    t = struct.unpack('3f', data)
        temp1 = t[0]
    temp2 = t[1]
    temp3 = t[2]
        self.user1 = self.user1[1:] + [temp1]
    print temp2
    self.user2 = self.user2[1:] + [temp2]
    self.user3 = self.user3[1:] + [temp3]
        # update the plots
        self.l_user1.set_ydata(self.user1)
    self.l_user2.set_ydata(self.user2)
    self.l_user3.set_ydata(self.user3)
        # just draw the "animated" objects
        self.ax.draw_artist(self.l_user1)
    self.ax.draw_artist(self.l_user2)
    self.ax.draw_artist(self.l_user3)# It is used to efficiently update Axes data (axis ticks, labels, etc are not updated)
        self.canvas.blit(self.ax.bbox)
    print 'onTimer ends'
    def __del__(self):
    self.ser.close()
    t.Stop()

if __name__ == '__main__':
    app = wx.PySimpleApp()
    frame = PlotFigure()
    t = wx.Timer(frame, TIMER_ID)
    t.Start(1)
    print 'new test'
    frame.Show()
    print 'after frame show '
    app.MainLoop()

I wonder if it is the thread problem.so I do an another test: Here is part of my code:

^{pr2}$

在` 读取方法失败!

然后是另一个简单的测试,我去掉了线程,它工作正常! 以下是我的部分代码: `在

ser = serial.Serial('COM1', 115200)
num=0
try:
    while True:

        data = ser.read(12)
        print num
        time.sleep(1)    
        ser.flushInput()


finally:
    ser.close()

`


Tags: thetoselfidreadaxsercanvas
1条回答
网友
1楼 · 发布于 2024-03-29 02:38:00
#
# The following is an example of an application I have running using the 
# serial port.  Although the baud is lower, it should not matter.  This application
# will capture each byte and append them into a buffer before processing.  All bytes 
# are captured including non-printable characters.  The terminator in my application
# is the RETURN character (0x0A) or the (0x0D) denoting end of line.
# I think your port is hung waiting on the full twelve (12) bytes that you are 
# expecting on the read.  The hang condition is because only a portion of 
# the 12 bytes have occurred.  I suggest you add a timeout value of 1 second 
# to the serial read, then test for the length of what was read to validate 
# what you think you just read.  If zero, you have a serial cable or problem 
# at the other end of the cable.  If greater than zero, then the device is 
# not sending what you think it should.

import serial
import string
import binascii
import threading
import thread
import fileinput
import datetime
import time
import sys

from configobj import ConfigObj


# Insert following code at top in the import area of your application
# 
hdrFile  = 'AppName.ini'
config   = ConfigObj(hdrFile)
hdrVal   = 'COM'
comport  = int(config[hdrVal])
hdrBaud  = 'Baudrate'
baudrate = config[hdrBaud]

print "Com Config = ",config
print "Com Port   = ",comport

boolSerOnline = False
bool_IsALIVE  = False      # thread has active serial port

class myThread( threading.Thread ):
    def __init__(self, intThreadID, strName, intCountVal ):
        #                    - Constructor
        threading.Thread.__init__(self)    # <<  MUST be first in Init
        self.threadID = intThreadID
        self.threadName = strName
        self.intStartCounter = intCountVal
    def run(self):
        #                    - Tooltalk
        # Thread to Listen to the RS-232 RX Port
        #                - TOOLTALK ()
        #                - THREAD
        global bool_IsALIVE, boolThreadRun
        #
        #
        inbuf  = []          # Clear Buffer
        buffer = ""          # Clear Buffer
        strFirstByte = ""
        boolWriteCRLF = 0
        #
        print "\n*** Starting %s: %s" % ( self.threadName, time.ctime(time.time()))
        print "\n"
        #
        bool_IsALIVE = False
        #
        #                  - Start of Thread Loop
        while ( boolThreadRun ):
            #               - THREAD is Running

            try:
                #        Read 1 byte at a time in thread
                # Note: reads all characters, Even non-printable characters
                strByte = str(ser.read( 1 ))
                #
                intLength = len( strByte )
                #
                if ( intLength > 0 ):
                    bool_IsALIVE = True          # Indicate connected
                    # Process byte
                    #
                    if (strFirstByte == ""):
                        boolWriteCRLF  = 0
                        #
                        if ( strByte == '\r' ):
                            strByte = ''          # Skip until we have first byte
                        elif ( strByte == '\n' ):
                            strByte = ''          # Skip until we have first byte
                        elif ( strByte == '0' ):
                            inbuf = []                     # Clear buffer first
                            inbuf.append( strByte )        # Add to inbuf
                            buffer = ''.join(inbuf)        # join the two buffers
                            strFirstByte = strByte
                            boolWriteCRLF  = 1
                        elif ( strByte == 'S' ):
                            inbuf = []                     # Clear buffer first
                            inbuf.append( strByte )        # Add to inbuf
                            buffer = ''.join(inbuf)        # join the two buffers
                            strFirstByte = strByte
                            boolWriteCRLF  = 1
                        elif ( strByte == 'C' ):
                            inbuf = []                     # Clear buffer first
                            inbuf.append( strByte )        # Add to inbuf
                            buffer = ''.join(inbuf)        # join the two buffers
                            strFirstByte = strByte
                            boolWriteCRLF  = 1
                        elif ( strByte == '*' ):
                            inbuf = []                     # Clear buffer first
                            inbuf.append( strByte )        # Add to inbuf
                            buffer = ''.join(inbuf)        # join the two buffers
                            strFirstByte = strByte     # save first byte of string
                            boolWriteCRLF  = 1
                        elif ( strByte == '>' ):
                            inbuf = []                     # Clear buffer first
                            inbuf.append( strByte )        # Add to inbuf
                            buffer = ''.join(inbuf)        # join the two buffers
                            strFirstByte = strByte     # save first byte of string
                            boolWriteCRLF  = 1
                        elif ( strByte == 'M' ):
                            inbuf = []                     # Clear buffer first
                            inbuf.append( strByte )        # Add to inbuf
                            buffer = ''.join(inbuf)        # join the two buffers
                            strFirstByte = strByte     # save first byte of string
                            boolWriteCRLF  = 1
                        else:
                            pass
                    else:
                        inbuf.append( strByte )        # Add to inbuf
                        buffer = ''.join(inbuf)        # join the two buffers
                else:
                    # serial driver timed out producing No byte 
                    # print "SKIP - NO BYTE (NO POWER?)"
                    pass
            except:
                bool_IsALIVE = False
                print "[Tooltalk] Error reading Serial Port"



if __name__ == "__main__":
    try:
        # *** +++++++++++++ Notice the 1 second timeout ...
        ser = serial.Serial(comport,baudrate,8,'N',1,timeout=1)
        boolSerOnline = True
    except:
        print "[Main] Serial Port NOT Connected\n"
        boolSerOnline = False

    boolThreadRun = True        # Flag indicating "Tooltalk" to Run
    #
    thread1 = myThread(1, "Tooltalk-1", 1)   # Init Thread
    thread1.start()        # Start Tooltalk Thread

相关问题 更多 >