RomanNum3ral
/
plane-notify
mirror of https://github.com/RomanNum3ral/plane-notify.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Cleanup Whitespaces and remove json from NotifyBot

This commit is contained in:
Jxck-S 2020-08-15 12:22:16 -04:00
parent aa9953c6b9
commit a2a249086c
6 changed files with 44 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
NotifyBot.py
View File

@ -1,20 +1,19 @@
#Github Updated - NotifyBot 11 #Github Updated - NotifyBot 11
#Import Modules #Import Modules
#Clear Terminal #Clear Terminal
import os import os
os.system('cls' if os.name == 'nt' else 'clear') os.system('cls' if os.name == 'nt' else 'clear')
#Setup Geopy #Setup Geopy
from geopy.geocoders import Nominatim from geopy.geocoders import Nominatim
geolocator = Nominatim(user_agent="OpenSkyBot", timeout=5) geolocator = Nominatim(user_agent="OpenSkyBot", timeout=5)
import json
import time import time
from colorama import Fore, Back, Style from colorama import Fore, Back, Style
import datetime import datetime
from defOpenSky import pullOpenSky from defOpenSky import pullOpenSky
from defADSBX import pullADSBX from defADSBX import pullADSBX
#Setup Config File #Setup Config File
import configparser import configparser
config = configparser.ConfigParser() config = configparser.ConfigParser()
config.read('config.ini') config.read('config.ini')
@ -30,15 +29,15 @@ if config.getboolean('DISCORD', 'ENABLE'):
if config.getboolean('TWITTER', 'ENABLE'): if config.getboolean('TWITTER', 'ENABLE'):
from defTweet import tweepysetup from defTweet import tweepysetup
tweet_api = tweepysetup() tweet_api = tweepysetup()
else: else:
tweet_api = None tweet_api = None
#Setup PushBullet #Setup PushBullet
if config.getboolean('PUSHBULLET', 'ENABLE'): if config.getboolean('PUSHBULLET', 'ENABLE'):
from pushbullet import Pushbullet from pushbullet import Pushbullet
pb = Pushbullet(config['PUSHBULLET']['API_KEY']) pb = Pushbullet(config['PUSHBULLET']['API_KEY'])
pb_channel = pb.get_channel(config.get('PUSHBULLET', 'CHANNEL_TAG')) pb_channel = pb.get_channel(config.get('PUSHBULLET', 'CHANNEL_TAG'))
else: else:
pb_channel = None pb_channel = None
pb = None pb = None
#Set Plane ICAO #Set Plane ICAO
@ -49,7 +48,7 @@ geo_alt_ft = None
last_geo_alt_ft = None last_geo_alt_ft = None
last_below_desired_ft = None last_below_desired_ft = None
feeding = None feeding = None
last_feeding = None last_feeding = None
last_on_ground = None last_on_ground = None
on_ground = None on_ground = None
invalid_Location = None invalid_Location = None
@ -107,7 +106,7 @@ while True:
print ("Longitude: ", longitude) print ("Longitude: ", longitude)
print ("GEO Alitude Ft: ", geo_alt_ft) print ("GEO Alitude Ft: ", geo_alt_ft)
print(Style.RESET_ALL) print(Style.RESET_ALL)
#Lookup Location of coordinates #Lookup Location of coordinates
if longitude != None and latitude != None: if longitude != None and latitude != None:
combined = f"{latitude}, {longitude}" combined = f"{latitude}, {longitude}"
try: try:
@ -116,8 +115,8 @@ while True:
print ("Geopy API Error") print ("Geopy API Error")
print (Fore.YELLOW) print (Fore.YELLOW)
# print ("Geopy debug: ", location.raw) # print ("Geopy debug: ", location.raw)
print(Style.RESET_ALL) print(Style.RESET_ALL)
feeding = True feeding = True
else: else:
print (Fore.RED + 'No Location') print (Fore.RED + 'No Location')
feeding = False feeding = False
@ -136,13 +135,13 @@ while True:
invalid_Location = True invalid_Location = True
print ("Invalid Location: ", invalid_Location) print ("Invalid Location: ", invalid_Location)
if invalid_Location: if invalid_Location:
print (Fore.RED) print (Fore.RED)
print (geoError) print (geoError)
print ("Likely Over Water or Invalid Location") print ("Likely Over Water or Invalid Location")
print(Style.RESET_ALL) print(Style.RESET_ALL)
#Convert Full address to sep variables only if Valid Location #Convert Full address to sep variables only if Valid Location
elif invalid_Location is False: elif invalid_Location is False:
@ -153,7 +152,7 @@ while True:
county = address.get('county', '') county = address.get('county', '')
city = address.get('city', '') city = address.get('city', '')
town = address.get('town', '') town = address.get('town', '')
hamlet = address.get('hamlet', '') hamlet = address.get('hamlet', '')
# print (Fore.YELLOW) # print (Fore.YELLOW)
# print ("Address Fields debug: ", address) # print ("Address Fields debug: ", address)
# print(Style.RESET_ALL) # print(Style.RESET_ALL)
@ -167,7 +166,7 @@ while True:
print ("Hamlet: ", hamlet) print ("Hamlet: ", hamlet)
print ("County: ", county) print ("County: ", county)
print(Style.RESET_ALL) print(Style.RESET_ALL)
#Check if below desire ft #Check if below desire ft
if geo_alt_ft is None: if geo_alt_ft is None:
below_desired_ft = False below_desired_ft = False
@ -176,15 +175,15 @@ while True:
#Check if tookoff #Check if tookoff
tookoff = bool(invalid_Location is False and below_desired_ft and on_ground is False and ((last_feeding is False and feeding) or (last_on_ground))) tookoff = bool(invalid_Location is False and below_desired_ft and on_ground is False and ((last_feeding is False and feeding) or (last_on_ground)))
print ("Tookoff Just Now:", tookoff) print ("Tookoff Just Now:", tookoff)
#Check if Landed #Check if Landed
landed = bool(last_below_desired_ft and invalid_Location is False and ((last_feeding and feeding is False and last_on_ground is False) or (on_ground and last_on_ground is False))) landed = bool(last_below_desired_ft and invalid_Location is False and ((last_feeding and feeding is False and last_on_ground is False) or (on_ground and last_on_ground is False)))
print ("Landed Just Now:", landed) print ("Landed Just Now:", landed)
#Chose city town county or hamlet for location as not all are always avalible. #Chose city town county or hamlet for location as not all are always avalible.
if feeding and invalid_Location is False: if feeding and invalid_Location is False:
aera_hierarchy = city or town or county or hamlet aera_hierarchy = city or town or county or hamlet
#Takeoff Notifcation and Landed #Takeoff Notifcation and Landed
if tookoff: if tookoff:
tookoff_message = ("Just took off from" + " " + aera_hierarchy + ", " + state + ", " + country_code) tookoff_message = ("Just took off from" + " " + aera_hierarchy + ", " + state + ", " + country_code)
@ -196,7 +195,7 @@ while True:
getSS(icao) getSS(icao)
#Discord #Discord
if config.getboolean('DISCORD', 'ENABLE'): if config.getboolean('DISCORD', 'ENABLE'):
dis_message = icao + " " + tookoff_message dis_message = icao + " " + tookoff_message
sendDis(dis_message) sendDis(dis_message)
#PushBullet #PushBullet
if pb != None: if pb != None:
@ -211,7 +210,7 @@ while True:
os.remove("map.png") os.remove("map.png")
if landed: if landed:
landed_time_msg = "" landed_time_msg = ""
if takeoff_time != None: if takeoff_time != None:
landed_time = time.time() - takeoff_time landed_time = time.time() - takeoff_time
@ -221,7 +220,7 @@ while True:
#Google Map or tar1090 screenshot #Google Map or tar1090 screenshot
if config.getboolean('GOOGLE', 'STATICMAP_ENABLE'): if config.getboolean('GOOGLE', 'STATICMAP_ENABLE'):
getMap(aera_hierarchy + ", " + state + ", " + country_code) getMap(aera_hierarchy + ", " + state + ", " + country_code)
else: else:
getSS(icao) getSS(icao)
#Discord #Discord
if config.getboolean('DISCORD', 'ENABLE'): if config.getboolean('DISCORD', 'ENABLE'):
@ -248,7 +247,7 @@ while True:
last_below_desired_ft = below_desired_ft last_below_desired_ft = below_desired_ft
elif failed: elif failed:
print ("Failed to connect to data source rechecking in 15s") print ("Failed to connect to data source rechecking in 15s")
if takeoff_time != None: if takeoff_time != None:
elapsed_time = time.time() - takeoff_time elapsed_time = time.time() - takeoff_time
@ -256,7 +255,7 @@ while True:
print(time_since_tk) print(time_since_tk)
elapsed_calc_time = time.time() - start_time elapsed_calc_time = time.time() - start_time
print (Back.MAGENTA, "--------", running_Count, "------------------------Elapsed Time- ", elapsed_calc_time, "-------------------------------------", Style.RESET_ALL) print (Back.MAGENTA, "--------", running_Count, "------------------------Elapsed Time- ", elapsed_calc_time, "-------------------------------------", Style.RESET_ALL)

