navan
/
AutoSafe
mirror of https://github.com/navanchauhan/AutoSafe.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Optimised; PyLint score 9.44/10

This commit is contained in:
Navan Chauhan 2019-05-24 23:17:13 +05:30
parent fca01b7e7b
commit 656e8b59c4
1 changed files with 79 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

158
main.py
View File

@ -1,24 +1,21 @@
"""AutoSafe
Modular Road-Safety Toolkit
import overpy """
import sys #pylint: disable=C0103,C0111
import simplejson as sjson import argparse
import json
import time import time
from threading import Thread import overpy
import simplejson as sjson
from scipy.spatial import distance as dist from scipy.spatial import distance as dist
from imutils.video import VideoStream from imutils.video import VideoStream
from imutils import face_utils from imutils import face_utils
from threading import Thread #import playsound
import numpy as np
import playsound
import argparse
import imutils import imutils
import dlib import dlib
import cv2 import cv2
from uber_rides.session import Session #from uber_rides.session import Session
from uber_rides.client import UberRidesClient #from uber_rides.client import UberRidesClient
import signal
import argparse
import tweepy import tweepy
import requests import requests
@ -29,7 +26,7 @@ import requests
# Defining Variables # # Defining Variables #
###################### ######################
radius = str(100) # Radius for maxspeed function. taken as a string because gets concatenated in the function radius = str(100) # Radius for maxspeed function. taken as a string because gets concatenated
################ ################
# Dummy Values # # Dummy Values #
@ -81,37 +78,37 @@ def get_api(cfg):
auth.set_access_token(cfg['access_token'], cfg['access_token_secret']) auth.set_access_token(cfg['access_token'], cfg['access_token_secret'])
return tweepy.API(auth) return tweepy.API(auth)
def tweet(): def tweetMe():
cfg = { cfg = {
"consumer_key" : "knQFpTnjuSvr6OxYwebt3wyrd", "consumer_key":"knQFpTnjuSvr6OxYwebt3wyrd",
"consumer_secret" : "Mhex3oRkmaF7lD3hoMvHpAD6ctW0ugKYCopTlhc0JzOLOMIZ0w", "consumer_secret":"Mhex3oRkmaF7lD3hoMvHpAD6ctW0ugKYCopTlhc0JzOLOMIZ0w",
"access_token" : "2846631344-wEozinvHfEIFxFVy51I6te8SrN5OTFtU00wxsiz", "access_token":"2846631344-wEozinvHfEIFxFVy51I6te8SrN5OTFtU00wxsiz",
"access_token_secret" : "Nfx1U8a2TjAQXFLBrJIyy2p36sjBGAWFIthLc1cIoI56U" "access_token_secret":"Nfx1U8a2TjAQXFLBrJIyy2p36sjBGAWFIthLc1cIoI56U"
} }
api = get_api(cfg) api = get_api(cfg)
headers = { headers = {
'Accept': 'application/json', 'Accept': 'application/json',
'user-key': 'a530c1424d9abe5442fa22f77ce03d25', 'user-key': 'a530c1424d9abe5442fa22f77ce03d25',
} }
params = ( params = (
('lat', '28.546519'), ('lat', '28.546519'),
('lon', '77.179248'), ('lon', '77.179248'),
) )
url = 'https://developers.zomato.com/api/v2.1/geocode'
response = requests.get('https://developers.zomato.com/api/v2.1/geocode', headers=headers, params=params) response = requests.get(url, headers=headers, params=params)
loc = response.json()['location']['title'] loc = response.json()['location']['title']
tweet = ("Stay Alert! Sudden braking at: ", loc ) tweet = ("Stay Alert! Sudden braking at: ", loc)
status = api.update_status(status=tweet) status = api.update_status(status=tweet)
print(status)
################################ ################################
# Fetching Details from Zomato # # Fetching Details from Zomato #
################################ ################################
res = []
def getRes(): def getRes():
res = []
headers = { headers = {
'Accept': 'application/json', 'Accept': 'application/json',
'user-key': 'a530c1424d9abe5442fa22f77ce03d25', 'user-key': 'a530c1424d9abe5442fa22f77ce03d25',
@ -121,8 +118,8 @@ def getRes():
('lat', '28.546519'), ('lat', '28.546519'),
('lon', '77.179248'), ('lon', '77.179248'),
) )
url = 'https://developers.zomato.com/api/v2.1/geocode'
response = requests.get('https://developers.zomato.com/api/v2.1/geocode', headers=headers, params=params) response = requests.get(url, headers=headers, params=params)
res = response.json()['popularity']['nearby_res'] res = response.json()['popularity']['nearby_res']
return res return res
@ -133,12 +130,12 @@ def getDetails(res):
} }
url = "https://developers.zomato.com/api/v2.1/restaurant?res_id=" + str(res[0]) url = "https://developers.zomato.com/api/v2.1/restaurant?res_id=" + str(res[0])
newResponse = requests.get(url, headers=headers) newResponse = requests.get(url, headers=headers)
newRes = [] #newRes = []
resName = newResponse.json()['name'] resName = newResponse.json()['name']
resAddress = newResponse.json()['location']['address'] resAddress = newResponse.json()['location']['address']
print("You are feeling sleepy, why don't you take a break?\n") print("You are feeling sleepy, why don't you take a break?\n")
print("Your nearest eatery is " + resName,"\n") print("Your nearest eatery is " + resName, "\n")
print(resName + " is at " + resAddress,"\n") print(resName + " is at " + resAddress, "\n")
def zomato(): def zomato():
res = getRes() res = getRes()
@ -149,22 +146,22 @@ def zomato():
############################ ############################
def drunk(): def drunk():
print("Your Location Appears to be around places which sells \n alcohol, please take the breathalyser test ") print("Your appear to be around places which sell \n alcohol, take the breathalyser test")
bac_raw = str(open("./files/bac.txt","r").read()) bac_raw = str(open("./files/bac.txt", "r").read())
bac = float(float(bac_raw)/100) bac = float(float(bac_raw)/100)
print(bac) print(bac)
if(bac >= 0.08): if bac >= 0.08:
print("Please Do Not Drive!\n") print("Please Do Not Drive!\n")
print("I can call a cab if you want\n") print("I can call a cab if you want\n")
print("") print("")
answer = input("Say No, to disagree, else I'll book the cab: \n") answer = input("Say No, to disagree, else I'll book the cab: \n")
if(answer=="no"): if answer == "no":
print("You are not fit to drive") print("You are not fit to drive")
print("Text message to emergency contact sent") print("Text message to emergency contact sent")
print("Your cab has been booked, thank you for not driving") print("Your cab has been booked, thank you for not driving")
print("have a safe journey!") print("have a safe journey!")
def sound_alarm(): def sound_alarm():
print('You Sleep You Lose') print('You Sleep You Lose')
@ -175,7 +172,7 @@ def eye_aspect_ratio(eye):
A = dist.euclidean(eye[1], eye[5]) A = dist.euclidean(eye[1], eye[5])
B = dist.euclidean(eye[2], eye[4]) B = dist.euclidean(eye[2], eye[4])
C = dist.euclidean(eye[0], eye[3]) C = dist.euclidean(eye[0], eye[3])
ear = (A + B) / (2.0 * C) ear = (A + B) / (2.0 * C)
return ear return ear
@ -185,9 +182,8 @@ shape_predictor = "./files/shape_predictor_68_face_landmarks.dat"
EYE_AR_THRESH = 0.2 # If the EAR goes < this for 48 frames, it is counted as drowsiness EYE_AR_THRESH = 0.2 # If the EAR goes < this for 48 frames, it is counted as drowsiness
EYE_AR_CONSEC_FRAMES = 24 EYE_AR_CONSEC_FRAMES = 24
COUNTER = 0
ALERT = False
######################## ########################
@ -195,6 +191,10 @@ ALERT = False
######################## ########################
def sleepiness(): def sleepiness():
COUNTER = 0
#ALERT = False
print("Initialising Facial Landmark Predictor...") print("Initialising Facial Landmark Predictor...")
detector = dlib.get_frontal_face_detector() detector = dlib.get_frontal_face_detector()
predictor = dlib.shape_predictor(shape_predictor) predictor = dlib.shape_predictor(shape_predictor)
@ -209,17 +209,17 @@ def sleepiness():
frame = vs.read() frame = vs.read()
frame = imutils.resize(frame, width=450) frame = imutils.resize(frame, width=450)
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
rects = detector(gray, 0) rects = detector(gray, 0)
for rect in rects: for rect in rects:
shape = predictor(gray, rect) shape = predictor(gray, rect)
shape = face_utils.shape_to_np(shape) shape = face_utils.shape_to_np(shape)
leftEye = shape[lStart:lEnd] leftEye = shape[lStart:lEnd]
rightEye = shape[rStart:rEnd] rightEye = shape[rStart:rEnd]
leftEAR = eye_aspect_ratio(leftEye) leftEAR = eye_aspect_ratio(leftEye)
rightEAR = eye_aspect_ratio(rightEye) rightEAR = eye_aspect_ratio(rightEye)
ear = (leftEAR + rightEAR) / 2.0 ear = (leftEAR + rightEAR) / 2.0
leftEyeHull = cv2.convexHull(leftEye) leftEyeHull = cv2.convexHull(leftEye)
@ -229,41 +229,39 @@ def sleepiness():
if ear < EYE_AR_THRESH: if ear < EYE_AR_THRESH:
COUNTER += 1 COUNTER += 1
# if the eyes were closed for a sufficient number of # if the eyes were closed for a sufficient number of
# then sound the alarm # then sound the alarm
if COUNTER >= EYE_AR_CONSEC_FRAMES: if COUNTER >= EYE_AR_CONSEC_FRAMES:
# if the alarm is not on, turn it on # if the alarm is not on, turn it on
if not ALARM_ON: if not ALARM_ON:
ALARM_ON = True ALARM_ON = True
# draw an alarm on the frame # draw an alarm on the frame
cv2.putText(frame, "Sleepiness Detected!", (10, 30), cv2.putText(frame, "Sleepiness Detected!", (10, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2) cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2)
print("Sleepiness Detected!") print("Sleepiness Detected!")
print("\a");print("\a");print("\a");print("\a");print("\a") print("\n \n \n")
print("\a");print("\a");print("\a");print("\a");print("\a")
print("\a");print("\a");print("\a");print("\a");print("\a")
zomato() zomato()
time.sleep(5) time.sleep(5)
exit() exit()
else: else:
COUNTER = 0 COUNTER = 0
ALARM_ON = False ALARM_ON = False
cv2.putText(frame, "Ratio: {:.2f}".format(ear), (300, 30), cv2.putText(frame, "Ratio: {:.2f}".format(ear), (300, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2) cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2)
cv2.imshow("Frame", frame) cv2.imshow("Frame", frame)
key = cv2.waitKey(1) & 0xFF key = cv2.waitKey(1) & 0xFF
# breaks loop on q # breaks loop on q
if key == ord("q"): if key == ord("q"):
break break
cv2.destroyAllWindows() cv2.destroyAllWindows()
vs.stop() vs.stop()
@ -278,7 +276,7 @@ def maxspeed(coordinates, radius):
####################### #######################
# Query for Open Maps # # Query for Open Maps #
####################### #######################
result = api.query(""" result = api.query("""
way(around:""" + radius + """,""" + lat + """,""" + lon + """) ["maxspeed"]; way(around:""" + radius + """,""" + lat + """,""" + lon + """) ["maxspeed"];
@ -288,36 +286,37 @@ def maxspeed(coordinates, radius):
results_list = [] results_list = []
for way in result.ways: for way in result.ways:
road = {} road = {}
road["name"] = way.tags.get("name", "n/a") road["name"] = way.tags.get("name", "n/a")
road["speed_limit"] = way.tags.get("maxspeed", "n/a") road["speed_limit"] = way.tags.get("maxspeed", "n/a")
nodes = [] nodes = []
for node in way.nodes: for node in way.nodes:
nodes.append((node.lat, node.lon)) nodes.append((node.lat, node.lon))
road["nodes"] = nodes road["nodes"] = nodes
results_list.append(road) results_list.append(road)
return results_list return results_list
############################################################################################################################ #########################################################################
# Gives data to the function and gets json in return. This json is then parsed. Then the double quotes are stripped off it # # Gives data to the function and gets json in return. #
############################################################################################################################ #This json is then parsed. Then the double quotes are stripped off it #
#########################################################################
def speedlim(): def speedlim():
speedLimit = sjson.dumps(maxspeed((str(glat), str(glng)), radius)[0]['speed_limit']).strip('\"') speedLimit = sjson.dumps(maxspeed((str(glat), str(glng)), radius)[0]['speed_limit']).strip('\"')
while(True): while True:
carSpeedDummy = open("./files/carSpeed.txt","r") carSpeedDummy = open("./files/carSpeed.txt", "r")
carSpeed = carSpeedDummy.read() carSpeed = carSpeedDummy.read()
#print(carSpeed) #print(carSpeed)
#carSpeed = 29 #carSpeed = 29
while(int(carSpeed) > int(speedLimit)): while int(carSpeed) > int(speedLimit):
carSpeedDummy = open("./files/carSpeed.txt","r") carSpeedDummy = open("./files/carSpeed.txt", "r")
carSpeed = carSpeedDummy.read() carSpeed = carSpeedDummy.read()
print("Over The Speed Limit") print("Over The Speed Limit")
time.sleep(5) time.sleep(5)
while(int(carSpeed) <= int(speedLimit)): while int(carSpeed) <= int(speedLimit):
carSpeedDummy = open("./files/carSpeed.txt","r") carSpeedDummy = open("./files/carSpeed.txt", "r")
carSpeed = carSpeedDummy.read() carSpeed = carSpeedDummy.read()
print("Under the Speed Limit") print("Under the Speed Limit")
time.sleep(5) time.sleep(5)
@ -327,16 +326,17 @@ def speedlim():
################## ##################
def brakes(): def brakes():
while(True): while True:
tweet() tweetMe()
##################################################### #####################################################
# Using Argument Parse to run one command at a time # # Using Argument Parse to run one command at a time #
##################################################### #####################################################
parser = argparse.ArgumentParser() parser = argparse.ArgumentParser()
FUNCTION_MAP = {'overspeed' : speedlim, 'sleep-detector' : sleepiness, 'sudden-braking' : brakes, 'drunk' : drunk } FUNCTION_MAP = {'overspeed' : speedlim, 'sleep-detector' : sleepiness,
'sudden-braking' : brakes, 'drunk' : drunk}
parser.add_argument('function', choices=list(FUNCTION_MAP)) parser.add_argument('function', choices=list(FUNCTION_MAP))
args = parser.parse_args() args = parser.parse_args()
func = FUNCTION_MAP[args.function] FUNC = FUNCTION_MAP[args.function]
func() FUNC()