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

New Location Name Lookup, Althold Sel Alt 

-Scrap GeoPy Lookup, now using nearest airport city/state
-Update Nav mode notifications, include althold alitude,
screenshot only on approach
-Discord support for only message, image not required.
This commit is contained in:
Jxck-S 2021-01-16 13:38:25 -05:00
parent ecc80b89d0
commit a0d42b47e7
4 changed files with 60 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
NotifyBotMulti.py
View File

@ -10,7 +10,7 @@ from planeClass import Plane
from datetime import datetime from datetime import datetime
import pytz import pytz
import os import os
required_files = [("Roboto-Regular.ttf", 'https://github.com/google/fonts/raw/master/apache/roboto/static/Roboto-Regular.ttf'), ('airports.csv', 'https://ourairports.com/data/airports.csv')] required_files = [("Roboto-Regular.ttf", 'https://github.com/google/fonts/raw/master/apache/roboto/static/Roboto-Regular.ttf'), ('airports.csv', 'https://ourairports.com/data/airports.csv'), ('regions.csv', 'https://ourairports.com/data/regions.csv')]
for file in required_files: for file in required_files:
file_name = file[0] file_name = file[0]
url = file[1] url = file[1]

11
defAirport.py
View File

@ -25,7 +25,7 @@ def getClosestAirport(latitude, longitude, allowed_types):
import csv import csv
from geopy.distance import geodesic from geopy.distance import geodesic
plane = (latitude, longitude) plane = (latitude, longitude)
with open('airports.csv', 'r') as airport_csv: with open('airports.csv', 'r', encoding='utf-8') as airport_csv:
airport_csv_reader = csv.DictReader(filter(lambda row: row[0]!='#', airport_csv)) airport_csv_reader = csv.DictReader(filter(lambda row: row[0]!='#', airport_csv))
for airport in airport_csv_reader: for airport in airport_csv_reader:
if airport['type'] in allowed_types: if airport['type'] in allowed_types:
@ -37,8 +37,13 @@ def getClosestAirport(latitude, longitude, allowed_types):
elif airport_dist < closest_airport_dist: elif airport_dist < closest_airport_dist:
closest_airport_dict = airport closest_airport_dict = airport
closest_airport_dist = airport_dist closest_airport_dist = airport_dist
closest_airport_dict['distance'] = closest_airport_dist closest_airport_dict['distance_mi'] = closest_airport_dist
#Convert indent key to icao key as its labeled icao in other places not ident #Convert indent key to icao key as its labeled icao in other places not ident
closest_airport_dict['icao'] = closest_airport_dict.pop('ident') closest_airport_dict['icao'] = closest_airport_dict.pop('ident')
print("Closest Airport:", closest_airport_dict['icao'], closest_airport_dict['name'], closest_airport_dist, "Miles Away") #Get full region/state name from iso region name
with open('regions.csv', 'r', encoding='utf-8') as regions_csv:
regions_csv = csv.DictReader(filter(lambda row: row[0]!='#', regions_csv))
for region in regions_csv:
if region['code'] == closest_airport_dict['iso_region']:
closest_airport_dict['region'] = region['name']
return closest_airport_dict return closest_airport_dict

5
defDiscord.py
View File

@ -1,8 +1,9 @@
def sendDis(message, map_file_name, config): def sendDis(message, config, image_name = None):
from discord_webhook import DiscordWebhook from discord_webhook import DiscordWebhook
webhook = DiscordWebhook(url=config.get('DISCORD', 'URL'), content=message, username=config.get('DISCORD', 'USERNAME')) webhook = DiscordWebhook(url=config.get('DISCORD', 'URL'), content=message, username=config.get('DISCORD', 'USERNAME'))
with open(map_file_name, "rb") as f: if image_name != None:
with open(image_name, "rb") as f:
webhook.add_file(file=f.read(), filename='map.png') webhook.add_file(file=f.read(), filename='map.png')
webhook.execute() webhook.execute()

95
planeClass.py
View File

