Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def download_airports(self) -> None: # coverage: ignore
from .. import session
f = session.get(
"https://ourairports.com/data/airports.csv", stream=True
)
total = int(f.headers["Content-Length"])
buffer = io.BytesIO()
for chunk in tqdm(
f.iter_content(1024),
total=total // 1024 + 1 if total % 1024 > 0 else 0,
desc="airports @ourairports.com",
):
buffer.write(chunk)
buffer.seek(0)
df = pd.read_csv(buffer)
f = session.get("https://ourairports.com/data/countries.csv",)
buffer = io.BytesIO(f.content)
def download_data(dataset: Dict[str, str]) -> io.BytesIO:
from .. import session
f = session.get(dataset["url"], stream=True)
total = int(f.headers["Content-Length"])
buffer = io.BytesIO()
for chunk in tqdm(
f.iter_content(1024),
total=total // 1024 + 1 if total % 1024 > 0 else 0,
desc="download",
):
buffer.write(chunk)
buffer.seek(0)
compute_md5 = md5(buffer.getbuffer()).hexdigest()
if compute_md5 != dataset["md5sum"]:
raise RuntimeError(
f"Error in MD5 check: {compute_md5} instead of {dataset['md5sum']}"
)
def download_opensky(self): # coverage: ignore
"""Downloads the latest version of the OpenSky aircraft database.
url: https://opensky-network.org/aircraft-database
"""
from .. import session
logging.warning("Downloading OpenSky aircraft database")
file_url = (
"https://opensky-network.org/datasets/metadata/aircraftDatabase.csv"
)
f = session.get(file_url, stream=True)
total = int(f.headers["Content-Length"])
buffer = io.BytesIO()
for chunk in tqdm(
f.iter_content(1024),
total=total // 1024 + 1 if total % 1024 > 0 else 0,
desc="download",
):
buffer.write(chunk)
buffer.seek(0)
self._opensky = pd.read_csv(
buffer,
dtype={"icao24": str, "operator": str},
skiprows=[1],
engine="c",
keep_default_na=False,
def download_bluesky(self) -> None: # coverage: ignore
from .. import session
self._runways = dict()
c = session.get(base_url + "/apt.zip")
with ZipFile(BytesIO(c.content)).open("apt.dat", "r") as fh:
for line in fh.readlines():
elems = (
line.decode(encoding="ascii", errors="ignore")
.strip()
.split()
)
if len(elems) == 0:
continue
# 1: AIRPORT
if elems[0] == "1":
# Add airport to runway threshold database
cur: List[Tuple[Threshold, Threshold]] = list()
self._runways[elems[4]] = cur
# 30.580372 -094.384169 FAREL
fields = line.split()
navaids.append(
Navaid(
fields[2],
"FIX",
float(fields[0]),
float(fields[1]),
float("nan"),
None,
None,
None,
)
)
c = session.get(f"{base_url}/earth_nav.dat")
for line in c.iter_lines():
line = line.decode(encoding="ascii", errors="ignore").strip()
# Skip empty lines or comments
if len(line) == 0 or line[0] == "#":
continue
# Data line => Process fields of this record, separated by a comma
# Example lines:
# 2 58.61466599 125.42666626 451 522 30 0.0 A Aldan NDB
# 3 31.26894444 -085.72630556 334 11120 40 -3.0 OZR CAIRNS VOR-DME
# type lat lon elev freq ? var id desc
# 0 1 2 3 4 5 6 7 8
def skyvector(self) -> Dict[str, Any]:
from ..data import session
c = session.get(
"https://skyvector.com/api/routes?dep={}&dst={}".format(
self.origin, self.destination
)
)
c.raise_for_status()
c = session.get(
f"https://skyvector.com/api/fpl?cmd=route&route={self.repr}"
)
c.raise_for_status()
return c.json()
def download_junzis(self) -> None: # coverage: ignore
"""Downloads the latest version of the aircraft database by @junzis.
url: https://junzisun.com/adb/download
"""
from .. import session
f = session.get("https://junzisun.com/adb/download")
with zipfile.ZipFile(io.BytesIO(f.content)) as zfile:
with zfile.open("aircraft_db.csv", "r") as dbfile:
self._junzis = (
pd.read_csv(dbfile, dtype=str)
.assign(
regid=lambda df: df.regid.str.upper(),
mdl=lambda df: df.mdl.str.upper(),
)
.rename(
columns={
"icao": "icao24",
"regid": "registration",
"mdl": "typecode",
"type": "model",
}
)
def download_data(self) -> None:
from .. import session
logging.warning(
f"Downloading data from {self.website}. Please check terms of use."
)
c = session.get(self.json_url)
c.raise_for_status()
json_contents = c.json()
with self.cache_file.open("w") as fh:
json.dump(json_contents, fh)
def download_data(self) -> None: # coverage: ignore
"""Downloads the latest version of the navaid database from the
repository.
"""
from .. import session
navaids = []
c = session.get(f"{base_url}/earth_fix.dat")
for line in c.iter_lines():
line = line.decode(encoding="ascii", errors="ignore").strip()
# Skip empty lines or comments
if len(line) < 3 or line[0] == "#":
continue
# Start with valid 2 digit latitude -45. or 52.
if not ((line[0] == "-" and line[3] == ".") or line[2] == "."):
continue
# Data line => Process fields of this record, separated by a comma
# Example line:
# 30.580372 -094.384169 FAREL
def skyvector(self) -> Dict[str, Any]:
from ..data import session
c = session.get(
"https://skyvector.com/api/routes?dep={}&dst={}".format(
self.origin, self.destination
)
)
c.raise_for_status()
c = session.get(
f"https://skyvector.com/api/fpl?cmd=route&route={self.repr}"
)
c.raise_for_status()
return c.json()