app_v0.py

from __future__ import annotations

from typing import Iterable
import gradio as gr
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
import os
import math
import torch
from chronos import ChronosPipeline
import warnings

from seafoam import Seafoam

warnings.filterwarnings("ignore")

import numpy as np
import matplotlib.ticker as ticker

os.makedirs("example_files", exist_ok=True)

def process_csv(file):
    if file is None:
        return None, gr.Dropdown(choices=[])

    if not file.name.endswith('.csv'):
        raise gr.Error("Please upload a CSV file only")
    df = pd.read_csv(file.name)
    columns = df.columns.tolist()
    transformed_columns = list(map(lambda x: ' '.join([word.capitalize() for word in x.split('_')]), columns))
    data_columns = gr.Dropdown(choices=transformed_columns, value=None)
    return df, data_columns, data_columns


def process_data(csv_file, date_column_value, target_column_value):
    try:
        if not csv_file:
            return "Error: Upload Csv File"

        if not date_column_value or not target_column_value:
            return "Error: Both date and target columns must be selected"

        date_column = date_column_value.lower().replace(" ", "_")
        target_column = target_column_value.lower().replace(" ", "_")

        # Read the CSV file
        df = pd.read_csv(csv_file.name)

        numeric_mask = df[date_column].apply(lambda x: isinstance(x, (int, float)))
        if numeric_mask.any():
            return "Error: Found numeric values in column '{date_column}'. Please provide dates in string format like 'YYYY-MM-DD'."

        df['date'] = pd.to_datetime(df[date_column])

        df['month'] = df['date'].dt.month
        df['year'] = df['date'].dt.year
        df['sold_qty'] = df[target_column]

        monthly_sales = df.groupby(['year', 'month'])['sold_qty'].sum().reset_index()
        monthly_sales = monthly_sales.rename(columns={'year': 'year', 'month': 'month', 'sold_qty': 'y'})

        pipeline = ChronosPipeline.from_pretrained(
            "amazon/chronos-t5-base",
            device_map="cpu",
            torch_dtype=torch.float32,
            )
        context = torch.tensor(monthly_sales["y"])
        prediction_length = 12
        forecast = pipeline.predict(context, prediction_length)

        # Prepare forecast data
        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)

        df['month_name'] = df['date'].dt.month_name()
        month_order = [
            'January', 'February', 'March', 'April', 'May', 'June',
            'July', 'August', 'September', 'October', 'November', 'December'
        ]
        df['month_name'] = pd.Categorical(df['month_name'], categories=month_order, ordered=True)

        expanded_df = df.copy()
        year_month_sum = expanded_df.groupby(['year', 'month_name'])['sold_qty'].sum().reset_index()

        # Create a pivot table: sum of units sold per year and month
        pivot_table = year_month_sum.pivot(index='year', columns='month_name', values='sold_qty')

        new_data_list = [math.ceil(x) for x in median]

        # Add the new data list for the next year (incrementing the year by 1)
        next_year = pivot_table.index[-1] + 1  # Increment the year by 1
        pivot_table.loc[next_year] = new_data_list  # Add the new row for the next year

        # Visualization: Pivot Table Data (Second Plot)
        fig3, ax3 = plt.subplots(figsize=(18, 6))

        # Create a table inside the plot
        ax3.axis('off')  # Turn off the axis
        table = ax3.table(cellText=pivot_table.values, colLabels=pivot_table.columns, rowLabels=pivot_table.index, loc='center', cellLoc='center')

        # Style the table
        table.auto_set_font_size(False)
        table.set_fontsize(12)
        table.scale(1.2, 1.2)  # Scale the table for better visibility

        # Adjust table colors (optional)
        for (i, j), cell in table.get_celld().items():
            if i == 0:
                cell.set_text_props(weight='bold')
                cell.set_facecolor('#f2f2f2')
            elif j == 0:
                cell.set_text_props(weight='bold')
                cell.set_facecolor('#f2f2f2')
            else:
                cell.set_facecolor('white')

        # Visualization
        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(), fig3

    except Exception as e:
        print(f"Error: {str(e)}")
        return None

# Create Gradio interface
with gr.Blocks(theme=Seafoam()) as demo:
    gr.Markdown("# Chronos Forecasting - Tops infosolutions Pvt Ltd")
    gr.Markdown("Upload a CSV file and click 'Forecast' to generate sales forecast for next 12 months .")

    df_state = gr.State()

    with gr.Row():
        file_input = gr.File(label="Upload CSV File", file_types=[".csv"])


    with gr.Row():
        date_column = gr.Dropdown(
            choices=[],
            label="Select Date column",
            multiselect=False,
            value=None
        )

        target_column = gr.Dropdown(
            choices=[],
            label="Select Target column",
            multiselect=False,
            value=None
        )

    gr.Examples(
        examples=[
            ["example_files/test_tops_product_id_1.csv"],
            ["example_files/test_tops_product_id_2.csv"],
            ["example_files/test_tops_product_id_3.csv"],
            ["example_files/test_tops_product_id_4.csv"]
        ],
        inputs=file_input,
        outputs=[df_state, date_column, target_column],
        fn=process_csv,
        cache_examples=True
    )

    with gr.Row():
        visualize_btn = gr.Button("Forecast", variant="primary")

    with gr.Row():
        plot_output = gr.Plot(label="Chronos Forecasting Visualization")

    with gr.Row():
        pivot_plot_output = gr.Plot(label="Monthly Sales Pivot Table")

    file_input.upload(
        process_csv,
        inputs=[file_input],
        outputs=[df_state, date_column, target_column]
    )

    # Column selection handler
    date_column.change(
        lambda x: x if x else "",
        inputs=[date_column],
        outputs=[]
    )

    target_column.change(
        lambda x: x if x else "",
        inputs=[target_column],
        outputs=[]
    )

    visualize_btn.click(
        fn=process_data,
        inputs=[file_input, date_column, target_column],
        outputs=[plot_output, pivot_plot_output]
    )





# Launch the app
if __name__ == "__main__":
    demo.launch()