src/app.py

"""
Main interface for the FlightSure application. 
Preforms Gradio interface setup and calls the backend function.

Author: William Parker
"""

import os
import pickle
from datetime import datetime
from typing import Optional

import gradio as gr
import joblib
import pandas as pd
from dotenv import load_dotenv
from gradio_calendar import Calendar
from numpy import ndarray
from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LinearRegression

import open_weather

# Get variables
load_dotenv()

unvalidated_open_weather_api_key: str | None = os.getenv("OPEN_WEATHER_API_KEY")

env_open_weather_api_key: str

if unvalidated_open_weather_api_key is not None:
    env_open_weather_api_key = unvalidated_open_weather_api_key

with open(
    os.path.join(os.path.dirname(__file__), "helpers", "airport_codes.pickle"), "rb"
) as f:
    airport_codes = pickle.load(f)


# Main processing function
def backend(
    open_weather_api_key_input: str,
    airport: str,
    flight_time: str,
    date: datetime,
):
    """
    Main processing function for the FlightSure application.
    Uses the OpenWeather API to get the weather forecast for the given airport and time.
    Then predicts the flight delay based on the weather forecast.
    """
    # Format date time
    format_str = "%H:%M" if len(flight_time.split(":")) == 2 else "%H:%M:%S"
    time_obj = datetime.strptime(flight_time, format_str).time()
    scheduled_time = datetime.combine(date.date(), time_obj)

    if unvalidated_open_weather_api_key is not None:
        open_weather_api_key = env_open_weather_api_key
    elif open_weather_api_key_input != "":
        open_weather_api_key = open_weather_api_key_input
    else:
        raise ValueError(
            "OpenWeather API Key is missing. Please enter it in the text box or in the .env file."
        )

    # Get weather
    raw_forecast = open_weather.get_weather(
        airport, scheduled_time, open_weather_api_key
    )
    input_dict = open_weather.parse_weather(raw_forecast)

    # Create a text summary of the weather

    weather_forecast = f"""Temperature: {round(input_dict["Temperature"][0])} F
Pressure: {input_dict["Altimeter_Pressure"][0]} hPa
Visibility: {input_dict["Visibility"][0]} m
Wind Speed: {input_dict["Wind_Speed"][0]} mph
Rain: {round(input_dict["Precipitation"][0], 2)} mm"""

    input_values = pd.DataFrame(input_dict)

    # Run the models

    model_path = os.path.join(os.path.dirname(__file__), "..", "models", airport)

    weather_cancellation_classification_path = os.path.join(
        model_path, "weather_cancellation_classification.pkl"
    )
    weather_cancellation_classification: RandomForestClassifier = joblib.load(
        weather_cancellation_classification_path
    )

    cancellation_prediction: bool = weather_cancellation_classification.predict(
        input_values
    )[0]
    cancellation_chance: ndarray = weather_cancellation_classification.predict_proba(
        input_values
    )[0]

    cancellation_text = f"We predict that your flight will {'' if cancellation_prediction else 'not'} be canceled. Chance of cancellation: {round(cancellation_chance[1] * 100, 2)}%."

    weather_delay_classification_path = os.path.join(
        model_path, "weather_delay_classification.pkl"
    )
    weather_delay_classification: RandomForestClassifier = joblib.load(
        weather_delay_classification_path
    )

    delay_bool_prediction: bool = weather_delay_classification.predict(input_values)[0]
    delay_bool_chance: ndarray = weather_delay_classification.predict_proba(
        input_values
    )[0]

    delay_text = f"We predict that your flight will {'' if delay_bool_prediction else 'not'} be delayed. Chance of delay: {round(delay_bool_chance[1] * 100, 2)}%. "

    weather_delay_regression_path = os.path.join(
        model_path, "weather_delay_regression.pkl"
    )
    weather_delay_regression: LinearRegression = joblib.load(
        weather_delay_regression_path
    )

    delay_length: float = weather_delay_regression.predict(input_values)[0]

    if delay_bool_prediction:
        delay_length_text = (
            f"The predicted delay length is {round(delay_length)} minutes."
        )
    else:
        delay_length_text = "N/A"

    return weather_forecast, cancellation_text, delay_text, delay_length_text


# Interface Components
open_weather_api_textbox = gr.Textbox(
    label="OpenWeather API Key:",
    info="Leave blank if you set this in the .env file."
    )

dropdown = gr.components.Dropdown(
    choices=airport_codes, label="Select your airport code:"
)

time_textbox = gr.components.Textbox(label="Enter departure time:")

calendar = Calendar(
    type="datetime",
    label="Select date of departure:",
    info="Click the calendar icon to bring up the calendar.",
)

inputs = [open_weather_api_textbox, dropdown, time_textbox, calendar]

weather_stats_textbox = gr.Textbox(label="Weather Forecast:")

cancellation_chance_textbox = gr.Textbox(label="Chance of Cancellation:")

delay_chance_textbox = gr.Textbox(label="Chance of Delay:")

delay_length_textbox = gr.Textbox(label="Length of Delay:")

outputs = [
    weather_stats_textbox,
    cancellation_chance_textbox,
    delay_chance_textbox,
    delay_length_textbox,
]

# Interface
interface = gr.Interface(
    fn=backend,
    inputs=inputs,
    outputs=outputs,
    title="FlightSure",
    allow_flagging=False,
)

# Launch
if __name__ == "__main__":
    interface.launch()