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

New Resolution Advisory system, Mictronics DB, PID File 

-Makes PID file on Linux and SIGTERM detection, mess with logging
-Better DIR handling
-Downloads Micrtronics db and unzip, parsing library
-Downloads RA list of day from ADSBX and looks for new ones
-ADSBX Screenshoting, timebased, tracks, labels
This commit is contained in:
Jxck-S 2021-05-02 13:27:05 -04:00
parent e09061289b
commit 1d7bc62699
5 changed files with 204 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
__main__.py
View File

@ -6,17 +6,29 @@ import traceback
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"
def write_pid_file(filepath):
import os
pid = str(os.getpid())
f = open(filepath, 'w')
f.write(pid)
f.close()
write_pid_file(pid_file_path)
print("Made PIDFile")
from planeClass import Plane
from datetime import datetime
import pytz
import os
if 'plane-notify' not in os.getcwd():
os.chdir('./plane-notify')
import signal
abspath = os.path.abspath(__file__)
dname = os.path.dirname(abspath)
os.chdir(dname)
if not os.path.isdir("./dependencies/"):
os.mkdir("./dependencies/")
import sys
sys.path.extend([os.getcwd()])
required_files = [("Roboto-Regular.ttf", 'https://github.com/googlefonts/roboto/blob/main/src/hinted/Roboto-Regular.ttf?raw=true'), ('airports.csv', 'https://ourairports.com/data/airports.csv'), ('regions.csv', 'https://ourairports.com/data/regions.csv'), ('ADSBX_Logo.png', "https://www.adsbexchange.com/wp-content/uploads/cropped-Stealth.png")]
required_files = [("Roboto-Regular.ttf", 'https://github.com/googlefonts/roboto/blob/main/src/hinted/Roboto-Regular.ttf?raw=true'), ('airports.csv', 'https://ourairports.com/data/airports.csv'), ('regions.csv', 'https://ourairports.com/data/regions.csv'), ('ADSBX_Logo.png', "https://www.adsbexchange.com/wp-content/uploads/cropped-Stealth.png"), ('Mictronics_db.zip', "https://www.mictronics.de/aircraft-database/indexedDB.php")]
for file in required_files:
file_name = file[0]
url = file[1]
@ -33,12 +45,26 @@ for file in required_files:
print("Successfully got", file_name)
else:
print("Already have", file_name, "continuing")
if os.path.isfile("./dependencies/" + required_files[4][0]) and not os.path.isfile("./dependencies/aircrafts.json"):
print("Extracting Mictronics DB")
from zipfile import ZipFile
with ZipFile("./dependencies/" + required_files[4][0], 'r') as mictronics_db:
mictronics_db.extractall("./dependencies/")
main_config = configparser.ConfigParser()
print(os.getcwd())
main_config.read('./configs/mainconf.ini')
source = main_config.get('DATA', 'SOURCE')
if main_config.getboolean('DISCORD', 'ENABLE'):
from defDiscord import sendDis
sendDis("Started", main_config)
def service_exit(signum, frame):
if main_config.getboolean('DISCORD', 'ENABLE'):
from defDiscord import sendDis
sendDis("Service Stop", main_config)
os.remove(pid_file_path)
raise SystemExit("Service Stop")
signal.signal(signal.SIGTERM, service_exit)
try:
print("Source is set to", source)
import sys
@ -60,7 +86,7 @@ try:
tz = pytz.timezone(main_config.get('DATA', 'TZ'))
except pytz.exceptions.UnknownTimeZoneError:
tz = pytz.UTC
last_ra_count = None
while True:
datetime_tz = datetime.now(tz)
if datetime_tz.hour == 0 and datetime_tz.minute == 0:
@ -70,6 +96,38 @@ try:
header = ("-------- " + str(running_Count) + " -------- " + str(datetime_tz.strftime("%I:%M:%S %p")) + " ---------------------------------------------------------------------------")
print (Back.GREEN + Fore.BLACK + header[0:100] + Style.RESET_ALL)
if source == "ADSBX":
#ACAS data
from defADSBX import pull_date_ras
import ast
today = datetime.utcnow()
date = today.strftime("%Y/%m/%d")
ras, failed = pull_date_ras(date)
sorted_ras = {}
if failed is False and ras != None:
#Testing RAs
#if last_ra_count is not None:
# with open('./testing/acastest.json') as f:
# data = f.readlines()
# ras += data
ra_count = len(ras)
if last_ra_count is not None and ra_count != last_ra_count:
print(abs(ra_count - last_ra_count), "new Resolution Advisories")
for ra_num, ra in enumerate(ras[last_ra_count:]):
ra = ast.literal_eval(ra)
if ra['hex'].upper() in planes.keys():
if ra['hex'].upper() not in sorted_ras.keys():
sorted_ras[ra['hex'].upper()] = [ra]
else:
sorted_ras[ra['hex'].upper()].append(ra)
else:
print("No new Resolution Advisories")
last_ra_count = ra_count
for key, obj in planes.items():
if sorted_ras != {} and key in sorted_ras.keys():
print(key, "has", len(sorted_ras[key]), "RAs")
obj.check_new_ras(sorted_ras[key])
obj.expire_ra_types()
#Normal API data
api_version = int(main_config.get('ADSBX', 'API_VERSION'))
if api_version == 2:
icao_key = 'hex'
@ -152,4 +210,10 @@ except Exception as e:
if main_config.getboolean('DISCORD', 'ENABLE'):
from defDiscord import sendDis
sendDis(str("Error Exiting: " + str(traceback.format_exc())), main_config)
raise e
import logging
logging.basicConfig(filename='crash.log', filemode='a', format='%(asctime)s - %(message)s')
logging.error(e)
raise e
finally:
if platform.system() == "Linux":
os.remove(pid_file_path)

