Hardware code update
This commit is contained in:
LEYAO LI
parent
850e91f209
commit
14dd6f7e98
6 changed files with 218 additions and 59 deletions
4
Main.py
4
Main.py
|
|
@ -1,5 +1,5 @@
|
|||
import DBHelper
|
||||
import Start_Engine
|
||||
import start_engine
|
||||
import Upload_Face
|
||||
import Remove_Face
|
||||
import Check_Up
|
||||
|
|
@ -11,7 +11,7 @@ if __name__ == "__main__":
|
|||
print("Starting the program.")
|
||||
while True:
|
||||
if DBHelper.get_power() == "on":
|
||||
Start_Engine.start()
|
||||
start_engine.start()
|
||||
|
||||
if None not in (DBHelper.get_register_firstname(),
|
||||
DBHelper.get_register_lastname(),
|
||||
|
|
|
|||
Binary file not shown.
120
final_demo.py
Normal file
120
final_demo.py
Normal file
|
|
@ -0,0 +1,120 @@
|
|||
import os, signal
|
||||
import RPi.GPIO as GPIO
|
||||
import pyrebase # u need to install Pyrebase module firstly
|
||||
from time import sleep
|
||||
|
||||
class Motor:
|
||||
print("Starting of the program")
|
||||
def __init__(self):
|
||||
config = {
|
||||
"apiKey": "AIzaSyAdL0W5HscjEDFPK4BDi6Cnc7FLa30GPYY",
|
||||
"authDomain": "vehicleantitheftrecognition.firebaseapp.com",
|
||||
"databaseURL": "https://vehicleantitheftrecognition.firebaseio.com/",
|
||||
"storageBucket": "vehicleantitheftrecognition.firebaseapp.com"
|
||||
}
|
||||
self.firebase = pyrebase.initialize_app(config)
|
||||
GPIO.setmode(GPIO.BCM)
|
||||
GPIO.setwarnings(False)
|
||||
#preset GPIO ports for 2 motors
|
||||
self.Motor1 = {'EN': 25, 'input1': 24, 'input2': 23}
|
||||
self.Motor2 = {'EN': 17, 'input1': 27, 'input2': 22}
|
||||
# preset the port for buttons and alarm
|
||||
GPIO.setup(5,GPIO.IN) # start motor button, initially True
|
||||
GPIO.setup(13,GPIO.IN) # stop motor button, initially True
|
||||
GPIO.setup(16,GPIO.IN) # start alarm button, initially True
|
||||
GPIO.setup(26,GPIO.OUT) # alarm output
|
||||
|
||||
for x in self.Motor1:
|
||||
GPIO.setup(self.Motor1[x], GPIO.OUT)
|
||||
GPIO.setup(self.Motor2[x], GPIO.OUT)
|
||||
#utilize PWM function, enable motors and frequency is 100Hz
|
||||
self.EN1 = GPIO.PWM(self.Motor1['EN'], 100)
|
||||
self.EN2 = GPIO.PWM(self.Motor2['EN'], 100)
|
||||
|
||||
self.EN1.start(0)
|
||||
self.EN2.start(0)
|
||||
|
||||
#stop signals for motors and alarm
|
||||
self.motorStop=False
|
||||
self.alarmStop=False
|
||||
|
||||
|
||||
|
||||
# new update motor and alarm functions, are able to connect embedded system throught firebase
|
||||
|
||||
def start_motor(self):
|
||||
|
||||
self.motorStop=self.stop_motor()
|
||||
|
||||
while (not self.motorStop): #break the loop when motor stop signal is detected
|
||||
|
||||
self.motorStop=self.stop_motor()
|
||||
|
||||
|
||||
def stop_motor(self):
|
||||
|
||||
database = self.firebase.database() # get alarm on/off signal from firebase
|
||||
signals = database.child("signal")
|
||||
motorSignal = signals.child(1).child("motor").get().val()
|
||||
|
||||
|
||||
if (motorSignal=="off") or (not GPIO.input(13)):
|
||||
print("stopping motor...")
|
||||
self.EN1.ChangeDutyCycle(0)
|
||||
self.EN2.ChangeDutyCycle(0)
|
||||
print("motor stops")
|
||||
return True
|
||||
elif (motorSignal=="on") or (not GPIO.input(5)):
|
||||
|
||||
self.EN1.ChangeDutyCycle(50)
|
||||
self.EN2.ChangeDutyCycle(50)
|
||||
|
||||
GPIO.output(self.Motor1['input1'], GPIO.HIGH)
|
||||
GPIO.output(self.Motor1['input2'], GPIO.LOW)
|
||||
|
||||
GPIO.output(self.Motor2['input1'], GPIO.HIGH)
|
||||
GPIO.output(self.Motor2['input2'], GPIO.LOW)
|
||||
|
||||
print("motor is turned on")
|
||||
return False
|
||||
|
||||
|
||||
def start_alarm(self):
|
||||
|
||||
self.alarmStop=self.stop_alarm()
|
||||
|
||||
while (not self.alarmStop): # if alarmStop is False or button is pressed
|
||||
# # enter the loop
|
||||
self.alarmStop=self.stop_alarm() # infinitely check if alarmStop True
|
||||
|
||||
# break the loop if alarm is turned off
|
||||
def stop_alarm(self):
|
||||
|
||||
database = self.firebase.database() # get alarm on/off signal from firebase
|
||||
signals = database.child("signal")
|
||||
alarmSignal = signals.child(1).child("alarm").get().val()
|
||||
|
||||
if (alarmSignal=="on" or (not GPIO.input(16))):
|
||||
print("Alarm is turned on")
|
||||
GPIO.output(26,True)
|
||||
return False
|
||||
elif alarmSignal=="off":
|
||||
print("Alarm turning off...")
|
||||
self.alarmStop=True
|
||||
GPIO.output(26,False)
|
||||
print("Alarm is off")
|
||||
return True
|
||||
|
||||
|
||||
|
||||
if __name__=="__main__":
|
||||
|
||||
#print("Execute function...")
|
||||
|
||||
|
||||
motor1=Motor()
|
||||
|
||||
while True: # turn on the system forever
|
||||
|
||||
motor1.start_alarm() # alarm on/off test
|
||||
motor1.start_motor() # motor on/off test
|
||||
2
hfs.comments.txt
Normal file
2
hfs.comments.txt
Normal file
|
|
@ -0,0 +1,2 @@
|
|||
final_demo.py=uploaded by lli
|
||||
remote_camera.py=uploaded by lli
|
||||
|
|
@ -1,57 +0,0 @@
|
|||
# import the RPi library
|
||||
import RPi.GPIO as GPIO
|
||||
import time
|
||||
|
||||
# initialize the ports
|
||||
GPIO.setwarnings(False)
|
||||
GPIO.setmode(GPIO.BCM)
|
||||
|
||||
# preset the port for button and alarm
|
||||
GPIO.setup(5,GPIO.IN) # button 1
|
||||
GPIO.setup(6,GPIO.IN) # button 2
|
||||
GPIO.setup(13,GPIO.IN) # button 3
|
||||
GPIO.setup(16,GPIO.IN) # button 4
|
||||
GPIO.setup(26,GPIO.OUT) # alarm
|
||||
|
||||
# preset the port for the L293D
|
||||
Motor1 = {'EN': 25, 'input1': 24, 'input2': 23}
|
||||
Motor2 = {'EN': 17, 'input1': 27, 'input2': 22}
|
||||
for x in Motor1:
|
||||
GPIO.setup(Motor1[x], GPIO.OUT)
|
||||
GPIO.setup(Motor2[x], GPIO.OUT)
|
||||
EN1 = GPIO.PWM(Motor1['EN'],100)
|
||||
EN2 = GPIO.PWM(Motor2['EN'],100)
|
||||
EN1.start(0)
|
||||
EN2.start(0)
|
||||
|
||||
while True:
|
||||
if not GPIO.input(5):
|
||||
# If press button 1, two DC motors running full speed forward
|
||||
EN1.ChangeDutyCycle(100)
|
||||
EN2.ChangeDutyCycle(100)
|
||||
print ("FORWARD MOTION")
|
||||
GPIO.output(Motor1['input1'],GPIO.HIGH)
|
||||
GPIO.output(Motor1['input2'],GPIO.LOW)
|
||||
GPIO.output(Motor2['input2'],GPIO.HIGH)
|
||||
GPIO.output(Motor2['input1'],GPIO.LOW)
|
||||
elif not GPIO.input(6):
|
||||
# If press button 2, two DC motors running full speed backword
|
||||
EN1.ChangeDutyCycle(100)
|
||||
EN2.ChangeDutyCycle(100)
|
||||
print ("BACKWORD MOTION")
|
||||
GPIO.output(Motor1['input1'],GPIO.LOW)
|
||||
GPIO.output(Motor1['input2'],GPIO.HIGH)
|
||||
GPIO.output(Motor2['input2'],GPIO.LOW)
|
||||
GPIO.output(Motor2['input1'],GPIO.HIGH)
|
||||
elif not GPIO.input(13):
|
||||
# If press button 3, two DC motors stop running
|
||||
print ("STOP")
|
||||
EN1.ChangeDutyCycle(0)
|
||||
EN2.ChangeDutyCycle(0)
|
||||
elif not GPIO.input(16):
|
||||
# If press button 4, alarm starts beeping
|
||||
print ("ALARM")
|
||||
GPIO.output(26,True)
|
||||
else:
|
||||
# If no button been pressed, alarm stop beeping
|
||||
GPIO.output(26,False)
|
||||
94
remote_camera.py
Normal file
94
remote_camera.py
Normal file
|
|
@ -0,0 +1,94 @@
|
|||
# Web streaming example
|
||||
# Source code from the official PiCamera package
|
||||
# http://picamera.readthedocs.io/en/latest/recipes2.html#web-streaming
|
||||
|
||||
import io
|
||||
import picamera
|
||||
import logging
|
||||
import socketserver
|
||||
from threading import Condition
|
||||
from http import server
|
||||
|
||||
PAGE="""\
|
||||
<html>
|
||||
<head>
|
||||
<title>Raspberry Pi - Remote Camera</title>
|
||||
</head>
|
||||
<body>
|
||||
<center><h1>Raspberry Pi - Remote Camera</h1></center>
|
||||
<center><img src="stream.mjpg" width="1080" height="720"></center>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
class StreamingOutput(object):
|
||||
def __init__(self):
|
||||
self.frame = None
|
||||
self.buffer = io.BytesIO()
|
||||
self.condition = Condition()
|
||||
|
||||
def write(self, buf):
|
||||
if buf.startswith(b'\xff\xd8'):
|
||||
# New frame, copy the existing buffer's content and notify all
|
||||
# clients it's available
|
||||
self.buffer.truncate()
|
||||
with self.condition:
|
||||
self.frame = self.buffer.getvalue()
|
||||
self.condition.notify_all()
|
||||
self.buffer.seek(0)
|
||||
return self.buffer.write(buf)
|
||||
|
||||
class StreamingHandler(server.BaseHTTPRequestHandler):
|
||||
def do_GET(self):
|
||||
if self.path == '/':
|
||||
self.send_response(301)
|
||||
self.send_header('Location', '/index.html')
|
||||
self.end_headers()
|
||||
elif self.path == '/index.html':
|
||||
content = PAGE.encode('utf-8')
|
||||
self.send_response(200)
|
||||
self.send_header('Content-Type', 'text/html')
|
||||
self.send_header('Content-Length', len(content))
|
||||
self.end_headers()
|
||||
self.wfile.write(content)
|
||||
elif self.path == '/stream.mjpg':
|
||||
self.send_response(200)
|
||||
self.send_header('Age', 0)
|
||||
self.send_header('Cache-Control', 'no-cache, private')
|
||||
self.send_header('Pragma', 'no-cache')
|
||||
self.send_header('Content-Type', 'multipart/x-mixed-replace; boundary=FRAME')
|
||||
self.end_headers()
|
||||
try:
|
||||
while True:
|
||||
with output.condition:
|
||||
output.condition.wait()
|
||||
frame = output.frame
|
||||
self.wfile.write(b'--FRAME\r\n')
|
||||
self.send_header('Content-Type', 'image/jpeg')
|
||||
self.send_header('Content-Length', len(frame))
|
||||
self.end_headers()
|
||||
self.wfile.write(frame)
|
||||
self.wfile.write(b'\r\n')
|
||||
except Exception as e:
|
||||
logging.warning(
|
||||
'Removed streaming client %s: %s',
|
||||
self.client_address, str(e))
|
||||
else:
|
||||
self.send_error(404)
|
||||
self.end_headers()
|
||||
|
||||
class StreamingServer(socketserver.ThreadingMixIn, server.HTTPServer):
|
||||
allow_reuse_address = True
|
||||
daemon_threads = True
|
||||
|
||||
with picamera.PiCamera(resolution='1080x720', framerate=24) as camera:
|
||||
output = StreamingOutput()
|
||||
#Uncomment the next line to change your Pi's Camera rotation (in degrees)
|
||||
camera.rotation = 90
|
||||
camera.start_recording(output, format='mjpeg')
|
||||
try:
|
||||
address = ('', 8000)
|
||||
server = StreamingServer(address, StreamingHandler)
|
||||
server.serve_forever()
|
||||
finally:
|
||||
camera.stop_recording()
|
||||
Loading…
Reference in a new issue