@ -69,9 +69,9 @@ class Plane:
#Parse ADBSX V2 Vector #Parse ADBSX V2 Vector
from colorama import Fore, Back, Style from colorama import Fore, Back, Style
self.printheader("head") self.printheader("head")
#print (Fore.YELLOW +"ADSBX Sourced Data: ", ac_dict, Style.RESET_ALL) print (Fore.YELLOW +"ADSBX Sourced Data: ", ac_dict, Style.RESET_ALL)
try: try:
self.__dict__.update({'icao' : ac_dict['hex'].upper(), 'reg' : ac_dict['r'], 'squawk' : ac_dict['squawk'], 'latitude' : float(ac_dict['lat']), 'longitude' : float(ac_dict['lon'])}) self.__dict__.update({'icao' : ac_dict['hex'].upper(), 'reg' : ac_dict['r'], 'latitude' : float(ac_dict['lat']), 'longitude' : float(ac_dict['lon'])})
if 'alt_geom' in ac_dict: if 'alt_geom' in ac_dict:
self.geo_alt_ft = int(ac_dict['alt_geom']) self.geo_alt_ft = int(ac_dict['alt_geom'])
elif 'alt_geom' not in ac_dict and 'alt_baro' in ac_dict: elif 'alt_geom' not in ac_dict and 'alt_baro' in ac_dict:
@ -86,10 +86,21 @@ class Plane:
if'nav_modes' in ac_dict: if'nav_modes' in ac_dict:
self.nav_modes = ac_dict['nav_modes'] self.nav_modes = ac_dict['nav_modes']
for idx, mode in enumerate(self.nav_modes): for idx, mode in enumerate(self.nav_modes):
if mode == ('tcas' or 'lnav' or 'vnav'): if mode.upper() in ['TCAS', 'LNAV', 'VNAV']:
self.nav_modes[idx] = self.nav_modes[idx].upper() self.nav_modes[idx] = self.nav_modes[idx].upper()
else: else:
self.nav_modes[idx] = self.nav_modes[idx].capitalize() self.nav_modes[idx] = self.nav_modes[idx].capitalize()
if 'squawk' in ac_dict:
self.squawk = ac_dict['squawk']
else:
self.squawk = None
if "nav_altitude_fms" in ac_dict:
self.nav_altitude = ac_dict['nav_altitude_fms']
if "nav_altitude_mcp" in ac_dict:
self.nav_altitude = ac_dict['nav_altitude_mcp']
else:
self.nav_altitude = None
from datetime import datetime, timedelta from datetime import datetime, timedelta
#Create last seen timestamp from how long ago in secs a pos was rec #Create last seen timestamp from how long ago in secs a pos was rec
now = datetime.now() now = datetime.now()
@ -131,9 +142,6 @@ class Plane:
#print("\033[H\033[J") #print("\033[H\033[J")
#Ability to Remove old Map #Ability to Remove old Map
import os import os
#Setup Geopy
from geopy.geocoders import Nominatim
geolocator = Nominatim(user_agent="NotifyBot", timeout=5)
import time import time
from colorama import Fore, Back, Style from colorama import Fore, Back, Style
#Platform for determining OS for strftime #Platform for determining OS for strftime
@ -179,13 +187,14 @@ class Plane:
[(Fore.CYAN + "ICAO" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.icao + Style.RESET_ALL)], [(Fore.CYAN + "ICAO" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.icao + Style.RESET_ALL)],
[(Fore.CYAN + "Callsign" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.callsign + Style.RESET_ALL)] if self.callsign != None else None, [(Fore.CYAN + "Callsign" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.callsign + Style.RESET_ALL)] if self.callsign != None else None,
[(Fore.CYAN + "Reg" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.reg + Style.RESET_ALL)] if "reg" in self.__dict__ else None, [(Fore.CYAN + "Reg" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.reg + Style.RESET_ALL)] if "reg" in self.__dict__ else None,
[(Fore.CYAN + "Squawk" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.squawk + Style.RESET_ALL)] if "squawk" in self.__dict__ else None, [(Fore.CYAN + "Squawk" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.squawk + Style.RESET_ALL)] if "squawk" in self.__dict__ and self.squawk != None else None,
[(Fore.CYAN + "Latitude" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(self.latitude) + Style.RESET_ALL)], [(Fore.CYAN + "Latitude" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(self.latitude) + Style.RESET_ALL)],
[(Fore.CYAN + "Longitude" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(self.longitude) + Style.RESET_ALL)], [(Fore.CYAN + "Longitude" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(self.longitude) + Style.RESET_ALL)],
[(Fore.CYAN + "Last Contact" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(time_since_contact).split(".")[0]+ Style.RESET_ALL)], [(Fore.CYAN + "Last Contact" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(time_since_contact).split(".")[0]+ Style.RESET_ALL)],
[(Fore.CYAN + "On Ground" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(self.on_ground) + Style.RESET_ALL)], [(Fore.CYAN + "On Ground" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(self.on_ground) + Style.RESET_ALL)],
[(Fore.CYAN + "GEO Alitude" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str("{:,} ft".format(self.geo_alt_ft)) + Style.RESET_ALL)], [(Fore.CYAN + "GEO Alitude" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str("{:,} ft".format(self.geo_alt_ft)) + Style.RESET_ALL)],
[(Fore.CYAN + "Nav Modes" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + ', '.join(self.nav_modes) + Style.RESET_ALL)] if "nav_modes" in self.__dict__ and self.nav_modes != None else None, [(Fore.CYAN + "Nav Modes" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + ', '.join(self.nav_modes) + Style.RESET_ALL)] if "nav_modes" in self.__dict__ and self.nav_modes != None else None,
[(Fore.CYAN + "Sel Alt Ft" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str("{:,} ft".format(self.nav_altitude)) + Style.RESET_ALL)] if "nav_altitude" in self.__dict__ and self.nav_altitude != None else None
] ]
output = list(filter(None, output)) output = list(filter(None, output))
print(tabulate(output, [], 'fancy_grid')) print(tabulate(output, [], 'fancy_grid'))
@ -243,58 +252,28 @@ class Plane:
#Lookup Location of coordinates #Lookup Location of coordinates
if self.landed or self.tookoff: if self.landed or self.tookoff:
if self.trigger_type == "now on ground" or "data acquisition" and self.longitude != None and self.latitude != None: if self.trigger_type == "now on ground" or "data acquisition" and self.longitude != None and self.latitude != None:
combined = f"{self.latitude} , {self.longitude}"
nearest_airport_dict = getClosestAirport(self.latitude, self.longitude, self.config.get("AIRPORT", "TYPES")) nearest_airport_dict = getClosestAirport(self.latitude, self.longitude, self.config.get("AIRPORT", "TYPES"))
has_coords = True has_coords = True
elif self.trigger_type == "data loss" or "no longer on ground" and self.last_longitude != None and self.last_latitude != None: elif self.trigger_type == "data loss" or "no longer on ground" and self.last_longitude != None and self.last_latitude != None:
combined = f"{self.last_latitude}, {self.last_longitude}"
nearest_airport_dict = getClosestAirport(self.last_latitude, self.last_longitude, self.config.get("AIRPORT", "TYPES")) nearest_airport_dict = getClosestAirport(self.last_latitude, self.last_longitude, self.config.get("AIRPORT", "TYPES"))
has_coords = True has_coords = True
else: else:
print (Fore.RED + 'No Location, No coordinates') print (Fore.RED + 'No Location, No coordinates')
invalid_Location = True has_coords = False
print(Style.RESET_ALL) print(Style.RESET_ALL)
if has_coords: if has_coords:
try: #Convert dictionary keys to sep variables
location = geolocator.reverse(combined) country_code = nearest_airport_dict['iso_country']
except BaseException: state = nearest_airport_dict['region']
print ("Geopy API Error") municipality = nearest_airport_dict['municipality']
else:
# print (Fore.YELLOW, "Geopy debug: ", location.raw, Style.RESET_ALL)
#Checking for invalid location, where GeoPy doesn't find coordinates info
try:
geoError = location.raw['error']
except KeyError:
invalid_Location = False
#Convert Full address to sep variables only if Valid Location
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: ", self.address)
# print(Style.RESET_ALL)
print (Fore.GREEN) print (Fore.GREEN)
print("Entire Address: ", location.address)
print ("Country Code: ", country_code) print ("Country Code: ", country_code)
print ("Country: ", country)
print ("State: ", state) print ("State: ", state)
print ("City: ", city) print ("Municipality: ", municipality)
print ("Town: ", town)
print ("Hamlet: ", hamlet)
print ("County: ", county)
print(Style.RESET_ALL) print(Style.RESET_ALL)
#Chose city town county or hamlet for location as not all are always avalible.
aera_hierarchy = city or town or county or hamlet
else: else:
invalid_Location = True
print (Fore.RED) print (Fore.RED)
print (geoError) print ("Invalid Location")
print ("Invalid Location, Likely Over Water ")
print(Style.RESET_ALL) print(Style.RESET_ALL)
#Set Discord Title #Set Discord Title
@ -318,18 +297,18 @@ class Plane:
self.landed_time = None self.landed_time = None
elif self.landed: elif self.landed:
self.landed_time_msg = None self.landed_time_msg = None
if invalid_Location is False: if has_coords:
message = (self.type_header + aera_hierarchy + ", " + state + ", " + country_code + ".") + ((" " + self.landed_time_msg) if self.landed_time_msg != None else "") message = (self.type_header + municipality + ", " + state + ", " + country_code + ".") + ((" " + self.landed_time_msg) if self.landed_time_msg != None else "")
else: else:
message = ("Landed" + ((" " + self.landed_time_msg) if self.landed_time_msg != None else "") if self.landed else "Tookoff" if self.tookoff else "") message = ("Landed" + ((" " + self.landed_time_msg) if self.landed_time_msg != None else "") if self.landed else "Tookoff" if self.tookoff else "")
print (message) print (message)
#Google Map or tar1090 screenshot #Google Map or tar1090 screenshot
if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP": if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP":
getMap((aera_hierarchy + ", " + state + ", " + country_code), self.icao) getMap((municipality + ", " + state + ", " + country_code), self.icao)
elif self.config.get('MAP', 'OPTION') == "ADSBX": elif self.config.get('MAP', 'OPTION') == "ADSBX":
getSS(self.icao, self.overlays) getSS(self.icao, self.overlays)
if nearest_airport_dict != None: if nearest_airport_dict != None:
append_airport(self.map_file_name, nearest_airport_dict['icao'], nearest_airport_dict['name'], nearest_airport_dict['distance']) append_airport(self.map_file_name, nearest_airport_dict['icao'], nearest_airport_dict['name'], nearest_airport_dict['distance_mi'])
airport_string = nearest_airport_dict['icao'] + ", " + nearest_airport_dict["name"] airport_string = nearest_airport_dict['icao'] + ", " + nearest_airport_dict["name"]
else: else:
airport_string = "" airport_string = ""
@ -338,11 +317,11 @@ class Plane:
#Discord #Discord
if self.config.getboolean('DISCORD', 'ENABLE'): if self.config.getboolean('DISCORD', 'ENABLE'):
dis_message = (self.dis_title + " " + message + " " + airport_string).strip() dis_message = (self.dis_title + " " + message + " " + airport_string).strip()
sendDis(dis_message, self.map_file_name, self.config) sendDis(dis_message, self.config, self.map_file_name)
#PushBullet #PushBullet
if self.config.getboolean('PUSHBULLET', 'ENABLE'): if self.config.getboolean('PUSHBULLET', 'ENABLE'):
with open(self.map_file_name, "rb") as pic: with open(self.map_file_name, "rb") as pic:
map_data = self.pb.upload_file(pic, "Tookoff IMG") map_data = self.pb.upload_file(pic, "Tookoff IMG" if self.tookoff else "Landed IMG")
self.pb_channel.push_note(self.config.get('PUSHBULLET', 'TITLE'), message) self.pb_channel.push_note(self.config.get('PUSHBULLET', 'TITLE'), message)
self.pb_channel.push_file(**map_data) self.pb_channel.push_file(**map_data)
#Twitter #Twitter
@ -369,21 +348,23 @@ class Plane:
#Discord #Discord
if self.config.getboolean('DISCORD', 'ENABLE'): if self.config.getboolean('DISCORD', 'ENABLE'):
dis_message = (self.dis_title + " " + squawk_message) dis_message = (self.dis_title + " " + squawk_message)
sendDis(dis_message, self.map_file_name, self.config) sendDis(dis_message, self.config, self.map_file_name)
os.remove(self.map_file_name) os.remove(self.map_file_name)
#Nav Modes Notifications #Nav Modes Notifications
if self.nav_modes != None and self.last_nav_modes != None: if self.nav_modes != None and self.last_nav_modes != None:
new_modes = []
for mode in self.nav_modes: for mode in self.nav_modes:
if mode not in self.last_nav_modes: if mode not in self.last_nav_modes:
new_modes.append(mode)
#Discord #Discord
if new_modes != []: print(mode, "enabled")
modes_string = ", ".join(new_modes)
if self.config.getboolean('DISCORD', 'ENABLE'): if self.config.getboolean('DISCORD', 'ENABLE'):
dis_message = (self.dis_title + " " + modes_string + " mode enabled.").strip() dis_message = (self.dis_title + " " + mode + " mode enabled.")
if mode == "Approach":
getSS(self.icao, self.overlays) getSS(self.icao, self.overlays)
sendDis(dis_message, self.map_file_name, self.config) sendDis(dis_message, self.config, self.map_file_name)
elif mode == "Althold" and self.nav_altitude != None:
sendDis((dis_message + ", Sel Alt. " + str(self.nav_altitude) + ", Current Alt. " + str(self.geo_alt_ft)), self.config)
else:
sendDis(dis_message, self.config)
#Set Variables to compare to next check #Set Variables to compare to next check