Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Clean up low-hanging fruit. #11

Open
wants to merge 2 commits into
base: multi
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .python-version
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
plane-notify
107 changes: 107 additions & 0 deletions ADSBX.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,107 @@
import requests
import json
import configparser
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is configparser ever used?

from datetime import datetime
from http.client import IncompleteRead
import urllib3
import socket

from config import MAIN_CONFIG

API_VERSION = int(MAIN_CONFIG.get("ADSBX", "API_VERSION"))
API_KEY = MAIN_CONFIG.get("ADSBX", "API_KEY")


def pull(url, headers):
try:
response = requests.get(url, headers=headers, timeout=30)
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)
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

could this be done better with the python stdlib logging module?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't like Python's logging module. Maybe something like https://github.com/Delgan/loguru?

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What's wrong with the logging module? It's in stdlib and it's mostly the same as prints but more organised when in default settings

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've found https://github.com/Delgan/loguru to be much more pleasing to use, I think the stdlib logging requires more configuration. To be honest both are suitable for this, loguru is just what I prefer

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):
if API_VERSION not in [1, 2]:
raise ValueError("Bad ADSBX API Version")

if MAIN_CONFIG.getboolean("ADSBX", "ENABLE_PROXY") is False:
if API_VERSION == 1:
if len(planes) > 1:
url = "https://adsbexchange.com/api/aircraft/json/"

elif len(planes) == 1:
url = (
f"https://adsbexchange.com/api/aircraft/icao/{next(planes.keys())}/"
)

elif API_VERSION == 2:
url = "https://adsbexchange.com/api/aircraft/v2/all"

else:
if MAIN_CONFIG.has_option("ADSBX", "PROXY_HOST"):
url = MAIN_CONFIG.get("ADSBX", "PROXY_HOST")
if API_VERSION == 1:
url += "/api/aircraft/json/all"
if API_VERSION == 2:
url += "/api/aircraft/v2/all"
else:
raise ValueError("Proxy enabled but no host")
headers = {"api-auth": API_KEY, "Accept-Encoding": "gzip"}
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

adding some whitespace in the below code could make it more readable

response = pull(url, headers)
if response is not None:
try:
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"])
if "ctime" in data.keys():
data_ctime = float(data["ctime"]) / 1000
print("Data ctime:", datetime.utcfromtimestamp(data_ctime))
if "now" in data.keys():
data_now = float(data["now"]) / 1000
print("Data now time:", datetime.utcfromtimestamp(data_now))
print("Current UTC:", datetime.utcnow())
else:
data = None
return data


def pull_date_ras(date):
url = f"https://globe.adsbexchange.com/globe_history/{date}/acas/acas.json"
headers = {"Accept-Encoding": "gzip"}
response = pull(url, headers)
if response is not None:
data = response.text.splitlines()
else:
data = None
return data
Loading