私はシリアルを使用して簡単なシリアル コントローラーを作成しました。実際には、それはより多くのフランクシュタイン コードであり、他の人が作成したものと私がつなぎ合わせたものです。私の無能さを気軽にからかってください。バックトラック (Python 2.6) を実行している Linux では問題なく動作しますが、Raspberry Pi (Python 2.7) で実行しようとすると、次のエラーが発生します。
Traceback (most recent call last):Exception in thread Thread-2:
Traceback (most recent call last):
File "/usr/lib/python2.7/threading.py", line 551, in __bootstrap_inner
self.run()
File "/usr/lib/python2.7/threading.py", line 504, in run
self.__target(*self.__args, **self.__kwargs)
File "/home/pi/Zombie Python 1.2 (Modified for Pi) (from Adapt 1.7).py", line 147, in rx
self.ser = serial.Serial('/dev/ttyUSB0', 9600, timeout=1, bytesize=8, stopbits=1)
File "/usr/local/lib/python2.7/dist-packages/serial/serialutil.py", line 260, in __init__
self.open()
File "/usr/local/lib/python2.7/dist-packages/serial/serialposix.py", line 276, in open
raise SerialException("could not open port %s: %s" % (self._port, msg))
SerialException: could not open port /dev/ttyUSB0: [Errno 2] No such file or directory: '/dev/ttyUSB0'
File "/home/pi/Zombie Python 1.2 (Modified for Pi) (from Adapt 1.7).py", line 185, in <module>
client = ThreadedClient(root)
File "/home/pi/Zombie Python 1.2 (Modified for Pi) (from Adapt 1.7).py", line 113, in __init__
self.periodicCall()
File "/home/pi/Zombie Python 1.2 (Modified for Pi) (from Adapt 1.7).py", line 119, in periodicCall
self.gui.processIncoming()
AttributeError: GuiPart instance has no attribute 'processIncoming'
私のコードは次のとおりです。
import Tkinter
import time
import threading
import random
import Queue
import serial
import readline
#import xbee
import sys
x=""
class GuiPart:
def __init__(self, master, queue, endApplication):
self.sonar = Tkinter.StringVar() # Feeds sonar sensor data to label
self.lm = Tkinter.StringVar() # Feeds left motor speed to label
self.rm = Tkinter.StringVar() # Feeds right motor speed to label
self.queue = queue
# Set up the GUI
frame1 = Tkinter.Frame(master, bd=200) #Setup frame.
frame1.bind("<Key>", key) # Allow frame to handle keypresses.
frame1.focus_set() #Set focus of frame so that keypresses activate event.
frame1.pack() #Show it.
#Button
console = Tkinter.Button(frame1, text='Close', command=endApplication)
console.pack()
# Add more GUI stuff here
self.lm.set(0) # Initializes left motor label
self.rm.set(0) # Initializes right motor label
self.sonar.set(0) # Initializes sonar label
#Sonar label
sonarLbl = Tkinter.Label(frame1, textvariable=self.sonar)
sonarLbl.pack()
#Right motor label
rmLbl = Tkinter.Label(frame1, text="Left Motor Speed: ", textvariable=self.rm)
rmLbl.pack()
#Left motor label
lmLbl = Tkinter.Label(frame1, textvariable=self.lm)
lmLbl.pack()
def key(self, event):
#print "pressed", repr(event.char)
#self.sonar = repr(event.char) <------ This should be the line to handle keypresses
global x
x = repr(event.char)
def processIncoming(self):
"""
Handle all the messages currently in the queue (if any).
"""
while self.queue.qsize():
try:
msg = self.queue.get(0)
# Check contents of message and do what it says
# As a test, we simply print it
# Below is where I will parse the "msg" variable, splitting
# it (msg.rsplit) to pull out sensor data and update labels.
lm, rm, sonar, mknt = msg.rsplit(",")
lm = "Left Motor Speed: "+lm
rm = "Right Motor Speed: "+rm
sonar = "Sonar: "+sonar+" CMs away"
self.sonar.set(sonar) # Setting the labels with latest sensor info.
self.lm.set(lm) # Setting the labels with latest sensor info.
self.rm.set(rm) # Setting the labels with latest sensor info.
except Queue.Empty:
pass
class ThreadedClient:
"""
Launch the main part of the GUI and the worker thread. periodicCall and
endApplication could reside in the GUI part, but putting them here
means that you have all the thread controls in a single place.
"""
def __init__(self, master):
"""
Start the GUI and the asynchronous threads. We are in the main
(original) thread of the application, which will later be used by
the GUI. We spawn a new thread for the worker.
"""
self.master = master
# Create the queue
self.queue = Queue.Queue()
# Set up the GUI part
self.gui = GuiPart(master, self.queue, self.endApplication)
# Set up the thread to do asynchronous I/O
# More can be made if necessary
self.running = 1
self.thread1 = threading.Thread(target=self.workerThread1)
self.thread1.start()
#Start receiving thread.
self.rx = threading.Thread(target=self.rx)
self.rx.start()
# Start the periodic call in the GUI to check if the queue contains
# anything
self.periodicCall()
def periodicCall(self):
"""
Check every 100 ms if there is something new in the queue.
"""
self.gui.processIncoming()
if not self.running:
# This is the brutal stop of the system. You may want to do
# some cleanup before actually shutting it down.
import sys
sys.exit(1)
self.master.after(100, self.periodicCall)
def workerThread1(self):
"""
This is where we handle the asynchronous I/O. For example, it may be
a 'select()'.
One important thing to remember is that the thread has to yield
control.
"""
while self.running:
# To simulate asynchronous I/O, we create a random number at
# random intervals. Replace the following 2 lines with the real
# thing.
time.sleep(rand.random() * 0.3)
msg = rand.random()
#self.queue.put(msg)
# Continuously read and print packets
def rx(self):
global x
self.ser = serial.Serial('/dev/ttyUSB0', 9600, timeout=1, bytesize=8, stopbits=1)
self.ser.flush()
while(1):
response = str(self.ser.readline())
# Send movement codes.
if x == "'a'" or x == "'A'": # or "'A'": # Turn left.
self.ser.write('4')
elif x == "'w'" or x == "'W'": #Go forward.
self.ser.write("3")
elif x == "'d'" or x == "'D'": #Turn right.
self.ser.write("2")
elif x == "'s'" or x == "'S'": #Go back.
self.ser.write("1")
elif x == "'x'" or x == "'X'": #Stop.
self.ser.write("5")
elif x == "'1'": #Raise speed.
self.ser.write("7")
elif x == "'2'": #Lower speed.
self.ser.write("6")
x = ""
if len(response) > 10:
self.ser.flushInput() #If you don't flush the buffer, then it'll try to read out all
#the sensor readings, so the reaction time will be extrordinary.
time.sleep(.1)# This scales back the CPU usage.
self.queue.put(response)
def endApplication(self):
print "Closing"
self.running = 0
self.ser.close()
sys.exit(1)
rand = random.Random()
root = Tkinter.Tk()
client = ThreadedClient(root)
root.mainloop()