Spaces:
Sleeping
Sleeping
File size: 4,060 Bytes
ef02bce |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 |
"""
This module contains functions to fetch weather data from the OpenWeather API and parse the data.
Author: William Parker
"""
import os
import pickle
from datetime import datetime, timezone
from typing import List, Tuple
import pytz
import requests
# Constants
with open(
os.path.join(os.path.dirname(__file__), "helpers", "airport_coordinates.pickle"),
"rb",
) as f:
airport_codes_with_coordinates: List[Tuple[str, float, float]] = pickle.load(f)
with open(
os.path.join(os.path.dirname(__file__), "helpers", "airport_timezones.pickle"),
"rb",
) as f:
airport_codes_with_timezones: List[Tuple[str, str]] = pickle.load(f)
# Private functions
def _get_lat_lon(airport_code: str) -> Tuple[float, float]:
"""
Get the latitude and longitude of an airport from the airport code.
Args:
airport_code (str): The airport code.
Returns:
Tuple[float, float]: The latitude and longitude of the airport.
"""
for tup in airport_codes_with_coordinates:
if tup[0] == airport_code:
return tup[1], tup[2]
raise ValueError(
f"Airport '{airport_code}' not found in airport_codes_with_coordinates"
)
def _closest_forecast(open_weather_data: dict, scheduled_time: datetime, airport: str):
"""
Get the closest weather forecast by time to the scheduled time of
the flight from the list of forecasts in the OpenWeather API response.
Args:
open_weather_data (dict): The OpenWeather API response.
scheduled_time (datetime): The scheduled time of the flight.
airport (str): The airport code.
Returns:
dict: The closest weather forecast.
"""
forecasts: List[dict] = open_weather_data["list"]
airport_timezone_str: str
for tup in airport_codes_with_timezones:
if tup[0] == airport:
airport_timezone_str = tup[1]
airport_timezone = pytz.timezone(airport_timezone_str)
for forecast in forecasts:
utc_datetime = datetime.fromtimestamp(forecast["dt"], tz=timezone.utc)
forecast["date_time"] = utc_datetime.replace(tzinfo=pytz.utc).astimezone(
airport_timezone
)
scheduled_time_localized = airport_timezone.localize(scheduled_time)
closest_forecast = min(
forecasts, key=lambda x: abs(x["date_time"] - scheduled_time_localized)
)
return closest_forecast
def get_weather(airport: str, scheduled_time: datetime, api_key: str):
"""
Connects to the OpenWeather API and fetches the weather forecast for the given airport and time.
Args:
airport (str): The airport code.
scheduled_time (datetime): The scheduled time of the flight.
api_key (str): The OpenWeather API key.
Returns:
dict: The weather forecast data.
"""
lat, lon = _get_lat_lon(airport)
url = f"https://api.openweathermap.org/data/2.5/forecast?lat={lat}&lon={lon}&appid={api_key}&units=imperial"
response = requests.get(url, timeout=5)
if response.status_code == 200:
data = response.json()
return _closest_forecast(data, scheduled_time, airport)
raise ValueError(f"OpenWeather response Code: {response.status_code}")
def parse_weather(weather_data: dict) -> dict[str, List[float]]:
"""
Get the relevant weather data from the OpenWeather API response's forecast.
Args:
weather_data (dict): A single forecast from the OpenWeather API.
Returns:
dict[str, float]: The parsed weather data.
"""
prediction_dict = {}
prediction_dict["Temperature"] = [weather_data["main"]["temp"]]
prediction_dict["Altimeter_Pressure"] = [weather_data["main"]["pressure"]]
prediction_dict["Visibility"] = [weather_data["visibility"]]
prediction_dict["Wind_Speed"] = [weather_data["wind"]["speed"]]
if "rain" in weather_data and "3h" in weather_data["rain"]:
prediction_dict["Precipitation"] = [weather_data["rain"]["3h"] / 3]
else:
prediction_dict["Precipitation"] = [0]
return prediction_dict
|