2
defDiscord.py
View File

@ -6,5 +6,5 @@ def sendDis(message):
webhook = DiscordWebhook(url=config.get('DISCORD', 'URL'), content=message, username="plane-notify") webhook = DiscordWebhook(url=config.get('DISCORD', 'URL'), content=message, username="plane-notify")
with open("map.png", "rb") as f: with open("map.png", "rb") as f:
webhook.add_file(file=f.read(), filename='map.png') webhook.add_file(file=f.read(), filename='map.png')
response = webhook.execute() response = webhook.execute()

28
defMap.py
View File

@ -1,5 +1,5 @@
def getMap(mapLocation): def getMap(mapLocation):
import requests import requests
import configparser import configparser
config = configparser.ConfigParser() config = configparser.ConfigParser()
config.read('config.ini') config.read('config.ini')
@ -8,18 +8,18 @@ def getMap(mapLocation):
center = str(mapLocation) center = str(mapLocation)
zoom = 9 zoom = 9
r = requests.get(url + "center=" + center + "&zoom=" + r = requests.get(url + "center=" + center + "&zoom=" +
str(zoom) + "&size=800x800 &key=" + str(zoom) + "&size=800x800 &key=" +
api_key + "&sensor=false") api_key + "&sensor=false")
# wb mode is stand for write binary mode # wb mode is stand for write binary mode
f = open('map.png', 'wb') f = open('map.png', 'wb')
# r.content gives content, # r.content gives content,
# in this case gives image # in this case gives image
f.write(r.content) f.write(r.content)
# close method of file object # close method of file object
# save and close the file # save and close the file
f.close() f.close()

