#!/usr/bin/env python3 # coding: Latin-1 # Creates a web-page interface for DiddyBorg # Import library functions we need import PicoBorgRev3 as PicoBorgRev import time import sys import threading import socketserver #import picamera #import picamera.array #import cv2 import datetime import os import csv # Settings for the web-page webPort = 8080 # Port number for the web-page, 80 is what web-pages normally use #imageWidth = 240 # Width of the captured image in pixels #imageHeight = 180 # Height of the captured image in pixels #frameRate = 10 # Number of images to capture per second displayRate = 2 # Number of images to request per second #photoDirectory = '/home/pi' # Directory to save photos to # Global values global PBR #global lastFrame #global lockFrame #global camera #global processor global running global watchdog running = True # Setup the PicoBorg Reverse PBR1 = PicoBorgRev.PicoBorgRev() PBR1.i2cAddress = 0x12 # Uncomment and change the value if you have changed the board address PBR2 = PicoBorgRev.PicoBorgRev() PBR2.i2cAddress = 0x11 # Uncomment and change the value if you have changed the board address PBR1.Init() PBR2.Init() if not PBR1.foundChip: boards = PicoBorgRev.ScanForPicoBorgReverse() if len(boards) == 0: print ('No PicoBorg Reverse found, check you are attached :)') else: print ('No PicoBorg Reverse at address %02X, but we did find boards:' % (PBR.i2cAddress)) for board in boards: print (' %02X (%d)' % (board, board)) print ('If you need to change the I²C address change the setup line so it is correct, e.g.') print ('PBR.i2cAddress = 0x%02X' % (boards[0])) sys.exit() #PBR.SetEpoIgnore(True) # Uncomment to disable EPO latch, needed if you do not have a switch / jumper PBR1.SetCommsFailsafe(False) # Disable the communications failsafe PBR2.SetCommsFailsafe(False) # Disable the communications failsafe PBR1.ResetEpo() PBR2.ResetEpo() # Power settings voltageIn = 1.8 * 10 # Total battery voltage to the PicoBorg Reverse voltageOut = 12.0 # Maximum motor voltage # Setup the power limits if voltageOut > voltageIn: maxPower = 1.0 else: maxPower = voltageOut / float(voltageIn) #wifi strength def WiFiSignalStrength(): # Get the WiFi signal strength using iwconfig wifiQuery = open("data.csv", "r") wifiResult = wifiQuery.readlines() wifiQuery.close() wifiStrength = wifiResult # See if we are connected if len(wifiResult) == 0: #Not connected wifiStrength = 'Not connected' else : # Connected, the value we get here will be a division, e.g. 5/20 wifiResult = wifiResult[0] # Get the text wifiResult = wifiResult[:-1] # Remove the '\n' symbol from the end wifiValues = wifiResult.split('/') if len(wifiValues) == 2: # Get the percentage from the values wifiPercent = (float(wifiValues[0]) / float(wifiValues[1])) * 100.0 wifiStrength = '%.0f %%'(wifiPercent) else: # Unexpected, show the returned text instead wifiStrength = wifiResult # Return the value return wifiStrength # Timeout thread class Watchdog(threading.Thread): def __init__(self): super(Watchdog, self).__init__() self.event = threading.Event() self.terminated = False self.start() self.timestamp = time.time() def run(self): timedOut = True # This method runs in a separate thread while not self.terminated: # Wait for a network event to be flagged for up to one second if timedOut: if self.event.wait(10): # Connection print ('Reconnected...') timedOut = False self.event.clear() else: if self.event.wait(10): self.event.clear() else: # Timed out print ('Timed out...') timedOut = True PBR1.MotorsOff() PBR2.MotorsOff() # Class used to implement the web server class WebServer(socketserver.BaseRequestHandler): def handle(self): global PBR #global lastFrame global watchdog # Get the HTTP request data reqData = self.request.recv(1024).strip() reqData = reqData.decode('utf-8') reqData = reqData.split('\n') # Get the URL requested getPath = '' for line in reqData: if line.startswith('GET'): parts = line.split(' ') getPath = parts[1] break watchdog.event.set() if getPath.startswith('/shutdown'): # Shut the Raspberry Pi down httpText = '
' httpText += 'Shutting down...' httpText += '
' self.send(httpText) os.system('sudo shutdown -h now') elif getPath.startswith('/wifi-once'): # Display the WiFi signal strength httpText = '
' httpText += WiFiSignalStrength() httpText += '
' self.send(httpText) elif getPath.startswith('/off'): # Turn the drives off httpText = '
' httpText += 'Speeds: 0 %, 0 %' httpText += '
' self.send(httpText) PBR1.MotorsOff() PBR2.MotorsOff() elif getPath.startswith('/set/'): # Motor power setting: /set/driveLeft/driveRight parts = getPath.split('/') # Get the power levels if len(parts) >= 4: try: driveLeft = float(parts[2]) driveRight = float(parts[3]) except: # Bad values driveRight = 0.0 driveLeft = 0.0 else: # Bad request driveRight = 0.0 driveLeft = 0.0 # Ensure settings are within limits if driveRight < -1: driveRight = -1 elif driveRight > 1: driveRight = 1 if driveLeft < -1: driveLeft = -1 elif driveLeft > 1: driveLeft = 1 # Report the current settings percentLeft = driveLeft * 100.0; percentRight = driveRight * 100.0; httpText = '
' httpText += 'Speeds: %.0f %%, %.0f %%' % (percentLeft, percentRight) httpText += '
' self.send(httpText) # Set the outputs driveLeft *= maxPower driveRight *= maxPower PBR1.SetMotor1(driveRight) PBR1.SetMotor2(-driveLeft) elif getPath.startswith('/set2/'): # Motor power setting: /set/driveLeft/driveRight parts = getPath.split('/') # Get the power levels if len(parts) >= 4: try: driveLeft = float(parts[2]) driveRight = float(parts[3]) except: # Bad values driveRight = 0.0 driveLeft = 0.0 else: # Bad request driveRight = 0.0 driveLeft = 0.0 # Ensure settings are within limits if driveRight < -1: driveRight = -1 elif driveRight > 1: driveRight = 1 if driveLeft < -1: driveLeft = -1 elif driveLeft > 1: driveLeft = 1 # Report the current settings percentLeft = driveLeft * 100.0; percentRight = driveRight * 100.0; httpText = '
' httpText += 'Speeds: %.0f %%, %.0f %%' % (percentLeft, percentRight) httpText += '
' self.send(httpText) # Set the outputs driveLeft *= maxPower driveRight *= maxPower PBR2.SetMotor1(driveRight) PBR2.SetMotor2(-driveLeft) elif getPath == ('/wifi'): # Repeated reading of the WiFi strength, set a delayed refresh # We use AJAX to avoid screen refreshes caused by refreshing a frame displayDelay = int(1000 / displayRate) httpText = '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' % (displayDelay) httpText += '
Waiting for first distance reading...
\n' httpText += '\n' httpText += '\n' self.send(httpText) elif getPath == '/': # Main page, click buttons to move and to stop httpText = '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '
\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '

