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

Merge branch 'multi' into feature/docker-enviroment

This commit is contained in:
Raymond Bernardo 2021-07-27 20:19:42 -04:00 committed by GitHub
parent b82a25f248 b76a7fe2ec
commit c9eb3efff0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 350 additions and 322 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Pipfile
View File

@ -13,10 +13,10 @@ pytz = "*"
pillow = "*" pillow = "*"
tweepy = "*" tweepy = "*"
"pushbullet.py" = "*" "pushbullet.py" = "*"
discord-webhook = "*"
selenium = "*" selenium = "*"
opensky-api = {editable = true, git = "https://github.com/openskynetwork/opensky-api.git", subdirectory = "python"} opensky-api = {editable = true, git = "https://github.com/openskynetwork/opensky-api.git", subdirectory = "python"}
webdriver-manager = "*" webdriver-manager = "*"
discord-webhook = "*"
[requires] [requires]
python_version = "3.9" python_version = "3.9"

108
Pipfile.lock generated
View File

@ -56,18 +56,18 @@
}, },
"discord-webhook": { "discord-webhook": {
"hashes": [ "hashes": [
"sha256:d7774538321e62967b4248accfcad56972f05b9e07bd6599cd0d1c2ba376b325", "sha256:17e475d8a52fe0bfa26b071925f55087600e9bb96e821b611dc463f4b4998c89",
"sha256:f946aa5cc820e3cb95575f43ff0ac1e37d655a25cb214ca1a3b74a6af71d7e13" "sha256:f3d660df572caaa9c2621edd7e8634a70d6d8295ce9256c365838312457069a1"
], ],
"index": "pypi", "index": "pypi",
"version": "==0.13.0" "version": "==0.14.0"
}, },
"geographiclib": { "geographiclib": {
"hashes": [ "hashes": [
"sha256:12bd46ee7ec25b291ea139b17aa991e7ef373e21abd053949b75c0e9ca55c632", "sha256:8f441c527b0b8a26cd96c965565ff0513d1e4d9952b704bf449409e5015c77b7",
"sha256:51cfa698e7183792bce27d8fb63ac8e83689cd8170a730bf35e1a5c5bf8849b9" "sha256:ac400d672b8954b0306bca890b088bb8ba2a757dc8133cca0b878f34b33b2740"
], ],
"version": "==1.50" "version": "==1.52"
}, },
"geopy": { "geopy": {
"hashes": [ "hashes": [
@ -87,11 +87,11 @@
}, },
"oauthlib": { "oauthlib": {
"hashes": [ "hashes": [
"sha256:bee41cc35fcca6e988463cacc3bcb8a96224f470ca547e697b604cc697b2f889", "sha256:42bf6354c2ed8c6acb54d971fce6f88193d97297e18602a3a886603f9d7730cc",
"sha256:df884cd6cbe20e32633f1db1072e9356f53638e4361bef4e8b03c9127c9328ea" "sha256:8f0215fcc533dd8dd1bee6f4c412d4f0cd7297307d43ac61666389e3bc3198a3"
], ],
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'", "markers": "python_version >= '3.6'",
"version": "==3.1.0" "version": "==3.1.1"
}, },
"opensky-api": { "opensky-api": {
"editable": true, "editable": true,
@ -101,43 +101,43 @@
}, },
"pillow": { "pillow": {
"hashes": [ "hashes": [
"sha256:01425106e4e8cee195a411f729cff2a7d61813b0b11737c12bd5991f5f14bcd5", "sha256:0b2efa07f69dc395d95bb9ef3299f4ca29bcb2157dc615bae0b42c3c20668ffc",
"sha256:031a6c88c77d08aab84fecc05c3cde8414cd6f8406f4d2b16fed1e97634cc8a4", "sha256:114f816e4f73f9ec06997b2fde81a92cbf0777c9e8f462005550eed6bae57e63",
"sha256:083781abd261bdabf090ad07bb69f8f5599943ddb539d64497ed021b2a67e5a9", "sha256:147bd9e71fb9dcf08357b4d530b5167941e222a6fd21f869c7911bac40b9994d",
"sha256:0d19d70ee7c2ba97631bae1e7d4725cdb2ecf238178096e8c82ee481e189168a", "sha256:15a2808e269a1cf2131930183dcc0419bc77bb73eb54285dde2706ac9939fa8e",
"sha256:0e04d61f0064b545b989126197930807c86bcbd4534d39168f4aa5fda39bb8f9", "sha256:196560dba4da7a72c5e7085fccc5938ab4075fd37fe8b5468869724109812edd",
"sha256:12e5e7471f9b637762453da74e390e56cc43e486a88289995c1f4c1dc0bfe727", "sha256:1c03e24be975e2afe70dfc5da6f187eea0b49a68bb2b69db0f30a61b7031cee4",
"sha256:22fd0f42ad15dfdde6c581347eaa4adb9a6fc4b865f90b23378aa7914895e120", "sha256:1fd5066cd343b5db88c048d971994e56b296868766e461b82fa4e22498f34d77",
"sha256:238c197fc275b475e87c1453b05b467d2d02c2915fdfdd4af126145ff2e4610c", "sha256:29c9569049d04aaacd690573a0398dbd8e0bf0255684fee512b413c2142ab723",
"sha256:3b570f84a6161cf8865c4e08adf629441f56e32f180f7aa4ccbd2e0a5a02cba2", "sha256:2b6dfa068a8b6137da34a4936f5a816aba0ecc967af2feeb32c4393ddd671cba",
"sha256:463822e2f0d81459e113372a168f2ff59723e78528f91f0bd25680ac185cf797", "sha256:2cac53839bfc5cece8fdbe7f084d5e3ee61e1303cccc86511d351adcb9e2c792",
"sha256:4d98abdd6b1e3bf1a1cbb14c3895226816e666749ac040c4e2554231068c639b", "sha256:2ee77c14a0299d0541d26f3d8500bb57e081233e3fa915fa35abd02c51fa7fae",
"sha256:5afe6b237a0b81bd54b53f835a153770802f164c5570bab5e005aad693dab87f", "sha256:37730f6e68bdc6a3f02d2079c34c532330d206429f3cee651aab6b66839a9f0e",
"sha256:5b70110acb39f3aff6b74cf09bb4169b167e2660dabc304c1e25b6555fa781ef", "sha256:3f08bd8d785204149b5b33e3b5f0ebbfe2190ea58d1a051c578e29e39bfd2367",
"sha256:5cbf3e3b1014dddc45496e8cf38b9f099c95a326275885199f427825c6522232", "sha256:4bc3c7ef940eeb200ca65bd83005eb3aae8083d47e8fcbf5f0943baa50726856",
"sha256:624b977355cde8b065f6d51b98497d6cd5fbdd4f36405f7a8790e3376125e2bb", "sha256:660a87085925c61a0dcc80efb967512ac34dbb256ff7dd2b9b4ee8dbdab58cf4",
"sha256:63728564c1410d99e6d1ae8e3b810fe012bc440952168af0a2877e8ff5ab96b9", "sha256:67b3666b544b953a2777cb3f5a922e991be73ab32635666ee72e05876b8a92de",
"sha256:66cc56579fd91f517290ab02c51e3a80f581aba45fd924fcdee01fa06e635812", "sha256:70af7d222df0ff81a2da601fab42decb009dc721545ed78549cb96e3a1c5f0c8",
"sha256:6c32cc3145928c4305d142ebec682419a6c0a8ce9e33db900027ddca1ec39178", "sha256:8960a8a9f4598974e4c2aeb1bff9bdd5db03ee65fd1fce8adf3223721aa2a636",
"sha256:8b56553c0345ad6dcb2e9b433ae47d67f95fc23fe28a0bde15a120f25257e291", "sha256:9364c81b252d8348e9cc0cb63e856b8f7c1b340caba6ee7a7a65c968312f7dab",
"sha256:8bb1e155a74e1bfbacd84555ea62fa21c58e0b4e7e6b20e4447b8d07990ac78b", "sha256:969cc558cca859cadf24f890fc009e1bce7d7d0386ba7c0478641a60199adf79",
"sha256:95d5ef984eff897850f3a83883363da64aae1000e79cb3c321915468e8c6add5", "sha256:9a211b663cf2314edbdb4cf897beeb5c9ee3810d1d53f0e423f06d6ebbf9cd5d",
"sha256:a013cbe25d20c2e0c4e85a9daf438f85121a4d0344ddc76e33fd7e3965d9af4b", "sha256:a17ca41f45cf78c2216ebfab03add7cc350c305c38ff34ef4eef66b7d76c5229",
"sha256:a787ab10d7bb5494e5f76536ac460741788f1fbce851068d73a87ca7c35fc3e1", "sha256:a2f381932dca2cf775811a008aa3027671ace723b7a38838045b1aee8669fdcf",
"sha256:a7d5e9fad90eff8f6f6106d3b98b553a88b6f976e51fce287192a5d2d5363713", "sha256:a4eef1ff2d62676deabf076f963eda4da34b51bc0517c70239fafed1d5b51500",
"sha256:aac00e4bc94d1b7813fe882c28990c1bc2f9d0e1aa765a5f2b516e8a6a16a9e4", "sha256:c088a000dfdd88c184cc7271bfac8c5b82d9efa8637cd2b68183771e3cf56f04",
"sha256:b91c36492a4bbb1ee855b7d16fe51379e5f96b85692dc8210831fbb24c43e484", "sha256:c11003197f908878164f0e6da15fce22373ac3fc320cda8c9d16e6bba105b844",
"sha256:c03c07ed32c5324939b19e36ae5f75c660c81461e312a41aea30acdd46f93a7c", "sha256:c2a5ff58751670292b406b9f06e07ed1446a4b13ffced6b6cab75b857485cbc8",
"sha256:c5236606e8570542ed424849f7852a0ff0bce2c4c8d0ba05cc202a5a9c97dee9", "sha256:c35d09db702f4185ba22bb33ef1751ad49c266534339a5cebeb5159d364f6f82",
"sha256:c6b39294464b03457f9064e98c124e09008b35a62e3189d3513e5148611c9388", "sha256:c379425c2707078dfb6bfad2430728831d399dc95a7deeb92015eb4c92345eaf",
"sha256:cb7a09e173903541fa888ba010c345893cd9fc1b5891aaf060f6ca77b6a3722d", "sha256:cc866706d56bd3a7dbf8bac8660c6f6462f2f2b8a49add2ba617bc0c54473d83",
"sha256:d68cb92c408261f806b15923834203f024110a2e2872ecb0bd2a110f89d3c602", "sha256:d0da39795049a9afcaadec532e7b669b5ebbb2a9134576ebcc15dd5bdae33cc0",
"sha256:dc38f57d8f20f06dd7c3161c59ca2c86893632623f33a42d592f097b00f720a9", "sha256:f4ebde71785f8bceb39dcd1e7f06bcc5d5c3cf48b9f69ab52636309387b097c8",
"sha256:e98eca29a05913e82177b3ba3d198b1728e164869c613d76d0de4bde6768a50e", "sha256:fc214a6b75d2e0ea7745488da7da3c381f41790812988c7a92345978414fad37",
"sha256:f217c3954ce5fd88303fc0c317af55d5e0204106d86dea17eb8205700d47dec2" "sha256:ff04c373477723430dce2e9d024c708a047d44cf17166bf16e604b379bf0ca14"
], ],
"index": "pypi", "index": "pypi",
"version": "==8.2.0" "version": "==8.3.1"
}, },
"pushbullet.py": { "pushbullet.py": {
"hashes": [ "hashes": [
@ -157,11 +157,11 @@
}, },
"python-magic": { "python-magic": {
"hashes": [ "hashes": [
"sha256:8551e804c09a3398790bd9e392acb26554ae2609f29c72abb0b9dee9a5571eae", "sha256:4fec8ee805fea30c07afccd1592c0f17977089895bdfaae5fec870a84e997626",
"sha256:ca884349f2c92ce830e3f498c5b7c7051fe2942c3ee4332f65213b8ebff15a62" "sha256:de800df9fb50f8ec5974761054a708af6e4246b03b4bdaee993f948947b0ebcf"
], ],
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'", "markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4'",
"version": "==0.4.22" "version": "==0.4.24"
}, },
"pytz": { "pytz": {
"hashes": [ "hashes": [
@ -224,11 +224,11 @@
}, },
"urllib3": { "urllib3": {
"hashes": [ "hashes": [
"sha256:753a0374df26658f99d826cfe40394a686d05985786d946fbe4165b5148f5a7c", "sha256:39fb8672126159acb139a7718dd10806104dec1e2f0f6c88aab05d17df10c8d4",
"sha256:a7acd0977125325f516bda9735fa7142b909a8d01e8b2e4c8108d0984e6e0098" "sha256:f57b4c16c62fa2760b7e3d97c35b255512fb6b59a259730f36ba32ce9f8e342f"
], ],
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4' and python_version < '4'", "markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4' and python_version < '4'",
"version": "==1.26.5" "version": "==1.26.6"
}, },
"webdriver-manager": { "webdriver-manager": {
"hashes": [ "hashes": [
@ -240,11 +240,11 @@
}, },
"websocket-client": { "websocket-client": {
"hashes": [ "hashes": [
"sha256:3e2bf58191d4619b161389a95bdce84ce9e0b24eb8107e7e590db682c2d0ca81", "sha256:b68e4959d704768fa20e35c9d508c8dc2bbc041fd8d267c0d7345cffe2824568",
"sha256:abf306dc6351dcef07f4d40453037e51cc5d9da2ef60d0fc5d0fe3bcda255372" "sha256:e5c333bfa9fa739538b652b6f8c8fc2559f1d364243c8a689d7c0e1d41c2e611"
], ],
"markers": "python_version >= '3.6'", "markers": "python_version >= '3.6'",
"version": "==1.0.1" "version": "==1.1.0"
} }
}, },
"develop": {} "develop": {}

56
__main__.py
View File

@ -7,7 +7,7 @@ if platform.system() == "Windows":
from colorama import init from colorama import init
init(convert=True) init(convert=True)
elif platform.system() == "Linux": 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): def write_pid_file(filepath):
import os import os
pid = str(os.getpid()) pid = str(os.getpid())
@ -24,10 +24,11 @@ import signal
abspath = os.path.abspath(__file__) abspath = os.path.abspath(__file__)
dname = os.path.dirname(abspath) dname = os.path.dirname(abspath)
os.chdir(dname) os.chdir(dname)
if not os.path.isdir("./dependencies/"):
os.mkdir("./dependencies/")
import sys import sys
sys.path.extend([os.getcwd()]) sys.path.extend([os.getcwd()])
#Dependency Handling
if not os.path.isdir("./dependencies/"):
os.mkdir("./dependencies/")
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")] 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: for file in required_files:
file_name = file[0] file_name = file[0]
@ -50,6 +51,7 @@ if os.path.isfile("./dependencies/" + required_files[4][0]) and not os.path.isfi
from zipfile import ZipFile from zipfile import ZipFile
with ZipFile("./dependencies/" + required_files[4][0], 'r') as mictronics_db: with ZipFile("./dependencies/" + required_files[4][0], 'r') as mictronics_db:
mictronics_db.extractall("./dependencies/") mictronics_db.extractall("./dependencies/")
main_config = configparser.ConfigParser() main_config = configparser.ConfigParser()
print(os.getcwd()) print(os.getcwd())
main_config.read('./configs/mainconf.ini') main_config.read('./configs/mainconf.ini')
@ -64,6 +66,10 @@ def service_exit(signum, frame):
os.remove(pid_file_path) os.remove(pid_file_path)
raise SystemExit("Service Stop") raise SystemExit("Service Stop")
signal.signal(signal.SIGTERM, service_exit) signal.signal(signal.SIGTERM, service_exit)
if os.path.isfile("lookup_route.py"):
print("Route lookup is enabled")
else:
print("Route lookup is disabled")
try: try:
print("Source is set to", source) print("Source is set to", source)
@ -101,9 +107,9 @@ try:
import ast import ast
today = datetime.utcnow() today = datetime.utcnow()
date = today.strftime("%Y/%m/%d") date = today.strftime("%Y/%m/%d")
ras, failed = pull_date_ras(date) ras = pull_date_ras(date)
sorted_ras = {} sorted_ras = {}
if failed is False and ras != None: if ras is not None:
#Testing RAs #Testing RAs
#if last_ra_count is not None: #if last_ra_count is not None:
# with open('./testing/acastest.json') as f: # with open('./testing/acastest.json') as f:
@ -136,25 +142,24 @@ try:
else: else:
raise ValueError("Invalid API Version") raise ValueError("Invalid API Version")
from defADSBX import pull_adsbx from defADSBX import pull_adsbx
data, failed = pull_adsbx(planes) data = pull_adsbx(planes)
if failed == False: if data is not None:
if data['ac'] != None: if data['ac'] is not None:
data_indexed = {}
for planeData in data['ac']:
data_indexed[planeData[icao_key].upper()] = planeData
for key, obj in planes.items(): for key, obj in planes.items():
has_data = False try:
for planeData in data['ac']: if api_version == 1:
if planeData[icao_key].upper() == key: obj.run_adsbx_v1(data_indexed[key.upper()])
if api_version == 1: elif api_version == 2:
obj.run_adsbx_v1(planeData) obj.run_adsbx_v2(data_indexed[key.upper()])
elif api_version == 2: except KeyError:
obj.run_adsbx_v2(planeData)
has_data = True
break
if has_data is False:
obj.run_empty() obj.run_empty()
else: else:
for obj in planes.values(): for obj in planes.values():
obj.run_empty() obj.run_empty()
elif failed: else:
failed_count += 1 failed_count += 1
elif source == "OPENS": elif source == "OPENS":
from defOpenSky import pull_opensky from defOpenSky import pull_opensky
@ -190,7 +195,6 @@ try:
footer = "-------- " + str(running_Count) + " -------- " + str(datetime_tz.strftime("%I:%M:%S %p")) + " ------------------------Elapsed Time- " + str(round(elapsed_calc_time, 3)) + " -------------------------------------" footer = "-------- " + str(running_Count) + " -------- " + str(datetime_tz.strftime("%I:%M:%S %p")) + " ------------------------Elapsed Time- " + str(round(elapsed_calc_time, 3)) + " -------------------------------------"
print (Back.GREEN + Fore.BLACK + footer[0:100] + Style.RESET_ALL) print (Back.GREEN + Fore.BLACK + footer[0:100] + Style.RESET_ALL)
sleep_sec = 30 sleep_sec = 30
for i in range(sleep_sec,0,-1): for i in range(sleep_sec,0,-1):
if i < 10: if i < 10:
@ -208,11 +212,17 @@ except KeyboardInterrupt as e:
sendDis(str("Manual Exit: " + str(e)), main_config) sendDis(str("Manual Exit: " + str(e)), main_config)
except Exception as e: except Exception as e:
if main_config.getboolean('DISCORD', 'ENABLE'): if main_config.getboolean('DISCORD', 'ENABLE'):
from defDiscord import sendDis try:
sendDis(str("Error Exiting: " + str(traceback.format_exc())), main_config) os.remove('crash_latest.log')
except OSError:
pass
import logging import logging
logging.basicConfig(filename='crash.log', filemode='a', format='%(asctime)s - %(message)s') logging.basicConfig(filename='crash_latest.log', filemode='w', format='%(asctime)s - %(message)s')
logging.Formatter.converter = time.gmtime
logging.error(e) logging.error(e)
logging.error(str(traceback.format_exc()))
from defDiscord import sendDis
sendDis(str("Error Exiting: " + str(e) + "Failed on " + key), main_config, "crash_latest.log")
raise e raise e
finally: finally:
if platform.system() == "Linux": if platform.system() == "Linux":

3
configs/mainconf.ini
View File

@ -4,7 +4,8 @@
#By default configured with OpenSky which anyone can use without a login #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. #ADS-B Exchange has better data but is not avalible unless you feed their network or pay.
SOURCE = OPENS 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. #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 #List of TZs names https://en.wikipedia.org/wiki/List_of_tz_database_time_zones
TZ = UTC TZ = UTC

11
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. #Plane to track, based of ICAO or ICAO24 which is the unique transponder address of a plane.
ICAO = icaohere ICAO = icaohere
#Optional Per Plane Override
#DATA_LOSS_MINS = 20
[MAP] [MAP]
#Map to create from Google Static Maps or screenshot global tar1090 from globe.adsbexchange.com #Map to create from Google Static Maps or screenshot global tar1090 from globe.adsbexchange.com
#Enter GOOGLESTATICMAP or ADSBX #Enter GOOGLESTATICMAP or ADSBX
@ -30,8 +33,10 @@ API_KEY = apikey
CHANNEL_TAG = channeltag CHANNEL_TAG = channeltag
[DISCORD] [DISCORD]
ENABLE = TRUE ENABLE = FALSE
#WEBHOOK URL https://support.discord.com/hc/en-us/articles/228383668-Intro-to-Webhooks #WEBHOOK URL https://support.discord.com/hc/en-us/articles/228383668-Intro-to-Webhooks
URL = webhookurl URL = webhookurl
Title = #Role to tag optional, the role ID
USERNAME = plane-notify ROLE_ID =
Title =
USERNAME = plane-notify

132
defADSBX.py
View File

@ -9,6 +9,26 @@ import socket
main_config = configparser.ConfigParser() main_config = configparser.ConfigParser()
main_config.read('./configs/mainconf.ini') main_config.read('./configs/mainconf.ini')
api_version = main_config.get('ADSBX', 'API_VERSION') 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")
print(error_message)
response = None
except (requests.RequestException, IncompleteRead, ValueError, socket.timeout, socket.gaierror) as error_message:
print("Connection Error")
print(error_message)
response = None
except Exception as error_message:
print("Connection Error uncaught, basic exception for all")
print(error_message)
response = None
return response
def pull_adsbx(planes): def pull_adsbx(planes):
api_version = int(main_config.get('ADSBX', 'API_VERSION')) api_version = int(main_config.get('ADSBX', 'API_VERSION'))
if api_version not in [1, 2]: if api_version not in [1, 2]:
@ -29,97 +49,43 @@ def pull_adsbx(planes):
url = main_config.get('ADSBX', 'PROXY_HOST') + "/api/aircraft/v2/all" url = main_config.get('ADSBX', 'PROXY_HOST') + "/api/aircraft/v2/all"
else: else:
raise ValueError("Proxy enabled but no host") raise ValueError("Proxy enabled but no host")
return pull(url)
def pull(url):
headers = { headers = {
'api-auth': main_config.get('ADSBX', 'API_KEY'), 'api-auth': main_config.get('ADSBX', 'API_KEY'),
'Accept-Encoding': 'gzip' 'Accept-Encoding': 'gzip'
} }
try: response = pull(url, headers)
response = requests.get(url, headers = headers) if response is not None:
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:
try:
data = json.loads(response.text)
except (json.decoder.JSONDecodeError, ValueError) as error_message:
print("Error with JSON")
if 'data' in locals() and data != None:
print (json.dumps(data, indent = 2))
print(error_message)
failed = True
data = None
except TypeError as error_message:
print("Type Error", error_message)
failed = True
data = None
else:
failed = False
else:
failed = True
data = None
if "response" in locals():
print ("HTTP Status Code:", response.status_code)
if failed is False:
try: try:
if data['msg'] != "No error": data = json.loads(response.text)
except (json.decoder.JSONDecodeError, ValueError) as error_message:
print("Error with JSON")
print(error_message)
data = None
except TypeError as error_message:
print("Type Error", error_message)
data = None
else:
if "msg" in data.keys() and data['msg'] != "No error":
raise ValueError("Error from ADSBX: msg = ", data['msg']) raise ValueError("Error from ADSBX: msg = ", data['msg'])
except KeyError: if "ctime" in data.keys():
pass data_ctime = float(data['ctime']) / 1000.0
if "ctime" in data.keys(): print("Data ctime:",datetime.utcfromtimestamp(data_ctime))
data_ctime = float(data['ctime']) / 1000.0 if "now" in data.keys():
print("Data ctime:",datetime.utcfromtimestamp(data_ctime)) data_now = float(data['now']) / 1000.0
if "now" in data.keys(): print("Data now time:",datetime.utcfromtimestamp(data_now))
data_now = float(data['now']) / 1000.0
print("Data now time:",datetime.utcfromtimestamp(data_now))
print("Current UTC:", datetime.utcnow()) print("Current UTC:", datetime.utcnow())
return data, failed else:
data = None
return data
def pull_date_ras(date): def pull_date_ras(date):
url = f"https://globe.adsbexchange.com/globe_history/{date}/acas/acas.json" url = f"https://globe.adsbexchange.com/globe_history/{date}/acas/acas.json"
headers = { headers = {
'Accept-Encoding': 'gzip' 'Accept-Encoding': 'gzip'
} }
try: response = pull(url, headers)
response = requests.get(url, headers = headers) if response is not None:
response.raise_for_status() data = response.text.splitlines()
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: else:
if "response" in locals() and response.status_code == 200: data = None
failed = False return data
data = response.text.splitlines()
else:
failed = True
data = None
if "response" in locals():
print ("HTTP Status Code:", response.status_code)
return data, failed

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): def getClosestAirport(latitude, longitude, allowed_types):
import csv
from geopy.distance import geodesic from geopy.distance import geodesic
plane = (latitude, longitude) plane = (latitude, longitude)
with open('./dependencies/airports.csv', 'r', encoding='utf-8') as airport_csv: 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 closest_airport_dict['distance_mi'] = closest_airport_dist
#Convert indent key to icao key as its labeled icao in other places not ident #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') closest_airport_dict['icao'] = closest_airport_dict.pop('gps_code')
#Get full region/state name from iso region name closest_airport_dict = add_airport_region(closest_airport_dict)
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']
return closest_airport_dict return closest_airport_dict
def get_airport_by_icao(icao): def get_airport_by_icao(icao):
import csv
with open('./dependencies/airports.csv', 'r', encoding='utf-8') as airport_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)) airport_csv_reader = csv.DictReader(filter(lambda row: row[0]!='#', airport_csv))
for airport in airport_csv_reader: 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 #Convert indent key to icao key as its labeled icao in other places not ident
matching_airport['icao'] = matching_airport.pop('gps_code') matching_airport['icao'] = matching_airport.pop('gps_code')
break break
matching_airport = add_airport_region(matching_airport)
return matching_airport return matching_airport

10
defDiscord.py
View File

@ -1,10 +1,12 @@
def sendDis(message, config, image_name = None): def sendDis(message, config, file_name = None, role_id = None):
import requests import requests
from discord_webhook import DiscordWebhook from discord_webhook import DiscordWebhook
if role_id != None:
message += f" <@&{role_id}>"
webhook = DiscordWebhook(url=config.get('DISCORD', 'URL'), content=message[0:1999], username=config.get('DISCORD', 'USERNAME')) webhook = DiscordWebhook(url=config.get('DISCORD', 'URL'), content=message[0:1999], username=config.get('DISCORD', 'USERNAME'))
if image_name != None: if file_name != None:
with open(image_name, "rb") as f: with open(file_name, "rb") as f:
webhook.add_file(file=f.read(), filename='map.png') webhook.add_file(file=f.read(), filename=file_name)
try: try:
webhook.execute() webhook.execute()
except requests.Exceptions: except requests.Exceptions:

4
defOpenSky.py
View File

@ -11,7 +11,7 @@ def pull_opensky(planes):
icao_array.append(key.lower()) icao_array.append(key.lower())
try: try:
planeData = opens_api.get_states(time_secs=0, icao24=icao_array) planeData = opens_api.get_states(time_secs=0, icao24=icao_array)
except: except Exception as e:
print ("OpenSky Error") print ("OpenSky Error", e)
failed = True failed = True
return planeData, failed return planeData, failed

50
defSS.py
View File

@ -19,23 +19,53 @@ def get_adsbx_screenshot(file_path, url_params, enable_labels=False, enable_trac
browser.get(url) browser.get(url)
remove_id_elements = ["show_trace", "credits", 'infoblock_close', 'selected_photo_link', "history_collapse"] remove_id_elements = ["show_trace", "credits", 'infoblock_close', 'selected_photo_link', "history_collapse"]
for element in remove_id_elements: for element in remove_id_elements:
element = browser.find_element_by_id(element) try:
browser.execute_script("""var element = arguments[0]; element.parentNode.removeChild(element); """, element) element = browser.find_element_by_id(element)
element = browser.find_elements_by_class_name("infoHeading") browser.execute_script("""var element = arguments[0]; element.parentNode.removeChild(element); """, element)
browser.execute_script("""var element = arguments[0]; element.parentNode.removeChild(element); """, element[19]) except:
print("issue removing", element, "from map")
#Remove watermark on data #Remove watermark on data
browser.execute_script("document.getElementById('selected_infoblock').className = 'none';") try:
browser.execute_script("document.getElementById('selected_infoblock').className = 'none';")
except:
print("Couldn't remove watermark from map")
#Disable slidebar #Disable slidebar
browser.execute_script("$('#infoblock-container').css('overflow', 'hidden');") try:
browser.execute_script("$('#infoblock-container').css('overflow', 'hidden');")
except:
print("Couldn't disable sidebar on map")
#Remove share #Remove share
element = browser.find_element_by_xpath("//*[contains(text(), 'Share')]") try:
browser.execute_script("""var element = arguments[0]; element.parentNode.removeChild(element); """, element) element = browser.find_element_by_xpath("//*[contains(text(), 'Share')]")
browser.execute_script("""var element = arguments[0]; element.parentNode.removeChild(element); """, element)
except:
print("Couldn't remove share button from map")
#browser.execute_script("toggleFollow()") #browser.execute_script("toggleFollow()")
if enable_labels: if enable_labels:
browser.find_element_by_tag_name('body').send_keys('l') browser.find_element_by_tag_name('body').send_keys('l')
if enable_track_labels: if enable_track_labels:
browser.find_element_by_tag_name('body').send_keys('k') browser.find_element_by_tag_name('body').send_keys('k')
WebDriverWait(browser, 40).until(lambda d: d.execute_script("return jQuery.active == 0")) WebDriverWait(browser, 40).until(lambda d: d.execute_script("return jQuery.active == 0"))
try:
photo_box = browser.find_element_by_id("silhouette")
except:
pass
else:
import requests, json
photo_list = json.loads(requests.get("https://raw.githubusercontent.com/Jxck-S/aircraft-photos/main/photo-list.json").text)
if "icao" in url_params:
import re
icao = re.search('icao=(.+?)&', url_params).group(1).lower()
print(icao)
if icao in photo_list.keys():
browser.execute_script("arguments[0].id = 'airplanePhoto';", photo_box)
browser.execute_script(f"arguments[0].src = 'https://raw.githubusercontent.com/Jxck-S/aircraft-photos/main/images/{photo_list[icao]['reg']}.jpg';", photo_box)
copyright = browser.find_element_by_id("copyrightInfo")
browser.execute_script("arguments[0].id = 'copyrightInfoFreeze';", copyright)
browser.execute_script("$('#copyrightInfoFreeze').css('font-size', '12px');")
browser.execute_script(f"arguments[0].appendChild(document.createTextNode('Image © {photo_list[icao]['photographer']}'))", copyright)
time.sleep(5) time.sleep(5)
browser.save_screenshot(file_path) browser.save_screenshot(file_path)
browser.quit() browser.quit()
@ -46,6 +76,4 @@ def generate_adsbx_screenshot_time_params(timestamp):
print(timestamp_dt) print(timestamp_dt)
start_time = timestamp_dt - timedelta(minutes=1) 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") 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 return time_params
def generate_adsbx_overlay_param(overlays):
return "&overlays=" + ",".join(overlays)

276
planeClass.py
View File

@ -1,5 +1,8 @@
from datetime import datetime from datetime import datetime, timedelta
class Plane: class Plane:
import configparser
main_config = configparser.ConfigParser()
main_config.read('./configs/mainconf.ini')
def __init__(self, icao, config_path, config): def __init__(self, icao, config_path, config):
"""Initializes a plane object from its config file and given icao.""" """Initializes a plane object from its config file and given icao."""
self.icao = icao.upper() self.icao = icao.upper()
@ -8,7 +11,6 @@ class Plane:
self.config = config self.config = config
self.conf_file_path = config_path self.conf_file_path = config_path
self.alt_ft = None self.alt_ft = None
self.last_alt_ft = None
self.below_desired_ft = None self.below_desired_ft = None
self.last_below_desired_ft = None self.last_below_desired_ft = None
self.feeding = None self.feeding = None
@ -19,18 +21,27 @@ class Plane:
self.latitude = None self.latitude = None
self.takeoff_time = None self.takeoff_time = None
import tempfile 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_latitude = None
self.last_longitude = None self.last_longitude = None
self.last_contact = None self.last_pos_datetime = None
self.landing_plausible = False self.landing_plausible = False
self.squawks = [None, None, None, None]
self.nav_modes = None self.nav_modes = None
self.last_nav_modes = None self.last_nav_modes = None
self.recheck_to = None self.recheck_to = None
self.speed = None self.speed = None
self.nearest_airport_dict = None self.nearest_airport_dict = None
self.recent_ra_types = {} 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 #Setup Tweepy
if self.config.getboolean('TWITTER', 'ENABLE'): if self.config.getboolean('TWITTER', 'ENABLE'):
from defTweet import tweepysetup from defTweet import tweepysetup
@ -51,11 +62,6 @@ class Plane:
self.alt_ft = round(float(ac_dict.baro_altitude) * 3.281) self.alt_ft = round(float(ac_dict.baro_altitude) * 3.281)
elif self.on_ground: elif self.on_ground:
self.alt_ft = 0 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: except ValueError as e:
print("Got data but some data is invalid!") print("Got data but some data is invalid!")
print(e) 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)}) 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: if self.on_ground:
self.alt_ft = 0 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: except ValueError as e:
print("Got data but some data is invalid!") print("Got data but some data is invalid!")
@ -92,6 +93,7 @@ class Plane:
#Parse ADBSX V2 Vector #Parse ADBSX V2 Vector
from colorama import Fore, Back, Style from colorama import Fore, Back, Style
self.printheader("head") self.printheader("head")
print(ac_dict)
try: try:
self.__dict__.update({'icao' : ac_dict['hex'].upper(), 'latitude' : float(ac_dict['lat']), 'longitude' : float(ac_dict['lon']), 'speed': ac_dict['gs']}) 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: if "r" in ac_dict:
@ -104,7 +106,10 @@ class Plane:
elif ac_dict['alt_baro'] == "ground": elif ac_dict['alt_baro'] == "ground":
self.alt_ft = 0 self.alt_ft = 0
self.on_ground = True self.on_ground = True
self.callsign = ac_dict.get('flight') 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: if 'nav_modes' in ac_dict:
self.nav_modes = ac_dict['nav_modes'] self.nav_modes = ac_dict['nav_modes']
for idx, mode in enumerate(self.nav_modes): for idx, mode in enumerate(self.nav_modes):
@ -112,23 +117,16 @@ class Plane:
self.nav_modes[idx] = self.nav_modes[idx].upper() self.nav_modes[idx] = self.nav_modes[idx].upper()
else: else:
self.nav_modes[idx] = self.nav_modes[idx].capitalize() self.nav_modes[idx] = self.nav_modes[idx].capitalize()
#Insert newest sqwauk at 0, sqwuak length should be 4 long 0-3 self.squawk = ac_dict.get('squawk')
self.squawks.insert(0, ac_dict.get('squawk'))
#Removes oldest sqwauk index 4 5th sqwauk
if len(self.squawks) == 5:
self.squawks.pop(4)
if "nav_altitude_fms" in ac_dict: if "nav_altitude_fms" in ac_dict:
self.nav_altitude = ac_dict['nav_altitude_fms'] self.sel_nav_alt = ac_dict['nav_altitude_fms']
if "nav_altitude_mcp" in ac_dict: elif "nav_altitude_mcp" in ac_dict:
self.nav_altitude = ac_dict['nav_altitude_mcp'] self.sel_nav_alt = ac_dict['nav_altitude_mcp']
else: 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 #Create last seen timestamp from how long ago in secs a pos was rec
now = datetime.now() self.last_pos_datetime = datetime.now() - timedelta(seconds= ac_dict["seen_pos"])
last_pos_datetime = now - timedelta(seconds= ac_dict["seen_pos"])
self.last_contact = datetime.timestamp(last_pos_datetime)
except (ValueError, KeyError) as e: except (ValueError, KeyError) as e:
print("Got data but some data is invalid!") print("Got data but some data is invalid!")
@ -145,51 +143,61 @@ class Plane:
elif type == "foot": elif type == "foot":
header = "----------------------------------------------------------------------------------------------------" header = "----------------------------------------------------------------------------------------------------"
print(Back.MAGENTA + header[0:100] + Style.RESET_ALL) print(Back.MAGENTA + header[0:100] + Style.RESET_ALL)
def get_time_since(self, last_contact): def get_time_since(self, datetime_obj):
if last_contact != None: if datetime_obj != None:
last_contact_dt = datetime.fromtimestamp(last_contact) time_since = datetime.now() - datetime_obj
time_since_contact = datetime.now() - last_contact_dt
else: else:
time_since_contact = None time_since = None
return time_since_contact 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): def run_empty(self):
self.printheader("head") self.printheader("head")
self.feeding = False self.feeding = False
self.run_check() self.run_check()
def run_check(self): def run_check(self):
"""Runs a check of a plane module to see if its landed or takenoff using plane data, and takes action if so.""" """Runs a check of a plane module to see if its landed or takenoff using plane data, and takes action if so."""
#Import Modules
#Ability to Remove old Map #Ability to Remove old Map
import os import os
from colorama import Fore, Style from colorama import Fore, Style
#Platform for determining OS for strftime
import platform
from tabulate import tabulate from tabulate import tabulate
from modify_image import append_airport #Proprietary Route Lookup
from defAirport import getClosestAirport if os.path.isfile("lookup_route.py") and (self.db_flags is None or not self.db_flags & 1):
#Propritary
ENABLE_ROUTE_LOOKUP = False
if ENABLE_ROUTE_LOOKUP:
from lookup_route import lookup_route from lookup_route import lookup_route
ENABLE_ROUTE_LOOKUP = True
else: else:
#Dead Place function ENABLE_ROUTE_LOOKUP = False
def lookup_route(*args):
return None
if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP":
from defMap import getMap
elif self.config.get('MAP', 'OPTION') == "ADSBX":
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:
self.overlays = ""
else:
raise ValueError("Map option not set correctly in this planes conf")
if self.config.getboolean('DISCORD', 'ENABLE'): if self.config.getboolean('DISCORD', 'ENABLE'):
from defDiscord import sendDis from defDiscord import sendDis
if self.feeding == False: 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 = [ output = [
[(Fore.CYAN + "ICAO" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.icao + Style.RESET_ALL)], [(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 [(Fore.CYAN + "Last Contact" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + str(time_since_contact) + Style.RESET_ALL)] if time_since_contact != None else None
@ -198,19 +206,19 @@ class Plane:
print(tabulate(output, [], 'fancy_grid')) print(tabulate(output, [], 'fancy_grid'))
print("No Data") print("No Data")
elif self.feeding == True: 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 = [ output = [
[(Fore.CYAN + "ICAO" + Style.RESET_ALL), (Fore.LIGHTGREEN_EX + self.icao + Style.RESET_ALL)], [(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 + "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, [(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 #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 + "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 + "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 + "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 + "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 + "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)) output = list(filter(None, output))
print(tabulate(output, [], 'fancy_grid')) print(tabulate(output, [], 'fancy_grid'))
@ -225,9 +233,10 @@ class Plane:
if self.below_desired_ft and self.on_ground is False: if self.below_desired_ft and self.on_ground is False:
if self.last_on_ground: if self.last_on_ground:
self.tookoff = True self.tookoff = True
self.trigger_type = "no longer on ground" trigger_type = "no longer on ground"
type_header = "Took off from" type_header = "Took off from"
elif self.last_feeding is False and self.feeding and self.landing_plausible == False: 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")) nearest_airport_dict = getClosestAirport(self.latitude, self.longitude, self.config.get("AIRPORT", "TYPES"))
if nearest_airport_dict['elevation_ft'] != "": if nearest_airport_dict['elevation_ft'] != "":
alt_above_airport = (self.alt_ft - int(nearest_airport_dict['elevation_ft'])) alt_above_airport = (self.alt_ft - int(nearest_airport_dict['elevation_ft']))
@ -236,7 +245,7 @@ class Plane:
alt_above_airport = None alt_above_airport = None
if (alt_above_airport != None and alt_above_airport <= 10000) or self.alt_ft <= 15000: if (alt_above_airport != None and alt_above_airport <= 10000) or self.alt_ft <= 15000:
self.tookoff = True self.tookoff = True
self.trigger_type = "data acquisition" trigger_type = "data acquisition"
type_header = "Took off near" type_header = "Took off near"
else: else:
self.tookoff = False self.tookoff = False
@ -246,15 +255,16 @@ class Plane:
#Check if Landed #Check if Landed
if self.on_ground and self.last_on_ground is False and self.last_below_desired_ft: if self.on_ground and self.last_on_ground is False and self.last_below_desired_ft:
self.landed = True self.landed = True
self.trigger_type = "now on ground" trigger_type = "now on ground"
type_header = "Landed in" type_header = "Landed in"
self.landing_plausible = False self.landing_plausible = False
#Set status for landing plausible #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: 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 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")) nearest_airport_dict = getClosestAirport(self.latitude, self.longitude, self.config.get("AIRPORT", "TYPES"))
if nearest_airport_dict['elevation_ft'] != "": if nearest_airport_dict['elevation_ft'] != "":
alt_above_airport = (self.alt_ft - int(nearest_airport_dict['elevation_ft'])) alt_above_airport = (self.alt_ft - int(nearest_airport_dict['elevation_ft']))
@ -264,7 +274,7 @@ class Plane:
if (alt_above_airport != None and alt_above_airport <= 10000) or self.alt_ft <= 15000: if (alt_above_airport != None and alt_above_airport <= 10000) or self.alt_ft <= 15000:
self.landing_plausible = False self.landing_plausible = False
self.landed = True self.landed = True
self.trigger_type = "data loss" trigger_type = "data loss"
type_header = "Landed near" type_header = "Landed near"
else: else:
print("Alt greater then 10k AGL") print("Alt greater then 10k AGL")
@ -273,16 +283,17 @@ class Plane:
self.landed = False self.landed = False
if self.landed: if self.landed:
print ("Landed by", self.trigger_type) print ("Landed by", trigger_type)
if self.tookoff: if self.tookoff:
print("Tookoff by", self.trigger_type) print("Tookoff by", trigger_type)
#Find nearest airport, and location #Find nearest airport, and location
if self.landed or self.tookoff: if self.landed or self.tookoff:
from defAirport import getClosestAirport
if "nearest_airport_dict" in globals(): if "nearest_airport_dict" in globals():
pass #Airport already set 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")) 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")) nearest_airport_dict = getClosestAirport(self.last_latitude, self.last_longitude, self.config.get("AIRPORT", "TYPES"))
#Convert dictionary keys to sep variables #Convert dictionary keys to sep variables
country_code = nearest_airport_dict['iso_country'] country_code = nearest_airport_dict['iso_country']
@ -298,11 +309,7 @@ class Plane:
else: else:
area = f"{municipality}, {state}" area = f"{municipality}, {state}"
location_string = (f"{area}, {country_code}") location_string = (f"{area}, {country_code}")
print (Fore.GREEN) print (Fore.GREEN + "Country Code:", country_code, "State:", state, "Municipality:", municipality + Style.RESET_ALL)
print ("Country Code: ", country_code)
print ("State: ", state)
print ("Municipality: ", municipality)
print(Style.RESET_ALL)
title_switch = { title_switch = {
"reg": self.reg, "reg": self.reg,
"callsign": self.callsign, "callsign": self.callsign,
@ -320,37 +327,22 @@ class Plane:
if self.tookoff: if self.tookoff:
self.takeoff_time = datetime.utcnow() self.takeoff_time = datetime.utcnow()
landed_time_msg = None landed_time_msg = None
#Route Lookup | Proprietary #Proprietary Route Lookup
if ENABLE_ROUTE_LOOKUP: if ENABLE_ROUTE_LOOKUP:
extra_route_info = lookup_route(self.reg, (self.latitude, self.longitude), self.type, self.alt_ft) self.nearest_airport_dict = nearest_airport_dict
if extra_route_info == None: route_to = self.route_info()
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[11])
code = to_airport['iata_code'] if to_airport['iata_code'] != "" else to_airport['icao']
if extra_route_info[11] != nearest_airport_dict['icao']:
route_to = "Going to"
else:
route_to = "Will be returning to"
route_to += f" {code}, {to_airport['name']}"
elif self.landed and self.takeoff_time != None: elif self.landed and self.takeoff_time != None:
landed_time = datetime.utcnow() - self.takeoff_time landed_time = datetime.utcnow() - self.takeoff_time
if platform.system() == "Linux": if trigger_type == "data loss":
strftime_splitter = "-" landed_time -= timedelta(seconds=time_since_contact.total_seconds())
elif platform.system() == "Windows":
strftime_splitter = "#"
hours, remainder = divmod(landed_time.total_seconds(), 3600) hours, remainder = divmod(landed_time.total_seconds(), 3600)
minutes, seconds = divmod(remainder, 60) minutes, seconds = divmod(remainder, 60)
min_syntax = "Mins" if minutes > 1 else "Min" min_syntax = "Mins" if minutes > 1 else "Min"
if hours > 0: if hours > 0:
hour_syntax = "Hours" if hours > 1 else "Hour" hour_syntax = "Hours" if hours > 1 else "Hour"
landed_time_msg = (f"Apx. flt. time {int(hours)} {hour_syntax}: {int(minutes)} {min_syntax}. ") landed_time_msg = (f"Apx. flt. time {int(hours)} {hour_syntax}" + (f" : {int(minutes)} {min_syntax}. " if minutes > 0 else "."))
else: else:
landed_time_msg = (f"Apx. flt. time {int(minutes)} {min_syntax}. ") landed_time_msg = (f"Apx. flt. time {int(minutes)} {min_syntax}.")
# landed_time_msg = time.strftime(f"Apx. flt. time %{strftime_splitter}H Hours : %{strftime_splitter}M Mins. ", time.gmtime(landed_time))
# landed_time_msg = landed_time_msg.replace("0 Hours : ", "")
self.takeoff_time = None self.takeoff_time = None
elif self.landed: elif self.landed:
landed_time_msg = None landed_time_msg = None
@ -358,18 +350,22 @@ class Plane:
print (message) print (message)
#Google Map or tar1090 screenshot #Google Map or tar1090 screenshot
if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP": if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP":
from defMap import getMap
getMap((municipality + ", " + state + ", " + country_code), self.map_file_name) getMap((municipality + ", " + state + ", " + country_code), self.map_file_name)
elif self.config.get('MAP', 'OPTION') == "ADSBX": 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) get_adsbx_screenshot(self.map_file_name, url_params)
from modify_image import append_airport
append_airport(self.map_file_name, nearest_airport_dict) append_airport(self.map_file_name, nearest_airport_dict)
#airport_string = nearest_airport_dict['icao'] + ", " + nearest_airport_dict["name"]
else: else:
raise ValueError("Map option not set correctly in this planes conf") raise ValueError("Map option not set correctly in this planes conf")
#Discord #Discord
if self.config.getboolean('DISCORD', 'ENABLE'): if self.config.getboolean('DISCORD', 'ENABLE'):
dis_message = f"{self.dis_title} {message}".strip() dis_message = f"{self.dis_title} {message}".strip()
sendDis(dis_message, self.config, self.map_file_name) role_id = self.config.get('DISCORD', 'ROLE_ID') if self.config.has_option('DISCORD', 'ROLE_ID') else None
sendDis(dis_message, self.config, self.map_file_name, role_id = role_id)
#PushBullet #PushBullet
if self.config.getboolean('PUSHBULLET', 'ENABLE'): if self.config.getboolean('PUSHBULLET', 'ENABLE'):
with open(self.map_file_name, "rb") as pic: with open(self.map_file_name, "rb") as pic:
@ -379,51 +375,57 @@ class Plane:
#Twitter #Twitter
if self.config.getboolean('TWITTER', 'ENABLE'): if self.config.getboolean('TWITTER', 'ENABLE'):
twitter_media_map_obj = self.tweet_api.media_upload(self.map_file_name) 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.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]) 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) os.remove(self.map_file_name)
#To Location #Recheck Proprietary Route Lookup a minute later if infomation was not available on takeoff.
if self.recheck_to and (datetime.utcnow() - self.takeoff_time).total_seconds() > 60: 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 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 self.nearest_takeoff_airport = None
if extra_route_info != None: if route_to != None:
from defAirport import get_airport_by_icao print(route_to)
to_airport = get_airport_by_icao(extra_route_info[11]) #Discord
code = to_airport['iata_code'] if to_airport['iata_code'] != "" else to_airport['icao']
if extra_route_info[11] != nearest_airport_dict['icao']:
route_to = "Going to"
else:
route_to = "Will be returning to"
route_to += f" {code}, {to_airport['name']}"
if self.config.getboolean('DISCORD', 'ENABLE'): if self.config.getboolean('DISCORD', 'ENABLE'):
dis_message = (self.dis_title + route_to).strip() dis_message = f"{self.dis_title} {route_to}".strip()
sendDis(dis_message, self.config) role_id = self.config.get('DISCORD', 'ROLE_ID') if self.config.has_option('DISCORD', 'ROLE_ID') else None
sendDis(dis_message, self.config, role_id = role_id)
#Twitter #Twitter
if self.config.getboolean('TWITTER', 'ENABLE'): if self.config.getboolean('TWITTER', 'ENABLE'):
tweet = self.tweet_api.user_timeline(count = 1)[0] tweet = self.tweet_api.user_timeline(count = 1)[0]
self.tweet_api.update_status(status = f"{self.twitter_title} {route_to}".strip(), in_reply_to_status_id = tweet.id) self.tweet_api.update_status(status = f"{self.twitter_title} {route_to}".strip(), in_reply_to_status_id = tweet.id)
#Squawks
squawks =[("7500", "Hijacking"), ("7600", "Radio Failure"), ("7700", "Emergency")]
if self.feeding: if self.feeding:
for squawk in squawks: #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: emergency_squawks ={"7500" : "Hijacking", "7600" :"Radio Failure", "7700" : "General Emergency"}
squawk_message = ("Squawking " + squawk[0] + ", " + squawk[1]) 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) print(squawk_message)
#Google Map or tar1090 screenshot #Google Map or tar1090 screenshot
if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP": if self.config.get('MAP', 'OPTION') == "GOOGLESTATICMAP":
getMap((municipality + ", " + state + ", " + country_code), self.map_file_name) getMap((municipality + ", " + state + ", " + country_code), self.map_file_name)
if self.config.get('MAP', 'OPTION') == "ADSBX": 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) get_adsbx_screenshot(self.map_file_name, url_params)
#Discord
if self.config.getboolean('DISCORD', 'ENABLE'): if self.config.getboolean('DISCORD', 'ENABLE'):
dis_message = (self.dis_title + " " + squawk_message) dis_message = (self.dis_title + " " + squawk_message)
sendDis(dis_message, self.config, self.map_file_name) sendDis(dis_message, self.config, self.map_file_name)
os.remove(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 #Nav Modes Notifications
if self.nav_modes != None and self.last_nav_modes != None: if self.nav_modes != None and self.last_nav_modes != None:
for mode in self.nav_modes: for mode in self.nav_modes:
@ -433,13 +435,21 @@ class Plane:
if self.config.getboolean('DISCORD', 'ENABLE'): if self.config.getboolean('DISCORD', 'ENABLE'):
dis_message = (self.dis_title + " " + mode + " mode enabled.") dis_message = (self.dis_title + " " + mode + " mode enabled.")
if mode == "Approach": 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) get_adsbx_screenshot(self.map_file_name, url_params)
sendDis(dis_message, self.config, self.map_file_name) sendDis(dis_message, self.config, self.map_file_name)
elif mode in ["Althold", "VNAV", "LNAV"] and self.nav_altitude != None: #elif mode in ["Althold", "VNAV", "LNAV"] and self.sel_nav_alt != None:
sendDis((dis_message + ", Sel Alt. " + str(self.nav_altitude) + ", Current Alt. " + str(self.alt_ft)), self.config) # sendDis((dis_message + ", Sel Alt. " + str(self.sel_nav_alt) + ", Current Alt. " + str(self.alt_ft)), self.config)
else: else:
sendDis(dis_message, self.config) 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 # #Power Up
# if self.last_feeding == False and self.speed == 0 and self.on_ground: # if self.last_feeding == False and self.speed == 0 and self.on_ground:
# if self.config.getboolean('DISCORD', 'ENABLE'): # if self.config.getboolean('DISCORD', 'ENABLE'):
@ -449,12 +459,12 @@ class Plane:
#Set Variables to compare to next check #Set Variables to compare to next check
self.last_feeding = self.feeding self.last_feeding = self.feeding
self.last_alt_ft = self.alt_ft
self.last_on_ground = self.on_ground self.last_on_ground = self.on_ground
self.last_below_desired_ft = self.below_desired_ft self.last_below_desired_ft = self.below_desired_ft
self.last_longitude = self.longitude self.last_longitude = self.longitude
self.last_latitude = self.latitude self.last_latitude = self.latitude
self.last_nav_modes = self.nav_modes self.last_nav_modes = self.nav_modes
self.last_sel_alt = self.sel_nav_alt
if self.takeoff_time != None: if self.takeoff_time != None:
@ -472,22 +482,24 @@ class Plane:
ra_message += f", {ra['acas_ra']['advisory_complement']}" ra_message += f", {ra['acas_ra']['advisory_complement']}"
if bool(int(ra['acas_ra']['MTE'])): if bool(int(ra['acas_ra']['MTE'])):
ra_message += ", Multi threat" ra_message += ", Multi threat"
from defSS import get_adsbx_screenshot, generate_adsbx_screenshot_time_params, generate_adsbx_overlay_param from defSS import get_adsbx_screenshot, generate_adsbx_screenshot_time_params
url_params = generate_adsbx_screenshot_time_params(ra['acas_ra']['unix_timestamp']) + f"&zoom=11&largeMode=2&hideButtons&hideSidebar&mapDim=0&overlays={self.overlays}" url_params = f"&lat={ra['lat']}&lon={ra['lon']}&zoom=11&largeMode=2&hideButtons&hideSidebar&mapDim=0&overlays={self.get_adsbx_map_overlays()}"
if "threat_id_hex" in ra['acas_ra'].keys(): if "threat_id_hex" in ra['acas_ra'].keys():
from mictronics_parse import get_aircraft_by_icao from mictronics_parse import get_aircraft_by_icao
threat_reg = get_aircraft_by_icao(ra['acas_ra']['threat_id_hex'])[0] 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'] threat_id = threat_reg if threat_reg is not None else "ICAO: " + ra['acas_ra']['threat_id_hex']
ra_message += f", invader: {threat_id}" ra_message += f", invader: {threat_id}"
url_params += f"&icao={self.icao.lower()},{ra['acas_ra']['threat_id_hex']}" url_params += generate_adsbx_screenshot_time_params(ra['acas_ra']['unix_timestamp']) + f"&icao={ra['acas_ra']['threat_id_hex']},{self.icao.lower()}&timestamp={ra['acas_ra']['unix_timestamp']}"
else: else:
url_params += f"&icao={self.icao.lower()}&noIsolation" url_params += f"&icao={self.icao.lower()}&noIsolation"
print(url_params)
get_adsbx_screenshot(self.map_file_name, url_params, True, True) get_adsbx_screenshot(self.map_file_name, url_params, True, True)
if self.config.getboolean('DISCORD', 'ENABLE'): if self.config.getboolean('DISCORD', 'ENABLE'):
from defDiscord import sendDis from defDiscord import sendDis
dis_message = f"{self.dis_title} {ra_message}" dis_message = f"{self.dis_title} {ra_message}"
sendDis(dis_message, self.config, self.map_file_name) role_id = self.config.get('DISCORD', 'ROLE_ID') if self.config.has_option('DISCORD', 'ROLE_ID') else None
sendDis(dis_message, self.config, self.map_file_name, role_id = role_id)
#if twitter #if twitter
def expire_ra_types(self): def expire_ra_types(self):
if self.recent_ra_types != {}: if self.recent_ra_types != {}: