371 lines
19 KiB
Python
371 lines
19 KiB
Python
class Plane:
|
|
def __init__(self, icao, conf_file):
|
|
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.reg = 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
|
|
def getICAO(self):
|
|
return self.icao
|
|
def run(self, ac_dict):
|
|
|
|
#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))
|
|
main_config = configparser.ConfigParser()
|
|
main_config.read('./configs/mainconf.ini')
|
|
|
|
#Platform for determining OS for strftime
|
|
import platform
|
|
from defAirport import getAirport
|
|
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
|
|
#Parse OpenSky Vector
|
|
self.plane_Dict = None
|
|
if main_config.get('DATA', 'SOURCE') == "OPENS":
|
|
self.val_error = False
|
|
if ac_dict != None:
|
|
#print (Fore.YELLOW + "OpenSky Sourced Data: ", ac_dict)
|
|
try:
|
|
self.plane_Dict ={'icao' : ac_dict.icao24.upper(), 'callsign' : ac_dict.callsign, 'latitude' : ac_dict.latitude, 'longitude' : ac_dict.longitude, 'on_ground' : bool(ac_dict.on_ground)}
|
|
if ac_dict.geo_altitude != None:
|
|
self.plane_Dict['geo_alt_ft'] = float(ac_dict.geo_altitude) * 3.281
|
|
elif 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
|
|
|
|
#Parse ADBSX Vector
|
|
elif main_config.get('DATA', 'SOURCE') == "ADSBX":
|
|
self.val_error = False
|
|
if ac_dict != None:
|
|
#print (Fore.YELLOW +"ADSBX Sourced Data: ", ac_dict + Style.RESET_ALL)
|
|
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)
|
|
if "to" in ac_dict.keys():
|
|
self.plane_Dict['to_location'] = ac_dict["to"]
|
|
if "from" in ac_dict.keys():
|
|
self.plane_Dict['from_location'] = ac_dict["from"]
|
|
else:
|
|
self.plane_Dict = None
|
|
print (Fore.CYAN + "ICAO:", self.icao + Style.RESET_ALL)
|
|
#Print out data, and convert to locals
|
|
if self.val_error is False:
|
|
if self.plane_Dict == None:
|
|
self.feeding = False
|
|
print("No Data")
|
|
elif self.plane_Dict != None:
|
|
self.feeding = True
|
|
self.__dict__.update(self.plane_Dict)
|
|
if "reg" in self.plane_Dict.keys():
|
|
print(Fore.CYAN + "Registration: ", self.reg)
|
|
if "from_location" in self.plane_Dict.keys():
|
|
print("From: ", self.from_location)
|
|
if "to_location" in self.plane_Dict.keys():
|
|
print("To: ", self.to_location)
|
|
print (Fore.CYAN + "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, Style.RESET_ALL)
|
|
#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:
|
|
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.last_below_desired_ft:
|
|
if self.on_ground and self.last_on_ground is False:
|
|
self.landed = True
|
|
self.trigger_type = "now on ground"
|
|
self.landed_header = "Landed in "
|
|
elif self.last_feeding and self.feeding is False and self.last_on_ground is False:
|
|
self.landed = True
|
|
self.trigger_type = "data loss"
|
|
self.landed_header = "Landed near "
|
|
else:
|
|
self.landed = False
|
|
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.landed and self.last_longitude != None and self.last_latitude != None:
|
|
self.combined = f"{self.last_latitude}, {self.last_longitude}"
|
|
self.has_coords = True
|
|
elif self.tookoff and self.longitude != None and self.latitude != None:
|
|
self.combined = f"{self.latitude} , {self.longitude}"
|
|
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:
|
|
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)
|
|
else:
|
|
raise Exception("Map option not set correctly in this planes conf")
|
|
#Discord
|
|
if self.config.getboolean('DISCORD', 'ENABLE'):
|
|
nearest = getAirport(self.latitude, self.longitude)
|
|
self.dis_message = (self.dis_title + " " + self.tookoff_message + nearest['icao'] + ", " + nearest["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")
|
|
push = self.pb_channel.push_note(self.config.get('PUSHBULLET', 'TITLE'), self.tookoff_message)
|
|
push = self.pb_channel.push_file(**map_data)
|
|
#Twitter
|
|
if self.config.getboolean('TWITTER', 'ENABLE'):
|
|
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)
|
|
else:
|
|
raise Exception("Map option not set correctly in this planes conf")
|
|
#Discord
|
|
if self.config.getboolean('DISCORD', 'ENABLE'):
|
|
nearest = getAirport(self.last_latitude, self.last_longitude)
|
|
self.dis_message = (self.dis_title + " " +self.landed_message + nearest['icao'] + ", " + nearest["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")
|
|
push = self.pb_channel.push_note(self.config.get('PUSHBULLET', 'TITLE'), self.landed_message)
|
|
push = self.pb_channel.push_file(**map_data)
|
|
#Twitter
|
|
if self.config.getboolean('TWITTER', 'ENABLE'):
|
|
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)
|
|
|
|
#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)
|