34
defADSBX.py
View File

@ -5,6 +5,7 @@ from datetime import datetime
from http.client import IncompleteRead
import http.client as http
import urllib3
import socket
main_config = configparser.ConfigParser()
main_config.read('./configs/mainconf.ini')
api_version = main_config.get('ADSBX', 'API_VERSION')
@ -88,4 +89,37 @@ def pull(url):
data_now = float(data['now']) / 1000.0
print("Data now time:",datetime.utcfromtimestamp(data_now))
print("Current UTC:", datetime.utcnow())
return data, failed
def pull_date_ras(date):
url = f"https://globe.adsbexchange.com/globe_history/{date}/acas/acas.json"
headers = {
'Accept-Encoding': 'gzip'
}
try:
response = requests.get(url, headers = headers)
response.raise_for_status()
except (requests.HTTPError, ConnectionError, requests.Timeout, urllib3.exceptions.ConnectionError) as error_message:
print("Basic Connection Error")
print(error_message)
failed = True
data = None
except (requests.RequestException, IncompleteRead, ValueError, socket.timeout, socket.gaierror) as error_message:
print("Connection Error")
print(error_message)
failed = True
data = None
except Exception as error_message:
print("Connection Error uncaught, basic exception for all")
print(error_message)
failed = True
data = None
else:
if "response" in locals() and response.status_code == 200:
failed = False
data = response.text.splitlines()
else:
failed = True
data = None
if "response" in locals():
print ("HTTP Status Code:", response.status_code)
return data, failed

20
defSS.py
View File

@ -3,7 +3,7 @@ from webdriver_manager.chrome import ChromeDriverManager
import time
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.common.by import By
def getSS(icao, file_path, overlays):
def get_adsbx_screenshot(file_path, url_params, enable_labels=False, enable_track_labels=False):
chrome_options = webdriver.ChromeOptions()
chrome_options.headless = True
chrome_options.add_argument('window-size=800,800')
@ -14,7 +14,7 @@ def getSS(icao, file_path, overlays):
if platform.system() == "Linux" and os.geteuid()==0:
chrome_options.add_argument('--no-sandbox') # required when running as root user. otherwise you would get no sandbox errors.
browser = webdriver.Chrome(ChromeDriverManager().install(), options=chrome_options)
url = "https://globe.adsbexchange.com/?largeMode=2&hideButtons&hideSidebar&mapDim=0&zoom=9&icao=" + icao + "&" + overlays
url = f"https://globe.adsbexchange.com/?{url_params}"
browser.set_page_load_timeout(80)
browser.get(url)
WebDriverWait(browser, 40).until(lambda d: d.execute_script("return jQuery.active == 0"))
@ -33,5 +33,19 @@ def getSS(icao, file_path, overlays):
element = browser.find_element_by_xpath("//*[contains(text(), 'Share')]")
browser.execute_script("""var element = arguments[0]; element.parentNode.removeChild(element); """, element)
#browser.execute_script("toggleFollow()")
if enable_labels:
browser.find_element_by_tag_name('body').send_keys('l')
if enable_track_labels:
browser.find_element_by_tag_name('body').send_keys('k')
browser.save_screenshot(file_path)
browser.quit()
browser.quit()
def generate_adsbx_screenshot_time_params(timestamp):
from datetime import datetime
from datetime import timedelta
timestamp_dt = datetime.utcfromtimestamp(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)

