cpp代码：import threading
import SocketServer
import RPi.GPIO as GPIO
from SocketServer import StreamRequestHandler as SRH
from AlphaBot import AlphaBot
from PCA9685 import PCA9685
from time import ctime
Ab = AlphaBot()
pwm = PCA9685(0x40)
pwm.setPWMFreq(50)
BUZ = 4
GPIO.setmode(GPIO.BCM)
GPIO.setwarnings(False)
GPIO.setup(BUZ,GPIO.OUT)
#Set the Horizontal servo parameters
HPulse = 1500 #Sets the initial Pulse
HStep = 0 #Sets the initial step length
pwm.setServoPulse(0,HPulse)
#Set the vertical servo parameters
VPulse = 1500 #Sets the initial Pulse
VStep = 0 #Sets the initial step length
pwm.setServoPulse(1,VPulse)
host = '192.168.10.235'
port = 8000
addr = (host,port)
def beep_on():
GPIO.output(BUZ,GPIO.HIGH)
def beep_off():
GPIO.output(BUZ,GPIO.LOW)
class Servers(SRH):
def handle(self):
global HStep,VStep
print 'got connection from ',self.client_address
self.wfile.write('connection %s:%s at %s succeed!' % (host,port,ctime()))
while True:
data = self.request.recv(1024)
if not data:
break
if data == "Stop":
HStep = 0
VStep = 0
Ab.stop()
elif data == "Forward":
Ab.forward()
elif data == "Backward":
Ab.backward()
elif data == "TurnLeft":
Ab.left()
elif data == "TurnRight":
Ab.right()
elif data == "Up":
VStep = -5
elif data == "Down":
VStep = 5
elif data == "Left":
HStep = 5
elif data == "Right":
HStep = -5
elif data == "BuzzerOn":
beep_on()
elif data == "BuzzerOff":
beep_off()
else:
value = 0
try:
value = int(data)
if(value >= 0 and value <= 100):
print(value)
Ab.setPWMA(value);
Ab.setPWMB(value);
except:
print("Command error")
print data
#print "recv from ", self.client_address[0]
self.request.send(data)
def timerfunc():
global HPulse,VPulse,HStep,VStep,pwm
if(HStep != 0):
HPulse += HStep
if(HPulse >= 2500):
HPulse = 2500
if(HPulse <= 500):
HPulse = 500
#set channel 2, the Horizontal servo
pwm.setServoPulse(0,HPulse)
if(VStep != 0):
VPulse += VStep
if(VPulse >= 2500):
VPulse = 2500
if(VPulse <= 500):
VPulse = 500
#set channel 3, the vertical servo
pwm.setServoPulse(1,VPulse)
global t #Notice: use global variable!
t = threading.Timer(0.02, timerfunc)
t.start()
t = threading.Timer(0.02, timerfunc)
t.setDaemon(True)
t.start()
print 'server is running....'
server = SocketServer.ThreadingTCPServer(addr,Servers)
server.serve_forever()
如果觉得《树莓派python编程小车_树莓派智能车AlphaBot教程11：Python 网络编程》对你有帮助，请点赞、收藏，并留下你的观点哦！
