140 lines
6.6 KiB
Python
140 lines
6.6 KiB
Python
# FlightMonitor/data/live_fetcher.py
|
|
import requests # External library for making HTTP requests
|
|
import json # For parsing JSON responses
|
|
|
|
# It's good practice to define the API endpoint URL as a constant
|
|
OPENSKY_API_URL = "https://opensky-network.org/api/states/all"
|
|
|
|
class LiveFetcher:
|
|
"""
|
|
Fetches live flight data from an external API (e.g., OpenSky Network).
|
|
"""
|
|
|
|
def __init__(self, timeout_seconds=10):
|
|
"""
|
|
Initializes the LiveFetcher.
|
|
|
|
Args:
|
|
timeout_seconds (int): Timeout for the API request in seconds.
|
|
"""
|
|
self.timeout = timeout_seconds
|
|
|
|
def fetch_flights(self, bounding_box: dict):
|
|
"""
|
|
Fetches current flight states within a given geographical bounding box.
|
|
|
|
Args:
|
|
bounding_box (dict): A dictionary containing the keys
|
|
'lat_min', 'lon_min', 'lat_max', 'lon_max'.
|
|
|
|
Returns:
|
|
list: A list of dictionaries, where each dictionary represents a flight
|
|
and contains keys like 'icao24', 'callsign', 'longitude', 'latitude',
|
|
'altitude', 'velocity', 'heading'.
|
|
Returns None if an error occurs (e.g., network issue, API error).
|
|
"""
|
|
if not bounding_box:
|
|
print("Error: Bounding box not provided to fetch_flights.")
|
|
return None
|
|
|
|
params = {
|
|
"lamin": bounding_box["lat_min"],
|
|
"lomin": bounding_box["lon_min"],
|
|
"lamax": bounding_box["lat_max"],
|
|
"lomax": bounding_box["lon_max"],
|
|
}
|
|
|
|
try:
|
|
response = requests.get(OPENSKY_API_URL, params=params, timeout=self.timeout)
|
|
response.raise_for_status() # Raises an HTTPError for bad responses (4XX or 5XX)
|
|
|
|
data = response.json()
|
|
# print(f"Raw data from OpenSky: {json.dumps(data, indent=2)}") # For debugging
|
|
|
|
if data and "states" in data and data["states"] is not None:
|
|
flights_list = []
|
|
for state_vector in data["states"]:
|
|
# According to OpenSky API documentation for /states/all:
|
|
# 0: icao24
|
|
# 1: callsign
|
|
# 2: origin_country
|
|
# 5: longitude
|
|
# 6: latitude
|
|
# 7: baro_altitude (meters)
|
|
# 8: on_ground (boolean)
|
|
# 9: velocity (m/s over ground)
|
|
# 10: true_track (degrees, 0-360, clockwise from North)
|
|
# 13: geo_altitude (meters)
|
|
# Ensure vector has enough elements and longitude/latitude are not None
|
|
if len(state_vector) > 13 and state_vector[5] is not None and state_vector[6] is not None:
|
|
flight_info = {
|
|
"icao24": state_vector[0].strip() if state_vector[0] else "N/A",
|
|
"callsign": state_vector[1].strip() if state_vector[1] else "N/A",
|
|
"origin_country": state_vector[2],
|
|
"longitude": float(state_vector[5]),
|
|
"latitude": float(state_vector[6]),
|
|
"baro_altitude": float(state_vector[7]) if state_vector[7] is not None else None,
|
|
"on_ground": bool(state_vector[8]),
|
|
"velocity": float(state_vector[9]) if state_vector[9] is not None else None,
|
|
"true_track": float(state_vector[10]) if state_vector[10] is not None else None,
|
|
"geo_altitude": float(state_vector[13]) if state_vector[13] is not None else None,
|
|
}
|
|
# Use callsign if available and not empty, otherwise icao24 for display label
|
|
flight_info["display_label"] = flight_info["callsign"] if flight_info["callsign"] != "N/A" and flight_info["callsign"] else flight_info["icao24"]
|
|
|
|
flights_list.append(flight_info)
|
|
return flights_list
|
|
else:
|
|
# No states found or states is null (can happen if area is empty or over water with no traffic)
|
|
print("No flight states returned from API or 'states' field is null.")
|
|
return [] # Return empty list if no flights, not None
|
|
except requests.exceptions.Timeout:
|
|
print(f"Error: API request timed out after {self.timeout} seconds.")
|
|
return None
|
|
except requests.exceptions.HTTPError as http_err:
|
|
print(f"Error: HTTP error occurred: {http_err} - Status: {response.status_code}")
|
|
# You could inspect response.text or response.json() here for more details if needed
|
|
# print(f"Response content: {response.text}")
|
|
return None
|
|
except requests.exceptions.RequestException as req_err:
|
|
print(f"Error: An error occurred during API request: {req_err}")
|
|
return None
|
|
except json.JSONDecodeError:
|
|
print("Error: Could not decode JSON response from API.")
|
|
return None
|
|
except ValueError as val_err: # Handles potential float conversion errors
|
|
print(f"Error: Could not parse flight data value: {val_err}")
|
|
return None
|
|
|
|
if __name__ == '__main__':
|
|
# Example usage for testing the fetcher directly
|
|
print("Testing LiveFetcher...")
|
|
# Example bounding box (Switzerland)
|
|
# Note: OpenSky may return empty for very small or specific unloaded areas
|
|
# Larger areas like Central Europe are more likely to have data.
|
|
example_bbox = {
|
|
"lat_min": 45.8389, "lon_min": 5.9962,
|
|
"lat_max": 47.8229, "lon_max": 10.5226
|
|
}
|
|
# example_bbox = { # Broader Central Europe
|
|
# "lat_min": 45.0, "lon_min": 5.0,
|
|
# "lat_max": 55.0, "lon_max": 15.0
|
|
# }
|
|
|
|
fetcher = LiveFetcher(timeout_seconds=15)
|
|
flights = fetcher.fetch_flights(example_bbox)
|
|
|
|
if flights is not None:
|
|
if flights: # Check if list is not empty
|
|
print(f"Successfully fetched {len(flights)} flights.")
|
|
for i, flight in enumerate(flights):
|
|
if i < 5: # Print details of first 5 flights
|
|
print(
|
|
f" Flight {i+1}: {flight.get('display_label', 'N/A')} "
|
|
f"Lat={flight.get('latitude')}, Lon={flight.get('longitude')}, "
|
|
f"Alt={flight.get('baro_altitude')}m, Vel={flight.get('velocity')}m/s"
|
|
)
|
|
else:
|
|
print("No flights found in the specified area.")
|
|
else:
|
|
print("Failed to fetch flights.") |