Spaces:
Sleeping
Sleeping
| import pandas as pd | |
| import numpy as np | |
| import tensorflow as tf | |
| from tensorflow.keras.models import Sequential | |
| from tensorflow.keras.layers import Dense, LSTM | |
| from sklearn.preprocessing import MinMaxScaler | |
| from sklearn.metrics import mean_squared_error | |
| import matplotlib.pyplot as plt | |
| import gradio as gr | |
| # fix random seed for reproducibility | |
| tf.random.set_seed(7) | |
| def train_and_predict(file, epochs): | |
| # Load the dataset | |
| dataframe = pd.read_csv(file.name, usecols=[1], engine='python', encoding="big5") | |
| dataset = dataframe.values.astype('float32') | |
| # Normalize the dataset | |
| scaler = MinMaxScaler(feature_range=(0, 1)) | |
| dataset = scaler.fit_transform(dataset) | |
| # Split into train and test sets | |
| train_size = int(len(dataset) * 0.8) | |
| train, test = dataset[0:train_size,:], dataset[train_size:len(dataset),:] | |
| # Convert an array of values into a dataset matrix | |
| def create_dataset(dataset, look_back=1): | |
| dataX, dataY = [], [] | |
| for i in range(len(dataset)-look_back-1): | |
| a = dataset[i:(i+look_back), 0] | |
| dataX.append(a) | |
| dataY.append(dataset[i + look_back, 0]) | |
| return np.array(dataX), np.array(dataY) | |
| # Reshape into X=t and Y=t+1 | |
| look_back = 1 | |
| trainX, trainY = create_dataset(train, look_back) | |
| testX, testY = create_dataset(test, look_back) | |
| trainX = np.reshape(trainX, (trainX.shape[0], 1, trainX.shape[1])) | |
| testX = np.reshape(testX, (testX.shape[0], 1, testX.shape[1])) | |
| # Create and fit the LSTM network | |
| model = Sequential() | |
| model.add(LSTM(4, input_shape=(1, look_back))) | |
| model.add(Dense(1)) | |
| model.compile(loss='mean_squared_error', optimizer='adam') | |
| model.fit(trainX, trainY, epochs=epochs, batch_size=1, verbose=2) | |
| # Make predictions | |
| trainPredict = model.predict(trainX) | |
| testPredict = model.predict(testX) | |
| # Invert predictions | |
| trainPredict = scaler.inverse_transform(trainPredict) | |
| trainY = scaler.inverse_transform([trainY]) | |
| testPredict = scaler.inverse_transform(testPredict) | |
| testY = scaler.inverse_transform([testY]) | |
| # Calculate root mean squared error | |
| trainScore = np.sqrt(mean_squared_error(trainY[0], trainPredict[:,0])) | |
| testScore = np.sqrt(mean_squared_error(testY[0], testPredict[:,0])) | |
| # Plot predictions | |
| trainPredictPlot = np.empty_like(dataset) | |
| trainPredictPlot[:, :] = np.nan | |
| trainPredictPlot[look_back:len(trainPredict)+look_back, :] = trainPredict | |
| testPredictPlot = np.empty_like(dataset) | |
| testPredictPlot[:, :] = np.nan | |
| testPredictPlot[len(trainPredict)+(look_back*2)+1:len(dataset)-1, :] = testPredict | |
| plt.figure(figsize=(12, 8)) | |
| plt.plot(scaler.inverse_transform(dataset), label='Original Data') | |
| plt.plot(trainPredictPlot, label='Training Predictions', linestyle='--') | |
| plt.plot(testPredictPlot, label='Test Predictions', linestyle='--') | |
| plt.xlabel('Time') | |
| plt.ylabel('Scaled Values') | |
| plt.title('Original Data and Predictions') | |
| plt.legend() | |
| return (f'Train Score: {trainScore:.2f} RMSE\nTest Score: {testScore:.2f} RMSE'), plt | |
| # Gradio interface | |
| file_input = gr.File(label="Upload CSV File") | |
| epochs_input = gr.Slider(minimum=1, maximum=100, value=50, label="Epochs") | |
| output_text = gr.Textbox(label="Training and Testing RMSE Scores") | |
| output_plot = gr.Plot(label="Original Data and Predictions") | |
| gr.Interface( | |
| fn=train_and_predict, | |
| inputs=[file_input, epochs_input], | |
| outputs=[output_text, output_plot], | |
| title="LSTM Model for Time Series Prediction", | |
| description="Upload a CSV file with time series data and specify the number of epochs to train the model." | |
| ).launch() | |