class Plane: def __init__(self, icao, conf_file): """Initializes a plane object from its config file and given icao.""" self.icao = icao.upper() self.conf_file = conf_file self.geo_alt_ft = None self.last_geo_alt_ft = None self.last_below_desired_ft = None self.feeding = None self.last_feeding = None self.last_on_ground = None self.on_ground = None self.invalid_Location = None self.longitude = None self.latitude = None self.callsign = None self.takeoff_time = None self.map_file_name = icao.upper() + "_map.png" self.last_latitude = None self.last_longitude = None self.recheck_needed = None self.last_recheck_needed = None self.last_contact = None self.last_feed_data = None def getICAO(self): return self.icao def run(self, ac_dict, source): """Runs a check of a plane module to see if its landed or takenoff using plane data, and takes action if so.""" #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="NotifyBot", timeout=5) import time from colorama import Fore, Back, Style #Setup Config File import configparser self.config = configparser.ConfigParser() self.config.read(("./configs/"+ self.conf_file)) #Platform for determining OS for strftime import platform from datetime import datetime from tabulate import tabulate from AppendAirport import append_airport from defAirport import getClosestAirport if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP": from defMap import getMap elif self.config.get('MAP', 'OPTION') == "ADSBX": from defSS import getSS else: raise Exception("Map option not set correctly in this planes conf") if self.config.getboolean('DISCORD', 'ENABLE'): from defDiscord import sendDis #Setup Tweepy if self.config.getboolean('TWITTER', 'ENABLE'): from defTweet import tweepysetup self.tweet_api = tweepysetup(self.config) #Setup PushBullet if self.config.getboolean('PUSHBULLET', 'ENABLE'): from pushbullet import Pushbullet self.pb = Pushbullet(self.config['PUSHBULLET']['API_KEY']) self.pb_channel = self.pb.get_channel(self.config.get('PUSHBULLET', 'CHANNEL_TAG')) print (Back.MAGENTA, "---------", self.conf_file, "---------------------------- ICAO:", self.icao, "--------------------------", Style.RESET_ALL) #Reset Variables self.below_desired_ft = None self.geo_alt_ft = None self.longitude = None self.latitude = None self.on_ground = None self.has_location = None time_since_contact = None #Parse OpenSky Vector if source == "OPENS": self.val_error = False if ac_dict != None: #print (Fore.YELLOW + "OpenSky Sourced Data: ", ac_dict) try: self.__dict__.update({'icao' : ac_dict.icao24.upper(), 'callsign' : ac_dict.callsign, 'latitude' : ac_dict.latitude, 'longitude' : ac_dict.longitude, 'on_ground' : bool(ac_dict.on_ground), 'last_contact' : ac_dict.last_contact}) if ac_dict.geo_altitude != None: self.geo_alt_ft = round(float(ac_dict.geo_altitude) * 3.281) elif self.on_ground: self.geo_alt_ft = 0 except ValueError as e: self.val_error = True print("Got data but some data is invalid!") print(e) #Parse ADBSX Vector elif source == "ADSBX": self.val_error = False if ac_dict != None: #print (Fore.YELLOW +"ADSBX Sourced Data: ", ac_dict, Style.RESET_ALL) try: #postime is divided by 1000 to get seconds from milliseconds, from timestamp expects secs. self.__dict__.update({'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"])), 'last_contact' : round(float(ac_dict["postime"])/1000)}) if self.on_ground: self.geo_alt_ft = 0 if "to" in ac_dict.keys(): self.to_location = ac_dict["to"] if "from" in ac_dict.keys(): self.from_location = ac_dict["from"] except ValueError as e: self.val_error = True print("Got data but some data is invalid!") print(e) #print (Fore.CYAN + "ICAO:", self.icao + Style.RESET_ALL) #Print out data, and convert to locals if self.val_error is False: if self.last_contact != None: last_contact_dt = datetime.fromtimestamp(self.last_contact) time_since_contact = datetime.now() - last_contact_dt else: time_since_contact = None if ac_dict == None: self.feeding = False output = [ [(Fore.CYAN + "ICAO" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.icao + Style.RESET_ALL)], [(Fore.CYAN + "Last Contact" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(time_since_contact) + Style.RESET_ALL)] if time_since_contact != None else None ] output = list(filter(None, output)) print(tabulate(output, [], 'fancy_grid')) print("No Data") elif ac_dict != None: self.feeding = True output = [ [(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)], [(Fore.CYAN + "Reg" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.reg + Style.RESET_ALL)] if "reg" in self.__dict__ else None, [(Fore.CYAN + "From" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.from_location + Style.RESET_ALL)] if "from_location" in self.__dict__ else None, [(Fore.CYAN + "To" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.to_location + Style.RESET_ALL)] if "to_location" in self.__dict__ else None, [(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 + "Last Contact" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(time_since_contact) + Style.RESET_ALL)], [(Fore.CYAN + "On Ground" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(self.on_ground) + Style.RESET_ALL)], [(Fore.CYAN + "GEO Alitude Ft" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(self.geo_alt_ft) + Style.RESET_ALL)] ] output = list(filter(None, output)) print(tabulate(output, [], 'fancy_grid')) #Set Check for inconsistancy in data if not self.last_recheck_needed: #Recheck needed if feeding state changes if self.feeding == False and self.last_feeding: self.recheck_needed = True print("Recheck needed, feeding status changed") else: self.recheck_needed = False elif self.last_recheck_needed: self.recheck_needed = False #Run a Check compares new data to last flagged(check) data if self.last_recheck_needed: if self.recheck_feeding == self.feeding: print("Data Feeding change consistent") elif self.recheck_feeding != self.feeding: print("Data Feeding change was inconsistent last data ignored") self.recheck_feeding = self.feeding self.last_recheck_needed = self.recheck_needed if self.recheck_needed is False: #Check if below desire ft 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 if self.below_desired_ft and self.on_ground is False: if self.last_on_ground: self.tookoff = True self.trigger_type = "no longer on ground" self.tookoff_header = "Took off from " elif self.last_feeding is False and self.feeding and self.last_feed_data == None: self.tookoff = True self.trigger_type = "data acquisition" self.tookoff_header = "Took off near " else: self.tookoff = False else: self.tookoff = False #self.tookoff = bool(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 if self.on_ground and self.last_on_ground is False and self.last_below_desired_ft: self.landed = True self.trigger_type = "now on ground" self.landed_header = "Landed in " self.last_feed_data = None #Store a dictionary when data is lost near landing conditions, elif self.last_below_desired_ft and self.last_feeding and self.feeding is False and self.last_on_ground is False: self.last_feed_data = {} self.last_feed_data.update(self.__dict__) print("Latest data stored") elif self.last_feed_data != None and self.feeding is False and time_since_contact.seconds >= 300: self.__dict__.update(self.last_feed_data) self.last_feed_data = None self.landed = True self.trigger_type = "data loss" self.landed_header = "Landed near " else: self.landed = False #self.landed = bool(self.last_below_desired_ft 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) if self.landed: print ("Landed by", self.trigger_type) if self.tookoff: print("Tookoff by", self.trigger_type) #Lookup Location of coordinates if self.landed or self.tookoff: if self.trigger_type == "now on ground" or "data acquisition" and self.longitude != None and self.latitude != None: self.combined = f"{self.latitude} , {self.longitude}" nearest_airport_dict = getClosestAirport(self.latitude, self.longitude, self.config.get("AIRPORT", "TYPES")) self.has_coords = True elif self.trigger_type == "data loss" or "no longer on ground" and self.last_longitude != None and self.last_latitude != None: self.combined = f"{self.last_latitude}, {self.last_longitude}" nearest_airport_dict = getClosestAirport(self.last_latitude, self.last_longitude, self.config.get("AIRPORT", "TYPES")) self.has_coords = True else: print (Fore.RED + 'No Location') self.has_location = False self.invalid_Location = True self.has_coords = False print(Style.RESET_ALL) if self.has_coords: try: self.location = geolocator.reverse(self.combined) except BaseException: print ("Geopy API Error") else: # print (Fore.YELLOW, "Geopy debug: ", location.raw, Style.RESET_ALL) self.has_location = True #Figure if valid location, valid being geopy finds a location if self.has_location: try: self.geoError = self.location.raw['error'] except KeyError: self.invalid_Location = False self.geoError = None else: self.invalid_Location = True print ("Invalid Location: ", self.invalid_Location) if self.invalid_Location: print (Fore.RED) 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 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: ", self.address) # print(Style.RESET_ALL) print (Fore.GREEN) 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) #Chose city town county or hamlet for location as not all are always avalible. if self.invalid_Location is False: self.aera_hierarchy = self.city or self.town or self.county or self.hamlet #Set Discord Title if self.config.getboolean('DISCORD', 'ENABLE'): self.dis_title = self.icao if self.config.get('DISCORD', 'TITLE') == "icao" else self.callsign if self.config.get('DISCORD', 'TITLE') == "callsign" else self.config.get('DISCORD', 'TITLE') #Set Twitter Title if self.config.getboolean('TWITTER', 'ENABLE'): self.twitter_title = self.icao if self.config.get('TWITTER', 'TITLE') == "icao" else self.callsign if self.config.get('TWITTER', 'TITLE') == "callsign" else self.config.get('TWITTER', 'TITLE') #Takeoff Notifcation and Landed if self.tookoff: if self.invalid_Location is False: self.tookoff_message = (self.tookoff_header + self.aera_hierarchy + ", " + self.state + ", " + self.country_code + ". ") else: self.tookoff_message = ("Took off") print (self.tookoff_message) #Google Map or tar1090 screenshot if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP": getMap((self.aera_hierarchy + ", " + self.state + ", " + self.country_code), self.icao) elif self.config.get('MAP', 'OPTION') == "ADSBX": getSS(self.icao) append_airport(self.map_file_name, nearest_airport_dict['icao'], nearest_airport_dict['name'], nearest_airport_dict['distance']) else: raise Exception("Map option not set correctly in this planes conf") #Discord if self.config.getboolean('DISCORD', 'ENABLE'): self.dis_message = (self.dis_title + " " + self.tookoff_message + nearest_airport_dict['icao'] + ", " + nearest_airport_dict["name"]).strip() sendDis(self.dis_message, self.map_file_name, self.config) #PushBullet if self.config.getboolean('PUSHBULLET', 'ENABLE'): with open(self.map_file_name, "rb") as pic: map_data = self.pb.upload_file(pic, "Tookoff IMG") self.pb_channel.push_note(self.config.get('PUSHBULLET', 'TITLE'), self.tookoff_message) self.pb_channel.push_file(**map_data) #Twitter if self.config.getboolean('TWITTER', 'ENABLE'): twitter_media_map_obj = self.tweet_api.media_upload(self.map_file_name) alt_text = "Call: " + self.callsign + " On Ground: " + str(self.on_ground) + " Alt: " + str(self.geo_alt_ft) + " Last Contact: " + str(time_since_contact) + " Trigger: " + self.trigger_type self.tweet_api.create_media_metadata(media_id= twitter_media_map_obj.media_id, alt_text= alt_text) self.tweet_api.update_status(status = ((self.twitter_title + " " + self.tookoff_message).strip()), media_ids=[twitter_media_map_obj.media_id]) #self.tweet_api.update_with_media(self.map_file_name, status = (self.twitter_title + " " + self.tookoff_message).strip()) self.takeoff_time = time.time() os.remove(self.map_file_name) if self.landed: self.landed_time_msg = "" if self.takeoff_time != None: self.landed_time = time.time() - self.takeoff_time if platform.system() == "Linux": self.landed_time_msg = time.strftime("Apx. flt. time %-H Hours : %-M Mins. ", time.gmtime(self.landed_time)) elif platform.system() == "Windows": self.landed_time_msg = time.strftime("Apx. flt. time %#H Hours : %#M Mins. ", time.gmtime(self.landed_time)) if self.invalid_Location is False: self.landed_message = (self.landed_header + self.aera_hierarchy + ", " + self.state + ", " + self.country_code + ". " + self.landed_time_msg) else: self.landed_message = ("Landed", self.landed_time_msg) print (self.landed_message) #Google Map or tar1090 screenshot if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP": getMap((self.aera_hierarchy + ", " + self.state + ", " + self.country_code), self.icao) elif self.config.get('MAP', 'OPTION') == "ADSBX": getSS(self.icao) append_airport(self.map_file_name, nearest_airport_dict['icao'], nearest_airport_dict['name'], nearest_airport_dict['distance']) else: raise Exception("Map option not set correctly in this planes conf") #Discord if self.config.getboolean('DISCORD', 'ENABLE'): self.dis_message = (self.dis_title + " " +self.landed_message + nearest_airport_dict['icao'] + ", " + nearest_airport_dict["name"]).strip() sendDis(self.dis_message, self.map_file_name, self.config) #PushBullet if self.config.getboolean('PUSHBULLET', 'ENABLE'): with open(self.map_file_name, "rb") as pic: map_data = self.pb.upload_file(pic, "Landed IMG") self.pb_channel.push_note(self.config.get('PUSHBULLET', 'TITLE'), self.landed_message) self.pb_channel.push_file(**map_data) #Twitter if self.config.getboolean('TWITTER', 'ENABLE'): twitter_media_map_obj = self.tweet_api.media_upload(self.map_file_name) alt_text = "Call: " + self.callsign + " On Ground: " + str(self.on_ground) + " Alt: " + str(self.geo_alt_ft) + " Last Contact: " + str(time_since_contact) + " Trigger: " + self.trigger_type self.tweet_api.create_media_metadata(media_id= twitter_media_map_obj.media_id, alt_text= alt_text) self.tweet_api.update_status(status = ((self.twitter_title + " " + self.landed_message).strip()), media_ids=[twitter_media_map_obj.media_id]) #self.tweet_api.update_with_media(self.map_file_name, status = (self.twitter_title + " " + self.landed_message).strip()) self.takeoff_time = None self.landed_time = None self.time_since_tk = None os.remove(self.map_file_name) self.last_contact = None #Set Variables to compare to next check 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 self.last_longitude = self.longitude self.last_latitude = self.latitude elif self.val_error: print ("Failed to Parse Will Recheck this Plane After new data") if self.takeoff_time != None: self.elapsed_time = time.time() - self.takeoff_time self.time_since_tk = time.strftime("Time Since Take off %H Hours : %M Mins : %S Secs", time.gmtime(self.elapsed_time)) print(self.time_since_tk) print (Back.MAGENTA, "---------------------------------------------------------------------------", Style.RESET_ALL)