from MultiModalTimer import MultiModalTimerConfig, MultiModalTimerModel from safetensors.torch import load_file import gradio as gr import torch import numpy as np import pandas as pd import matplotlib.pyplot as plt import io from PIL import Image import pickle from transformers import CLIPImageProcessor with open('example/inputs.pkl', 'rb') as f: inputs = pickle.load(f) with open('example/targets.pkl', 'rb') as f: targets = pickle.load(f) descriptions = { "NN5 Daily": "Daily cash withdrawal volumes from automated teller machines (ATMs) in the United Kingdom, originally used in the NN5 forecasting competition.", "Australian Electricity": "Half-hourly electricity demand data across five Australian states.", "CIF 2016": "Monthly banking time series used in the CIF 2016 forecasting challenge, reflecting customer financial behaviours.", "Tourism Monthly": "Monthly tourism-related time series used in the Kaggle Tourism forecasting competition, covering various regions and visitor types." } models = {} for dataset in ["NN5_Daily", "Australian_Electricity"]: config = MultiModalTimerConfig.from_pretrained(f"ckpt/CLIPQwenTimer/{dataset}/config.json") model = MultiModalTimerModel(config) state_dict = load_file(f"ckpt/CLIPQwenTimer/{dataset}/model.safetensors") model.load_state_dict(state_dict, strict=False) # device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu") device = torch.device("cpu") model.to(device) models[dataset.replace("_", " ")] = model context_length = { "NN5 Daily": 56, "Australian Electricity": 48, "CIF 2016": 12, "Tourism Monthly": 24 } def predict(dataset, text, df, example_index): time_series = np.array(df.iloc[:, -1]) mean = np.mean(time_series) std = np.std(time_series) time_series_normalized = (time_series-mean)/std input_ids = torch.tensor(time_series_normalized).to(torch.float32).to(device) input_ids = input_ids.unsqueeze(0) plt.figure(figsize=(384/100, 384/100), dpi=100) plt.plot(time_series_normalized, color="black", linestyle="-", linewidth=1, marker="*", markersize=1) plt.xticks([]) plt.yticks([]) plt.subplots_adjust(top=1, bottom=0, right=1, left=0, hspace=0, wspace=0) plt.margins(0,0) buf = io.BytesIO() plt.savefig(buf, format='png') buf.seek(0) plot_img = Image.open(buf).convert('RGB') processor = CLIPImageProcessor() images = processor.preprocess(plot_img)['pixel_values'][0] images = torch.tensor(images).to(device) images = images.unsqueeze(0) text = None if text == '' else text out = models[dataset](**{'input_ids': input_ids, 'images': images, 'texts': text})['logits'] cl = context_length[dataset] out = out[0, :cl] out = out.detach().cpu().numpy() out = out*std+mean input_dates_series = pd.to_datetime(df["Timestamp"]) time_diff = input_dates_series.diff().mode()[0] start_time = input_dates_series.iloc[-1] + time_diff forecast_dates_series = pd.date_range(start=start_time, periods=len(input_dates_series), freq=time_diff) plt.style.use("seaborn-v0_8") fig, ax = plt.subplots() ax.plot(input_dates_series, time_series, color="black", alpha=0.7, linewidth=3, label='Input') ax.plot(forecast_dates_series, out, color='C2', alpha=0.7, linewidth=3, label='Forecast') if example_index == 3: # Custom Input pass else: true = targets[dataset][example_index].iloc[:, -1] ax.plot(forecast_dates_series, true, color='C0', alpha=0.7, linewidth=3, label='True') pass ax.legend() return fig def selected_dataset(dataset): gallery_items = [(Image.open(f'example/img/{dataset.replace(" ", "_")}/{i}.png').convert('RGB'), str(i+1)) for i in range(3)] gallery_items.append((np.ones((64, 64)), 'Custom Input')) return gr.Gallery(gallery_items, interactive=True, columns=2, height="350px", object_fit="contain"), gr.Textbox(value=descriptions[dataset], label="Dataset Description", interactive=False) def selected_example(evt: gr.SelectData): return evt.index def update_time_series_dataframe(dataset, example_index): if example_index is None: return None, None elif example_index == 3: # Custom Input return gr.File(label="Time Series CSV File", file_types=[".csv"], visible=True), gr.Dataframe(value=None, datatype="str", label="Time Series Input", interactive=True) else: df = inputs[dataset][example_index] return gr.File(value=None, visible=False), gr.Dataframe(value=df, label="Time Series Input", interactive=False) def load_csv(file): if file is None: return pd.DataFrame() return pd.read_csv(file.name) with gr.Blocks() as demo: gr.Image( value="logo.png", show_label=False, show_download_button=False, show_fullscreen_button=False, show_share_button=False, interactive=False, height=128, container=False, elem_id="logo" ) gr.HTML(""" """) with gr.Row(): with gr.Column(): dataset_dropdown = gr.Dropdown(["NN5 Daily", "Australian Electricity"], value=None, label="Datasets", interactive=True) dataset_description_textbox = gr.Textbox(label="Dataset Description", interactive=False) example_gallery = gr.Gallery( None, interactive=False ) example_index = gr.State(value=None) example_gallery.select(selected_example, None, example_index) time_series_csv = gr.File(label="Time Series CSV File", file_types=[".csv"], visible=False) time_series_dataframe = gr.Dataframe(value=None, headers=["Timestamp", "Value"], label="Time Series Input", interactive=False) dataset_dropdown.change(selected_dataset, inputs=dataset_dropdown, outputs=[example_gallery, dataset_description_textbox]) dataset_dropdown.change(update_time_series_dataframe, inputs=[dataset_dropdown, example_index], outputs=[time_series_csv, time_series_dataframe]) example_index.change(update_time_series_dataframe, inputs=[dataset_dropdown, example_index], outputs=[time_series_csv, time_series_dataframe]) time_series_csv.change(load_csv, inputs=time_series_csv, outputs=time_series_dataframe) btn = gr.Button("Run") with gr.Column(): forecast_plot = gr.Plot(label="Forecast", format="png") btn.click(predict, inputs=[dataset_dropdown, dataset_description_textbox, time_series_dataframe, example_index], outputs=forecast_plot) if __name__ == "__main__": demo.launch(ssr_mode=False)