# Loading packages import datetime import joblib import pandas as pd import numpy as np import matplotlib.pyplot as plt import warnings import hopsworks import streamlit as st import json import os import seaborn as sns import time import random # Configuring the web page and setting the page title and icon st.set_page_config( page_title='Parking Occupacy Detection', page_icon='🅿️', initial_sidebar_state='expanded') # Ignoring filtering warnings warnings.filterwarnings("ignore") def color_prediction(val): color = 'red' if val=='detection' else 'green' if val=='no_detection' else 'yellow' return f'background-color: {color}' # Setting the title and adding text st.title('Parking Occupancy Detection') # Creating tabs for the different features of the application tab1,tab2,tab3,tab4, tab5 = st.tabs(['Parking lot status', 'Magnetic Field Explorer', 'About', 'Dataset and visualisations', 'Model performance']) with tab1: # Logging in to Hopsworks and loading the feature store project = hopsworks.login(project = "miknie20", api_key_value=os.environ['HOPSWORKS_API_KEY']) fs = project.get_feature_store() col1, col2 = st.columns(2) with col1: st.markdown("**Parking place near building:**") # Function to load the building model @st.cache_data() def get_building_model(project=project): mr = project.get_model_registry() building_model = mr.get_model("building_hist_model", version = 2) building_model_dir = building_model.download() return joblib.load(building_model_dir + "/building_hist_model.pkl") # Retrieving model building_hist_model = get_building_model() # Loading the feature group with latest data for building api_building_newest_fg = fs.get_feature_group(name = 'api_building_newest', version = 1) # Function to loading the feature group with latest data for building as a dataset @st.cache_data() def retrieve_building(feature_group=api_building_newest_fg): api_building_newest_fg = feature_group.select(["time", "x", "y", "z"]) df_building = api_building_newest_fg.read(read_options={"use_hive": True}) return df_building # Retrieving building data building_new = retrieve_building() building_most_recent_prediction = building_new[['x', 'y', 'z']] building_most_recent_prediction = building_hist_model.predict(building_most_recent_prediction) building_new['prediction'] = building_most_recent_prediction building_new = building_new.set_index(['time']) st.dataframe(building_new[['prediction']].tail(5)) with col2: st.markdown("**Parking place near bikelane:**") # Function to load the bikelane model @st.cache_data() def get_bikelane_model(project=project): mr = project.get_model_registry() bikelane_model = mr.get_model("bikelane_hist_model", version = 1) bikelane_model_dir = bikelane_model.download() return joblib.load(bikelane_model_dir + "/bikelane_hist_model.pkl") # Retrieving model bikelane_hist_model = get_bikelane_model() # Loading the feature group with latest data for bikelane api_bikelane_newest_fg = fs.get_feature_group(name = 'api_bikelane_newest', version = 1) # Function to loading the feature group with latest data for building as a dataset @st.cache_data() def retrieve_bikelane(feature_group=api_bikelane_newest_fg): api_bikelane_newest_fg = feature_group.select(["time", "x", "y", "z"]) df_bikelane = api_bikelane_newest_fg.read(read_options={"use_hive": True}) return df_bikelane # Retrieving building data bikelane_new = retrieve_bikelane() bikelane_most_recent_prediction = bikelane_new[['x', 'y', 'z']] bikelane_most_recent_prediction = bikelane_hist_model.predict(bikelane_most_recent_prediction) bikelane_new['prediction'] = bikelane_most_recent_prediction bikelane_new = bikelane_new.set_index(['time']) st.dataframe(bikelane_new[['prediction']].tail(5)) if st.button("Update status"): # values cannot be used in st.session_state!! if 'my_values' not in st.session_state: st.session_state.my_values = list() if not st.session_state.my_values: st.session_state.my_values.append(0) new_value = st.session_state.my_values[-1] + random.randrange(-100, 100) / 100 st.session_state.my_values.append(new_value) time.sleep(.2) st.experimental_rerun() with tab2: def explore_magnetic_field(model, x, y, z): input_list = [x, y, z] res = model.predict(np.asarray(input_list).reshape(1,-1)) explorer_prediction = res[0] if explorer_prediction == 'detection': label = "Vehicle detected" else: label = "No vehicle detected" return label st.markdown('**Experiment with building model:**') x_input_building = st.slider("Choose your x-value", -232, 909, 0) y_input_building = st.slider("Choose your y-value", -1112, 435, 0) z_input_building = st.slider("Choose your z-value", -1648, 226, 0) if st.button("Predict building input"): building_input_prediction = explore_magnetic_field(building_hist_model, x_input_building, y_input_building, z_input_building) st.write(building_input_prediction) st.markdown('**Experiment with bikelane model:**') x_input_bikelane = st.slider("Choose your x-value", -547, 288, 0) y_input_bikelane = st.slider("Choose your y-value", -1007, 786, 0) z_input_bikelane = st.slider("Choose your z-value", -1475, 16, 0) if st.button("Predict bikelane input"): bikelane_input_prediction = explore_magnetic_field(bikelane_hist_model, x_input_bikelane, y_input_bikelane, z_input_bikelane) st.write(bikelane_input_prediction) with tab3: st.markdown('This application is made as part of the module "Data Engineering and Machine Learning Operations in Business - F2024" in Business Data Science 2nd Semester at Aalborg University Business School.') st.markdown('The application is made by Annika and Mikkel and is divided into 4 tabs:') st.markdown('* **Parking lot status:** The first tab includes the actual interface, where the goal has been to make a simple UI which shows if 3 parking spaces are occupied or available.') st.markdown('* **About:** In the second tab (the current tab) you can get some information about the interface.') st.markdown('* **Dataset and visualisations:** The third tab contains an overview of the training data and also includes EDAs for each individual parking space. The goal with these EDAs is to give you some information about when the parking spaces usually are occupied.') st.markdown('* **Model Performance:** The fourth tab explains how the underlying Machine Learning Model performs and how the predictor works.') with tab4: st.markdown('...') # Loading the parking detection feature view #parking_detection_fv = fs.get_feature_view( # name = 'parking_detection_fv', # version = 1) # Function to loading the parking detection feature view as a dataset #@st.cache_data() #def retrieve_batch_data(feature_view=parking_detection_fv): # batch_data = feature_view.get_batch_data() # return batch_data # Retrieving batch data #batch_data = retrieve_batch_data() # Display dataset overview #st.subheader("Dataset Overview") #st.dataframe(batch_data.head()) with tab5: st.markdown('The predictions are made on the basis of a KNearestNeighbours model.') #st.write(model) # Making a countplot of the predictions #predictions = model.predict(batch_data) #df_test = batch_data.copy() #df_test['predictions'] = predictions #st.dataframe(df_test.head()) #plt.figure(figsize=(10, 6)) #sns.set_style("darkgrid") #sns.countplot(data=df_test, x="predictions") #plt.title('Distribution of Predictions') #st.pyplot(plt) # Confusion Matrix st.subheader("Confusion Matrix") st.markdown('In this confusion matrix...') # Loading the feature view with latest data for building #building_new_fv = fs.get_feature_view( # name = 'building_new_fv', # version = 1) # Function to loading the feature view with latest data for building as a dataset #@st.cache_data() #def retrieve_building(feature_view=building_new_fv): # building_new_fv = feature_view.get_batch_data() # return building_new_fv # Retrieving building data #building_new = retrieve_building() # .style.applymap(color_prediction, subset=['prediction']) # Loading the feature view with latest data for bikelane #bikelane_new_fv = fs.get_feature_view( # name = 'bikelane_new_fv', # version = 1) # Function to loading the feature view with latest data for bikelane as a dataset #@st.cache_data() #def retrieve_bikelane(feature_view=bikelane_new_fv): # bikelane_new_fv = feature_view.get_batch_data() # return bikelane_new_fv # Retrieving bikelane data #bikelane_new = retrieve_bikelane() #st.markdown('Parking Space near Bikelane:') #bikelane_most_recent_prediction = bikelane_new[['x', 'y', 'z']] #bikelane_most_recent_prediction = bikelane_hist_model.predict(bikelane_most_recent_prediction) #bikelane_new['prediction'] = bikelane_most_recent_prediction #bikelane_new = bikelane_new.set_index(['time']) #st.dataframe(bikelane_new[['prediction']].tail(5).style.applymap(color_prediction, subset=['prediction']))