import os
import re
import json
import time
import traceback
from pathlib import Path
from typing import Dict, Any, List, Tuple
import pandas as pd
import gradio as gr
import papermill as pm
import plotly.graph_objects as go
# Optional LLM (HuggingFace Inference API)
try:
from huggingface_hub import InferenceClient
except Exception:
InferenceClient = None
# =========================================================
# CONFIG
# =========================================================
BASE_DIR = Path(__file__).resolve().parent
NB1 = os.environ.get("NB1", "datacreation.ipynb").strip()
NB2 = os.environ.get("NB2", "pythonanalysis.ipynb").strip()
RUNS_DIR = BASE_DIR / "runs"
ART_DIR = BASE_DIR / "artifacts"
PY_FIG_DIR = ART_DIR / "py" / "figures"
PY_TAB_DIR = ART_DIR / "py" / "tables"
PAPERMILL_TIMEOUT = int(os.environ.get("PAPERMILL_TIMEOUT", "1800"))
MAX_PREVIEW_ROWS = int(os.environ.get("MAX_FILE_PREVIEW_ROWS", "50"))
MAX_LOG_CHARS = int(os.environ.get("MAX_LOG_CHARS", "8000"))
HF_API_KEY = os.environ.get("HF_API_KEY", "").strip()
MODEL_NAME = os.environ.get("MODEL_NAME", "deepseek-ai/DeepSeek-R1").strip()
HF_PROVIDER = os.environ.get("HF_PROVIDER", "novita").strip()
N8N_WEBHOOK_URL = os.environ.get("N8N_WEBHOOK_URL", "").strip()
LLM_ENABLED = bool(HF_API_KEY) and InferenceClient is not None
llm_client = (
InferenceClient(provider=HF_PROVIDER, api_key=HF_API_KEY)
if LLM_ENABLED
else None
)
# =========================================================
# HELPERS
# =========================================================
def ensure_dirs():
for p in [RUNS_DIR, ART_DIR, PY_FIG_DIR, PY_TAB_DIR]:
p.mkdir(parents=True, exist_ok=True)
def stamp():
return time.strftime("%Y%m%d-%H%M%S")
def _ls(dir_path: Path, exts: Tuple[str, ...]) -> List[str]:
if not dir_path.is_dir():
return []
return sorted(p.name for p in dir_path.iterdir() if p.is_file() and p.suffix.lower() in exts)
def _read_csv(path: Path) -> pd.DataFrame:
return pd.read_csv(path, nrows=MAX_PREVIEW_ROWS)
def _read_json(path: Path):
with path.open(encoding="utf-8") as f:
return json.load(f)
def artifacts_index() -> Dict[str, Any]:
return {
"python": {
"figures": _ls(PY_FIG_DIR, (".png", ".jpg", ".jpeg")),
"tables": _ls(PY_TAB_DIR, (".csv", ".json")),
},
}
# =========================================================
# PIPELINE RUNNERS
# =========================================================
def run_notebook(nb_name: str) -> str:
ensure_dirs()
nb_in = BASE_DIR / nb_name
if not nb_in.exists():
return f"ERROR: {nb_name} not found."
nb_out = RUNS_DIR / f"run_{stamp()}_{nb_name}"
pm.execute_notebook(
input_path=str(nb_in),
output_path=str(nb_out),
cwd=str(BASE_DIR),
log_output=True,
progress_bar=False,
request_save_on_cell_execute=True,
execution_timeout=PAPERMILL_TIMEOUT,
)
return f"Executed {nb_name}"
def run_datacreation() -> str:
try:
log = run_notebook(NB1)
csvs = [f.name for f in BASE_DIR.glob("*.csv")]
return f"OK {log}\n\nCSVs now in /app:\n" + "\n".join(f" - {c}" for c in sorted(csvs))
except Exception as e:
return f"FAILED {e}\n\n{traceback.format_exc()[-2000:]}"
def run_pythonanalysis() -> str:
try:
log = run_notebook(NB2)
idx = artifacts_index()
figs = idx["python"]["figures"]
tabs = idx["python"]["tables"]
return (
f"OK {log}\n\n"
f"Figures: {', '.join(figs) or '(none)'}\n"
f"Tables: {', '.join(tabs) or '(none)'}"
)
except Exception as e:
return f"FAILED {e}\n\n{traceback.format_exc()[-2000:]}"
def run_full_pipeline() -> str:
logs = []
logs.append("=" * 50)
logs.append("STEP 1/2: Data Creation")
logs.append("=" * 50)
logs.append(run_datacreation())
logs.append("")
logs.append("=" * 50)
logs.append("STEP 2/2: Python Analysis")
logs.append("=" * 50)
logs.append(run_pythonanalysis())
return "\n".join(logs)
# =========================================================
# GALLERY LOADERS
# =========================================================
def _load_all_figures() -> List[Tuple[str, str]]:
items = []
for p in sorted(PY_FIG_DIR.glob("*.png")):
items.append((str(p), p.stem.replace("_", " ").title()))
return items
def _load_table_safe(path: Path) -> pd.DataFrame:
try:
if path.suffix == ".json":
obj = _read_json(path)
if isinstance(obj, dict):
return pd.DataFrame([obj])
return pd.DataFrame(obj)
return _read_csv(path)
except Exception as e:
return pd.DataFrame([{"error": str(e)}])
def refresh_gallery():
figures = _load_all_figures()
idx = artifacts_index()
table_choices = list(idx["python"]["tables"])
default_df = pd.DataFrame()
if table_choices:
default_df = _load_table_safe(PY_TAB_DIR / table_choices[0])
return (
figures if figures else [],
gr.update(choices=table_choices, value=table_choices[0] if table_choices else None),
default_df,
)
def on_table_select(choice: str):
if not choice:
return pd.DataFrame([{"hint": "Select a table above."}])
path = PY_TAB_DIR / choice
if not path.exists():
return pd.DataFrame([{"error": f"File not found: {choice}"}])
return _load_table_safe(path)
# =========================================================
# KPI LOADER
# =========================================================
def load_kpis() -> Dict[str, Any]:
candidate = PY_TAB_DIR / "kpis.json"
if candidate.exists():
try:
return _read_json(candidate)
except Exception:
pass
return {}
# =========================================================
# AI DASHBOARD
# =========================================================
DASHBOARD_SYSTEM = """You are an AI dashboard assistant for a food and nutrition analytics app.
The user asks questions about food products, calories, sugar, fat, protein, fiber, salt,
Nutri-Score, health labels, and nutrition trends.
You have access to pre-computed artifacts from a Python analysis pipeline.
AVAILABLE ARTIFACTS (only reference ones that exist):
{artifacts_json}
KPI SUMMARY:
{kpis_json}
YOUR JOB:
1. Answer the user's question conversationally using the KPIs and available artifacts.
2. At the END of your response, output a JSON block fenced with ```json ... ```.
3. The JSON must have this shape:
{{"show": "figure"|"table"|"none", "scope": "python", "filename": "..."}}
Rules:
- If the user asks about calories, energy, or nutrition overview, prefer food_dashboard.csv or calorie charts.
- If the user asks about sugar, fat, salt, protein, or fiber by health label, show the relevant table or figure.
- If the user asks about health label distribution, show the health label figure or table.
- If the user asks about Nutri-Score versus health label, show the comparison figure or table.
- If the user asks for recommendations, show recommendations.csv.
- If no artifact is relevant, return show = none.
- Keep your answer concise.
"""
JSON_BLOCK_RE = re.compile(r"```json\s*(\{.*?\})\s*```", re.DOTALL)
FALLBACK_JSON_RE = re.compile(r"\{[^{}]*\"show\"[^{}]*\}", re.DOTALL)
def _parse_display_directive(text: str) -> Dict[str, str]:
m = JSON_BLOCK_RE.search(text)
if m:
try:
return json.loads(m.group(1))
except json.JSONDecodeError:
pass
m = FALLBACK_JSON_RE.search(text)
if m:
try:
return json.loads(m.group(0))
except json.JSONDecodeError:
pass
return {"show": "none"}
def _clean_response(text: str) -> str:
return JSON_BLOCK_RE.sub("", text).strip()
def _n8n_call(msg: str):
import requests as req
try:
resp = req.post(N8N_WEBHOOK_URL, json={"question": msg}, timeout=20)
data = resp.json()
answer = data.get("answer", "No response from n8n workflow.")
chart = data.get("chart", "none")
if chart and chart != "none":
return answer, {"show": "figure", "chart": chart}
return answer, {"show": "none"}
except Exception as e:
return f"n8n error: {e}. Falling back to keyword matching.", None
def ai_chat(user_msg: str, history: list):
if not user_msg or not user_msg.strip():
return history, "", None, None
idx = artifacts_index()
kpis = load_kpis()
if N8N_WEBHOOK_URL:
reply, directive = _n8n_call(user_msg)
if directive is None:
reply_fb, directive = _keyword_fallback(user_msg, idx, kpis)
reply += "\n\n" + reply_fb
elif not LLM_ENABLED:
reply, directive = _keyword_fallback(user_msg, idx, kpis)
else:
system = DASHBOARD_SYSTEM.format(
artifacts_json=json.dumps(idx, indent=2),
kpis_json=json.dumps(kpis, indent=2) if kpis else "(no KPIs yet, run the pipeline first)",
)
msgs = [{"role": "system", "content": system}]
for entry in (history or [])[-6:]:
msgs.append(entry)
msgs.append({"role": "user", "content": user_msg})
try:
r = llm_client.chat_completion(
model=MODEL_NAME,
messages=msgs,
temperature=0.3,
max_tokens=600,
stream=False,
)
raw = (
r["choices"][0]["message"]["content"]
if isinstance(r, dict)
else r.choices[0].message.content
)
directive = _parse_display_directive(raw)
reply = _clean_response(raw)
except Exception as e:
reply = f"LLM error: {e}. Falling back to keyword matching."
reply_fb, directive = _keyword_fallback(user_msg, idx, kpis)
reply += "\n\n" + reply_fb
chart_out = None
tab_out = None
show = directive.get("show", "none")
fname = directive.get("filename", "")
chart_name = directive.get("chart", "")
chart_builders = {
"calories": build_calories_chart,
"health_label": build_health_label_chart,
"nutriscore": build_nutriscore_chart,
"macros": build_macros_chart,
}
if chart_name and chart_name in chart_builders:
chart_out = chart_builders[chart_name]()
elif show == "figure" and fname:
if "calorie" in fname or "energy" in fname:
chart_out = build_calories_chart()
elif "health_label" in fname:
chart_out = build_health_label_chart()
elif "nutriscore" in fname:
chart_out = build_nutriscore_chart()
elif "macro" in fname or "nutrition" in fname:
chart_out = build_macros_chart()
if show == "table" and fname:
fp = PY_TAB_DIR / fname
if fp.exists():
tab_out = _load_table_safe(fp)
else:
reply += f"\n\n*(Could not find table: {fname})*"
new_history = (history or []) + [
{"role": "user", "content": user_msg},
{"role": "assistant", "content": reply},
]
return new_history, "", chart_out, tab_out
def _keyword_fallback(msg: str, idx: Dict, kpis: Dict):
msg_lower = msg.lower()
if not idx["python"]["figures"] and not idx["python"]["tables"]:
return (
"No artifacts found yet. Please run the pipeline first, then come back here.",
{"show": "none"},
)
kpi_text = ""
if kpis:
kpi_text = (
f"Quick summary: **{kpis.get('n_products', '?')}** food products, "
f"average calories **{kpis.get('avg_calories_per_100g', '?')} kcal/100g**, "
f"and **{kpis.get('healthy_count', '?')}** products labelled healthy."
)
if any(w in msg_lower for w in ["calorie", "calories", "energy"]):
return (
f"Here is the calorie overview for your food dataset. {kpi_text}",
{"show": "figure", "chart": "calories"},
)
if any(w in msg_lower for w in ["health label", "healthy", "unhealthy", "moderate"]):
return (
f"Here is the health label distribution. {kpi_text}",
{"show": "figure", "chart": "health_label"},
)
if any(w in msg_lower for w in ["nutriscore", "nutri-score", "grade"]):
return (
f"Here is the Nutri-Score overview. {kpi_text}",
{"show": "figure", "chart": "nutriscore"},
)
if any(w in msg_lower for w in ["protein", "fat", "sugar", "salt", "fiber", "nutrition", "macros"]):
return (
f"Here is the nutrition breakdown across health labels. {kpi_text}",
{"show": "figure", "chart": "macros"},
)
if any(w in msg_lower for w in ["recommendation", "recommend", "action"]):
return (
f"Here are the recommendation actions for the products. {kpi_text}",
{"show": "table", "scope": "python", "filename": "recommendations.csv"},
)
if any(w in msg_lower for w in ["overview", "dashboard", "summary", "kpi"]):
return (
f"Dashboard overview: {kpi_text}",
{"show": "table", "scope": "python", "filename": "food_dashboard.csv"},
)
return (
f"I can help with calories, protein, fat, sugar, salt, fiber, Nutri-Score, health labels, and recommendations. {kpi_text}",
{"show": "none"},
)
# =========================================================
# KPI CARDS
# =========================================================
def render_kpi_cards() -> str:
kpis = load_kpis()
if not kpis:
return (
''
'
🍽️
'
'
No data yet
'
'
Run the pipeline to populate these cards.
'
'
'
for key, icon, label, colour in kpi_config:
val = kpis.get(key)
if val is None:
continue
if isinstance(val, (int, float)) and val > 100:
val = f"{val:,.0f}"
html += card(icon, label, str(val), colour)
html += "
"
return html
# =========================================================
# CHARTS
# =========================================================
CHART_PALETTE = ["#7c5cbf", "#2ec4a0", "#e8537a", "#e8a230", "#5e8fef"]
def _styled_layout(**kwargs) -> dict:
defaults = dict(
template="plotly_white",
paper_bgcolor="rgba(255,255,255,0.95)",
plot_bgcolor="rgba(255,255,255,0.98)",
font=dict(family="system-ui, sans-serif", color="#2d1f4e", size=12),
margin=dict(l=60, r=20, t=70, b=70),
title=dict(font=dict(size=15, color="#4b2d8a")),
)
defaults.update(kwargs)
return defaults
def _empty_chart(title: str) -> go.Figure:
fig = go.Figure()
fig.update_layout(
title=title,
height=420,
template="plotly_white",
paper_bgcolor="rgba(255,255,255,0.95)",
annotations=[dict(
text="Run the pipeline to generate data",
x=0.5, y=0.5, xref="paper", yref="paper",
showarrow=False,
font=dict(size=14, color="rgba(124,92,191,0.5)")
)],
)
return fig
def build_calories_chart() -> go.Figure:
path = PY_TAB_DIR / "food_dashboard.csv"
if not path.exists():
return _empty_chart("Calories Overview — run the pipeline first")
df = pd.read_csv(path).sort_values("energy-kcal_100g", ascending=False).head(15)
fig = go.Figure(go.Bar(
x=df["energy-kcal_100g"],
y=df["product_name"],
orientation="h"
))
fig.update_layout(**_styled_layout(
height=500,
title=dict(text="Top 15 Products by Calories (per 100g)")
))
fig.update_yaxes(autorange="reversed")
fig.update_xaxes(title="Calories per 100g")
return fig
def build_health_label_chart() -> go.Figure:
path = PY_TAB_DIR / "health_label_counts.csv"
if not path.exists():
return _empty_chart("Health Label Distribution — run the pipeline first")
df = pd.read_csv(path)
fig = go.Figure(go.Bar(
x=df["health_label"],
y=df["count"]
))
fig.update_layout(**_styled_layout(
height=420,
title=dict(text="Health Label Distribution")
))
fig.update_xaxes(title="Health Label")
fig.update_yaxes(title="Count")
return fig
def build_nutriscore_chart() -> go.Figure:
path = PY_TAB_DIR / "nutriscore_vs_health.csv"
if not path.exists():
return _empty_chart("Nutri-Score vs Health Label — run the pipeline first")
df = pd.read_csv(path)
fig = go.Figure()
for label in df["health_label"].unique():
sub = df[df["health_label"] == label]
fig.add_trace(go.Bar(
x=sub["nutriscore_grade"],
y=sub["count"],
name=label
))
fig.update_layout(**_styled_layout(
height=450,
barmode="stack",
title=dict(text="Nutri-Score vs Health Label")
))
fig.update_xaxes(title="Nutri-Score Grade")
fig.update_yaxes(title="Count")
return fig
def build_macros_chart() -> go.Figure:
path = PY_TAB_DIR / "nutrition_by_health_label.csv"
if not path.exists():
return _empty_chart("Nutrition by Health Label — run the pipeline first")
df = pd.read_csv(path)
fig = go.Figure()
for col in ["sugars_100g", "fat_100g", "salt_100g", "proteins_100g", "fiber_100g"]:
if col in df.columns:
fig.add_trace(go.Bar(
x=df["health_label"],
y=df[col],
name=col.replace("_100g", "").replace("_", " ").title()
))
fig.update_layout(**_styled_layout(
height=450,
barmode="group",
title=dict(text="Nutrition by Health Label")
))
fig.update_xaxes(title="Health Label")
fig.update_yaxes(title="Average per 100g")
return fig
def refresh_dashboard():
return render_kpi_cards(), build_calories_chart(), build_health_label_chart(), build_nutriscore_chart()
# =========================================================
# UI
# =========================================================
ensure_dirs()
def load_css() -> str:
css_path = BASE_DIR / "style.css"
return css_path.read_text(encoding="utf-8") if css_path.exists() else ""
with gr.Blocks(title="Food Nutrition Dashboard") as demo:
gr.Markdown(
"# SE21 App Template\n"
"*Food and nutrition analytics dashboard*",
elem_id="escp_title",
)
with gr.Tab("Pipeline Runner"):
with gr.Row():
with gr.Column(scale=1):
btn_nb1 = gr.Button("Step 1: Data Creation", variant="secondary")
with gr.Column(scale=1):
btn_nb2 = gr.Button("Step 2: Python Analysis", variant="secondary")
with gr.Row():
btn_all = gr.Button("Run Full Pipeline (Both Steps)", variant="primary")
run_log = gr.Textbox(label="Execution Log", lines=18, max_lines=30, interactive=False)
btn_nb1.click(run_datacreation, outputs=[run_log])
btn_nb2.click(run_pythonanalysis, outputs=[run_log])
btn_all.click(run_full_pipeline, outputs=[run_log])
with gr.Tab("Dashboard"):
kpi_html = gr.HTML(value=render_kpi_cards)
refresh_btn = gr.Button("Refresh Dashboard", variant="primary")
gr.Markdown("#### Interactive Charts")
chart_calories = gr.Plot(label="Calories Overview")
chart_health = gr.Plot(label="Health Label Distribution")
chart_nutri = gr.Plot(label="Nutri-Score Comparison")
gr.Markdown("#### Static Figures (from notebooks)")
gallery = gr.Gallery(label="Generated Figures", columns=2, height=480, object_fit="contain")
gr.Markdown("#### Data Tables")
table_dropdown = gr.Dropdown(label="Select a table to view", choices=[], interactive=True)
table_display = gr.Dataframe(label="Table Preview", interactive=False)
def _on_refresh():
kpi, c1, c2, c3 = refresh_dashboard()
figs, dd, df = refresh_gallery()
return kpi, c1, c2, c3, figs, dd, df
refresh_btn.click(
_on_refresh,
outputs=[kpi_html, chart_calories, chart_health, chart_nutri, gallery, table_dropdown, table_display],
)
table_dropdown.change(on_table_select, inputs=[table_dropdown], outputs=[table_display])
with gr.Tab('"AI" Dashboard'):
_ai_status = (
"Connected to your **n8n workflow**." if N8N_WEBHOOK_URL
else "**LLM active.**" if LLM_ENABLED
else "Using **keyword matching**."
)
gr.Markdown(
"### Ask questions, get interactive visualisations\n\n"
f"Type a question and the system will pick the right chart or table. {_ai_status}"
)
with gr.Row(equal_height=True):
with gr.Column(scale=1):
chatbot = gr.Chatbot(label="Conversation", height=380)
user_input = gr.Textbox(
label="Ask about your data",
placeholder="e.g. Show me calories / Which foods have the most protein? / Show health label distribution",
lines=1,
)
gr.Examples(
examples=[
"Show me calories",
"Which products are highest in sugar?",
"Show health label distribution",
"Compare Nutri-Score and health label",
"Show me protein and fat by health label",
"Give me a dashboard overview",
],
inputs=user_input,
)
with gr.Column(scale=1):
ai_figure = gr.Plot(label="Interactive Chart")
ai_table = gr.Dataframe(label="Data Table", interactive=False)
user_input.submit(
ai_chat,
inputs=[user_input, chatbot],
outputs=[chatbot, user_input, ai_figure, ai_table],
)
demo.launch(css=load_css(), allowed_paths=[str(BASE_DIR)])