Spaces:
Runtime error
Runtime error
| import json | |
| import io | |
| import os | |
| import tempfile | |
| import datetime | |
| import gradio as gr | |
| import pandas as pd | |
| import matplotlib.pyplot as plt | |
| import seaborn as sns | |
| import numpy as np | |
| from sktime.utils.plotting import plot_series | |
| from statsmodels.tsa.seasonal import seasonal_decompose | |
| from statsmodels.graphics.tsaplots import plot_acf, plot_pacf | |
| from src.forecaster import Forecaster | |
| from src.forecaster.models import XGBoost | |
| from src.analyser import Analyser | |
| from src.idsc import IDSC | |
| from src.forecaster.models import ProphetForecaster | |
| class GradioApp(): | |
| def __init__( | |
| self | |
| ) -> None: | |
| self.forecaster = Forecaster() | |
| self.analyser = Analyser() | |
| self.idsc = IDSC() | |
| self.historical_demo_data = 'data/multivariate/demo_historical.csv' | |
| self.future_demo_data = 'data/multivariate/demo_future.csv' | |
| self.data: pd.DataFrame = None | |
| self.n_predict = 3 | |
| self.window_length = 7 | |
| self.target_column = 'y' | |
| self.exog_columns = [] | |
| # Define if the model's result is going to be rounded | |
| self.round_results = True | |
| # Delete old temp files oder than n minutes | |
| self.delete_file_old_than_n_minutes = 10 | |
| self.plot_figsize_full_screen = (20, 4) | |
| # -------------------- # | |
| # Model Related Params # | |
| # -------------------- # | |
| # XGBoost # | |
| self.xgboost = XGBoost() | |
| self.xgboost_cv = False | |
| self.xgboost_params = self.xgboost.cv_params | |
| self.xgboost_strategy = 'recursive' | |
| self.xgboost_forecast = None | |
| self.xgboost_test = None | |
| print('Init Gradio app') | |
| # Prophet # | |
| self.prophet = ProphetForecaster() | |
| self.prophet__seasonality_mode = 'multiplicative' | |
| self.prophet__add_country_holidays = {'country_name': 'Singapore'} | |
| self.prophet__yearly_seasonality = True | |
| self.prophet__weekly_seasonality = False | |
| self.prophet__daily_seasonality = False | |
| def checkbox__round_results__change(self, val): | |
| self.round_results = val | |
| def textbox__target_column__change(self, val): | |
| print('Updating textbox__target_column:', val) | |
| self.target_column = val | |
| def btn__profiling__click(self): | |
| self.analyser.fit(self.data) | |
| self.analyser.profiling() | |
| return ( | |
| self.update__md__profiling(), | |
| self.update__plot__changepoints()) | |
| def btn__plot_correlation__click(self): | |
| return (self.update__plot__correlation()) | |
| def file__historical__upload( | |
| self, | |
| file | |
| ): | |
| self.data = pd.read_csv( | |
| file.name, | |
| index_col='datetime', | |
| parse_dates=['datetime']) | |
| print('[file__historical__upload]') | |
| return ( | |
| self.update__df__table_view(), | |
| self.update__dropdown__chart_view_filter(), | |
| self.update__dropdown__seasonality_decompose(), | |
| self.update__plot__chart_view()) | |
| def file__future__upload( | |
| self, | |
| file | |
| ): | |
| self.__handle_future_data_upload(file.name) | |
| return ( | |
| self.update__df__table_view(), | |
| self.update__dropdown__chart_view_filter(), | |
| self.update__dropdown__seasonality_decompose(), | |
| self.update__plot__chart_view(), | |
| self.update__number__n_predict()) | |
| def btn__load_future_demo__click( | |
| self | |
| ): | |
| self.__handle_future_data_upload(self.future_demo_data) | |
| # [df__table_view, number__n_predict] | |
| return ( | |
| self.update__df__table_view(), | |
| self.update__dropdown__chart_view_filter(), | |
| self.update__dropdown__seasonality_decompose(), | |
| self.update__plot__chart_view(), | |
| self.update__number__n_predict()) | |
| def __handle_future_data_upload( | |
| self, | |
| path | |
| ): | |
| data = pd.read_csv( | |
| path, | |
| index_col='datetime', | |
| parse_dates=['datetime']) | |
| self.exog_columns = data.columns.tolist() | |
| self.n_predict = len(data) | |
| print( | |
| f"[file__future__upload] with {self.exog_columns} columns") | |
| self.data = pd.concat( | |
| [self.data, data], | |
| axis=0) | |
| def number__n_predict__change( | |
| self, | |
| val | |
| ): | |
| print(f'[number__n_predict__change], {val}') | |
| self.n_predict = val | |
| def number__window_length__change( | |
| self, | |
| val): | |
| print(f'[number__window_length__change], {val}') | |
| self.window_length = val | |
| def btn__fit_data__click( | |
| self): | |
| data = self.data.drop(columns=self.exog_columns).dropna(how='any') | |
| self.forecaster.fit( | |
| data, | |
| target_col=self.target_column, | |
| n_predict=self.n_predict, | |
| window_length=self.window_length, | |
| exog=None if len( | |
| self.exog_columns) == 0 else self.data[self.exog_columns]) | |
| return ( | |
| gr.Number(interactive=False), # number__n_predict | |
| gr.Number(interactive=False), # number__window_length | |
| gr.File(interactive=False), # file__historical | |
| gr.File(interactive=False), # file__future | |
| gr.Button(visible=False), # btn__fit_data | |
| gr.Column(visible=True), # column__models | |
| gr.Button(visible=False), # btn__load_historical_demo | |
| gr.Button(visible=False), # btn__load_future_demo | |
| self.update__md__forecast_data_info() | |
| ) | |
| def btn__load_historical_demo__click( | |
| self | |
| ): | |
| self.data = pd.read_csv( | |
| self.historical_demo_data, | |
| index_col='datetime', | |
| parse_dates=['datetime']) | |
| return ( | |
| self.update__df__table_view(), | |
| self.update__dropdown__chart_view_filter(), | |
| self.update__dropdown__seasonality_decompose(), | |
| self.update__plot__chart_view() | |
| ) | |
| def dropdown__chart_view_filter__change(self, options): | |
| return (self.update__plot__chart_view(options)) | |
| def dropdown__seasonality_decompose__change(self, col): | |
| return ( | |
| self.update__plot__seasonality_decompose(col), | |
| self.update__plot_acg_pacf(col)) | |
| # ------------------------ # | |
| # XGboost Model Operations # | |
| # ------------------------ # | |
| def btn__train_xgboost__click(self): | |
| (test, forecast, best_params) = self.xgboost.fit_predict( | |
| y=self.forecaster.y, | |
| y_train=self.forecaster.y_train, | |
| window_length=self.forecaster.window_length, | |
| fh=self.forecaster.fh, | |
| fh_test=self.forecaster.fh_test, | |
| params=self.xgboost_params, | |
| X=self.forecaster.X, | |
| X_train=self.forecaster.X_train, | |
| X_test=self.forecaster.X_test, | |
| X_future=self.forecaster.X_future | |
| ) | |
| print(test, forecast, best_params) | |
| self.xgboost_forecast = forecast | |
| self.xgboost_test = test | |
| return ( | |
| self.update__plot__xgboost_result(test, forecast), | |
| self.update__file__xgboost_result(), | |
| self.update__df__xgboost_result()) | |
| def btn__set_xgboost_params__click(self, text): | |
| params = json.loads(text.replace("'", '"')) | |
| self.xgboost_params = params | |
| return ( | |
| self.update__json_xgboost_params() | |
| ) | |
| def checkbox__xgboost_round__change(self, val): | |
| self.xgboost.round_result = val | |
| # ----------------------------------- # | |
| # Prophet Model Operations & Updaters # | |
| # ----------------------------------- # | |
| def btn__forecast_with_prophet__click(self): | |
| self.prophet.fit_predict( | |
| self.forecaster.y_train, | |
| self.forecaster.y, | |
| self.forecaster.fh, | |
| self.forecaster.fh_test, | |
| self.forecaster.period, | |
| self.forecaster.freq, | |
| X=self.forecaster.exog, | |
| seasonality_mode=self.prophet__seasonality_mode, | |
| add_country_holidays=self.prophet__add_country_holidays, | |
| yearly_seasonality=self.prophet__yearly_seasonality, | |
| weekly_seasonality=self.prophet__weekly_seasonality, | |
| daily_seasonality=self.prophet__daily_seasonality, | |
| round_val=self.round_results) | |
| return ( | |
| self.update__plot__prophet_result(), | |
| self.update__file__prophet_result(), | |
| self.update__df__prophet_result()) | |
| def update__plot__prophet_result(self): | |
| fig, ax = plt.subplots(figsize=self.plot_figsize_full_screen) | |
| plot_series( | |
| self.forecaster.y_train[-2 * self.forecaster.period:], | |
| self.forecaster.y_test, | |
| self.prophet.predict, | |
| self.prophet.forecast, | |
| pred_interval=self.prophet.forecast_interval, | |
| labels=['Train', 'Test', 'Predicted - Test', 'Forecast'], | |
| ax=ax) | |
| ax.set_title('Prophet Forecast Result') | |
| ax.legend(loc='upper left') | |
| fig.tight_layout() | |
| return gr.Plot(fig) | |
| def update__file__prophet_result(self): | |
| prophet_forecast_df = pd.DataFrame(self.prophet.forecast) | |
| path = self.__create_temp_csv_file(prophet_forecast_df) | |
| return gr.File(path) | |
| def update__df__prophet_result(self): | |
| prophet_forecast_df = self.prophet.forecast.reset_index() | |
| return gr.Dataframe(value=prophet_forecast_df) | |
| # =============================== # | |
| # || Gradio Component Updaters || # | |
| # =============================== # | |
| def update__plot__changepoints(self): | |
| fig, axs = plt.subplots(2, 1, figsize=(20, 8)) | |
| axs[0].plot(self.data[['y']]) | |
| axs[0].text(self.data.index[0], | |
| axs[0].get_ylim()[1]*0.9, | |
| self.analyser.quantity_predictability[0], | |
| fontsize=20) | |
| for i, p in enumerate(self.analyser.quantity_change_points): | |
| axs[0].axvline(x=p) | |
| axs[0].text(p, | |
| axs[0].get_ylim()[1]*0.9, | |
| self.analyser.quantity_predictability[i+1], | |
| fontsize=20) | |
| axs[1].plot(self.data[['y']]) | |
| axs[1].text(self.data.index[0], | |
| axs[1].get_ylim()[1]*0.9, | |
| self.analyser.intermittent_predictability[0], | |
| fontsize=20) | |
| for i, p in enumerate(self.analyser.intermittent_change_points): | |
| axs[1].axvline(x=p) | |
| axs[1].text(p, | |
| axs[1].get_ylim()[1]*0.9, | |
| self.analyser.intermittent_predictability[i+1], | |
| fontsize=20) | |
| axs[0].set_title('Quantity Change Points & Predictability') | |
| axs[1].set_title('Intermittent Change Points & Predictability') | |
| fig.tight_layout() | |
| return gr.Plot(fig) | |
| def update__md__profiling(self): | |
| return (f""" | |
| \n### Data Characteristic: | |
| \n # {self.analyser.characteristic} | |
| \n --- | |
| \n### Quantity Change Points: {self.analyser.quantity_change_points.astype(str).tolist()} | |
| \n### Quantity Predictability: {self.analyser.quantity_predictability} | |
| \n### Intermittent Change Points: {self.analyser.intermittent_change_points.astype(str).tolist()} | |
| \n### Intermittent Predictability: {self.analyser.intermittent_predictability} | |
| """) | |
| def update__md__forecast_data_info(self): | |
| return gr.Markdown(value=f' \ | |
| **Forecasting for these timestamps**: \ | |
| {self.forecaster.fh.to_pandas().astype(str).tolist()} \ | |
| \n **Data Period**: {self.forecaster.period} \ | |
| \n **Data Frequency**: {self.forecaster.freq} \ | |
| ') | |
| def update__plot__correlation(self): | |
| fig, ax = plt.subplots(figsize=(20, 8)) | |
| corr = self.data.corr(numeric_only=True) | |
| mask = np.triu(np.ones_like(corr, dtype=bool)) | |
| sns.heatmap( | |
| corr, | |
| mask=mask, | |
| square=True, | |
| annot=True, | |
| cmap='coolwarm', | |
| linewidths=.5, | |
| cbar_kws={"shrink": .5}, | |
| ax=ax) | |
| fig.tight_layout() | |
| return gr.Plot(fig) | |
| def update__df__table_view( | |
| self | |
| ): | |
| data = self.data.reset_index() | |
| return gr.Dataframe(value=data) | |
| def update__number__n_predict( | |
| self | |
| ): | |
| return gr.Number(self.n_predict, interactive=False) | |
| def update__dropdown__chart_view_filter(self): | |
| options = self.data.columns.tolist() | |
| return gr.Dropdown(options, value=options) | |
| def update__dropdown__seasonality_decompose(self): | |
| options = self.data.columns.tolist() | |
| return gr.Dropdown(options) | |
| def update__plot__seasonality_decompose(self, col): | |
| seasonal = seasonal_decompose(self.data[col].dropna()) | |
| fig = seasonal.plot() | |
| return gr.Plot(fig) | |
| def update__plot_acg_pacf(self, col): | |
| fig, axs = plt.subplots(2, 1, sharex=True, sharey=True) | |
| plot_acf(self.data[col].dropna(), ax=axs[0], zero=False) | |
| plot_pacf(self.data[col].dropna(), ax=axs[1], zero=False) | |
| axs[0].set_title('Auto Correlation') | |
| axs[1].set_title('Partial Auto Correlation') | |
| return gr.Plot(fig) | |
| # ---------------------- # | |
| # Update XGboost Results # | |
| # ---------------------- # | |
| def update__json_xgboost_params(self): | |
| return gr.JSON(value=self.xgboost_params) | |
| def update__plot__xgboost_result(self, test, predict): | |
| fig, ax = plt.subplots(figsize=self.plot_figsize_full_screen) | |
| plot_series( | |
| self.forecaster.y_train[-2*self.forecaster.period:], | |
| self.forecaster.y_test, | |
| test, | |
| predict, | |
| labels=["y_train (part)", "y_test", "y_pred", 'y_forecast'], | |
| x_label='Date', | |
| ax=ax) | |
| ax.set_xticklabels(ax.get_xticklabels(), rotation=45) | |
| fig.tight_layout() | |
| return gr.Plot(fig) | |
| def update__plot__chart_view(self, cols=None): | |
| fig, ax = plt.subplots(figsize=self.plot_figsize_full_screen) | |
| _cols = cols | |
| if _cols is None: | |
| _cols = self.data.columns | |
| print('[update__plot__chart_view]') | |
| for col in _cols: | |
| ax.plot(self.data[[col]], label=col) | |
| fig.legend() | |
| fig.tight_layout() | |
| return gr.Plot(fig) | |
| def update__file__xgboost_result(self): | |
| path = self.__create_temp_csv_file(self.xgboost_forecast) | |
| return gr.File(path) | |
| def update__df__xgboost_result(self): | |
| # xgboost_forecast is actually a Series instead of proper DataFrame | |
| # Re constructing a proper dataframe for gradio to take | |
| data = pd.DataFrame( | |
| {"datetime": self.xgboost_forecast.index, | |
| "y": self.xgboost_forecast.values}) | |
| return gr.Dataframe(value=data) | |
| # ------------- # | |
| # Util Function # | |
| # ------------- # | |
| def __create_temp_csv_file(self, df) -> str: | |
| time_format = "%Y%m%d%H%M%S" | |
| directory = 'temp' | |
| now = datetime.datetime.now() | |
| # Check if there are old files, remove them # | |
| for filename in os.listdir(directory): | |
| file_path = os.path.join(directory, filename) | |
| file_time = datetime.datetime.strptime( | |
| filename.split('.')[0], time_format) | |
| # If the file is older than 3 minutes, delete the file | |
| if now > datetime.timedelta( | |
| minutes=self.delete_file_old_than_n_minutes) + file_time: | |
| print('deleting olde file: ', filename) | |
| os.remove(file_path) | |
| new_file_name = now.strftime(format=time_format) + '.csv' | |
| new_file_path = os.path.join(directory, new_file_name) | |
| df.to_csv(new_file_path) | |
| return new_file_path | |