Create app.py

app.py

@@ -0,0 +1,297 @@
+import pandas as pd
+import gradio as gr
+from pathlib import Path
+import plotly.express as px
+import numpy as np
+import torch
+from chronos import ChronosPipeline
+from datetime import datetime
+import matplotlib.pyplot as plt
+import matplotlib.ticker as ticker
+
+
+def filter_data(start, end, df_state, select_product_column, date_column, target_column):
+
+  if not date_column:
+    raise gr.Error("Please select a Date column")
+
+  if not target_column:
+    raise gr.Error("Please select a target column")
+
+  start_datetime = pd.to_datetime(datetime.utcfromtimestamp(start))
+  end_datetime = pd.to_datetime(datetime.utcfromtimestamp(end))
+  
+  original_date_column = None
+  original_target_column = None
+
+  column_mapping = {
+        ' '.join([word.capitalize() for word in col.split('_')]): col
+        for col in df_state.columns
+    }
+  if date_column in column_mapping:
+    original_date_column = column_mapping[date_column]
+
+  if target_column in column_mapping:
+    original_target_column = column_mapping[target_column]
+
+  df_state[original_date_column] = pd.to_datetime(df_state[original_date_column])
+  filtered_df = df_state[(df_state[original_date_column] >= start_datetime) & (df_state[original_date_column] <= end_datetime)]
+
+
+  fig = px.line(filtered_df, x=original_date_column, y=original_target_column, title="Historical Sales Data")
+  return [filtered_df, fig]
+
+
+def upload_file(filepath):
+  name = Path(filepath).name
+  df = pd.read_csv(filepath.name)
+  datetime_columns = []
+  numeric_columns = []
+
+  for col in df.columns:
+    try:
+      if all(isinstance(float(x), float) for x in df[col].head(3)):
+        numeric_columns.append(col)
+    except ValueError:
+        continue
+
+  for col in df.columns:
+    if df[col].dtype == 'object':
+      try:
+        df[col] = pd.to_datetime(df[col])
+      except:
+        pass
+    if df[col].dtype == 'datetime64[ns]':
+      datetime_columns.append(col)
+  datetime_columns = list(map(lambda x: ' '.join([word.capitalize() for word in x.split('_')]), datetime_columns))
+  columns = df.columns.tolist()
+  transformed_columns = list(map(lambda x: ' '.join([word.capitalize() for word in x.split('_')]), columns))
+  target_col = list(map(lambda x: ' '.join([word.capitalize() for word in x.split('_')]), numeric_columns))
+
+  transformed_columns.insert(0, "")
+  data_columns = gr.Dropdown(choices=transformed_columns, value=None)
+  date_columns = gr.Dropdown(choices=datetime_columns, value=None)
+  target_columns = gr.Dropdown(choices=target_col, value=None)
+
+  return [df, data_columns, date_columns, target_columns]
+
+def download_file():
+  return [gr.UploadButton(visible=True), gr.DownloadButton(visible=False)]
+
+
+def set_products(selected_column, df_state):
+  column_mapping = {
+        ' '.join([word.capitalize() for word in col.split('_')]): col
+        for col in df_state.columns
+    }
+  if selected_column in column_mapping:
+    original_column = column_mapping[selected_column]
+    unique_values = df_state[original_column].dropna().unique().tolist()
+    return unique_values
+  return []
+
+
+def set_dates(selected_column, df_state):
+  column_mapping = {
+        ' '.join([word.capitalize() for word in col.split('_')]): col
+        for col in df_state.columns
+    }
+
+  if selected_column in column_mapping:
+    original_column = column_mapping[selected_column]
+    min_date = df_state[original_column].min()
+    max_date = df_state[original_column].max()
+    return min_date, max_date
+  return None, None
+
+
+def forecast_chronos_data(df_state, date_column, target_column, select_period, forecasting_type):
+  if not date_column:
+    raise gr.Error("Please select a Date column")
+
+  if not target_column:
+    raise gr.Error("Please select a target column")
+
+  original_date_column = None
+  original_target_column = None
+
+  column_mapping = {
+        ' '.join([word.capitalize() for word in col.split('_')]): col
+        for col in df_state.columns
+    }
+  if date_column in column_mapping:
+    original_date_column = column_mapping[date_column]
+
+  if target_column in column_mapping:
+    original_target_column = column_mapping[target_column]
+
+  df_forecast = pd.DataFrame()
+  df_forecast['date'] = df_state[original_date_column]
+  df_forecast['month'] = df_forecast['date'].dt.month
+  df_forecast['year'] = df_forecast['date'].dt.year
+  df_forecast['sold_qty'] = df_state[original_target_column]
+
+  monthly_sales = df_forecast.groupby(['year', 'month'])['sold_qty'].sum().reset_index()
+  monthly_sales = monthly_sales.rename(columns={'year': 'year', 'month': 'month', 'sold_qty': 'y'})
+
+  device = "cuda" if torch.cuda.is_available() else "cpu"
+
+  pipeline = ChronosPipeline.from_pretrained(
+      "amazon/chronos-t5-base",
+      device_map=device,
+      torch_dtype=torch.float32,
+      )
+  context = torch.tensor(monthly_sales["y"])
+  prediction_length = select_period
+  forecast = pipeline.predict(context, prediction_length)
+
+  forecast_index = range(len(monthly_sales), len(monthly_sales) + prediction_length)
+  low, median, high = np.quantile(forecast[0].numpy(), [0.1, 0.5, 0.9], axis=0)
+
+  plt.figure(figsize=(30, 10))
+  plt.plot(monthly_sales["y"], color="royalblue", label="Historical Data", linewidth=2)
+  plt.plot(forecast_index, median, color="tomato", label="Median Forecast", linewidth=2)
+  plt.fill_between(forecast_index, low, high, color="tomato", alpha=0.3, label="80% Prediction Interval")
+  plt.title("Sales Forecasting Visualization", fontsize=16)
+  plt.xlabel("Months", fontsize=20)
+  plt.ylabel("Sold Qty", fontsize=20)
+
+  plt.xticks(fontsize=18)
+  plt.yticks(fontsize=18)
+
+  ax = plt.gca()
+  ax.xaxis.set_major_locator(ticker.MultipleLocator(3))
+  ax.yaxis.set_major_locator(ticker.MultipleLocator(5))
+  ax.grid(which='major', linestyle='--', linewidth=1.2, color='gray', alpha=0.7)
+
+  plt.legend(fontsize=18)
+  plt.grid(linestyle='--', linewidth=1.2, color='gray', alpha=0.7)
+  plt.tight_layout()
+  return plt.gcf()
+
+
+def home_page():
+    content = """
+    ### **Sales Forecasting with Chronos**
+
+    Welcome to the future of sales optimization with **Chronos**.
+    Say goodbye to guesswork and unlock the power of **data-driven insights** with our advanced forecasting platform.
+
+    - **Seamless CSV Upload**: Quickly upload your sales data in CSV format—no technical expertise needed.
+    - **AI-Powered Predictions**: Harness the power of state-of-the-art machine learning models to uncover trends and forecast future sales performance.
+    - **Interactive Visualizations**: Gain actionable insights with intuitive charts and graphs that make data easy to understand.
+
+    Start making smarter, data-backed business decisions today with **Chronos**!
+
+    """
+    return content
+
+def about_page():
+    content = """
+    ### 📧 **Contact Us:**
+    - **Email**:  [email protected] ✉️
+    - **Website**: [https://www.topsinfosolutions.com/](https://www.topsinfosolutions.com/) 🌐
+
+    ### 🛠 **What We Offer:**
+    - **Custom AI Solutions**: Tailored to your business needs 🤖
+    - **Chatbot Development**: Build intelligent conversational agents 💬
+    - **Vision Models**: Computer vision solutions for various applications 🖼️
+    - **AI Agents**: Personalized agents powered by advanced LLMs 🤖
+
+    ### 🤔 **How We Can Help:**
+    Reach out to us for bespoke AI services. Whether you need chatbots, vision models, or AI-powered agents, we’re here to build solutions that make a difference! 🌟
+
+    ### 💬 **Get in Touch:**
+    If you have any questions or need a custom solution, click the button below to schedule a consultation with us. 📅
+    """
+    return content
+
+
+with gr.Blocks() as demo:
+  with gr.Tabs():
+    with gr.TabItem("Home"):
+      df_state = gr.State()
+
+      gr.Image("/content/chronos-logo.png", interactive=False)
+      home_output = gr.Markdown(value=home_page(), label="Playground")
+
+      gr.Markdown("## Step 1: Historical/Training Data (currently supports *.csv only)")
+
+      with gr.Row():
+        file_input = gr.File(label="Upload Historical (Training Data) Sales Data", file_types=[".csv"])
+
+      with gr.Row():
+        date_column = gr.Dropdown(choices=[], label="Select Date column (*Required)", multiselect=False, value=None)
+        target_column = gr.Dropdown(choices=[], label="Select Target column (*Required)", multiselect=False, value=None)
+        select_product_column = gr.Dropdown(choices=[], label="Select Product column (Optional)", multiselect=False, value=None)
+        select_product = gr.Dropdown(choices=[], label="Select Product (Optional)", multiselect=False, value=None)
+        
+        
+
+      with gr.Row():
+        start = gr.DateTime("2021-01-01 00:00:00", label="Training data Start date")
+        end = gr.DateTime("2021-01-05 00:00:00", label="Training data End date")
+        apply_btn = gr.Button("Visualize Data", scale=0)
+
+      with gr.Row():
+        historical_data_plot = gr.Plot()
+
+      apply_btn.click(
+          filter_data,
+          inputs=[start, end, df_state, select_product_column, date_column, target_column],
+          outputs=[df_state, historical_data_plot]
+          )
+      
+      gr.Markdown("## Step 2: Forecast")
+      with gr.Row():
+        forecasting_type = gr.Radio(["day", "monthly", "year"], value="monthly", label="Forecasting Type", interactive=False)
+        select_period =  gr.Slider(2, 60, value=12, label="Select Period", info="Check Selected Forecast Type", interactive =True, step=1)
+        forecast_btn = gr.Button("Forecast")
+
+      
+      with gr.Row():
+        plot_forecast_output = gr.Plot(label="Chronos Forecasting Visualization")
+
+      forecast_btn.click(
+          forecast_chronos_data,
+          inputs=[df_state, date_column, target_column, select_period],
+          outputs=[plot_forecast_output]
+          )
+    
+
+
+      file_input.upload(
+          upload_file,
+          inputs=[file_input],
+          outputs=[df_state, select_product_column, date_column, target_column]
+          )
+
+      select_product_column.change(
+          set_products,
+          inputs=[select_product_column, df_state],
+          outputs=[]
+          )
+
+      date_column.change(
+          set_dates,
+          inputs=[date_column, df_state],
+          outputs=[start, end]
+          )
+
+      target_column.change(
+          lambda x: x if x else [],
+          inputs=[target_column],
+          outputs=[]
+          )
+        
+    with gr.TabItem("About Tops"):
+      df_state = gr.State()
+
+      gr.Image("/content/chronos-logo.png", interactive=False)
+      about_output = gr.Markdown(value=about_page(), label="About Tops")
+
+
+if __name__ == "__main__":
+    demo.launch()
+
+