Upload 3 files

app.py

@@ -0,0 +1,140 @@
+import streamlit as st
+import hopsworks
+import joblib
+import pandas as pd
+
+import numpy as np
+import folium
+from streamlit_folium import st_folium, folium_static
+import json
+import time
+from datetime import timedelta, datetime
+from branca.element import Figure
+
+from functions import decode_features, get_model
+
+
+def fancy_header(text, font_size=24):
+    res = f'<span style="color:#ff5f27; font-size: {font_size}px;">{text}</span>'
+    st.markdown(res, unsafe_allow_html=True )
+
+
+st.title('⛅️Air Quality Prediction Project🌩')
+
+progress_bar = st.sidebar.header('⚙️ Working Progress')
+progress_bar = st.sidebar.progress(0)
+st.write(36 * "-")
+fancy_header('\n📡 Connecting to Hopsworks Feature Store...')
+
+project = hopsworks.login()
+fs = project.get_feature_store()
+feature_view = fs.get_feature_view(
+    name = 'air_quality_fv',
+    version = 1
+)
+
+st.write("Successfully connected!✔️")
+progress_bar.progress(20)
+
+st.write(36 * "-")
+fancy_header('\n☁️ Getting batch data from Feature Store...')
+
+start_date = datetime.now() - timedelta(days=1)
+start_time = int(start_date.timestamp()) * 1000
+
+X = feature_view.get_batch_data(start_time=start_time)
+progress_bar.progress(50)
+
+latest_date_unix = str(X.date.values[0])[:10]
+latest_date = time.ctime(int(latest_date_unix))
+
+st.write(f"⏱ Data for {latest_date}")
+
+X = X.drop(columns=["date"]).fillna(0)
+
+data_to_display = decode_features(X, feature_view=feature_view)
+
+progress_bar.progress(60)
+
+st.write(36 * "-")
+fancy_header(f"🗺 Processing the map...")
+
+fig = Figure(width=550,height=350)
+
+my_map = folium.Map(location=[58, 20], zoom_start=3.71)
+fig.add_child(my_map)
+folium.TileLayer('Stamen Terrain').add_to(my_map)
+folium.TileLayer('Stamen Toner').add_to(my_map)
+folium.TileLayer('Stamen Water Color').add_to(my_map)
+folium.TileLayer('cartodbpositron').add_to(my_map)
+folium.TileLayer('cartodbdark_matter').add_to(my_map)
+folium.LayerControl().add_to(my_map)
+
+data_to_display = data_to_display[["city", "temp", "humidity",
+                                            "conditions", "aqi"]]
+
+cities_coords = {("Sundsvall", "Sweden"): [62.390811, 17.306927],
+                 ("Stockholm", "Sweden"): [59.334591, 18.063240],
+                 ("Malmo", "Sweden"): [55.604981, 13.003822]}
+
+if "Kyiv" in data_to_display["city"]:
+    cities_coords[("Kyiv", "Ukraine")]: [50.450001, 30.523333]
+
+data_to_display = data_to_display.set_index("city")
+
+cols_names_dict = {"temp": "Temperature",
+                   "humidity": "Humidity",
+                   "conditions": "Conditions",
+                   "aqi": "AQI"}
+
+data_to_display = data_to_display.rename(columns=cols_names_dict)
+
+cols_ = ["Temperature", "Humidity", "AQI"]
+data_to_display[cols_] = data_to_display[cols_].apply(lambda x: round(x, 1))
+
+for city, country in cities_coords:
+    text = f"""
+            <h4 style="color:green;">{city}</h4>
+            <h5 style="color":"green">
+                <table style="text-align: right;">
+                    <tr>
+                        <th>Country:</th>
+                        <td><b>{country}</b></td>
+                    </tr>
+                    """
+    for column in data_to_display.columns:
+        text += f"""
+                    <tr>
+                        <th>{column}:</th>
+                        <td>{data_to_display.loc[city][column]}</td>
+                    </tr>"""
+    text += """</table>
+                    </h5>"""
+
+    folium.Marker(
+        cities_coords[(city, country)], popup=text, tooltip=f"<strong>{city}</strong>"
+    ).add_to(my_map)
+
+
+# call to render Folium map in Streamlit
+folium_static(my_map)
+progress_bar.progress(80)
+st.sidebar.write("-" * 36)
+
+
+model = get_model(project=project,
+                  model_name="gradient_boost_model",
+                  evaluation_metric="f1_score",
+                  sort_metrics_by="max")
+
+preds = model.predict(X)
+
+cities = [city_tuple[0] for city_tuple in cities_coords.keys()]
+
+next_day_date = datetime.today() + timedelta(days=1)
+next_day = next_day_date.strftime ('%d/%m/%Y')
+df = pd.DataFrame(data=preds, index=cities, columns=[f"AQI Predictions for {next_day}"], dtype=int)
+
+st.sidebar.write(df)
+progress_bar.progress(100)
+st.button("Re-run")

