Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |