diff --git a/examples/requests_api_openskynetwork_private.py b/examples/requests_api_openskynetwork_private.py new file mode 100644 index 0000000..a69eba8 --- /dev/null +++ b/examples/requests_api_openskynetwork_private.py @@ -0,0 +1,152 @@ +# SPDX-FileCopyrightText: 2023 DJDevon3 +# SPDX-License-Identifier: MIT +# Coded for Circuit Python 8.1 +# DJDevon3 ESP32-S3 OpenSkyNetwork_Private_API_Example + +import os +import time +import ssl +import json +import wifi +import socketpool +import circuitpython_base64 as base64 +import adafruit_requests + +# OpenSky-Network.org Login required for this API +# REST API: https://openskynetwork.github.io/opensky-api/rest.html +# All active flights JSON: https://opensky-network.org/api/states/all # PICK ONE! :) +# JSON order: transponder, callsign, country +# ACTIVE transpondes only, for multiple "c822af&icao24=cb3993&icao24=c63923" +transponder = "7c6b2d" + +# Initialize WiFi Pool (There can be only 1 pool & top of script) +pool = socketpool.SocketPool(wifi.radio) + +# Time between API refreshes +# 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour +# OpenSky-Networks IP bans for too many requests, check rate limit. +# https://openskynetwork.github.io/opensky-api/rest.html#limitations +sleep_time = 1800 + +# this example uses settings.toml for credentials +# timezone offset is in seconds plus or minus GMT +ssid = os.getenv("AP_SSID") +appw = os.getenv("AP_PASSWORD") +osnu = os.getenv("OSN_Username") +osnp = os.getenv("OSN_Password") + +osn_cred = str(osnu) + ":" + str(osnp) +bytes_to_encode = b" " + str(osn_cred) + " " +base64_string = base64.encodebytes(bytes_to_encode) +base64cred = repr(base64_string)[2:-1] + +Debug_Auth = False # STREAMER WARNING this will show your credentials! +if Debug_Auth: + osn_cred = str(osnu) + ":" + str(osnp) + bytes_to_encode = b" " + str(osn_cred) + " " + print(repr(bytes_to_encode)) + base64_string = base64.encodebytes(bytes_to_encode) + print(repr(base64_string)[2:-1]) + base64cred = repr(base64_string)[2:-1] + print("Decoded Bytes:", str(base64cred)) + +# Requests URL - icao24 is their endpoint required for a transponder +# example https://opensky-network.org/api/states/all?icao24=a808c5 +# OSN private requires your username:password to be base64 encoded +osn_header = {"Authorization": "Basic " + str(base64cred)} +OPENSKY_SOURCE = "https://opensky-network.org/api/states/all?" + "icao24=" + transponder + + +# Converts seconds to human readable minutes/hours/days +def time_calc(input_time): # input_time in seconds + if input_time < 60: + sleep_int = input_time + time_output = f"{sleep_int:.0f} seconds" + elif 60 <= input_time < 3600: + sleep_int = input_time / 60 + time_output = f"{sleep_int:.0f} minutes" + elif 3600 <= input_time < 86400: + sleep_int = input_time / 60 / 60 + time_output = f"{sleep_int:.1f} hours" + else: + sleep_int = input_time / 60 / 60 / 24 + time_output = f"{sleep_int:.1f} days" + return time_output + + +def _format_datetime(datetime): + return "{:02}/{:02}/{} {:02}:{:02}:{:02}".format( + datetime.tm_mon, + datetime.tm_mday, + datetime.tm_year, + datetime.tm_hour, + datetime.tm_min, + datetime.tm_sec, + ) + + +# Connect to Wi-Fi +print("\n===============================") +print("Connecting to WiFi...") +request = adafruit_requests.Session(pool, ssl.create_default_context()) +while not wifi.radio.ipv4_address: + try: + wifi.radio.connect(ssid, appw) + except ConnectionError as e: + print("Connection Error:", e) + print("Retrying in 10 seconds") + time.sleep(10) +print("Connected!\n") + +while True: + # STREAMER WARNING this will show your credentials! + debug_request = False # Set True to see full request + if debug_request: + print("Full API HEADER: ", str(osn_header)) + print("Full API GET URL: ", OPENSKY_SOURCE) + print("===============================") + + print("\nAttempting to GET OpenSky-Network Data!") + opensky_response = request.get(url=OPENSKY_SOURCE, headers=osn_header).json() + + # Print Full JSON to Serial (doesn't show credentials) + debug_response = False # Set True to see full response + if debug_response: + dump_object = json.dumps(opensky_response) + print("JSON Dump: ", dump_object) + + # Key:Value Serial Debug (doesn't show credentials) + osn_debug_keys = True # Set True to print Serial data + if osn_debug_keys: + try: + osn_flight = opensky_response["time"] + print("Current Unix Time: ", osn_flight) + + current_struct_time = time.localtime(osn_flight) + current_date = "{}".format(_format_datetime(current_struct_time)) + print(f"Unix to Readable Time: {current_date}") + + # Current flight data for single callsign (right now) + osn_single_flight_data = opensky_response["states"] + + if osn_single_flight_data is not None: + print("Flight Data: ", osn_single_flight_data) + transponder = opensky_response["states"][0][0] + print("Transponder: ", transponder) + callsign = opensky_response["states"][0][1] + print("Callsign: ", callsign) + country = opensky_response["states"][0][2] + print("Flight Country: ", country) + else: + print("Flight has no active data or you're polling too fast.") + + print("\nFinished!") + print("Board Uptime: ", time_calc(time.monotonic())) + print("Next Update: ", time_calc(sleep_time)) + time.sleep(sleep_time) + print("===============================") + + except (ConnectionError, ValueError, NameError) as e: + print("OSN Connection Error:", e) + print("Next Retry: ", time_calc(sleep_time)) + time.sleep(sleep_time) diff --git a/examples/requests_api_openskynetwork_private_area.py b/examples/requests_api_openskynetwork_private_area.py new file mode 100644 index 0000000..e474e48 --- /dev/null +++ b/examples/requests_api_openskynetwork_private_area.py @@ -0,0 +1,184 @@ +# SPDX-FileCopyrightText: 2023 DJDevon3 +# SPDX-License-Identifier: MIT +# Coded for Circuit Python 8.1 +# DJDevon3 ESP32-S3 OpenSkyNetwork_Private_Area_API_Example + +import os +import time +import ssl +import json +import wifi +import socketpool +import circuitpython_base64 as base64 +import adafruit_requests + +# OpenSky-Network.org Website Login required for this API +# REST API: https://openskynetwork.github.io/opensky-api/rest.html + +# Retrieves all traffic within a geographic area (Orlando example) +latmin = "27.22" # east bounding box +latmax = "28.8" # west bounding box +lonmin = "-81.46" # north bounding box +lonmax = "-80.40" # south bounding box + +# Initialize WiFi Pool (There can be only 1 pool & top of script) +pool = socketpool.SocketPool(wifi.radio) + +# Time between API refreshes +# 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour +# OpenSky-Networks IP bans for too many requests, check rate limit. +# https://openskynetwork.github.io/opensky-api/rest.html#limitations +sleep_time = 1800 + +# this example uses settings.toml for credentials +# No token required, only website login +ssid = os.getenv("AP_SSID") +appw = os.getenv("AP_PASSWORD") +osnu = os.getenv("OSN_Username") +osnp = os.getenv("OSN_Password") + +osn_cred = str(osnu) + ":" + str(osnp) +bytes_to_encode = b" " + str(osn_cred) + " " +base64_string = base64.encodebytes(bytes_to_encode) +base64cred = repr(base64_string)[2:-1] + +Debug_Auth = False # STREAMER WARNING this will show your credentials! +if Debug_Auth: + osn_cred = str(osnu) + ":" + str(osnp) + bytes_to_encode = b" " + str(osn_cred) + " " + print(repr(bytes_to_encode)) + base64_string = base64.encodebytes(bytes_to_encode) + print(repr(base64_string)[2:-1]) + base64cred = repr(base64_string)[2:-1] + print("Decoded Bytes:", str(base64cred)) + +# OSN requires your username:password to be base64 encoded +# so technically it's not transmitted in the clear but w/e +osn_header = {"Authorization": "Basic " + str(base64cred)} + +# Example request of all traffic over Florida, geographic areas cost less per call. +# https://opensky-network.org/api/states/all?lamin=25.21&lomin=-84.36&lamax=30.0&lomax=-78.40 +OPENSKY_SOURCE = ( + "https://opensky-network.org/api/states/all?" + + "lamin=" + + latmin + + "&lomin=" + + lonmin + + "&lamax=" + + latmax + + "&lomax=" + + lonmax +) + + +# Converts seconds to human readable minutes/hours/days +def time_calc(input_time): # input_time in seconds + if input_time < 60: + sleep_int = input_time + time_output = f"{sleep_int:.0f} seconds" + elif 60 <= input_time < 3600: + sleep_int = input_time / 60 + time_output = f"{sleep_int:.0f} minutes" + elif 3600 <= input_time < 86400: + sleep_int = input_time / 60 / 60 + time_output = f"{sleep_int:.1f} hours" + else: + sleep_int = input_time / 60 / 60 / 24 + time_output = f"{sleep_int:.1f} days" + return time_output + + +def _format_datetime(datetime): + return "{:02}/{:02}/{} {:02}:{:02}:{:02}".format( + datetime.tm_mon, + datetime.tm_mday, + datetime.tm_year, + datetime.tm_hour, + datetime.tm_min, + datetime.tm_sec, + ) + + +# Connect to Wi-Fi +print("\n===============================") +print("Connecting to WiFi...") +request = adafruit_requests.Session(pool, ssl.create_default_context()) +while not wifi.radio.ipv4_address: + try: + wifi.radio.connect(ssid, appw) + except ConnectionError as e: + print("Connection Error:", e) + print("Retrying in 10 seconds") + time.sleep(10) +print("Connected!\n") + +while True: + # STREAMER WARNING this will show your credentials! + debug_request = False # Set True to see full request + if debug_request: + print("Full API HEADER: ", str(osn_header)) + print("Full API GET URL: ", OPENSKY_SOURCE) + print("===============================") + + print("\nAttempting to GET OpenSky-Network Data!") + opensky_response = request.get(url=OPENSKY_SOURCE, headers=osn_header).json() + + # Print Full JSON to Serial (doesn't show credentials) + debug_response = False # Set True to see full response + if debug_response: + dump_object = json.dumps(opensky_response) + print("JSON Dump: ", dump_object) + + # Key:Value Serial Debug (doesn't show credentials) + osn_debug_keys = True # Set True to print Serial data + if osn_debug_keys: + try: + osn_flight = opensky_response["time"] + print("Current Unix Time: ", osn_flight) + + current_struct_time = time.localtime(osn_flight) + current_date = "{}".format(_format_datetime(current_struct_time)) + print(f"Unix to Readable Time: {current_date}") + + # Current flight data for single callsign (right now) + osn_all_flights = opensky_response["states"] + + if osn_all_flights is not None: + # print("Flight Data: ", osn_all_flights) + for flights in osn_all_flights: + osn_t = f"Trans:{flights[0]} " + osn_c = f"Sign:{flights[1]} " + osn_o = f"Origin:{flights[2]} " + osn_tm = f"Time:{flights[3]} " + osn_l = f"Last:{flights[4]} " + osn_lo = f"Lon:{flights[5]} " + osn_la = f"Lat:{flights[6]} " + osn_ba = f"BaroAlt:{flights[7]} " + osn_g = f"Ground:{flights[8]} " + osn_v = f"Vel:{flights[9]} " + osn_h = f"Head:{flights[10]} " + osn_vr = f"VertRate:{flights[11]} " + osn_s = f"Sens:{flights[12]} " + osn_ga = f"GeoAlt:{flights[13]} " + osn_sq = f"Squawk:{flights[14]} " + osn_pr = f"Task:{flights[15]} " + osn_ps = f"PosSys:{flights[16]} " + osn_ca = f"Cat:{flights[16]} " + # This is just because pylint complains about long lines + string1 = f"{osn_t}{osn_c}{osn_o}{osn_tm}{osn_l}{osn_lo}" + string2 = f"{osn_la}{osn_ba}{osn_g}{osn_v}{osn_h}{osn_vr}" + string3 = f"{osn_s}{osn_ga}{osn_sq}{osn_pr}{osn_ps}{osn_ca}" + print(f"{string1}{string2}{string3}") + else: + print("Flight has no active data or you're polling too fast.") + + print("\nFinished!") + print("Board Uptime: ", time_calc(time.monotonic())) + print("Next Update: ", time_calc(sleep_time)) + time.sleep(sleep_time) + print("===============================") + + except (ConnectionError, ValueError, NameError) as e: + print("OSN Connection Error:", e) + print("Next Retry: ", time_calc(sleep_time)) + time.sleep(sleep_time) diff --git a/examples/requests_api_openskynetwork_public.py b/examples/requests_api_openskynetwork_public.py new file mode 100644 index 0000000..e114cb7 --- /dev/null +++ b/examples/requests_api_openskynetwork_public.py @@ -0,0 +1,135 @@ +# SPDX-FileCopyrightText: 2023 DJDevon3 +# SPDX-License-Identifier: MIT +# Coded for Circuit Python 8.1 +# Adafruit Feather ESP32-S3 OpenSkyNetwork_Public_API_Example +import os +import time +import ssl +import json +import wifi +import socketpool +import adafruit_requests + +# No login necessary for Public API. Drastically reduced daily limit vs Private +# OpenSky-Networks.org REST API: https://openskynetwork.github.io/opensky-api/rest.html +# All active flights JSON: https://opensky-network.org/api/states/all PICK ONE! +# JSON order: transponder, callsign, country +# ACTIVE transpondes only, for multiple "c822af&icao24=cb3993&icao24=c63923" +transponder = "ab1644" + +# Initialize WiFi Pool (There can be only 1 pool & top of script) +pool = socketpool.SocketPool(wifi.radio) + +# Time between API refreshes +# 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour +# OpenSky-Networks will temp ban your IP for too many daily requests. +# https://openskynetwork.github.io/opensky-api/rest.html#limitations +sleep_time = 1800 + +# Wifi credentials pulled from settings.toml +ssid = os.getenv("AP_SSID") +appw = os.getenv("AP_PASSWORD") + +# Requests URL - icao24 is their endpoint required for a transponder +# example https://opensky-network.org/api/states/all?icao24=a808c5 +OPENSKY_SOURCE = "https://opensky-network.org/api/states/all?" + "icao24=" + transponder + + +# Converts seconds to human readable minutes/hours/days +def time_calc(input_time): # input_time in seconds + if input_time < 60: + sleep_int = input_time + time_output = f"{sleep_int:.0f} seconds" + elif 60 <= input_time < 3600: + sleep_int = input_time / 60 + time_output = f"{sleep_int:.0f} minutes" + elif 3600 <= input_time < 86400: + sleep_int = input_time / 60 / 60 + time_output = f"{sleep_int:.1f} hours" + else: + sleep_int = input_time / 60 / 60 / 24 + time_output = f"{sleep_int:.1f} days" + return time_output + + +def _format_datetime(datetime): + return "{:02}/{:02}/{} {:02}:{:02}:{:02}".format( + datetime.tm_mon, + datetime.tm_mday, + datetime.tm_year, + datetime.tm_hour, + datetime.tm_min, + datetime.tm_sec, + ) + + +# Connect to Wi-Fi +print("\n===============================") +print("Connecting to WiFi...") +requests = adafruit_requests.Session(pool, ssl.create_default_context()) +while not wifi.radio.ipv4_address: + try: + wifi.radio.connect(ssid, appw) + except ConnectionError as e: + print("Connection Error:", e) + print("Retrying in 10 seconds") + time.sleep(10) +print("Connected!\n") + +while True: + debug_request = True # Set true to see full request + if debug_request: + print("Full API GET URL: ", OPENSKY_SOURCE) + print("===============================") + try: + print("\nAttempting to GET OpenSky-Network Stats!") + opensky_response = requests.get(url=OPENSKY_SOURCE) + osn_json = opensky_response.json() + except (ConnectionError, ValueError, NameError) as e: + print("Host No Response Error:", e) + + # Print Full JSON to Serial + debug_response = False # Set true to see full response + if debug_response: + dump_object = json.dumps(osn_json) + print("JSON Dump: ", dump_object) + + # Print to Serial + osn_debug_keys = True # Set true to print Serial data + if osn_debug_keys: + try: + osn_flight = osn_json["time"] + print("Current Unix Time: ", osn_flight) + + current_struct_time = time.localtime(osn_flight) + current_date = "{}".format(_format_datetime(current_struct_time)) + print(f"Unix to Readable Time: {current_date}") + + osn_single_flight_data = osn_json["states"] + if osn_single_flight_data is not None: + print("Flight Data: ", osn_single_flight_data) + transponder = osn_json["states"][0][0] + print("Transponder: ", transponder) + callsign = osn_json["states"][0][1] + print("Callsign: ", callsign) + country = osn_json["states"][0][2] + print("Flight Country: ", country) + else: + print("This flight has no active data or you're polling too fast.") + print( + "Read: https://openskynetwork.github.io/opensky-api/rest.html#limitations" + ) + print( + "Public Limits: 10 second max poll rate & 400 weighted calls daily" + ) + + print("\nFinished!") + print("Board Uptime: ", time_calc(time.monotonic())) + print("Next Update: ", time_calc(sleep_time)) + time.sleep(sleep_time) + print("===============================") + + except (ConnectionError, ValueError, NameError) as e: + print("OSN Connection Error:", e) + print("Next Retry: ", time_calc(sleep_time)) + time.sleep(sleep_time)