diff --git a/__main__.py b/__main__.py index ae9e8a2..0177dae 100644 --- a/__main__.py +++ b/__main__.py @@ -7,7 +7,7 @@ if platform.system() == "Windows": from colorama import init init(convert=True) elif platform.system() == "Linux": - pid_file_path = "/var/run/plane-notify/plane-notify.pid" + pid_file_path = "/home/plane-notify/pid.pid" def write_pid_file(filepath): import os pid = str(os.getpid()) diff --git a/configs/mainconf.ini b/configs/mainconf.ini index 8f28eed..0fa2fd7 100644 --- a/configs/mainconf.ini +++ b/configs/mainconf.ini @@ -4,7 +4,8 @@ #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 - +#Default amount of time after data loss to trigger a landing when under 10k ft +DATA_LOSS_MINS = 5 #Timezone if you want your own time to show in the console, if invalid will be set to UTC. #List of TZs names https://en.wikipedia.org/wiki/List_of_tz_database_time_zones TZ = UTC diff --git a/configs/plane1.ini b/configs/plane1.ini index eb44f51..9e77914 100644 --- a/configs/plane1.ini +++ b/configs/plane1.ini @@ -2,6 +2,9 @@ #Plane to track, based of ICAO or ICAO24 which is the unique transponder address of a plane. ICAO = icaohere +#Optional Per Plane Override +#DATA_LOSS_MINS = 20 + [MAP] #Map to create from Google Static Maps or screenshot global tar1090 from globe.adsbexchange.com #Enter GOOGLESTATICMAP or ADSBX @@ -36,4 +39,4 @@ URL = webhookurl #Role to tag optional, the role ID ROLE_ID = Title = -USERNAME = plane-notify +USERNAME = plane-notify \ No newline at end of file diff --git a/defADSBX.py b/defADSBX.py index c833bd6..f2f1efd 100644 --- a/defADSBX.py +++ b/defADSBX.py @@ -13,6 +13,7 @@ api_version = main_config.get('ADSBX', 'API_VERSION') def pull(url, headers): try: response = requests.get(url, headers = headers) + print ("HTTP Status Code:", response.status_code) response.raise_for_status() except (requests.HTTPError, ConnectionError, requests.Timeout, urllib3.exceptions.ConnectionError) as error_message: print("Basic Connection Error") @@ -26,8 +27,6 @@ def pull(url, headers): print("Connection Error uncaught, basic exception for all") print(error_message) response = None - if "response" in locals(): - print ("HTTP Status Code:", response.status_code) return response def pull_adsbx(planes): @@ -68,12 +67,12 @@ def pull_adsbx(planes): else: if "msg" in data.keys() and data['msg'] != "No error": raise ValueError("Error from ADSBX: msg = ", data['msg']) - if "ctime" in data.keys(): - data_ctime = float(data['ctime']) / 1000.0 - print("Data ctime:",datetime.utcfromtimestamp(data_ctime)) - if "now" in data.keys(): - data_now = float(data['now']) / 1000.0 - print("Data now time:",datetime.utcfromtimestamp(data_now)) + if "ctime" in data.keys(): + data_ctime = float(data['ctime']) / 1000.0 + print("Data ctime:",datetime.utcfromtimestamp(data_ctime)) + if "now" in data.keys(): + data_now = float(data['now']) / 1000.0 + print("Data now time:",datetime.utcfromtimestamp(data_now)) print("Current UTC:", datetime.utcnow()) else: data = None diff --git a/defAirport.py b/defAirport.py index 1c29117..ed64da8 100644 --- a/defAirport.py +++ b/defAirport.py @@ -1,5 +1,14 @@ +import csv +import math +def add_airport_region(airport_dict): + #Get full region/state name from iso region name + with open('./dependencies/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'] == airport_dict['iso_region']: + airport_dict['region'] = region['name'] + return airport_dict def getClosestAirport(latitude, longitude, allowed_types): - import csv from geopy.distance import geodesic plane = (latitude, longitude) with open('./dependencies/airports.csv', 'r', encoding='utf-8') as airport_csv: @@ -17,15 +26,9 @@ def getClosestAirport(latitude, longitude, allowed_types): closest_airport_dict['distance_mi'] = closest_airport_dist #Convert indent key to icao key as its labeled icao in other places not ident closest_airport_dict['icao'] = closest_airport_dict.pop('gps_code') - #Get full region/state name from iso region name - with open('./dependencies/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'] + closest_airport_dict = add_airport_region(closest_airport_dict) return closest_airport_dict def get_airport_by_icao(icao): - import csv with open('./dependencies/airports.csv', 'r', encoding='utf-8') as airport_csv: airport_csv_reader = csv.DictReader(filter(lambda row: row[0]!='#', airport_csv)) for airport in airport_csv_reader: @@ -34,4 +37,5 @@ def get_airport_by_icao(icao): #Convert indent key to icao key as its labeled icao in other places not ident matching_airport['icao'] = matching_airport.pop('gps_code') break + matching_airport = add_airport_region(matching_airport) return matching_airport \ No newline at end of file diff --git a/defSS.py b/defSS.py index a7ce226..e1873b1 100644 --- a/defSS.py +++ b/defSS.py @@ -56,6 +56,4 @@ def generate_adsbx_screenshot_time_params(timestamp): print(timestamp_dt) start_time = timestamp_dt - timedelta(minutes=1) time_params = "&showTrace=" + timestamp_dt.strftime("%Y-%m-%d") + "&startTime=" + start_time.strftime("%H:%M:%S") + "&endTime=" + timestamp_dt.strftime("%H:%M:%S") - return time_params -def generate_adsbx_overlay_param(overlays): - return "&overlays=" + ",".join(overlays) \ No newline at end of file + return time_params \ No newline at end of file diff --git a/planeClass.py b/planeClass.py index 7a5d611..b526ab9 100644 --- a/planeClass.py +++ b/planeClass.py @@ -1,5 +1,8 @@ -from datetime import datetime +from datetime import datetime, timedelta class Plane: + import configparser + main_config = configparser.ConfigParser() + main_config.read('./configs/mainconf.ini') def __init__(self, icao, config_path, config): """Initializes a plane object from its config file and given icao.""" self.icao = icao.upper() @@ -8,7 +11,6 @@ class Plane: self.config = config self.conf_file_path = config_path self.alt_ft = None - self.last_alt_ft = None self.below_desired_ft = None self.last_below_desired_ft = None self.feeding = None @@ -19,18 +21,27 @@ class Plane: self.latitude = None self.takeoff_time = None import tempfile - self.map_file_name = f"{tempfile.gettempdir()}\\{icao.upper()}_map.png" + self.map_file_name = f"{tempfile.gettempdir()}/{icao.upper()}_map.png" self.last_latitude = None self.last_longitude = None - self.last_contact = None + self.last_pos_datetime = None self.landing_plausible = False - self.squawks = [None, None, None, None] self.nav_modes = None self.last_nav_modes = None self.recheck_to = None self.speed = None self.nearest_airport_dict = None self.recent_ra_types = {} + self.db_flags = None + self.sel_nav_alt = None + self.last_sel_alt = None + self.squawk = None + self.emergency_already_triggered = None + self.last_emergency = None + if self.config.has_option('DATA', 'DATA_LOSS_MINS'): + self.data_loss_mins = self.config.getint('DATA', 'DATA_LOSS_MINS') + else: + self.data_loss_mins = Plane.main_config.getint('DATA', 'DATA_LOSS_MINS') #Setup Tweepy if self.config.getboolean('TWITTER', 'ENABLE'): from defTweet import tweepysetup @@ -51,11 +62,6 @@ class Plane: self.alt_ft = round(float(ac_dict.baro_altitude) * 3.281) elif self.on_ground: self.alt_ft = 0 - #Insert newest sqwauk at 0, sqwuak length should be 4 long 0-3 - self.squawks.insert(0, ac_dict.squawk) - #Removes oldest sqwauk index 4 5th sqwauk - if len(self.squawks) == 5: - self.squawks.pop(4) except ValueError as e: print("Got data but some data is invalid!") print(e) @@ -73,11 +79,6 @@ class Plane: self.__dict__.update({'icao' : ac_dict['icao'].upper(), 'callsign' : ac_dict['call'], 'reg' : ac_dict['reg'], 'latitude' : float(ac_dict['lat']), 'longitude' : float(ac_dict['lon']), 'alt_ft' : int(ac_dict['alt']), 'on_ground' : bool(int(ac_dict["gnd"])), 'last_contact' : round(float(ac_dict["postime"])/1000)}) if self.on_ground: self.alt_ft = 0 - #Insert newest sqwauk at 0, sqwuak length should be 4 long 0-3 - self.squawks.insert(0, ac_dict.get('sqk')) - #Removes oldest sqwauk index 4 5th sqwauk - if len(self.squawks) == 5: - self.squawks.pop(4) except ValueError as e: print("Got data but some data is invalid!") @@ -92,6 +93,7 @@ class Plane: #Parse ADBSX V2 Vector from colorama import Fore, Back, Style self.printheader("head") + print(ac_dict) try: self.__dict__.update({'icao' : ac_dict['hex'].upper(), 'latitude' : float(ac_dict['lat']), 'longitude' : float(ac_dict['lon']), 'speed': ac_dict['gs']}) if "r" in ac_dict: @@ -106,6 +108,8 @@ class Plane: self.on_ground = True if ac_dict.get('flight') is not None: self.callsign = ac_dict.get('flight').strip() + if ac_dict.get('dbFlags') is not None: + self.db_flags = ac_dict['dbFlags'] if 'nav_modes' in ac_dict: self.nav_modes = ac_dict['nav_modes'] for idx, mode in enumerate(self.nav_modes): @@ -113,23 +117,16 @@ class Plane: self.nav_modes[idx] = self.nav_modes[idx].upper() else: self.nav_modes[idx] = self.nav_modes[idx].capitalize() - #Insert newest sqwauk at 0, sqwuak length should be 4 long 0-3 - self.squawks.insert(0, ac_dict.get('squawk')) - #Removes oldest sqwauk index 4 5th sqwauk - if len(self.squawks) == 5: - self.squawks.pop(4) + self.squawk = ac_dict.get('squawk') 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'] + self.sel_nav_alt = ac_dict['nav_altitude_fms'] + elif "nav_altitude_mcp" in ac_dict: + self.sel_nav_alt = ac_dict['nav_altitude_mcp'] else: - self.nav_altitude = None + self.sel_nav_alt = None - from datetime import datetime, timedelta #Create last seen timestamp from how long ago in secs a pos was rec - now = datetime.now() - last_pos_datetime = now - timedelta(seconds= ac_dict["seen_pos"]) - self.last_contact = datetime.timestamp(last_pos_datetime) + self.last_pos_datetime = datetime.now() - timedelta(seconds= ac_dict["seen_pos"]) except (ValueError, KeyError) as e: print("Got data but some data is invalid!") @@ -146,13 +143,41 @@ class Plane: elif type == "foot": header = "----------------------------------------------------------------------------------------------------" print(Back.MAGENTA + header[0:100] + Style.RESET_ALL) - def get_time_since(self, last_contact): - if last_contact != None: - last_contact_dt = datetime.fromtimestamp(last_contact) - time_since_contact = datetime.now() - last_contact_dt + def get_time_since(self, datetime_obj): + if datetime_obj != None: + time_since = datetime.now() - datetime_obj else: - time_since_contact = None - return time_since_contact + time_since = None + return time_since + def get_adsbx_map_overlays(self): + if self.config.has_option('MAP', 'OVERLAYS'): + overlays = self.config.get('MAP', 'OVERLAYS') + else: + overlays = "" + return overlays + def route_info(self): + from lookup_route import lookup_route + extra_route_info = lookup_route(self.reg, (self.latitude, self.longitude), self.type, self.alt_ft) + if extra_route_info is None and not self.recheck_to: + self.recheck_to = True + route_to = None + elif extra_route_info is not None: + from defAirport import get_airport_by_icao + to_airport = get_airport_by_icao(extra_route_info['apdstic']) + code = to_airport['iata_code'] if to_airport['iata_code'] != "" else to_airport['icao'] + airport_text = f"{code}, {to_airport['name']}" + if 'arrivalRelative' in extra_route_info.keys() and "In" in extra_route_info['arrivalRelative']: + arrival_rel = "in ~" + extra_route_info['arrivalRelative'].strip("In ") + else: + arrival_rel = None + if extra_route_info['apdstic'] != self.nearest_airport_dict['icao']: + area = f"{to_airport['municipality']}, {to_airport['region']}, {to_airport['iso_country']}" + route_to = f"Going to {area} ({airport_text})" + f" arriving {arrival_rel}" if arrival_rel is not None else "" + else: + route_to = f"Will be returning to {airport_text}" + f" {arrival_rel}" if arrival_rel is not None else "" + else: + route_to = None + return route_to def run_empty(self): self.printheader("head") self.feeding = False @@ -163,29 +188,16 @@ class Plane: import os from colorama import Fore, Style from tabulate import tabulate - from modify_image import append_airport - from defAirport import getClosestAirport - #Proprietary Route Lookup - if os.path.isfile("lookup_route.py"): + if os.path.isfile("lookup_route.py") and (self.db_flags is None or not self.db_flags & 1): from lookup_route import lookup_route ENABLE_ROUTE_LOOKUP = True else: ENABLE_ROUTE_LOOKUP = False - if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP": - from defMap import getMap - elif self.config.get('MAP', 'OPTION') == "ADSBX": - from defSS import get_adsbx_screenshot - if self.config.has_option('MAP', 'OVERLAYS'): - self.overlays = self.config.get('MAP', 'OVERLAYS') - else: - self.overlays = "" - else: - raise ValueError("Map option not set correctly in this planes conf") if self.config.getboolean('DISCORD', 'ENABLE'): from defDiscord import sendDis if self.feeding == False: - time_since_contact = self.get_time_since(self.last_contact) + time_since_contact = self.get_time_since(self.last_pos_datetime) 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 @@ -194,19 +206,19 @@ class Plane: print(tabulate(output, [], 'fancy_grid')) print("No Data") elif self.feeding == True: - time_since_contact = self.get_time_since(self.last_contact) + time_since_contact = self.get_time_since(self.last_pos_datetime) 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)] if self.callsign != None else None, [(Fore.CYAN + "Reg" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.reg + Style.RESET_ALL)] if self.reg != None else None, #Squawks are latest to oldest - [(Fore.CYAN + "Squawks" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + ', '.join("NA" if x == None else x for x in self.squawks) + Style.RESET_ALL)], + [(Fore.CYAN + "Squawk" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + (self.squawk if self.squawk is not None else "NA") + Style.RESET_ALL)], [(Fore.CYAN + "Coordinates" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(self.latitude) + ", " + 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 + "On Ground" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(self.on_ground) + Style.RESET_ALL)], [(Fore.CYAN + "Baro Altitude" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str("{:,} ft".format(self.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 + "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 + [(Fore.CYAN + "Sel Alt Ft" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str("{:,} ft".format(self.sel_nav_alt)) + Style.RESET_ALL)] if "sel_nav_alt" in self.__dict__ and self.sel_nav_alt != None else None ] output = list(filter(None, output)) print(tabulate(output, [], 'fancy_grid')) @@ -221,9 +233,10 @@ class Plane: 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" + trigger_type = "no longer on ground" type_header = "Took off from" elif self.last_feeding is False and self.feeding and self.landing_plausible == False: + from defAirport import getClosestAirport nearest_airport_dict = getClosestAirport(self.latitude, self.longitude, self.config.get("AIRPORT", "TYPES")) if nearest_airport_dict['elevation_ft'] != "": alt_above_airport = (self.alt_ft - int(nearest_airport_dict['elevation_ft'])) @@ -232,7 +245,7 @@ class Plane: alt_above_airport = None if (alt_above_airport != None and alt_above_airport <= 10000) or self.alt_ft <= 15000: self.tookoff = True - self.trigger_type = "data acquisition" + trigger_type = "data acquisition" type_header = "Took off near" else: self.tookoff = False @@ -242,15 +255,16 @@ class Plane: #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" + trigger_type = "now on ground" type_header = "Landed in" self.landing_plausible = False #Set status for landing plausible elif self.below_desired_ft and self.last_feeding and self.feeding is False and self.last_on_ground is False: self.landing_plausible = True - print("Near landing conditions, if contiuned data loss for 5 mins, and if under 10k AGL landing true") + print("Near landing conditions, if contiuned data loss for configured time, and if under 10k AGL landing true") - elif self.landing_plausible and self.feeding is False and time_since_contact.total_seconds() >= 300: + elif self.landing_plausible and self.feeding is False and time_since_contact.total_seconds() >= (self.data_loss_mins * 60): + from defAirport import getClosestAirport nearest_airport_dict = getClosestAirport(self.latitude, self.longitude, self.config.get("AIRPORT", "TYPES")) if nearest_airport_dict['elevation_ft'] != "": alt_above_airport = (self.alt_ft - int(nearest_airport_dict['elevation_ft'])) @@ -260,7 +274,7 @@ class Plane: if (alt_above_airport != None and alt_above_airport <= 10000) or self.alt_ft <= 15000: self.landing_plausible = False self.landed = True - self.trigger_type = "data loss" + trigger_type = "data loss" type_header = "Landed near" else: print("Alt greater then 10k AGL") @@ -269,16 +283,17 @@ class Plane: self.landed = False if self.landed: - print ("Landed by", self.trigger_type) + print ("Landed by", trigger_type) if self.tookoff: - print("Tookoff by", self.trigger_type) + print("Tookoff by", trigger_type) #Find nearest airport, and location if self.landed or self.tookoff: + from defAirport import getClosestAirport if "nearest_airport_dict" in globals(): pass #Airport already set - elif self.trigger_type in ["now on ground", "data acquisition", "data loss"]: + elif trigger_type in ["now on ground", "data acquisition", "data loss"]: nearest_airport_dict = getClosestAirport(self.latitude, self.longitude, self.config.get("AIRPORT", "TYPES")) - elif self.trigger_type == "no longer on ground": + elif trigger_type == "no longer on ground": nearest_airport_dict = getClosestAirport(self.last_latitude, self.last_longitude, self.config.get("AIRPORT", "TYPES")) #Convert dictionary keys to sep variables country_code = nearest_airport_dict['iso_country'] @@ -314,21 +329,12 @@ class Plane: landed_time_msg = None #Proprietary Route Lookup if ENABLE_ROUTE_LOOKUP: - extra_route_info = lookup_route(self.reg, (self.latitude, self.longitude), self.type, self.alt_ft) - if extra_route_info == None: - self.recheck_to = True - self.nearest_takeoff_airport = nearest_airport_dict - else: - from defAirport import get_airport_by_icao - to_airport = get_airport_by_icao(extra_route_info['apdstic']) - code = to_airport['iata_code'] if to_airport['iata_code'] != "" else to_airport['icao'] - airport_text = f" {code}, {to_airport['name']}" - if extra_route_info['apdstic'] != nearest_airport_dict['icao']: - route_to = "Going to" + airport_text + " arriving " + extra_route_info['arrivalRelative'] - else: - route_to = f"Will be returning to {airport_text} {extra_route_info['arrivalRelative']}" + self.nearest_airport_dict = nearest_airport_dict + route_to = self.route_info() elif self.landed and self.takeoff_time != None: landed_time = datetime.utcnow() - self.takeoff_time + if trigger_type == "data loss": + landed_time -= timedelta(seconds=time_since_contact.total_seconds()) hours, remainder = divmod(landed_time.total_seconds(), 3600) minutes, seconds = divmod(remainder, 60) min_syntax = "Mins" if minutes > 1 else "Min" @@ -344,10 +350,14 @@ class Plane: print (message) #Google Map or tar1090 screenshot if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP": + from defMap import getMap getMap((municipality + ", " + state + ", " + country_code), self.map_file_name) elif self.config.get('MAP', 'OPTION') == "ADSBX": - url_params = f"icao={self.icao}&zoom=9&largeMode=2&hideButtons&hideSidebar&mapDim=0&overlays=" + self.overlays + from defSS import get_adsbx_screenshot + + url_params = f"icao={self.icao}&zoom=9&largeMode=2&hideButtons&hideSidebar&mapDim=0&overlays=" + self.get_adsbx_map_overlays() get_adsbx_screenshot(self.map_file_name, url_params) + from modify_image import append_airport append_airport(self.map_file_name, nearest_airport_dict) else: raise ValueError("Map option not set correctly in this planes conf") @@ -365,25 +375,16 @@ class Plane: #Twitter if self.config.getboolean('TWITTER', 'ENABLE'): twitter_media_map_obj = self.tweet_api.media_upload(self.map_file_name) - alt_text = f"Reg: {self.reg} On Ground: {str(self.on_ground)} Alt: {str(self.alt_ft)} Last Contact: {str(time_since_contact)} Trigger: {self.trigger_type}" + alt_text = f"Reg: {self.reg} On Ground: {str(self.on_ground)} Alt: {str(self.alt_ft)} Last Contact: {str(time_since_contact)} Trigger: {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 + " " + message).strip()), media_ids=[twitter_media_map_obj.media_id]) os.remove(self.map_file_name) #Recheck Proprietary Route Lookup a minute later if infomation was not available on takeoff. if self.recheck_to and self.takeoff_time is not None and (datetime.utcnow() - self.takeoff_time).total_seconds() > 60: + route_to = self.route_info() self.recheck_to = False - extra_route_info = lookup_route(self.reg, (self.latitude, self.longitude), self.type, self.alt_ft) - nearest_airport_dict = self.nearest_takeoff_airport self.nearest_takeoff_airport = None - if extra_route_info != None: - from defAirport import get_airport_by_icao - to_airport = get_airport_by_icao(extra_route_info['apdstic']) - code = to_airport['iata_code'] if to_airport['iata_code'] != "" else to_airport['icao'] - airport_text = f" {code}, {to_airport['name']}" - if extra_route_info['apdstic'] != nearest_airport_dict['icao']: - route_to = "Going to" + airport_text + " arriving " + extra_route_info['arrivalRelative'] - else: - route_to = f"Will be returning to {airport_text} {extra_route_info['arrivalRelative']}" + if route_to != None: print(route_to) #Discord if self.config.getboolean('DISCORD', 'ENABLE'): @@ -397,22 +398,34 @@ class Plane: if self.feeding: #Squawks - squawks =[("7500", "Hijacking"), ("7600", "Radio Failure"), ("7700", "Emergency")] - for squawk in squawks: - if all(v == squawk[0] for v in (self.squawks[0:2])) and self.squawks[2] != self.squawks[3] and None not in self.squawks: - squawk_message = ("Squawking " + squawk[0] + ", " + squawk[1]) + emergency_squawks ={"7500" : "Hijacking", "7600" :"Radio Failure", "7700" : "General Emergency"} + seen = datetime.now() - self.last_pos_datetime + #Only run check if emergency data previously set + if self.last_emergency is not None and not self.emergency_already_triggered: + time_since_org_emer = datetime.now() - self.last_emergency[0] + #Checks times to see x time and still same squawk + if time_since_org_emer.total_seconds() >= 60 and self.last_emergency[1] == self.squawk and seen.total_seconds() <= 60: + self.emergency_already_triggered = True + squawk_message = (f"{self.dis_title} Squawking {self.last_emergency[1]} {emergency_squawks[self.squawk]}").strip() print(squawk_message) #Google Map or tar1090 screenshot if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP": getMap((municipality + ", " + state + ", " + country_code), self.map_file_name) if self.config.get('MAP', 'OPTION') == "ADSBX": - url_params = f"icao={self.icao}&zoom=9&largeMode=2&hideButtons&hideSidebar&mapDim=0&overlays=" + self.overlays + from defSS import get_adsbx_screenshot + url_params = f"icao={self.icao}&zoom=9&largeMode=2&hideButtons&hideSidebar&mapDim=0&overlays=" + self.get_adsbx_map_overlays() get_adsbx_screenshot(self.map_file_name, url_params) - #Discord if self.config.getboolean('DISCORD', 'ENABLE'): dis_message = (self.dis_title + " " + squawk_message) sendDis(dis_message, self.config, self.map_file_name) os.remove(self.map_file_name) + #Realizes first time seeing emergency, stores time and type + elif self.squawk in emergency_squawks.keys() and not self.emergency_already_triggered and not self.on_ground: + print("Emergency", self.squawk, "detected storing code and time and waiting to trigger") + self.last_emergency = (self.last_pos_datetime, self.squawk) + elif self.squawk not in emergency_squawks.keys() and self.emergency_already_triggered: + self.emergency_already_triggered = None + #Nav Modes Notifications if self.nav_modes != None and self.last_nav_modes != None: for mode in self.nav_modes: @@ -422,13 +435,21 @@ class Plane: if self.config.getboolean('DISCORD', 'ENABLE'): dis_message = (self.dis_title + " " + mode + " mode enabled.") if mode == "Approach": - url_params = f"icao={self.icao}&zoom=9&largeMode=2&hideButtons&hideSidebar&mapDim=0&overlays={self.overlays}" + from defSS import get_adsbx_screenshot + url_params = f"icao={self.icao}&zoom=9&largeMode=2&hideButtons&hideSidebar&mapDim=0&overlays={self.get_adsbx_map_overlays()}" get_adsbx_screenshot(self.map_file_name, url_params) sendDis(dis_message, self.config, self.map_file_name) - elif mode in ["Althold", "VNAV", "LNAV"] and self.nav_altitude != None: - sendDis((dis_message + ", Sel Alt. " + str(self.nav_altitude) + ", Current Alt. " + str(self.alt_ft)), self.config) + #elif mode in ["Althold", "VNAV", "LNAV"] and self.sel_nav_alt != None: + # sendDis((dis_message + ", Sel Alt. " + str(self.sel_nav_alt) + ", Current Alt. " + str(self.alt_ft)), self.config) else: sendDis(dis_message, self.config) + #Selected Altitude + if self.sel_nav_alt is not None and self.last_sel_alt is not None and self.last_sel_alt != self.sel_nav_alt: + #Discord + print("Nav altitude is now", self.sel_nav_alt) + if self.config.getboolean('DISCORD', 'ENABLE'): + dis_message = (self.dis_title + " Sel. alt. " + str("{:,} ft".format(self.sel_nav_alt))) + sendDis(dis_message,self.config) # #Power Up # if self.last_feeding == False and self.speed == 0 and self.on_ground: # if self.config.getboolean('DISCORD', 'ENABLE'): @@ -438,12 +459,12 @@ class Plane: #Set Variables to compare to next check self.last_feeding = self.feeding - self.last_alt_ft = self.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 self.last_nav_modes = self.nav_modes + self.last_sel_alt = self.sel_nav_alt if self.takeoff_time != None: @@ -461,8 +482,8 @@ class Plane: ra_message += f", {ra['acas_ra']['advisory_complement']}" if bool(int(ra['acas_ra']['MTE'])): ra_message += ", Multi threat" - from defSS import get_adsbx_screenshot, generate_adsbx_screenshot_time_params, generate_adsbx_overlay_param - url_params = generate_adsbx_screenshot_time_params(ra['acas_ra']['unix_timestamp']) + f"&lat={ra['lat']}&lon={ra['lon']}&zoom=11&largeMode=2&hideButtons&hideSidebar&mapDim=0&overlays={self.overlays}×tamp={ra['acas_ra']['unix_timestamp']}" + from defSS import get_adsbx_screenshot, generate_adsbx_screenshot_time_params + url_params = generate_adsbx_screenshot_time_params(ra['acas_ra']['unix_timestamp']) + f"&lat={ra['lat']}&lon={ra['lon']}&zoom=11&largeMode=2&hideButtons&hideSidebar&mapDim=0&overlays={self.get_adsbx_map_overlays()}×tamp={ra['acas_ra']['unix_timestamp']}" if "threat_id_hex" in ra['acas_ra'].keys(): from mictronics_parse import get_aircraft_by_icao threat_reg = get_aircraft_by_icao(ra['acas_ra']['threat_id_hex'])[0]