8
defOpenSky.py
View File

@ -10,7 +10,7 @@ def pullOpenSky(TRACK_PLANE):
planeData = opens_api.get_states(time_secs=0, icao24=TRACK_PLANE.lower()) planeData = opens_api.get_states(time_secs=0, icao24=TRACK_PLANE.lower())
except: except:
print ("OpenSky Error") print ("OpenSky Error")
failed = True failed = True
if failed is False and planeData != None: if failed is False and planeData != None:
plane_Dict = {} plane_Dict = {}
geo_alt_m = None geo_alt_m = None
@ -19,7 +19,7 @@ def pullOpenSky(TRACK_PLANE):
plane_Dict['callsign'] = (dataStates.callsign) plane_Dict['callsign'] = (dataStates.callsign)
plane_Dict['longitude'] = (dataStates.longitude) plane_Dict['longitude'] = (dataStates.longitude)
plane_Dict['latitude'] = (dataStates.latitude) plane_Dict['latitude'] = (dataStates.latitude)
plane_Dict['on_ground'] = (dataStates.on_ground) plane_Dict['on_ground'] = (dataStates.on_ground)
geo_alt_m = (dataStates.geo_altitude) geo_alt_m = (dataStates.geo_altitude)
try: try:
if geo_alt_m != None: if geo_alt_m != None:
@ -32,6 +32,4 @@ def pullOpenSky(TRACK_PLANE):
plane_Dict = None plane_Dict = None
else: else:
plane_Dict = None plane_Dict = None
return plane_Dict, failed return plane_Dict, failed

2
defSS.py
View File

@ -11,7 +11,7 @@ def getSS(icao):
chrome_options = webdriver.ChromeOptions() chrome_options = webdriver.ChromeOptions()
chrome_options.add_argument('--headless') chrome_options.add_argument('--headless')
chrome_options.add_argument('window-size=800,800') chrome_options.add_argument('window-size=800,800')
# chrome_options.add_argument('--no-sandbox') # required when running as root user. otherwise you would get no sandbox errors. #chrome_options.add_argument('--no-sandbox') # required when running as root user. otherwise you would get no sandbox errors.
browser = webdriver.Chrome(options=chrome_options) browser = webdriver.Chrome(options=chrome_options)
url = "https://globe.adsbexchange.com/?largeMode=2&hideButtons&hideSidebar&mapDim=0&zoom=9&icao=" + icao url = "https://globe.adsbexchange.com/?largeMode=2&hideButtons&hideSidebar&mapDim=0&zoom=9&icao=" + icao
browser.get(url) browser.get(url)

2
defTweet.py
View File

@ -4,7 +4,7 @@ config = configparser.ConfigParser()
config.read('config.ini') config.read('config.ini')
import tweepy import tweepy
def tweepysetup(): def tweepysetup():
#DOCU #DOCU
#https://realpython.com/twitter-bot-python-tweepy/ #https://realpython.com/twitter-bot-python-tweepy/
auth = tweepy.OAuthHandler(config.get('TWITTER', 'CONSUMER_KEY'), config.get('TWITTER', 'CONSUMER_SECRET')) auth = tweepy.OAuthHandler(config.get('TWITTER', 'CONSUMER_KEY'), config.get('TWITTER', 'CONSUMER_SECRET'))
auth.set_access_token(config.get('TWITTER', 'ACCESS_TOKEN'), config.get('TWITTER', 'ACCESS_TOKEN_SECRET')) auth.set_access_token(config.get('TWITTER', 'ACCESS_TOKEN'), config.get('TWITTER', 'ACCESS_TOKEN_SECRET'))