26
mictronics_parse.py Normal file
View File

@ -0,0 +1,26 @@
import json
import os
folder = os.getcwd() + "/dependencies"
def get_aircraft_by_icao(icao):
with open(folder + '/aircrafts.json') as aircrafts_json:
aircraft = json.load(aircrafts_json)
try:
reg = aircraft[icao.upper()]
except KeyError:
reg = None
return reg
def get_type_desc(t):
with open(folder + '/types.json') as types_json:
types = json.load(types_json)
return types[t.upper()]
def get_db_ver():
with open(folder + '/dbversion.json') as dbver_json:
dbver = json.load(dbver_json)
return dbver["version"]
def test():
print(get_aircraft_by_icao("A835AF"))
print(get_type_desc("GLF6"))
print(get_db_ver())
#test()

83
planeClass.py
View File

@ -29,8 +29,7 @@ class Plane:
self.recheck_to = None
self.speed = None
self.nearest_airport_dict = None
self.acas_ra = None
self.last_acas_ra = None
self.recent_ra_types = {}
#Setup Tweepy
if self.config.getboolean('TWITTER', 'ENABLE'):
from defTweet import tweepysetup
@ -92,7 +91,6 @@ class Plane:
#Parse ADBSX V2 Vector
from colorama import Fore, Back, Style
self.printheader("head")
print (Fore.YELLOW +"ADSBX Sourced Data: ", ac_dict, Style.RESET_ALL)
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:
@ -113,10 +111,6 @@ class Plane:
self.nav_modes[idx] = self.nav_modes[idx].upper()
else:
self.nav_modes[idx] = self.nav_modes[idx].capitalize()
if 'acas_ra_csvline' in ac_dict:
self.acas_ra = ac_dict['acas_ra_csvline']
else:
self.acas_ra = None
#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
@ -190,7 +184,7 @@ class Plane:
if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP":
from defMap import getMap
elif self.config.get('MAP', 'OPTION') == "ADSBX":
from defSS import getSS
from defSS import get_adsbx_screenshot, generate_adsbx_overlay_param
if self.config.has_option('MAP', 'OVERLAYS'):
self.overlays = self.config.get('MAP', 'OVERLAYS')
else:
@ -240,9 +234,12 @@ class Plane:
type_header = "Took off from"
elif self.last_feeding is False and self.feeding and self.landing_plausible == False:
nearest_airport_dict = getClosestAirport(self.latitude, self.longitude, self.config.get("AIRPORT", "TYPES"))
alt_above_airport = (self.alt_ft - int(nearest_airport_dict['elevation_ft']))
print(f"AGL nearest airport: {alt_above_airport}")
if alt_above_airport <= 10000:
if nearest_airport_dict['elevation_ft'] != "":
alt_above_airport = (self.alt_ft - int(nearest_airport_dict['elevation_ft']))
print(f"AGL nearest airport: {alt_above_airport}")
else:
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"
type_header = "Took off near"
@ -264,9 +261,12 @@ class Plane:
elif self.landing_plausible and self.feeding is False and time_since_contact.seconds >= 300:
nearest_airport_dict = getClosestAirport(self.latitude, self.longitude, self.config.get("AIRPORT", "TYPES"))
alt_above_airport = (self.alt_ft - int(nearest_airport_dict['elevation_ft']))
print(f"AGL nearest airport: {alt_above_airport}")
if alt_above_airport <= 10000:
if nearest_airport_dict['elevation_ft'] != "":
alt_above_airport = (self.alt_ft - int(nearest_airport_dict['elevation_ft']))
print(f"AGL nearest airport: {alt_above_airport}")
else:
alt_above_airport = None
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"
@ -347,7 +347,8 @@ class Plane:
if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP":
getMap((municipality + ", " + state + ", " + country_code), self.map_file_name)
elif self.config.get('MAP', 'OPTION') == "ADSBX":
getSS(self.icao, self.map_file_name, self.overlays)
url_params = f"icao={self.icao}&zoom=9&largeMode=2&hideButtons&hideSidebar&mapDim=0" + generate_adsbx_overlay_param(self.overlays)
get_adsbx_screenshot(self.map_file_name, url_params)
append_airport(self.map_file_name, nearest_airport_dict)
#airport_string = nearest_airport_dict['icao'] + ", " + nearest_airport_dict["name"]
else:
@ -393,7 +394,8 @@ class Plane:
if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP":
getMap((municipality + ", " + state + ", " + country_code), self.map_file_name)
if self.config.get('MAP', 'OPTION') == "ADSBX":
getSS(self.icao, self.map_file_name, self.overlays)
url_params = f"icao={self.icao}&zoom=9&largeMode=2&hideButtons&hideSidebar&mapDim=0" + generate_adsbx_overlay_param(self.overlays)
get_adsbx_screenshot(self.map_file_name, url_params)
#Discord
if self.config.getboolean('DISCORD', 'ENABLE'):
dis_message = (self.dis_title + " " + squawk_message)
@ -408,9 +410,10 @@ class Plane:
if self.config.getboolean('DISCORD', 'ENABLE'):
dis_message = (self.dis_title + " " + mode + " mode enabled.")
if mode == "Approach":
getSS(self.icao, self.map_file_name, self.overlays)
url_params = f"icao={self.icao}&zoom=9&largeMode=2&hideButtons&hideSidebar&mapDim=0" + generate_adsbx_overlay_param(self.overlays)
get_adsbx_screenshot(self.map_file_name, url_params)
sendDis(dis_message, self.config, self.map_file_name)
elif mode == "Althold" and self.nav_altitude != None:
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)
else:
sendDis(dis_message, self.config)
@ -419,15 +422,9 @@ class Plane:
# if self.config.getboolean('DISCORD', 'ENABLE'):
# dis_message = (self.dis_title + "Powered Up").strip()
# sendDis(dis_message, self.config)
#TCAS/ACAS
if self.acas_ra != None and self.last_acas_ra != self.acas_ra:
if self.config.getboolean('DISCORD', 'ENABLE'):
dis_message = f"{self.dis_title} {self.acas_ra}"
sendDis(dis_message, self.config)
#Set Variables to compare to next check
self.last_acas_ra = self.acas_ra
self.last_feeding = self.feeding
self.last_alt_ft = self.alt_ft
self.last_on_ground = self.on_ground
@ -441,4 +438,40 @@ class Plane:
elapsed_time = self.time_since(self.takeoff_time)
time_since_tk = time.strftime("Time Since Take off %H Hours : %M Mins : %S Secs", time.gmtime(elapsed_time))
print(time_since_tk)
self.printheader("foot")
self.printheader("foot")
def check_new_ras(self, ras):
for ra in ras:
if self.recent_ra_types == {} or ra['acas_ra']['advisory'] not in self.recent_ra_types.keys():
self.recent_ra_types[ra['acas_ra']['advisory']] = ra['acas_ra']['unix_timestamp']
ra_message = f"RA: {ra['acas_ra']['advisory']}"
if ra['acas_ra']['advisory_complement'] != "":
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"&zoom=14&largeMode=2&hideButtons&hideSidebar&mapDim=0" + generate_adsbx_overlay_param(self.overlays)
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]
threat_id = threat_reg if threat_reg is not None else "ICAO: " + ra['acas_ra']['threat_id_hex']
ra_message += f", invader: {threat_id}"
url_params += f"&icao={self.icao},{ra['acas_ra']['threat_id_hex']}"
else:
url_params += f"&icao={self.icao}&noIsolation"
get_adsbx_screenshot(self.map_file_name, url_params, True, True)
if self.config.getboolean('DISCORD', 'ENABLE'):
from defDiscord import sendDis
dis_message = f"{self.dis_title} {ra_message}"
sendDis(dis_message, self.config, self.map_file_name)
#if twitter
def expire_ra_types(self):
if self.recent_ra_types != {}:
for ra_type, postime in self.recent_ra_types.copy().items():
from datetime import datetime
timestamp = datetime.fromtimestamp(postime)
time_since_ra = datetime.now() - timestamp
print(time_since_ra)
if time_since_ra.seconds >= 600:
print(ra_type)
self.recent_ra_types.pop(ra_type)