|
import gradio as gr |
|
from huggingface_hub import from_pretrained_keras |
|
import pandas as pd |
|
import numpy as np |
|
import json |
|
from matplotlib import pyplot as plt |
|
|
|
f = open('scaler.json') |
|
scaler = json.load(f) |
|
|
|
TIME_STEPS = 288 |
|
|
|
|
|
def create_sequences(values, time_steps=TIME_STEPS): |
|
output = [] |
|
for i in range(len(values) - time_steps + 1): |
|
output.append(values[i : (i + time_steps)]) |
|
return np.stack(output) |
|
|
|
|
|
def normalize_data(data): |
|
df_test_value = (data - scaler["mean"]) / scaler["std"] |
|
return df_test_value |
|
|
|
def plot_test_data(df_test_value): |
|
fig, ax = plt.subplots() |
|
df_test_value.plot(legend=False, ax=ax) |
|
return fig |
|
|
|
def get_anomalies(df_test_value): |
|
|
|
x_test = create_sequences(df_test_value.values) |
|
model = from_pretrained_keras("remeajayi/timeseries-anomaly-detection") |
|
|
|
|
|
x_test_pred = model.predict(x_test) |
|
test_mae_loss = np.mean(np.abs(x_test_pred - x_test), axis=1) |
|
test_mae_loss = test_mae_loss.reshape((-1)) |
|
|
|
|
|
anomalies = test_mae_loss > scaler["threshold"] |
|
return anomalies |
|
|
|
def plot_anomalies(df_test_value, data, anomalies): |
|
|
|
anomalous_data_indices = [] |
|
for data_idx in range(TIME_STEPS - 1, len(df_test_value) - TIME_STEPS + 1): |
|
if np.all(anomalies[data_idx - TIME_STEPS + 1 : data_idx]): |
|
anomalous_data_indices.append(data_idx) |
|
df_subset = data.iloc[anomalous_data_indices] |
|
fig, ax = plt.subplots() |
|
data.plot(legend=False, ax=ax) |
|
df_subset.plot(legend=False, ax=ax, color="r") |
|
return fig |
|
|
|
def master(file): |
|
|
|
data = pd.read_csv(file, parse_dates=True, index_col="timestamp") |
|
df_test_value = normalize_data(data) |
|
|
|
plot1 = plot_test_data(df_test_value) |
|
|
|
anomalies = get_anomalies(df_test_value) |
|
|
|
plot2 = plot_anomalies(df_test_value, data, anomalies) |
|
return plot2 |
|
|
|
outputs = gr.Plot() |
|
|
|
|
|
iface = gr.Interface(master, |
|
gr.inputs.File(label="csv file"), |
|
outputs=outputs, |
|
examples=["art_daily_jumpsup.csv"], title="Timeseries Anomaly Detection Using an Autoencoder", |
|
description = "Anomaly detection of timeseries data.", |
|
article = "Space by: <a href=\"https://www.linkedin.com/in/olohireme-ajayi/\">Reme Ajayi</a> <br> Keras Example by <a href=\"https://github.com/pavithrasv/\"> Pavithra Vijay</a>") |
|
|
|
|
|
|
|
iface.launch() |