117 lines
4.9 KiB
Python
117 lines
4.9 KiB
Python
import logging
|
|
import os
|
|
import sqlite3
|
|
import time
|
|
from urllib.parse import urlencode
|
|
|
|
from flight_map.geo import Bounds
|
|
from flight_map.network import request_json
|
|
from flight_map.storage import Observation, Session, checkpoint, save_observations
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
TOKEN_URL = (
|
|
"https://auth.opensky-network.org/auth/realms/opensky-network/protocol/openid-connect/token"
|
|
)
|
|
|
|
|
|
class OpenSky:
|
|
def __init__(self) -> None:
|
|
self.client_id = os.environ.get("OPENSKY_CLIENT_ID")
|
|
self.secret = os.environ.get("OPENSKY_CLIENT_SECRET")
|
|
self.token = os.environ.get("OPENSKY_ACCESS_TOKEN")
|
|
self.expires = float("inf") if self.token else 0.0
|
|
if bool(self.client_id) != bool(self.secret):
|
|
raise ValueError("Set both OPENSKY_CLIENT_ID and OPENSKY_CLIENT_SECRET.")
|
|
if not self.client_id and not self.token:
|
|
LOGGER.warning("Using anonymous OpenSky access; daily credit limits apply.")
|
|
|
|
def observations(self, bounds: Bounds) -> list[Observation]:
|
|
if self.client_id and time.time() >= self.expires:
|
|
response = request_json(
|
|
TOKEN_URL,
|
|
data=urlencode(
|
|
{
|
|
"grant_type": "client_credentials",
|
|
"client_id": self.client_id,
|
|
"client_secret": self.secret,
|
|
}
|
|
).encode(),
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|
)
|
|
if not isinstance(response, dict) or not isinstance(response.get("access_token"), str):
|
|
raise RuntimeError("OpenSky authentication returned no access token.")
|
|
self.token = response["access_token"]
|
|
self.expires = time.time() + float(response.get("expires_in", 300)) - 30
|
|
headers = {"Authorization": f"Bearer {self.token}"} if self.token else {}
|
|
query = urlencode(
|
|
{
|
|
"lomin": bounds.west,
|
|
"lamin": bounds.south,
|
|
"lomax": bounds.east,
|
|
"lamax": bounds.north,
|
|
}
|
|
)
|
|
response = request_json(
|
|
"https://opensky-network.org/api/states/all?" + query, headers=headers
|
|
)
|
|
if not isinstance(response, dict) or "states" not in response:
|
|
raise RuntimeError("OpenSky response is missing aircraft states.")
|
|
states = response["states"]
|
|
if states is None:
|
|
return []
|
|
if not isinstance(states, list):
|
|
raise RuntimeError("OpenSky aircraft states are malformed.")
|
|
observations = []
|
|
for state in states:
|
|
if not isinstance(state, list) or len(state) < 14:
|
|
raise RuntimeError("OpenSky returned a malformed aircraft state.")
|
|
if any(state[index] is None for index in (3, 5, 6)):
|
|
continue
|
|
try:
|
|
observations.append(
|
|
Observation(
|
|
str(state[0]).lower(),
|
|
str(state[1]).strip() or None if state[1] else None,
|
|
float(state[3]),
|
|
float(state[5]),
|
|
float(state[6]),
|
|
float(state[7]) if state[7] is not None else None,
|
|
float(state[13]) if state[13] is not None else None,
|
|
bool(state[8]),
|
|
)
|
|
)
|
|
except (ValueError, TypeError) as exc:
|
|
raise RuntimeError("OpenSky returned invalid aircraft observation values.") from exc
|
|
return observations
|
|
|
|
|
|
def collect(
|
|
connection: sqlite3.Connection,
|
|
session: Session,
|
|
client: OpenSky,
|
|
duration: float | None = None,
|
|
) -> None:
|
|
started = time.monotonic()
|
|
checkpoint(connection, session.id, time.time(), "recording")
|
|
try:
|
|
while True:
|
|
tick = time.monotonic()
|
|
observations = client.observations(session.collection_bounds)
|
|
save_observations(connection, session, observations, time.time())
|
|
LOGGER.info("Session %s: received %d observations", session.id, len(observations))
|
|
remaining = None if duration is None else duration - (time.monotonic() - started)
|
|
if remaining is not None and remaining <= 0:
|
|
break
|
|
delay = max(0, session.interval - (time.monotonic() - tick))
|
|
time.sleep(delay if remaining is None else min(delay, remaining))
|
|
if duration is not None and time.monotonic() - started >= duration:
|
|
break
|
|
except KeyboardInterrupt:
|
|
checkpoint(connection, session.id, time.time(), "interrupted")
|
|
LOGGER.info("Recording interrupted; committed observations are preserved.")
|
|
return
|
|
except (RuntimeError, ValueError, OSError, sqlite3.Error):
|
|
checkpoint(connection, session.id, time.time(), "failed")
|
|
raise
|
|
checkpoint(connection, session.id, time.time(), "stopped")
|