Inital Multi Version untested
This commit is contained in:
Jxck-S
parent
0c89f4cd76
commit
490d33ae9d
258
NotifyBot.py
258
NotifyBot.py
|
|
@ -1,258 +0,0 @@
|
|||
#Github Updated - NotifyBot 11
|
||||
#Import Modules
|
||||
#Clear Terminal
|
||||
print("\033[H\033[J")
|
||||
|
||||
#Ability to Remove old Map
|
||||
import os
|
||||
#Setup Geopy
|
||||
from geopy.geocoders import Nominatim
|
||||
geolocator = Nominatim(user_agent="OpenSkyBot", timeout=5)
|
||||
|
||||
import time
|
||||
from colorama import Fore, Back, Style
|
||||
from defOpenSky import pullOpenSky
|
||||
from defADSBX import pullADSBX
|
||||
|
||||
#Setup Config File
|
||||
import configparser
|
||||
config = configparser.ConfigParser()
|
||||
config.read('config.ini')
|
||||
|
||||
if config.getboolean('GOOGLE', 'STATICMAP_ENABLE'):
|
||||
from defMap import getMap
|
||||
else:
|
||||
from defSS import getSS
|
||||
|
||||
if config.getboolean('DISCORD', 'ENABLE'):
|
||||
from defDiscord import sendDis
|
||||
#Setup Tweepy
|
||||
if config.getboolean('TWITTER', 'ENABLE'):
|
||||
from defTweet import tweepysetup
|
||||
tweet_api = tweepysetup()
|
||||
#Setup PushBullet
|
||||
if config.getboolean('PUSHBULLET', 'ENABLE'):
|
||||
from pushbullet import Pushbullet
|
||||
pb = Pushbullet(config['PUSHBULLET']['API_KEY'])
|
||||
pb_channel = pb.get_channel(config.get('PUSHBULLET', 'CHANNEL_TAG'))
|
||||
|
||||
#Set Plane ICAO
|
||||
icao = config.get('DATA', 'ICAO').upper()
|
||||
#Pre Set Non Reseting Variables
|
||||
geo_alt_ft = None
|
||||
last_geo_alt_ft = None
|
||||
last_below_desired_ft = None
|
||||
feeding = None
|
||||
last_feeding = None
|
||||
last_on_ground = None
|
||||
on_ground = None
|
||||
invalid_Location = None
|
||||
longitude = None
|
||||
latitude = None
|
||||
running_Count = 0
|
||||
callsign = None
|
||||
takeoff_time = None
|
||||
reg = None
|
||||
#Begin Looping program
|
||||
while True:
|
||||
running_Count += 1
|
||||
start_time = time.time()
|
||||
print (Back.MAGENTA, "--------", running_Count, "-------------------------------------------------------------", Style.RESET_ALL)
|
||||
#Reset Variables
|
||||
below_desired_ft = None
|
||||
geo_alt_ft = None
|
||||
longitude = None
|
||||
latitude = None
|
||||
on_ground = None
|
||||
#Get API States for Plane
|
||||
plane_Dict = None
|
||||
if config.get('DATA', 'SOURCE') == "OPENS":
|
||||
plane_Dict, failed = pullOpenSky(icao)
|
||||
print (Fore.YELLOW)
|
||||
print ("OpenSky Sourced Data: ", plane_Dict)
|
||||
print(Style.RESET_ALL)
|
||||
elif config.get('DATA', 'SOURCE') == "ADSBX":
|
||||
plane_Dict, failed = pullADSBX(icao)
|
||||
print (Fore.YELLOW)
|
||||
print ("ADSBX Sourced Data: ", plane_Dict)
|
||||
print(Style.RESET_ALL)
|
||||
|
||||
print (Fore.CYAN)
|
||||
print ("Failed:", failed)
|
||||
print ("ICAO:", icao)
|
||||
print(Style.RESET_ALL)
|
||||
|
||||
#Pull Variables from plane_Dict
|
||||
if failed is False:
|
||||
if plane_Dict == None:
|
||||
feeding = False
|
||||
elif plane_Dict != None:
|
||||
locals().update(plane_Dict)
|
||||
print (Fore.CYAN)
|
||||
if config.get('DATA', 'SOURCE') == "ADSBX":
|
||||
print("Registration: ", reg)
|
||||
else:
|
||||
print("Registration: ", "Only shows when using ADSBX!")
|
||||
print ("Callsign: ", callsign)
|
||||
print ("On Ground: ", on_ground)
|
||||
print ("Latitude: ", latitude)
|
||||
print ("Longitude: ", longitude)
|
||||
print ("GEO Alitude Ft: ", geo_alt_ft)
|
||||
print(Style.RESET_ALL)
|
||||
#Lookup Location of coordinates
|
||||
if longitude != None and latitude != None:
|
||||
combined = f"{latitude}, {longitude}"
|
||||
try:
|
||||
location = geolocator.reverse(combined)
|
||||
except:
|
||||
print ("Geopy API Error")
|
||||
print (Fore.YELLOW)
|
||||
# print ("Geopy debug: ", location.raw)
|
||||
print(Style.RESET_ALL)
|
||||
feeding = True
|
||||
else:
|
||||
print (Fore.RED + 'No Location')
|
||||
feeding = False
|
||||
print(Style.RESET_ALL)
|
||||
|
||||
|
||||
|
||||
#Figure if valid location, valid being geopy finds a location
|
||||
if feeding:
|
||||
try:
|
||||
geoError = location.raw['error']
|
||||
except KeyError:
|
||||
invalid_Location = False
|
||||
geoError = None
|
||||
else:
|
||||
invalid_Location = True
|
||||
|
||||
print ("Invalid Location: ", invalid_Location)
|
||||
|
||||
if invalid_Location:
|
||||
print (Fore.RED)
|
||||
print (geoError)
|
||||
print ("Likely Over Water or Invalid Location")
|
||||
print(Style.RESET_ALL)
|
||||
|
||||
|
||||
#Convert Full address to sep variables only if Valid Location
|
||||
elif invalid_Location is False:
|
||||
address = location.raw['address']
|
||||
country = address.get('country', '')
|
||||
country_code = address.get('country_code', '').upper()
|
||||
state = address.get('state', '')
|
||||
county = address.get('county', '')
|
||||
city = address.get('city', '')
|
||||
town = address.get('town', '')
|
||||
hamlet = address.get('hamlet', '')
|
||||
# print (Fore.YELLOW)
|
||||
# print ("Address Fields debug: ", address)
|
||||
# print(Style.RESET_ALL)
|
||||
print (Fore.GREEN)
|
||||
print("Entire Address: ", location.address)
|
||||
print ("Country Code: ", country_code)
|
||||
print ("Country: ", country)
|
||||
print ("State: ", state)
|
||||
print ("City: ", city)
|
||||
print ("Town: ", town)
|
||||
print ("Hamlet: ", hamlet)
|
||||
print ("County: ", county)
|
||||
print(Style.RESET_ALL)
|
||||
|
||||
#Check if below desire ft
|
||||
if geo_alt_ft is None:
|
||||
below_desired_ft = False
|
||||
elif geo_alt_ft < 10000:
|
||||
below_desired_ft = True
|
||||
#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)))
|
||||
print ("Tookoff Just Now:", tookoff)
|
||||
|
||||
|
||||
#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)))
|
||||
print ("Landed Just Now:", landed)
|
||||
|
||||
#Chose city town county or hamlet for location as not all are always avalible.
|
||||
if feeding and invalid_Location is False:
|
||||
aera_hierarchy = city or town or county or hamlet
|
||||
#Takeoff Notifcation and Landed
|
||||
if tookoff:
|
||||
tookoff_message = ("Just took off from" + " " + aera_hierarchy + ", " + state + ", " + country_code)
|
||||
print (tookoff_message)
|
||||
#Google Map or tar1090 screenshot
|
||||
if config.getboolean('GOOGLE', 'STATICMAP_ENABLE'):
|
||||
getMap(aera_hierarchy + ", " + state + ", " + country_code)
|
||||
else:
|
||||
getSS(icao)
|
||||
#Discord
|
||||
if config.getboolean('DISCORD', 'ENABLE'):
|
||||
dis_message = config.get('DISCORD', 'TITLE') + " " + tookoff_message
|
||||
sendDis(dis_message)
|
||||
#PushBullet
|
||||
if config.getboolean('PUSHBULLET', 'ENABLE'):
|
||||
with open("map.png", "rb") as pic:
|
||||
map_data = pb.upload_file(pic, "Tookoff IMG")
|
||||
push = pb_channel.push_note(config.get('PUSHBULLET', 'TITLE'), tookoff_message)
|
||||
push = pb_channel.push_file(**map_data)
|
||||
#Twitter
|
||||
if config.getboolean('TWITTER', 'ENABLE'):
|
||||
tweet_api.update_with_media("map.png", status = tookoff_message)
|
||||
takeoff_time = time.time()
|
||||
os.remove("map.png")
|
||||
|
||||
|
||||
if landed:
|
||||
landed_time_msg = ""
|
||||
if takeoff_time != None:
|
||||
landed_time = time.time() - takeoff_time
|
||||
landed_time_msg = time.strftime("Apx. flt. time %H Hours : %M Mins ", time.gmtime(landed_time))
|
||||
landed_message = ("Landed just now in" + " " + aera_hierarchy + ", " + state + ", " + country_code + ". " + landed_time_msg)
|
||||
print (landed_message)
|
||||
#Google Map or tar1090 screenshot
|
||||
if config.getboolean('GOOGLE', 'STATICMAP_ENABLE'):
|
||||
getMap(aera_hierarchy + ", " + state + ", " + country_code)
|
||||
else:
|
||||
getSS(icao)
|
||||
#Discord
|
||||
if config.getboolean('DISCORD', 'ENABLE'):
|
||||
dis_message = config.get('DISCORD', 'TITLE') + " " + landed_message
|
||||
sendDis(dis_message)
|
||||
#PushBullet
|
||||
if config.getboolean('PUSHBULLET', 'ENABLE'):
|
||||
with open("map.png", "rb") as pic:
|
||||
map_data = pb.upload_file(pic, "Landed IMG")
|
||||
push = pb_channel.push_note(config.get('PUSHBULLET', 'TITLE'), landed_message)
|
||||
push = pb_channel.push_file(**map_data)
|
||||
#Twitter
|
||||
if config.getboolean('TWITTER', 'ENABLE'):
|
||||
tweet_api.update_with_media("map.png", status = landed_message)
|
||||
takeoff_time = None
|
||||
landed_time = None
|
||||
time_since_tk = None
|
||||
os.remove("map.png")
|
||||
|
||||
#Set Variables to compare to next check
|
||||
last_feeding = feeding
|
||||
last_geo_alt_ft = geo_alt_ft
|
||||
last_on_ground = on_ground
|
||||
last_below_desired_ft = below_desired_ft
|
||||
|
||||
elif failed:
|
||||
print ("Failed to connect to data source rechecking in 15s")
|
||||
|
||||
if takeoff_time != None:
|
||||
elapsed_time = time.time() - takeoff_time
|
||||
time_since_tk = time.strftime("Time Since Take off %H Hours : %M Mins : %S Secs", time.gmtime(elapsed_time))
|
||||
print(time_since_tk)
|
||||
|
||||
|
||||
|
||||
elapsed_calc_time = time.time() - start_time
|
||||
|
||||
print (Back.MAGENTA, "--------", running_Count, "------------------------Elapsed Time- ", elapsed_calc_time, "-------------------------------------", Style.RESET_ALL)
|
||||
print ("")
|
||||
time.sleep(15)
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,60 @@
|
|||
import requests
|
||||
import configparser
|
||||
import json
|
||||
import time
|
||||
from planeClass import Plane
|
||||
main_config = configparser.ConfigParser()
|
||||
main_config.read('mainconf.ini')
|
||||
import os
|
||||
#Set ADSBX URL Based off amount of Conf files
|
||||
if main_config.get('DATA', 'SOURCE') == "ADSBX":
|
||||
planes = {}
|
||||
for filename in os.listdir(os. getcwd()):
|
||||
if filename.endswith(".ini") and filename != "mainconf.ini":
|
||||
plane_config = configparser.ConfigParser()
|
||||
plane_config.read(filename)
|
||||
planes[plane_config.get('DATA', 'ICAO').upper()] = Plane(plane_config.get('DATA', 'ICAO'), filename)
|
||||
if len(planes) > 1:
|
||||
url = "https://adsbexchange.com/api/aircraft/json/"
|
||||
elif len(planes) == 1:
|
||||
url = "https://adsbexchange.com/api/aircraft/icao/" + str(list(planes.keys())[0]) + "/"
|
||||
|
||||
headers = {
|
||||
'api-auth': main_config.get('ADSBX', 'API_KEY'),
|
||||
'Content-Encoding': 'gzip'
|
||||
}
|
||||
elif main_config.get('DATA', 'SOURCE') == "OPENS":
|
||||
raise NotImplementedError
|
||||
while True:
|
||||
if main_config.get('DATA', 'SOURCE') == "ADSBX":
|
||||
try:
|
||||
response = requests.get(url, headers = headers)
|
||||
data = response.text
|
||||
data = json.loads(data)
|
||||
failed = False
|
||||
except (requests.HTTPError, requests.ConnectionError, requests.Timeout) as error_message:
|
||||
print("ADSBX Connection Error")
|
||||
print(error_message)
|
||||
failed = True
|
||||
except json.decoder.JSONDecodeError as error_message:
|
||||
print("Error with JSON")
|
||||
print(error_message)
|
||||
failed = True
|
||||
|
||||
if failed == False:
|
||||
print(data)
|
||||
if data['ac'] != None:
|
||||
for planeData in data['ac']:
|
||||
for key, obj in planes.items():
|
||||
if planeData['icao'] == key:
|
||||
print(planeData['icao'])
|
||||
print(planeData)
|
||||
obj.run(planeData)
|
||||
else:
|
||||
obj.run(None)
|
||||
else:
|
||||
for obj in planes.values():
|
||||
obj.run(None)
|
||||
|
||||
|
||||
time.sleep(15)
|
||||
45
config.ini
45
config.ini
|
|
@ -1,45 +0,0 @@
|
|||
[DATA]
|
||||
#Plane to track, based of ICAO or ICAO24 which is the unique transponder address of a plane.
|
||||
ICAO = planeicaohere
|
||||
|
||||
#Source to pull data from
|
||||
#SHOULD BE ADSBX which is ADS-B Exchange or OPENS which is OpenSky
|
||||
#By default configured with OpenSky which anyone can use without a login
|
||||
#ADS-B Exchange has better data but is not avalible unless you feed their network or pay.
|
||||
SOURCE = OPENS
|
||||
|
||||
#ADS-B Exchange https://www.adsbexchange.com/data/
|
||||
[ADSBX]
|
||||
API_KEY = apikey
|
||||
|
||||
#OpenSky https://opensky-network.org/apidoc/index.html
|
||||
#When using without your own login user and pass should be None
|
||||
[OPENSKY]
|
||||
USERNAME = None
|
||||
PASSWORD = None
|
||||
|
||||
[GOOGLE]
|
||||
#API KEYS
|
||||
#If static map disabled will load up tar1090 ads-b exchange and take screenshot instead.
|
||||
STATICMAP_ENABLE = FALSE
|
||||
STATICMAPKEY = googleapikey
|
||||
|
||||
[TWITTER]
|
||||
ENABLE = FALSE
|
||||
CONSUMER_KEY = ckhere
|
||||
CONSUMER_SECRET = cshere
|
||||
ACCESS_TOKEN = athere
|
||||
ACCESS_TOKEN_SECRET = atshere
|
||||
|
||||
[PUSHBULLET]
|
||||
ENABLE = FALSE
|
||||
TITLE = Title Of Pushbullet message
|
||||
API_KEY = apikey
|
||||
CHANNEL_TAG = channeltag
|
||||
|
||||
[DISCORD]
|
||||
ENABLE = TRUE
|
||||
#WEBHOOK URL https://support.discord.com/hc/en-us/articles/228383668-Intro-to-Webhooks
|
||||
URL = webhookhere
|
||||
Title = title in front of message
|
||||
USERNAME = username of user in channel can be anything
|
||||
|
|
@ -2,9 +2,9 @@ from discord_webhook import DiscordWebhook
|
|||
import configparser
|
||||
config = configparser.ConfigParser()
|
||||
config.read('config.ini')
|
||||
def sendDis(message):
|
||||
def sendDis(message, map_file_name):
|
||||
webhook = DiscordWebhook(url=config.get('DISCORD', 'URL'), content=message, username=config.get('DISCORD', 'USERNAME'))
|
||||
|
||||
with open("map.png", "rb") as f:
|
||||
with open(map_file_name, "rb") as f:
|
||||
webhook.add_file(file=f.read(), filename='map.png')
|
||||
response = webhook.execute()
|
||||
3
defSS.py
3
defSS.py
|
|
@ -16,5 +16,6 @@ def getSS(icao):
|
|||
url = "https://globe.adsbexchange.com/?largeMode=2&hideButtons&hideSidebar&mapDim=0&zoom=9&icao=" + icao
|
||||
browser.get(url)
|
||||
time.sleep(30)
|
||||
browser.save_screenshot("map.png")
|
||||
file_name = icao + "_map.png"
|
||||
browser.save_screenshot(file_name)
|
||||
browser.quit()
|
||||
|
|
@ -3,8 +3,7 @@
|
|||
#SHOULD BE ADSBX which is ADS-B Exchange or OPENS which is OpenSky
|
||||
#By default configured with OpenSky which anyone can use without a login
|
||||
#ADS-B Exchange has better data but is not avalible unless you feed their network or pay.
|
||||
SOURCE = OPENS
|
||||
|
||||
SOURCE = ADSBX
|
||||
|
||||
#ADS-B Exchange https://www.adsbexchange.com/data/
|
||||
[ADSBX]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,23 @@
|
|||
[DATA]
|
||||
#Plane to track, based of ICAO or ICAO24 which is the unique transponder address of a plane.
|
||||
ICAO = icao
|
||||
|
||||
[TWITTER]
|
||||
ENABLE = FALSE
|
||||
CONSUMER_KEY = ckhere
|
||||
CONSUMER_SECRET = cshere
|
||||
ACCESS_TOKEN = athere
|
||||
ACCESS_TOKEN_SECRET = atshere
|
||||
|
||||
[PUSHBULLET]
|
||||
ENABLE = FALSE
|
||||
TITLE = Title Of Pushbullet message
|
||||
API_KEY = apikey
|
||||
CHANNEL_TAG = channeltag
|
||||
|
||||
[DISCORD]
|
||||
ENABLE = TRUE
|
||||
#WEBHOOK URL https://support.discord.com/hc/en-us/articles/228383668-Intro-to-Webhooks
|
||||
URL = webhookurl
|
||||
Title = title
|
||||
USERNAME = plane-notify
|
||||
286
planeClass.py
286
planeClass.py
|
|
@ -16,53 +16,51 @@ class Plane:
|
|||
self.callsign = None
|
||||
self.takeoff_time = None
|
||||
self.reg = None
|
||||
self.map_file_name = icao.upper() + "_map.png"
|
||||
def getICAO(self):
|
||||
return self.icao
|
||||
def run(self, plane_Dict):
|
||||
def run(self, ac_dict):
|
||||
|
||||
#Import Modules
|
||||
#Clear Terminal
|
||||
print("\033[H\033[J")
|
||||
#print("\033[H\033[J")
|
||||
|
||||
#Ability to Remove old Map
|
||||
import os
|
||||
#Setup Geopy
|
||||
from geopy.geocoders import Nominatim
|
||||
geolocator = Nominatim(user_agent="OpenSkyBot", timeout=5)
|
||||
|
||||
geolocator = Nominatim(user_agent="NotifyBot", timeout=5)
|
||||
import time
|
||||
from colorama import Fore, Back, Style
|
||||
from defOpenSky import pullOpenSky
|
||||
from defADSBX import pullADSBX
|
||||
|
||||
#Setup Config File
|
||||
import configparser
|
||||
config = configparser.ConfigParser()
|
||||
config.read(self.conf_file)
|
||||
self.config = configparser.ConfigParser()
|
||||
self.config.read(self.conf_file)
|
||||
main_config = configparser.ConfigParser()
|
||||
main_config.read('mainconf.ini')
|
||||
|
||||
if config.getboolean('GOOGLE', 'STATICMAP_ENABLE'):
|
||||
if self.config.getboolean('GOOGLE', 'STATICMAP_ENABLE'):
|
||||
from defMap import getMap
|
||||
else:
|
||||
from defSS import getSS
|
||||
|
||||
if config.getboolean('DISCORD', 'ENABLE'):
|
||||
if self.config.getboolean('DISCORD', 'ENABLE'):
|
||||
from defDiscord import sendDis
|
||||
#Setup Tweepy
|
||||
if config.getboolean('TWITTER', 'ENABLE'):
|
||||
if self.config.getboolean('TWITTER', 'ENABLE'):
|
||||
from defTweet import tweepysetup
|
||||
tweet_api = tweepysetup()
|
||||
self.tweet_api = tweepysetup()
|
||||
#Setup PushBullet
|
||||
if config.getboolean('PUSHBULLET', 'ENABLE'):
|
||||
if self.config.getboolean('PUSHBULLET', 'ENABLE'):
|
||||
from pushbullet import Pushbullet
|
||||
pb = Pushbullet(config['PUSHBULLET']['API_KEY'])
|
||||
pb_channel = pb.get_channel(config.get('PUSHBULLET', 'CHANNEL_TAG'))
|
||||
self.pb = Pushbullet(self.config['PUSHBULLET']['API_KEY'])
|
||||
self.pb_channel = pb.get_channel(self.config.get('PUSHBULLET', 'CHANNEL_TAG'))
|
||||
|
||||
#Set Plane ICAO
|
||||
icao = config.get('DATA', 'ICAO').upper()
|
||||
#Pre Set Non Reseting Variables
|
||||
|
||||
running_Count +=1
|
||||
start_time = time.time()
|
||||
print (Back.MAGENTA, "--------", running_Count, "-------------------------------------------------------------", Style.RESET_ALL)
|
||||
self.running_Count +=1
|
||||
self.start_time = time.time()
|
||||
print (Back.MAGENTA, "--------", self.running_Count, "----------------------------- ICAO:", self.icao, "--------------------------", Style.RESET_ALL)
|
||||
#Reset Variables
|
||||
self.below_desired_ft = None
|
||||
self.geo_alt_ft = None
|
||||
|
|
@ -71,189 +69,201 @@ class Plane:
|
|||
self.on_ground = None
|
||||
#Get API States for Plane
|
||||
self.plane_Dict = None
|
||||
if config.get('DATA', 'SOURCE') == "OPENS":
|
||||
plane_Dict, failed = pullOpenSky(icao)
|
||||
print (Fore.YELLOW)
|
||||
print ("OpenSky Sourced Data: ", plane_Dict)
|
||||
print(Style.RESET_ALL)
|
||||
elif config.get('DATA', 'SOURCE') == "ADSBX":
|
||||
plane_Dict, failed = pullADSBX(icao)
|
||||
print (Fore.YELLOW)
|
||||
print ("ADSBX Sourced Data: ", plane_Dict)
|
||||
print(Style.RESET_ALL)
|
||||
if main_config.get('DATA', 'SOURCE') == "OPENS":
|
||||
raise NotImplementedError
|
||||
#plane_Dict, failed = pullOpenSky(icao)
|
||||
#print (Fore.YELLOW)
|
||||
#print ("OpenSky Sourced Data: ", plane_Dict)
|
||||
#print(Style.RESET_ALL)
|
||||
elif main_config.get('DATA', 'SOURCE') == "ADSBX":
|
||||
self.val_error = False
|
||||
if ac_dict != None:
|
||||
try:
|
||||
self.plane_Dict = {'icao' : ac_dict['icao'], 'callsign' : ac_dict['call'], 'reg' : ac_dict['reg'], 'latitude' : float(ac_dict['lat']), 'longitude' : float(ac_dict['lon']), 'geo_alt_ft' : int(ac_dict['galt']), 'on_ground' : bool(int(ac_dict["gnd"]))}
|
||||
if self.plane_Dict['on_ground']:
|
||||
self.plane_Dict['geo_alt_ft'] = 0
|
||||
except ValueError as e:
|
||||
self.plane_Dict = None
|
||||
self.val_error = True
|
||||
print("Got data but some data is invalid!")
|
||||
print(e)
|
||||
else:
|
||||
self.plane_Dict = None
|
||||
|
||||
print (Fore.YELLOW)
|
||||
print ("ADSBX Sourced Data: ", self.plane_Dict)
|
||||
print(Style.RESET_ALL)
|
||||
print (Fore.CYAN)
|
||||
print ("Failed:", failed)
|
||||
print ("ICAO:", icao)
|
||||
print ("ICAO:", self.icao)
|
||||
print(Style.RESET_ALL)
|
||||
|
||||
#Pull Variables from plane_Dict
|
||||
if failed is False:
|
||||
if plane_Dict == None:
|
||||
feeding = False
|
||||
elif plane_Dict != None:
|
||||
locals().update(plane_Dict)
|
||||
if self.val_error is False:
|
||||
if self.plane_Dict == None:
|
||||
self.feeding = False
|
||||
elif self.plane_Dict != None:
|
||||
self.__dict__.update(self.plane_Dict)
|
||||
print (Fore.CYAN)
|
||||
if config.get('DATA', 'SOURCE') == "ADSBX":
|
||||
print("Registration: ", reg)
|
||||
if main_config.get('DATA', 'SOURCE') == "ADSBX":
|
||||
print("Registration: ", self.reg)
|
||||
else:
|
||||
print("Registration: ", "Only shows when using ADSBX!")
|
||||
print ("Callsign: ", callsign)
|
||||
print ("On Ground: ", on_ground)
|
||||
print ("Latitude: ", latitude)
|
||||
print ("Longitude: ", longitude)
|
||||
print ("GEO Alitude Ft: ", geo_alt_ft)
|
||||
print ("Callsign: ", self.callsign)
|
||||
print ("On Ground: ", self.on_ground)
|
||||
print ("Latitude: ", self.latitude)
|
||||
print ("Longitude: ", self.longitude)
|
||||
print ("GEO Alitude Ft: ", self.geo_alt_ft)
|
||||
print(Style.RESET_ALL)
|
||||
#Lookup Location of coordinates
|
||||
if longitude != None and latitude != None:
|
||||
combined = f"{latitude}, {longitude}"
|
||||
if self.longitude != None and self.latitude != None:
|
||||
self.combined = f"{self.latitude}, {self.longitude}"
|
||||
try:
|
||||
location = geolocator.reverse(combined)
|
||||
except:
|
||||
print ("Geopy API Error")
|
||||
self.location = geolocator.reverse(self.combined)
|
||||
except BaseException as e:
|
||||
print ("Geopy API Error", e)
|
||||
print (Fore.YELLOW)
|
||||
# print ("Geopy debug: ", location.raw)
|
||||
print(Style.RESET_ALL)
|
||||
feeding = True
|
||||
self.feeding = True
|
||||
else:
|
||||
print (Fore.RED + 'No Location')
|
||||
feeding = False
|
||||
self.feeding = False
|
||||
print(Style.RESET_ALL)
|
||||
|
||||
|
||||
|
||||
#Figure if valid location, valid being geopy finds a location
|
||||
if feeding:
|
||||
if self.feeding:
|
||||
try:
|
||||
geoError = location.raw['error']
|
||||
self.geoError = self.location.raw['error']
|
||||
except KeyError:
|
||||
invalid_Location = False
|
||||
geoError = None
|
||||
self.invalid_Location = False
|
||||
self.geoError = None
|
||||
else:
|
||||
invalid_Location = True
|
||||
self.invalid_Location = True
|
||||
|
||||
print ("Invalid Location: ", invalid_Location)
|
||||
print ("Invalid Location: ", self.invalid_Location)
|
||||
|
||||
if invalid_Location:
|
||||
if self.invalid_Location:
|
||||
print (Fore.RED)
|
||||
print (geoError)
|
||||
print (self.geoError)
|
||||
print ("Likely Over Water or Invalid Location")
|
||||
print(Style.RESET_ALL)
|
||||
|
||||
|
||||
#Convert Full address to sep variables only if Valid Location
|
||||
elif invalid_Location is False:
|
||||
address = location.raw['address']
|
||||
country = address.get('country', '')
|
||||
country_code = address.get('country_code', '').upper()
|
||||
state = address.get('state', '')
|
||||
county = address.get('county', '')
|
||||
city = address.get('city', '')
|
||||
town = address.get('town', '')
|
||||
hamlet = address.get('hamlet', '')
|
||||
elif self.invalid_Location is False:
|
||||
self.address = self.location.raw['address']
|
||||
self.country = self.address.get('country', '')
|
||||
self.country_code = self.address.get('country_code', '').upper()
|
||||
self.state = self.address.get('state', '')
|
||||
self.county = self.address.get('county', '')
|
||||
self.city = self.address.get('city', '')
|
||||
self.town = self.address.get('town', '')
|
||||
self.hamlet = self.address.get('hamlet', '')
|
||||
# print (Fore.YELLOW)
|
||||
# print ("Address Fields debug: ", address)
|
||||
# print ("Address Fields debug: ", self.address)
|
||||
# print(Style.RESET_ALL)
|
||||
print (Fore.GREEN)
|
||||
print("Entire Address: ", location.address)
|
||||
print ("Country Code: ", country_code)
|
||||
print ("Country: ", country)
|
||||
print ("State: ", state)
|
||||
print ("City: ", city)
|
||||
print ("Town: ", town)
|
||||
print ("Hamlet: ", hamlet)
|
||||
print ("County: ", county)
|
||||
print("Entire Address: ", self.location.address)
|
||||
print ("Country Code: ", self.country_code)
|
||||
print ("Country: ", self.country)
|
||||
print ("State: ", self.state)
|
||||
print ("City: ", self.city)
|
||||
print ("Town: ", self.town)
|
||||
print ("Hamlet: ", self.hamlet)
|
||||
print ("County: ", self.county)
|
||||
print(Style.RESET_ALL)
|
||||
|
||||
#Check if below desire ft
|
||||
if geo_alt_ft is None:
|
||||
below_desired_ft = False
|
||||
elif geo_alt_ft < 10000:
|
||||
below_desired_ft = True
|
||||
if self.geo_alt_ft is None:
|
||||
self.below_desired_ft = False
|
||||
elif self.geo_alt_ft < 10000:
|
||||
self.below_desired_ft = True
|
||||
#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)))
|
||||
print ("Tookoff Just Now:", tookoff)
|
||||
self.tookoff = bool(self.invalid_Location is False and self.below_desired_ft and self.on_ground is False and ((self.last_feeding is False and self.feeding) or (self.last_on_ground)))
|
||||
print ("Tookoff Just Now:", self.tookoff)
|
||||
|
||||
|
||||
#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)))
|
||||
print ("Landed Just Now:", landed)
|
||||
self.landed = bool(self.last_below_desired_ft and self.invalid_Location is False and ((self.last_feeding and self.feeding is False and self.last_on_ground is False) or (self.on_ground and self.last_on_ground is False)))
|
||||
print ("Landed Just Now:", self.landed)
|
||||
|
||||
#Chose city town county or hamlet for location as not all are always avalible.
|
||||
if feeding and invalid_Location is False:
|
||||
aera_hierarchy = city or town or county or hamlet
|
||||
if self.feeding and self.invalid_Location is False:
|
||||
self.aera_hierarchy = self.city or self.town or self.county or self.hamlet
|
||||
#Takeoff Notifcation and Landed
|
||||
if tookoff:
|
||||
tookoff_message = ("Just took off from" + " " + aera_hierarchy + ", " + state + ", " + country_code)
|
||||
if self.tookoff:
|
||||
tookoff_message = ("Just took off from" + " " + self.aera_hierarchy + ", " + self.state + ", " + self.country_code)
|
||||
print (tookoff_message)
|
||||
#Google Map or tar1090 screenshot
|
||||
if config.getboolean('GOOGLE', 'STATICMAP_ENABLE'):
|
||||
getMap(aera_hierarchy + ", " + state + ", " + country_code)
|
||||
if self.config.getboolean('GOOGLE', 'STATICMAP_ENABLE'):
|
||||
getMap(self.aera_hierarchy + ", " + self.state + ", " + self.country_code)
|
||||
else:
|
||||
getSS(icao)
|
||||
getSS(self.icao)
|
||||
#Discord
|
||||
if config.getboolean('DISCORD', 'ENABLE'):
|
||||
dis_message = config.get('DISCORD', 'TITLE') + " " + tookoff_message
|
||||
sendDis(dis_message)
|
||||
if self.config.getboolean('DISCORD', 'ENABLE'):
|
||||
self.dis_message = self.config.get('DISCORD', 'TITLE') + " " + self.tookoff_message
|
||||
sendDis(dis_message, self.map_file_name)
|
||||
#PushBullet
|
||||
if config.getboolean('PUSHBULLET', 'ENABLE'):
|
||||
with open("map.png", "rb") as pic:
|
||||
if self.config.getboolean('PUSHBULLET', 'ENABLE'):
|
||||
with open(self.map_file_name, "rb") as pic:
|
||||
map_data = pb.upload_file(pic, "Tookoff IMG")
|
||||
push = pb_channel.push_note(config.get('PUSHBULLET', 'TITLE'), tookoff_message)
|
||||
push = pb_channel.push_file(**map_data)
|
||||
push = self.pb_channel.push_note(self.config.get('PUSHBULLET', 'TITLE'), self.tookoff_message)
|
||||
push = self.pb_channel.push_file(**map_data)
|
||||
#Twitter
|
||||
if config.getboolean('TWITTER', 'ENABLE'):
|
||||
tweet_api.update_with_media("map.png", status = tookoff_message)
|
||||
takeoff_time = time.time()
|
||||
os.remove("map.png")
|
||||
if self.config.getboolean('TWITTER', 'ENABLE'):
|
||||
self.tweet_api.update_with_media(self.map_file_name, status = self.tookoff_message)
|
||||
self.takeoff_time = time.time()
|
||||
os.remove(self.map_file_name)
|
||||
|
||||
|
||||
if landed:
|
||||
landed_time_msg = ""
|
||||
if takeoff_time != None:
|
||||
landed_time = time.time() - takeoff_time
|
||||
landed_time_msg = time.strftime("Apx. flt. time %H Hours : %M Mins ", time.gmtime(landed_time))
|
||||
landed_message = ("Landed just now in" + " " + aera_hierarchy + ", " + state + ", " + country_code + ". " + landed_time_msg)
|
||||
if self.landed:
|
||||
self.landed_time_msg = ""
|
||||
if self.takeoff_time != None:
|
||||
self.landed_time = time.time() - self.takeoff_time
|
||||
self.landed_time_msg = time.strftime("Apx. flt. time %H Hours : %M Mins ", self.time.gmtime(landed_time))
|
||||
self.landed_message = ("Landed just now in" + " " + self.aera_hierarchy + ", " + self.state + ", " + self.country_code + ". " + self.landed_time_msg)
|
||||
print (landed_message)
|
||||
#Google Map or tar1090 screenshot
|
||||
if config.getboolean('GOOGLE', 'STATICMAP_ENABLE'):
|
||||
getMap(aera_hierarchy + ", " + state + ", " + country_code)
|
||||
if self.config.getboolean('GOOGLE', 'STATICMAP_ENABLE'):
|
||||
getMap(self.aera_hierarchy + ", " + self.state + ", " + self.country_code)
|
||||
else:
|
||||
getSS(icao)
|
||||
getSS(self.icao)
|
||||
#Discord
|
||||
if config.getboolean('DISCORD', 'ENABLE'):
|
||||
dis_message = config.get('DISCORD', 'TITLE') + " " + landed_message
|
||||
sendDis(dis_message)
|
||||
if self.config.getboolean('DISCORD', 'ENABLE'):
|
||||
self.dis_message = self.config.get('DISCORD', 'TITLE') + " " + self.landed_message
|
||||
sendDis(self.dis_message, self.map_file_name)
|
||||
#PushBullet
|
||||
if config.getboolean('PUSHBULLET', 'ENABLE'):
|
||||
with open("map.png", "rb") as pic:
|
||||
if self.config.getboolean('PUSHBULLET', 'ENABLE'):
|
||||
with open(self.map_file_name, "rb") as pic:
|
||||
map_data = pb.upload_file(pic, "Landed IMG")
|
||||
push = pb_channel.push_note(config.get('PUSHBULLET', 'TITLE'), landed_message)
|
||||
push = pb_channel.push_file(**map_data)
|
||||
push = self.pb_channel.push_note(self.config.get('PUSHBULLET', 'TITLE'), self.landed_message)
|
||||
push = self.pb_channel.push_file(**map_data)
|
||||
#Twitter
|
||||
if config.getboolean('TWITTER', 'ENABLE'):
|
||||
tweet_api.update_with_media("map.png", status = landed_message)
|
||||
takeoff_time = None
|
||||
landed_time = None
|
||||
time_since_tk = None
|
||||
os.remove("map.png")
|
||||
if self.config.getboolean('TWITTER', 'ENABLE'):
|
||||
self.tweet_api.update_with_media(self.map_file_name, status = self.landed_message)
|
||||
self.takeoff_time = None
|
||||
self.landed_time = None
|
||||
self.time_since_tk = None
|
||||
os.remove(self.map_file_name)
|
||||
|
||||
#Set Variables to compare to next check
|
||||
last_feeding = feeding
|
||||
last_geo_alt_ft = geo_alt_ft
|
||||
last_on_ground = on_ground
|
||||
last_below_desired_ft = below_desired_ft
|
||||
self.last_feeding = self.feeding
|
||||
self.last_geo_alt_ft = self.geo_alt_ft
|
||||
self.last_on_ground = self.on_ground
|
||||
self.last_below_desired_ft = self.below_desired_ft
|
||||
|
||||
elif failed:
|
||||
print ("Failed to connect to data source rechecking in 15s")
|
||||
elif self.val_error:
|
||||
print ("Failed to Parse Will Recheck this Plane After new data")
|
||||
|
||||
if takeoff_time != None:
|
||||
elapsed_time = time.time() - takeoff_time
|
||||
time_since_tk = time.strftime("Time Since Take off %H Hours : %M Mins : %S Secs", time.gmtime(elapsed_time))
|
||||
print(time_since_tk)
|
||||
if self.takeoff_time != None:
|
||||
self.elapsed_time = self.time.time() - self.takeoff_time
|
||||
self.time_since_tk = self.time.strftime("Time Since Take off %H Hours : %M Mins : %S Secs", time.gmtime(self.elapsed_time))
|
||||
print(self.time_since_tk)
|
||||
|
||||
|
||||
|
||||
elapsed_calc_time = time.time() - start_time
|
||||
self.elapsed_calc_time = time.time() - self.start_time
|
||||
|
||||
print (Back.MAGENTA, "--------", running_Count, "------------------------Elapsed Time- ", elapsed_calc_time, "-------------------------------------", Style.RESET_ALL)
|
||||
print (Back.MAGENTA, "--------", self.running_Count, "------------------------Elapsed Time- ", self.elapsed_calc_time, "-------------------------------------", Style.RESET_ALL)
|
||||
|
|
|
|||
|
|
@ -1,41 +0,0 @@
|
|||
import requests
|
||||
import configparser
|
||||
import json
|
||||
import time
|
||||
from planeClass import Plane
|
||||
main_config = configparser.ConfigParser()
|
||||
main_config.read('mainconf.ini')
|
||||
import os
|
||||
#Set ADSBX URL Based off amount of Conf files
|
||||
if main_config.get('DATA', 'SOURCE') == "ADSBX":
|
||||
planes = {}
|
||||
for filename in os.listdir(os. getcwd()):
|
||||
if filename.endswith(".ini") and filename != "mainconf.ini":
|
||||
plane_config = configparser.ConfigParser()
|
||||
plane_config.read(filename)
|
||||
planes[plane_config.get('DATA', 'ICAO').upper()] = Plane(plane_config.get('DATA', 'ICAO'), filename)
|
||||
if len(planes) > 1:
|
||||
url = "https://adsbexchange.com/api/aircraft/json/"
|
||||
elif len(planes) == 1:
|
||||
url = "https://adsbexchange.com/api/aircraft/icao/" + next(iter(planes.values())).upper() + "/"
|
||||
|
||||
headers = {
|
||||
'api-auth': main_config.get('ADSBX', 'API_KEY'),
|
||||
'Content-Encoding': 'gzip'
|
||||
}
|
||||
if main_config.get('DATA', 'SOURCE') == "OPENS":
|
||||
raise NotImplementedError
|
||||
while True:
|
||||
if main_config.get('DATA', 'SOURCE') == "ADSBX":
|
||||
response = requests.get(url, headers = headers)
|
||||
data = response.text
|
||||
data = json.loads(data)
|
||||
for planeData in data:
|
||||
for obj in planes:
|
||||
if planeData['icao'] == obj.getICAO:
|
||||
obj.run(planeData)
|
||||
else:
|
||||
obj.run(None)
|
||||
|
||||
|
||||
time.sleep(15)
|
||||
Loading…
Reference in New Issue