Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@pytest.mark.skipif(aircraft.opensky_db is not None, reason="OpenSky database")
def test_opensky_dl() -> None:
aircraft.download_opensky()
assert aircraft.opensky_db is not None
def test_getter() -> None:
a = aircraft["PH-BHA"]
assert a.icao24.iloc[0] == "4851ad"
assert a.typecode.iloc[0] == "B789"
a = aircraft["39b415"]
assert a.registration.iloc[0] == "F-HNAV"
assert a.typecode.iloc[0] == "BE20"
estarrivalairport="destination", estdepartureairport="origin",
)
)
.assign(
firstseen=lambda df: pd.to_datetime(
df.firstseen, unit="s",
).dt.tz_localize("utc"),
lastseen=lambda df: pd.to_datetime(
df.lastseen, unit="s",
).dt.tz_localize("utc"),
day=lambda df: pd.to_datetime(df.day, unit="s",).dt.tz_localize(
"utc"
),
)
)
df_merged = df.merge(aircraft.opensky_db, how="left")[
[
"callsign",
"icao24",
"registration",
"typecode",
"origin",
"destination",
"firstseen",
"lastseen",
"day",
"latitude_1",
"longitude_1",
"altitude_1",
"latitude_2",
"longitude_2",
"altitude_2",
def aircraft_data(self) -> "Traffic":
"""
Add registration and aircraft typecode based on the `aircraft database
`_.
"""
from ..data import aircraft
return self.merge(
aircraft.data[["icao24", "registration", "typecode"]]
.query('typecode != ""')
.drop_duplicates("icao24"),
on="icao24",
how="left",
)
if args.verbose == 1:
logger.setLevel(logging.INFO)
elif args.verbose >= 2:
logger.setLevel(logging.DEBUG)
from ..data import airports, aircraft, navaids
if args.aircraft:
for arg in args.args:
print(aircraft[arg])
if args.operator:
print(aircraft.operator(" ".join(args.args)))
if args.stats:
print(aircraft.stats(" ".join(args.args)))
if args.airport:
for arg in args.args:
print(airports.search(arg))
if args.navaid:
for arg in args.args:
for navaid in navaids.search(arg):
print(navaid)
def get_aircraft(*args):
from .data import aircraft as data_aircraft
if args[0] == "get":
print(data_aircraft[args[1]])
elif args[0] == "operator":
print(data_aircraft.operator(" ".join(args[1:])))
elif args[0] == "stats":
print(data_aircraft.stats(" ".join(args[1:])))
else:
raise RuntimeError("Usage: traffic aircraft [get|operator|stats]")
def aircraft(self) -> Optional[str]:
from ..data import aircraft
if not isinstance(self.icao24, str):
return None
res = str(self.icao24)
ac = aircraft.get_unique(res)
if ac is None:
return res
registration = ac["registration"]
typecode = ac["typecode"]
flag = ac["flag"]
if registration is not None:
res += f" · {flag} {registration}"
else:
res = f"{flag} {res}"
if typecode is not None:
res += f" ({typecode})"
if isinstance(filename, str):
filename = Path(filename)
if not filename.parent.exists():
filename.parent.mkdir(parents=True)
altitude = (
"baro_altitude"
if "baro_altitude" in traffic.data.columns
else "altitude"
)
if "typecode" not in traffic.data.columns:
traffic = Traffic(
traffic.data.merge(
aircraft.data[["icao24", "typecode"]].drop_duplicates("icao24"),
on="icao24",
how="inner",
)
)
if "cas" not in traffic.data.columns:
traffic = Traffic(
traffic.data.assign(
cas=vtas2cas(traffic.data.ground_speed, traffic.data[altitude])
)
)
with filename.open("w") as fh:
t_delta = traffic.data.timestamp - traffic.start_time
data = (
traffic.assign_id()
def _repr_html_(self):
from ....data import aircraft, airports
title = f"<h4><b>Flight {self.flight_id}</b> "
title += f"({self.aerodromeOfDeparture} → "
title += f"{self.aerodromeOfDestination})</h4>"
if hasattr(self, "aircraftId"):
title += f"callsign: {self.aircraftId}<br>"
title += f" from {airports[self.aerodromeOfDeparture]}<br>"
title += f" to {airports[self.aerodromeOfDestination]}<br><br>"
cumul = list()
if hasattr(self, "aircraftAddress"):
cumul.append(aircraft[self.aircraftAddress.lower()].T)
cumul.append(
pd.DataFrame.from_dict(
[
{
"EOBT": self.estimatedOffBlockTime
if hasattr(self, "estimatedOffBlockTime")
else None,
"ETOT": self.estimatedTakeOffTime
if hasattr(self, "estimatedTakeOffTime")
else None,
"ATOT": self.actualTakeOffTime
if hasattr(self, "actualTakeOffTime")
else None,
"ETOA": self.estimatedTimeOfArrival
if hasattr(self, "estimatedTimeOfArrival")
def get_aircraft(*args):
from .data import aircraft as data_aircraft
if args[0] == "get":
print(data_aircraft[args[1]])
elif args[0] == "operator":
print(data_aircraft.operator(" ".join(args[1:])))
elif args[0] == "stats":
print(data_aircraft.stats(" ".join(args[1:])))
else:
raise RuntimeError("Usage: traffic aircraft [get|operator|stats]")