Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def api_sensors(self) -> Set[str]:
"""The set of sensors serials you own (require authentication)."""
today = round_time(datetime.now(timezone.utc), by=timedelta(days=1))
c = self.session.get(
f"https://opensky-network.org/api/sensor/myStats"
f"?days={int(today.timestamp())}",
auth=self.auth,
)
c.raise_for_status()
return set(c.json()[0]["stats"].keys())
other_params += "and {}callsign in ({}) ".format(
"sv." if count_airports_params > 0 else "", callsign
)
if bounds is not None:
try:
# thinking of shapely bounds attribute (in this order)
# I just don't want to add the shapely dependency here
west, south, east, north = bounds.bounds # type: ignore
except AttributeError:
west, south, east, north = bounds
other_params += "and lon>={} and lon<={} ".format(west, east)
other_params += "and lat>={} and lat<={} ".format(south, north)
day_min = round_time(start, how="before", by=timedelta(days=1))
day_max = round_time(stop, how="after", by=timedelta(days=1))
if count_airports_params > 0:
where_clause = (
"on sv.icao24 = est.e_icao24 and "
"sv.callsign = est.e_callsign and "
"est.firstseen <= sv.time and "
"sv.time <= est.lastseen "
"where"
)
if arrival_airport is not None and departure_airport is not None:
if airport is not None:
raise RuntimeError(
"airport may not be set if "
"either arrival_airport or departure_airport is set"
def api_range(
self, serial: str, date: Optional[timelike] = None
) -> SensorRange:
"""Wraps a polygon representing a sensor's range.
By default, returns the current range. Otherwise, you may enter a
specific day (as a string, as an epoch or as a datetime)
"""
if date is None:
date = round_time(datetime.now(timezone.utc), by=timedelta(days=1))
else:
date = to_datetime(date)
date = int(date.timestamp())
c = self.session.get(
f"https://opensky-network.org/api/range/days"
f"?days={date}&serials={serial}"
)
c.raise_for_status()
return SensorRange(c.json())
if isinstance(serials, Iterable):
other_tables += f", {table_name}.sensors s "
other_params += "and s.serial in {} ".format(tuple(serials))
columns = "s.serial, s.mintime as time, " + columns
parse_columns = "serial, time, " + parse_columns
elif isinstance(serials, int):
other_tables += f", {table_name}.sensors s "
other_params += "and s.serial = {} ".format((serials))
columns = "s.serial, s.mintime as time, " + columns
parse_columns = "serial, time, " + parse_columns
other_params += "and rawmsg is not null "
day_min = round_time(start, how="before", by=timedelta(days=1))
day_max = round_time(stop, how="after", by=timedelta(days=1))
if (
count_airports_params > 0
or bounds is not None
or callsign is not None
):
where_clause = (
f"on {table_name}.icao24 = est.e_icao24 and "
f"est.firstseen <= {table_name}.mintime and "
f"{table_name}.mintime <= est.lastseen "
"where"
)
if callsign is not None:
if count_airports_params > 0 or bounds is not None:
raise RuntimeError(
body.
"""
if isinstance(airport, str):
from .. import airports
airport_handle = airports[airport]
if airport_handle is None:
raise RuntimeError(f"Unknown airport {airport}")
airport_code = airport_handle.icao
else:
airport_code = airport.icao
if begin is None:
begin = round_time(datetime.now(timezone.utc), by=timedelta(days=1))
begin = to_datetime(begin)
if end is None:
end = begin + timedelta(days=1)
else:
end = to_datetime(end)
begin = int(begin.timestamp())
end = int(end.timestamp())
c = self.session.get(
f"https://opensky-network.org/api/flights/departure"
f"?begin={begin}&airport={airport_code}&end={end}"
)
c.raise_for_status()
return (
other_params += f"and {table_name}.icao24 in ({icao24}) "
if isinstance(serials, Iterable):
other_tables += f", {table_name}.sensors s "
other_params += "and s.serial in {} ".format(tuple(serials))
columns = "s.serial, s.mintime as time, " + columns
parse_columns = "serial, time, " + parse_columns
elif isinstance(serials, int):
other_tables += f", {table_name}.sensors s "
other_params += "and s.serial = {} ".format((serials))
columns = "s.serial, s.mintime as time, " + columns
parse_columns = "serial, time, " + parse_columns
other_params += "and rawmsg is not null "
day_min = round_time(start, how="before", by=timedelta(days=1))
day_max = round_time(stop, how="after", by=timedelta(days=1))
if (
count_airports_params > 0
or bounds is not None
or callsign is not None
):
where_clause = (
f"on {table_name}.icao24 = est.e_icao24 and "
f"est.firstseen <= {table_name}.mintime and "
f"{table_name}.mintime <= est.lastseen "
"where"
)
if callsign is not None:
if count_airports_params > 0 or bounds is not None:
def sensors(self, before: timelike, after: timelike,
bounds: Tuple[float, float, float, float]) -> pd.DataFrame:
before_hour = round_time(before, "before")
after_hour = round_time(after, "after")
try:
# thinking of shapely bounds attribute (in this order)
# I just don't want to add the shapely dependency here
west, south, east, north = bounds.bounds # type: ignore
except AttributeError:
west, south, east, north = bounds
other_where = "and lon>={} and lon<={} ".format(west, east)
other_where += "and lat>={} and lat<={} ".format(south, north)
query = self.sensor_request.format(before_hour=before_hour,
after_hour=after_hour,
other_where=other_where)
end: Optional[timelike] = None,
) -> pd.DataFrame:
"""Returns a flight table associated to an aircraft.
Official documentation
----------------------
This API call retrieves flights for a particular aircraft within a
certain time interval. Resulting flights departed and arrived within
[begin, end]. If no flights are found for the given period, HTTP stats
404 - Not found is returned with an empty response body.
"""
if begin is None:
begin = round_time(datetime.now(timezone.utc), by=timedelta(days=1))
begin = to_datetime(begin)
if end is None:
end = begin + timedelta(days=1)
else:
end = to_datetime(end)
begin = int(begin.timestamp())
end = int(end.timestamp())
c = self.session.get(
f"https://opensky-network.org/api/flights/aircraft"
f"?icao24={icao24}&begin={begin}&end={end}"
)
c.raise_for_status()
return (
pd.DataFrame.from_records(c.json())[