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

Add region to going to info, new squawks, sel alt noti. 

-Move pid file on linux
-ADSBX request, fix when response is cleared and no status code to print
-Airports, common add region function
-Default and per plane configurable data loss time till land.
-New squawk system from emergency-notify
-Disable route lookup for military flags
-Removed duplication of route to message generation, created function
-Route to now puts city/region then airport name
-Seperate sel alt notifications
-Remove Unused old map overlay function
This commit is contained in:
Jack Sweeney 2021-07-13 23:50:57 -04:00
parent d53be29961
commit 12ca3ace90
7 changed files with 148 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
__main__.py
View File

@ -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())

3
configs/mainconf.ini
View File

@ -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

5
configs/plane1.ini
View File

@ -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

15
defADSBX.py
View File

@ -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

20
defAirport.py
View File

@ -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

4
defSS.py
View File

@ -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)
return time_params

221
planeClass.py
View File

@ -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}&timestamp={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()}&timestamp={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]