diff --git a/__main__.py b/__main__.py index 70fbbcb..02e3d72 100644 --- a/__main__.py +++ b/__main__.py @@ -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 \ No newline at end of file + 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) \ No newline at end of file diff --git a/defADSBX.py b/defADSBX.py index 8b7de87..592cc54 100644 --- a/defADSBX.py +++ b/defADSBX.py @@ -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 \ No newline at end of file diff --git a/defSS.py b/defSS.py index 7b91d0c..d6c79a3 100644 --- a/defSS.py +++ b/defSS.py @@ -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() \ No newline at end of file + 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) \ No newline at end of file diff --git a/mictronics_parse.py b/mictronics_parse.py new file mode 100644 index 0000000..df719e1 --- /dev/null +++ b/mictronics_parse.py @@ -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() \ No newline at end of file diff --git a/planeClass.py b/planeClass.py index 4898b54..547e5c8 100644 --- a/planeClass.py +++ b/planeClass.py @@ -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") \ No newline at end of file + 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) \ No newline at end of file