Create app.py

app.py

@@ -0,0 +1,405 @@
+# app.py
+# Inventory Management Assistant – HuggingFace / Gradio app
+
+import os
+from typing import Dict, List, Tuple
+
+import pandas as pd
+import gradio as gr
+
+# ----------------------------- Data Load ----------------------------- #
+
+DATA_FILES: Dict[str, List[str]] = {
+    # Re-use the same six datasets as your other two apps.
+    # Adjust candidate paths / filenames as needed.
+    "backlog": [
+        "data/Backlog w Customer Names.xlsx",
+        "data/Backlog_w_Customer_Names.xlsx",
+        "data/backlog.xlsx",
+    ],
+    "inventory": [
+        "data/INVENTORY.xlsx",
+        "data/Inventory.xlsx",
+        "data/inventory.xlsx",
+    ],
+    "forecast": [
+        "data/Forecast.xlsx",
+        "data/forecast.xlsx",
+    ],
+    "po": [
+        "data/Open_PO.xlsx",
+        "data/open_po.xlsx",
+    ],
+    "material_master": [
+        "data/Material_Master.xlsx",
+        "data/material_master.xlsx",
+    ],
+    "map": [
+        "data/Material_Plant_Map.xlsx",
+        "data/material_plant_map.xlsx",
+    ],
+}
+
+
+def _load_first_existing(path_candidates: List[str]) -> pd.DataFrame:
+    """Try all candidate paths and return the first one that exists."""
+    for p in path_candidates:
+        if os.path.exists(p):
+            if p.lower().endswith(".csv"):
+                return pd.read_csv(p)
+            else:
+                return pd.read_excel(p)
+    raise FileNotFoundError(f"None of these files were found: {path_candidates}")
+
+
+def load_all_data() -> Dict[str, pd.DataFrame]:
+    data = {}
+    missing = []
+    for key, paths in DATA_FILES.items():
+        try:
+            df = _load_first_existing(paths)
+            data[key] = df
+        except FileNotFoundError:
+            missing.append(key)
+
+    if missing:
+        # Fail loudly so HF logs show which logical tables are missing
+        raise RuntimeError(f"Data load error – missing logical tables: {missing}")
+    return data
+
+
+DATA = load_all_data()
+INV = DATA["inventory"]
+BACKLOG = DATA["backlog"]
+
+# ----------------------------- Helper functions ----------------------------- #
+
+def _pick(colnames: List[str], candidates: List[str]):
+    for c in candidates:
+        if c in colnames:
+            return c
+    return None
+
+
+def _basic_inventory_view(df: pd.DataFrame, top_n: int = 20) -> pd.DataFrame:
+    """Return a light, generic view that won't break if columns differ."""
+    cols = df.columns.tolist()
+
+    mat_col = _pick(cols, ["SAP_MATERIAL_NO", "MATERIAL", "MATERIAL_NO"])
+    desc_col = _pick(cols, ["MATERIAL_DESCRIPTION", "MAT_DESC", "DESCRIPTION"])
+    plant_col = _pick(cols, ["PLANT", "LOCATION", "SITE"])
+    qty_col = _pick(cols, ["QOH", "QTY", "UNRESTRICTED_STOCK", "TOTAL_STOCK"])
+    age_col = _pick(cols, ["AGE_DAYS", "DAYS_ON_HAND", "DAYS_COVER"])
+
+    selected = [c for c in [mat_col, desc_col, plant_col, qty_col, age_col] if c]
+    if not selected:
+        return df.head(top_n)
+    return df[selected].head(top_n)
+
+
+# ----------------------------- Business Logic ----------------------------- #
+
+def get_fast_moving_materials(top_n: int = 25) -> pd.DataFrame:
+    """Very simple heuristic: lowest days cover / age, then highest demand/qty."""
+    df = INV.copy()
+    cols = df.columns.tolist()
+
+    days_cover_col = _pick(cols, ["DAYS_COVER", "AGE_DAYS", "DAYS_ON_HAND"])
+    demand_col = _pick(cols, ["AVG_DAILY_DEMAND", "DEMAND_PER_DAY", "ISSUES_PER_DAY"])
+    qty_col = _pick(cols, ["QOH", "QTY", "UNRESTRICTED_STOCK", "TOTAL_STOCK"])
+
+    if days_cover_col:
+        df = df.sort_values(by=days_cover_col, ascending=True)
+    elif demand_col:
+        df = df.sort_values(by=demand_col, ascending=False)
+    elif qty_col:
+        df = df.sort_values(by=qty_col, ascending=False)
+
+    return _basic_inventory_view(df, top_n=top_n)
+
+
+def get_dead_stock(top_n: int = 25) -> pd.DataFrame:
+    """Heuristic: highest age / lowest movement."""
+    df = INV.copy()
+    cols = df.columns.tolist()
+
+    age_col = _pick(cols, ["AGE_DAYS", "DAYS_ON_HAND", "DAYS_SINCE_MOVEMENT"])
+    if age_col:
+        df = df.sort_values(by=age_col, ascending=False)
+    else:
+        # fallback: just low-qty materials
+        qty_col = _pick(cols, ["QOH", "QTY", "UNRESTRICTED_STOCK", "TOTAL_STOCK"])
+        if qty_col:
+            df = df.sort_values(by=qty_col, ascending=True)
+
+    return _basic_inventory_view(df, top_n=top_n)
+
+
+def get_reallocation_opportunities(top_n: int = 25) -> pd.DataFrame:
+    """
+    Simple cross-plant reallocation view:
+    - Uses inventory + backlog.
+    - Marks surplus/shortage per material/plant.
+    """
+    inv = INV.copy()
+    bl = BACKLOG.copy()
+
+    inv_cols = inv.columns.tolist()
+    bl_cols = bl.columns.tolist()
+
+    mat_col_i = _pick(inv_cols, ["SAP_MATERIAL_NO", "MATERIAL", "MATERIAL_NO"])
+    plant_col_i = _pick(inv_cols, ["PLANT", "LOCATION", "SITE"])
+    qty_col_i = _pick(inv_cols, ["QOH", "QTY", "UNRESTRICTED_STOCK", "TOTAL_STOCK"])
+
+    mat_col_b = _pick(bl_cols, ["SAP_MATERIAL_NO", "MATERIAL", "MATERIAL_NO"])
+    plant_col_b = _pick(bl_cols, ["PLANT", "LOCATION", "SITE"])
+    demand_col_b = _pick(bl_cols, ["OPEN_QTY", "DEMAND_QTY", "BACKLOG_QTY"])
+
+    required = [mat_col_i, plant_col_i, qty_col_i, mat_col_b, plant_col_b, demand_col_b]
+    if any(c is None for c in required):
+        # If columns don't line up yet, just show generic message.
+        return pd.DataFrame(
+            {
+                "Message": [
+                    "Reallocation logic needs aligned columns in inventory & backlog.",
+                    f"Inventory columns: {inv_cols}",
+                    f"Backlog columns: {bl_cols}",
+                ]
+            }
+        )
+
+    inv_agg = (
+        inv.groupby([mat_col_i, plant_col_i])[qty_col_i]
+        .sum()
+        .reset_index()
+        .rename(columns={qty_col_i: "QOH"})
+    )
+
+    bl_agg = (
+        bl.groupby([mat_col_b, plant_col_b])[demand_col_b]
+        .sum()
+        .reset_index()
+        .rename(columns={mat_col_b: mat_col_i, plant_col_b: plant_col_i, demand_col_b: "DEMAND"})
+    )
+
+    merged = inv_agg.merge(bl_agg, on=[mat_col_i, plant_col_i], how="outer").fillna(0)
+    merged["NET"] = merged["QOH"] - merged["DEMAND"]
+
+    # Mark surplus / shortage
+    merged["STATUS"] = merged["NET"].apply(
+        lambda x: "Surplus" if x > 0 else ("Shortage" if x < 0 else "Balanced")
+    )
+
+    # Keep only materials which have at least one surplus and one shortage plant
+    mat_status = (
+        merged.groupby(mat_col_i)["STATUS"]
+        .agg(lambda s: set(s))
+        .reset_index()
+        .rename(columns={"STATUS": "STATUS_SET"})
+    )
+    interesting_mats = mat_status[
+        mat_status["STATUS_SET"].apply(lambda s: {"Surplus", "Shortage"}.issubset(s))
+    ][mat_col_i]
+
+    out = merged[merged[mat_col_i].isin(interesting_mats)]
+    out = out.sort_values(by=[mat_col_i, "STATUS", "NET"])
+    return out.head(top_n * 4)  # multiple rows per material
+
+
+def get_risk_recommendations(top_n: int = 25) -> pd.DataFrame:
+    """
+    Very simple 'at-risk' view:
+    - Net = demand – stock; positive = shortage.
+    """
+    inv = INV.copy()
+    bl = BACKLOG.copy()
+
+    inv_cols = inv.columns.tolist()
+    bl_cols = bl.columns.tolist()
+
+    mat_col_i = _pick(inv_cols, ["SAP_MATERIAL_NO", "MATERIAL", "MATERIAL_NO"])
+    plant_col_i = _pick(inv_cols, ["PLANT", "LOCATION", "SITE"])
+    qty_col_i = _pick(inv_cols, ["QOH", "QTY", "UNRESTRICTED_STOCK", "TOTAL_STOCK"])
+
+    mat_col_b = _pick(bl_cols, ["SAP_MATERIAL_NO", "MATERIAL", "MATERIAL_NO"])
+    plant_col_b = _pick(bl_cols, ["PLANT", "LOCATION", "SITE"])
+    demand_col_b = _pick(bl_cols, ["OPEN_QTY", "DEMAND_QTY", "BACKLOG_QTY"])
+
+    required = [mat_col_i, plant_col_i, qty_col_i, mat_col_b, plant_col_b, demand_col_b]
+    if any(c is None for c in required):
+        return pd.DataFrame(
+            {
+                "Message": [
+                    "Risk recommendations need aligned inventory & backlog columns.",
+                    f"Inventory columns: {inv_cols}",
+                    f"Backlog columns: {bl_cols}",
+                ]
+            }
+        )
+
+    inv_agg = (
+        inv.groupby([mat_col_i, plant_col_i])[qty_col_i]
+        .sum()
+        .reset_index()
+        .rename(columns={qty_col_i: "QOH"})
+    )
+    bl_agg = (
+        bl.groupby([mat_col_b, plant_col_b])[demand_col_b]
+        .sum()
+        .reset_index()
+        .rename(columns={mat_col_b: mat_col_i, plant_col_b: plant_col_i, demand_col_b: "DEMAND"})
+    )
+
+    merged = inv_agg.merge(bl_agg, on=[mat_col_i, plant_col_i], how="outer").fillna(0)
+    merged["SHORTAGE"] = merged["DEMAND"] - merged["QOH"]
+    merged = merged[merged["SHORTAGE"] > 0]
+
+    cols_out = [mat_col_i, plant_col_i, "QOH", "DEMAND", "SHORTAGE"]
+    return merged[cols_out].sort_values("SHORTAGE", ascending=False).head(top_n)
+
+
+def search_inventory(query: str) -> pd.DataFrame:
+    """Very light search by material number / description / plant."""
+    if not query:
+        return _basic_inventory_view(INV, top_n=25)
+
+    df = INV.copy()
+    cols = df.columns.tolist()
+    mat_col = _pick(cols, ["SAP_MATERIAL_NO", "MATERIAL", "MATERIAL_NO"])
+    desc_col = _pick(cols, ["MATERIAL_DESCRIPTION", "MAT_DESC", "DESCRIPTION"])
+    plant_col = _pick(cols, ["PLANT", "LOCATION", "SITE"])
+
+    mask = pd.Series([False] * len(df))
+    if mat_col:
+        mask |= df[mat_col].astype(str).str.contains(query, case=False, na=False)
+    if desc_col:
+        mask |= df[desc_col].astype(str).str.contains(query, case=False, na=False)
+    if plant_col:
+        mask |= df[plant_col].astype(str).str.contains(query, case=False, na=False)
+
+    results = df[mask]
+    if results.empty:
+        return pd.DataFrame({"Message": [f"No inventory rows found for '{query}'"]})
+    return _basic_inventory_view(results, top_n=50)
+
+
+# ----------------------------- Gradio Callbacks ----------------------------- #
+
+def handle_tile(tile: str, history: List[Tuple[str, str]]):
+    if history is None:
+        history = []
+
+    if tile == "fast":
+        user_msg = "Show me fast moving materials."
+        df = get_fast_moving_materials()
+        assistant_msg = "Here are the current fast-moving materials based on days cover / age."
+    elif tile == "reallocate":
+        user_msg = "Show stock reallocation possibilities."
+        df = get_reallocation_opportunities()
+        assistant_msg = "These materials have surplus at some plants and shortages at others."
+    elif tile == "risk":
+        user_msg = "Show inventory risk recommendations."
+        df = get_risk_recommendations()
+        assistant_msg = "These materials have net shortages based on backlog vs available stock."
+    elif tile == "dead":
+        user_msg = "Show dead / slow-moving stock."
+        df = get_dead_stock()
+        assistant_msg = "These materials appear to be slow-moving or dead stock."
+    else:
+        user_msg = "Unknown action."
+        df = pd.DataFrame({"Message": ["Unknown tile clicked."]})
+        assistant_msg = "I couldn't identify that tile."
+
+    history = history + [(("user"), user_msg), (("assistant"), assistant_msg)]
+    return history, df
+
+
+def handle_search(message: str, history: List[Tuple[str, str]]):
+    if history is None:
+        history = []
+
+    history = history + [("user", message)]
+    df = search_inventory(message)
+    assistant_msg = "Here is what I found in inventory for your search."
+    history = history + [("assistant", assistant_msg)]
+    return "", history, df
+
+
+# ----------------------------- UI Layout ----------------------------- #
+
+CUSTOM_CSS = """
+.gradio-container {font-family: 'Segoe UI', system-ui, -apple-system, BlinkMacSystemFont, sans-serif;}
+#header-bar {background-color: #002b5c; color: white; padding: 10px 16px; font-size: 20px; font-weight: 600;}
+.tile-row button {height: 60px; font-size: 16px; font-weight: 600;}
+#faq-bar {background-color: #003f87; color: white; padding: 8px 16px; margin-top: 8px;
+          border-radius: 8px; font-size: 15px; font-weight: 500;}
+"""
+
+with gr.Blocks(css=CUSTOM_CSS, title="Inventory Management Assistant") as demo:
+    gr.HTML('<div id="header-bar">Inventory Assistant</div>')
+
+    with gr.Row(elem_id="tile-row"):
+        btn_fast = gr.Button("Fast Moving Materials")
+        btn_reallocate = gr.Button("Stock Reallocation")
+        btn_risk = gr.Button("Risk Recommendations")
+        btn_dead = gr.Button("Dead Stock Materials")
+
+    gr.HTML(
+        '<div id="faq-bar">💡 FAQ: Where are we at risk on inventory, and where can we reallocate stock?</div>'
+    )
+
+    chatbot = gr.Chatbot(label="Inventory Assistant", height=260)
+    results_table = gr.Dataframe(
+        headers=[],
+        datatype="auto",
+        label="Results",
+        interactive=False,
+        visible=True,
+        wrap=True,
+        height=260,
+    )
+
+    with gr.Row():
+        txt = gr.Textbox(
+            placeholder="Ask about materials, plants or inventory…",
+            show_label=False,
+            scale=5,
+        )
+        btn_search = gr.Button("Search", scale=1)
+
+    # Wire the tiles
+    btn_fast.click(
+        fn=lambda h: handle_tile("fast", h),
+        inputs=chatbot,
+        outputs=[chatbot, results_table],
+    )
+    btn_reallocate.click(
+        fn=lambda h: handle_tile("reallocate", h),
+        inputs=chatbot,
+        outputs=[chatbot, results_table],
+    )
+    btn_risk.click(
+        fn=lambda h: handle_tile("risk", h),
+        inputs=chatbot,
+        outputs=[chatbot, results_table],
+    )
+    btn_dead.click(
+        fn=lambda h: handle_tile("dead", h),
+        inputs=chatbot,
+        outputs=[chatbot, results_table],
+    )
+
+    # Wire the search bar
+    btn_search.click(
+        fn=handle_search,
+        inputs=[txt, chatbot],
+        outputs=[txt, chatbot, results_table],
+    )
+    txt.submit(
+        fn=handle_search,
+        inputs=[txt, chatbot],
+        outputs=[txt, chatbot, results_table],
+    )
+
+if __name__ == "__main__":
+    demo.launch()