functions.py

@@ -0,0 +1,212 @@
+# The air quality data in this file are requested from IQAir website
+# The weather data in this file are requested from VisualCrossing website
+
+from datetime import datetime
+import requests
+import os
+import joblib
+import pandas as pd
+
+from dotenv import load_dotenv
+load_dotenv()
+
+# API key requested from the IQAir website
+# AIR_QUALITY_API_KEY = "794ac24a-edae-4d19-8ce9-7faf880cc303"
+
+# API key requested from VisualCrossing website
+# WEATHER_API_KEY = "582P2T7XUZWW6FGWEQK2W68LF"
+
+def decode_features(df, feature_view):
+    """Decodes features in the input DataFrame using corresponding Hopsworks Feature Store transformation functions"""
+    df_res = df.copy()
+
+    import inspect
+
+
+    td_transformation_functions = feature_view._batch_scoring_server._transformation_functions
+
+    res = {}
+    for feature_name in td_transformation_functions:
+        if feature_name in df_res.columns:
+            td_transformation_function = td_transformation_functions[feature_name]
+            sig, foobar_locals = inspect.signature(td_transformation_function.transformation_fn), locals()
+            param_dict = dict([(param.name, param.default) for param in sig.parameters.values() if param.default != inspect._empty])
+            if td_transformation_function.name == "min_max_scaler":
+                df_res[feature_name] = df_res[feature_name].map(
+                    lambda x: x * (param_dict["max_value"] - param_dict["min_value"]) + param_dict["min_value"])
+
+            elif td_transformation_function.name == "standard_scaler":
+                df_res[feature_name] = df_res[feature_name].map(
+                    lambda x: x * param_dict['std_dev'] + param_dict["mean"])
+            elif td_transformation_function.name == "label_encoder":
+                dictionary = param_dict['value_to_index']
+                dictionary_ = {v: k for k, v in dictionary.items()}
+                df_res[feature_name] = df_res[feature_name].map(
+                    lambda x: dictionary_[x])
+    return df_res
+
+
+def get_model(project, model_name, evaluation_metric, sort_metrics_by):
+    """Retrieve desired model or download it from the Hopsworks Model Registry.
+    In second case, it will be physically downloaded to this directory"""
+    TARGET_FILE = "model.pkl"
+    list_of_files = [os.path.join(dirpath,filename) for dirpath, _, filenames \
+                     in os.walk('.') for filename in filenames if filename == TARGET_FILE]
+
+    if list_of_files:
+        model_path = list_of_files[0]
+        model = joblib.load(model_path)
+    else:
+        if not os.path.exists(TARGET_FILE):
+            mr = project.get_model_registry()
+            # get best model based on custom metrics
+            model = mr.get_best_model(model_name,
+                                      evaluation_metric,
+                                      sort_metrics_by)
+            model_dir = model.download()
+            model = joblib.load(model_dir + "/model.pkl")
+
+    return model
+
+
+def get_air_json(city_name, AIR_QUALITY_API_KEY):
+    return requests.get(f'https://api.waqi.info/feed/{city_name}/?token={AIR_QUALITY_API_KEY}').json()['data']
+
+def get_air_quality_data(city_name):
+    AIR_QUALITY_API_KEY = os.getenv('AIR_QUALITY_API_KEY')
+    json = get_air_json(city_name, AIR_QUALITY_API_KEY)
+    iaqi = json['iaqi']
+    forecast = json['forecast']['daily']
+    return [
+        city_name,
+        json['aqi'],                 # AQI
+        json['time']['s'][:10],      # Date
+        iaqi['h']['v'],
+        iaqi['p']['v'],
+        iaqi['pm10']['v'],
+        iaqi['t']['v'],
+        forecast['o3'][0]['avg'],
+        forecast['o3'][0]['max'],
+        forecast['o3'][0]['min'],
+        forecast['pm10'][0]['avg'],
+        forecast['pm10'][0]['max'],
+        forecast['pm10'][0]['min'],
+        forecast['pm25'][0]['avg'],
+        forecast['pm25'][0]['max'],
+        forecast['pm25'][0]['min'],
+        # forecast['uvi'][0]['avg'],
+        # forecast['uvi'][0]['avg'],
+        # forecast['uvi'][0]['avg']
+    ]
+
+def get_air_quality_df(data):
+    col_names = [
+        'city',
+        'aqi',
+        'date&time',
+        'iaqi_h',
+        'iaqi_p',
+        'iaqi_pm10',
+        'iaqi_t',
+        'o3_avg',
+        'o3_max',
+        'o3_min',
+        'pm10_avg',
+        'pm10_max',
+        'pm10_min',
+        'pm25_avg',
+        'pm25_max',
+        'pm25_min',
+        # 'uvi_avg',
+        # 'uvi_max',
+        # 'uvi_min',
+    ]
+
+    new_data = pd.DataFrame(
+        data,
+        columns=col_names
+    )
+    new_data.date = new_data.date.apply(timestamp_2_time)
+
+    return new_data
+
+
+def get_weather_json(city, date, WEATHER_API_KEY):
+    return requests.get(f'https://weather.visualcrossing.com/VisualCrossingWebServices/rest/services/timeline/{city.lower()}/{date}?unitGroup=metric&include=days&key={WEATHER_API_KEY}&contentType=json').json()
+
+
+def get_weather_data(city_name, date):
+    WEATHER_API_KEY = os.getenv('WEATHER_API_KEY')
+    json = get_weather_json(city_name, date, WEATHER_API_KEY)
+    data = json['days'][0]
+
+    return [
+        json['address'].capitalize(),
+        data['datetime'],
+        data['tempmax'],
+        data['tempmin'],
+        data['temp'],
+        data['feelslikemax'],
+        data['feelslikemin'],
+        data['feelslike'],
+        data['dew'],
+        data['humidity'],
+        data['precip'],
+        data['precipprob'],
+        data['precipcover'],
+        data['snow'],
+        data['snowdepth'],
+        data['windgust'],
+        data['windspeed'],
+        data['winddir'],
+        data['pressure'],
+        data['cloudcover'],
+        data['visibility'],
+        data['solarradiation'],
+        data['solarenergy'],
+        data['uvindex'],
+        data['conditions']
+    ]
+
+
+def get_weather_df(data):
+    col_names = [
+        'city',
+        'date',
+        'tempmax',
+        'tempmin',
+        'temp',
+        'feelslikemax',
+        'feelslikemin',
+        'feelslike',
+        'dew',
+        'humidity',
+        'precip',
+        'precipprob',
+        'precipcover',
+        'snow',
+        'snowdepth',
+        'windgust',
+        'windspeed',
+        'winddir',
+        'pressure',
+        'cloudcover',
+        'visibility',
+        'solarradiation',
+        'solarenergy',
+        'uvindex',
+        'conditions'
+    ]
+
+    new_data = pd.DataFrame(
+        data,
+        columns=col_names
+    )
+    new_data.date = new_data.date.apply(timestamp_2_time)
+
+    return new_data
+
+def timestamp_2_time(x):
+    dt_obj = datetime.strptime(str(x), '%Y-%m-%d')
+    dt_obj = dt_obj.timestamp() * 1000
+    return int(dt_obj)

requirements.txt

@@ -0,0 +1,11 @@
+branca==0.6.0
+folium==0.14.0
+hopsworks==3.0.5
+joblib==1.2.0
+numpy==1.23.5
+pandas==1.5.2
+python-dotenv==0.21.0
+requests==2.28.1
+streamlit==1.17.0
+streamlit_folium==0.10.0
+dotenv