\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' #httpText += '\n' #httpText += '\n' httpText += '

\n' httpText += '\n' httpText += '

\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '
\n' httpText += '\n' httpText += '\n' self.send(httpText) elif getPath == '/joystick': # Alternate page, hold buttons to move (does not work with all devices) httpText = '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' httpText += '\n' #httpText += '\n' #httpText += '\n' #httpText += '\n' #httpText += '

\n' #httpText += '\n' #httpText += '\n' #httpText += '\n' #httpText += '\n' #httpText += '\n' #httpText += '\n' httpText += '

\n' httpText += '\n' httpText += '\n' #httpText += '

\n' httpText += '\n' #httpText += '\n' #httpText += '

\n' httpText += '\n' httpText += '\n' #httpText += '

\n' #httpText += '\n' httpText += '
\n' httpText += '\n' httpText += '\n' self.send(httpText) else: # Unexpected page self.send('Path : "%s"' % (getPath)) def send(self, content): reply = 'HTTP/1.0 200 OK\n\n%s' % (content) reply = reply.encode('utf-8') self.request.sendall(reply) print ('Setup the watchdog') watchdog = Watchdog() # Run the web server until we are told to close httpServer = socketserver.TCPServer(("0.0.0.0", webPort), WebServer) try: print ('Press CTRL+C to terminate the web-server') while running: httpServer.handle_request() except KeyboardInterrupt: # CTRL+C exit print ('\nUser shutdown') finally: # Turn the motors off under all scenarios PBR1.MotorsOff() PBR2.MotorsOff() print ('Motors off') # Tell each thread to stop, and wait for them to end running = False #captureThread.join() #processor.terminated = True watchdog.terminated = True #processor.join() watchdog.join() #del camera PBR1.SetLed(True) PBR2.SetLed(True) print ('Web-server terminated.')