""" 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