# Advanced Chart Generator - Plotly-Based Dynamic Charts
"""
Generates Plotly charts based on VisualizationDecision.
This module takes decisions from visualization_intelligence.py
and creates actual chart payloads for the frontend.
Features:
- Dynamic chart type rendering
- Consistent styling
- Proper axis labels
- Interactive tooltips
- Color schemes per mode
"""
from typing import Dict, List, Optional, Any, Tuple
import json
# Color schemes for different contexts - EXPANDED for 20+ items
COLOR_SCHEMES = {
"default": [
"#f97316", "#3b82f6", "#22c55e", "#a855f7", "#ef4444",
"#06b6d4", "#f59e0b", "#ec4899", "#8b5cf6", "#14b8a6",
"#84cc16", "#6366f1", "#f43f5e", "#0ea5e9", "#d946ef",
"#eab308", "#10b981", "#6b7280", "#78716c", "#0284c7"
],
"executive": ["#1e40af", "#3b82f6", "#60a5fa", "#93c5fd", "#1d4ed8", "#2563eb"],
"finance": ["#059669", "#10b981", "#34d399", "#6ee7b7", "#047857", "#0d9488"],
"warning": ["#dc2626", "#ef4444", "#f87171", "#fca5a5", "#b91c1c", "#991b1b"],
"neutral": ["#6b7280", "#9ca3af", "#d1d5db", "#e5e7eb", "#4b5563", "#374151"],
}
def generate_dynamic_chart(
df,
chart_type: str,
x_col: str,
y_col: str,
title: str,
group_col: Optional[str] = None,
currency_symbol: str = "₹",
color_scheme: str = "default",
limit: int = 10
) -> Dict[str, Any]:
"""
Generate a Plotly chart payload based on chart type.
Returns a JSON-serializable dict for frontend rendering.
"""
if df is None or df.empty:
return {"error": "No data available"}
colors = COLOR_SCHEMES.get(color_scheme, COLOR_SCHEMES["default"])
# Route to specific generator
generators = {
"line": _generate_line_chart,
"bar": _generate_bar_chart,
"grouped_bar": _generate_grouped_bar_chart,
"stacked_bar": _generate_stacked_bar_chart,
"pie": _generate_pie_chart,
"donut": _generate_donut_chart,
"area": _generate_area_chart,
"scatter": _generate_scatter_chart,
"forecast": _generate_forecast_chart,
"waterfall": _generate_waterfall_chart,
}
generator = generators.get(chart_type, _generate_bar_chart)
return generator(
df=df,
x_col=x_col,
y_col=y_col,
title=title,
group_col=group_col,
currency_symbol=currency_symbol,
colors=colors,
limit=limit
)
def _generate_line_chart(
df, x_col: str, y_col: str, title: str,
group_col: Optional[str], currency_symbol: str,
colors: List[str], limit: int
) -> Dict:
"""Generate line chart for trends."""
# Aggregate by x column if needed
if x_col and x_col in df.columns and y_col and y_col in df.columns:
aggregated = df.groupby(x_col)[y_col].sum().reset_index()
aggregated = aggregated.sort_values(x_col).tail(limit)
x_values = aggregated[x_col].astype(str).tolist()
y_values = aggregated[y_col].tolist()
else:
x_values = list(range(len(df)))
y_values = df[y_col].tolist() if y_col in df.columns else []
return {
"type": "plotly",
"chart_type": "line",
"data": [{
"type": "scatter",
"mode": "lines+markers",
"x": x_values,
"y": y_values,
"line": {"color": colors[0], "width": 3},
"marker": {"size": 8, "color": colors[0]},
"hovertemplate": f"%{{x}}
{currency_symbol}%{{y:,.0f}}"
}],
"layout": {
"title": {"text": title, "font": {"size": 16}},
"paper_bgcolor": "rgba(0,0,0,0)",
"plot_bgcolor": "rgba(0,0,0,0)",
"xaxis": {"title": x_col, "gridcolor": "rgba(128,128,128,0.2)"},
"yaxis": {"title": y_col, "gridcolor": "rgba(128,128,128,0.2)"},
"margin": {"l": 60, "r": 30, "t": 50, "b": 50},
"showlegend": False
}
}
def _generate_bar_chart(
df, x_col: str, y_col: str, title: str,
group_col: Optional[str], currency_symbol: str,
colors: List[str], limit: int
) -> Dict:
"""Generate bar chart for comparisons/rankings."""
if x_col and x_col in df.columns and y_col and y_col in df.columns:
aggregated = df.groupby(x_col)[y_col].sum().reset_index()
aggregated = aggregated.sort_values(y_col, ascending=False).head(limit)
x_values = aggregated[x_col].astype(str).tolist()
y_values = aggregated[y_col].tolist()
else:
x_values = []
y_values = []
# Assign colors per bar
bar_colors = [colors[i % len(colors)] for i in range(len(x_values))]
return {
"type": "plotly",
"chart_type": "bar",
"data": [{
"type": "bar",
"x": x_values,
"y": y_values,
"marker": {"color": bar_colors, "line": {"width": 0}},
"hovertemplate": f"%{{x}}
{currency_symbol}%{{y:,.0f}}"
}],
"layout": {
"title": {"text": title, "font": {"size": 16}},
"paper_bgcolor": "rgba(0,0,0,0)",
"plot_bgcolor": "rgba(0,0,0,0)",
"xaxis": {"title": "", "gridcolor": "rgba(128,128,128,0.2)", "tickangle": -45},
"yaxis": {"title": y_col, "gridcolor": "rgba(128,128,128,0.2)"},
"margin": {"l": 60, "r": 30, "t": 50, "b": 100},
"showlegend": False
}
}
def _generate_pie_chart(
df, x_col: str, y_col: str, title: str,
group_col: Optional[str], currency_symbol: str,
colors: List[str], limit: int
) -> Dict:
"""Generate pie chart for proportional data."""
print(f"[PIE CHART DEBUG] x_col={x_col}, y_col={y_col}, limit={limit}")
print(f"[PIE CHART DEBUG] df columns: {list(df.columns)}")
print(f"[PIE CHART DEBUG] df['{x_col}'] unique values: {df[x_col].nunique() if x_col in df.columns else 'N/A'}")
if x_col in df.columns:
print(f"[PIE CHART DEBUG] Sample values: {df[x_col].head(5).tolist()}")
if x_col and x_col in df.columns and y_col and y_col in df.columns:
aggregated = df.groupby(x_col)[y_col].sum().reset_index()
aggregated = aggregated.sort_values(y_col, ascending=False).head(limit)
labels = aggregated[x_col].astype(str).tolist()
values = aggregated[y_col].tolist()
print(f"[PIE CHART DEBUG] Final labels: {labels}")
else:
labels = []
values = []
pie_colors = [colors[i % len(colors)] for i in range(len(labels))]
return {
"type": "plotly",
"chart_type": "pie",
"data": [{
"type": "pie",
"labels": labels,
"values": values,
"marker": {"colors": pie_colors},
"textposition": "auto",
"textinfo": "label+percent",
"hovertemplate": f"%{{label}}
{currency_symbol}%{{value:,.0f}}
%{{percent}}"
}],
"layout": {
"title": {"text": title, "font": {"size": 16}},
"paper_bgcolor": "rgba(0,0,0,0)",
"showlegend": True,
"legend": {"font": {}},
"margin": {"l": 30, "r": 30, "t": 60, "b": 30}
}
}
def _generate_donut_chart(
df, x_col: str, y_col: str, title: str,
group_col: Optional[str], currency_symbol: str,
colors: List[str], limit: int
) -> Dict:
"""Generate donut chart (pie with hole)."""
pie = _generate_pie_chart(df, x_col, y_col, title, group_col, currency_symbol, colors, limit)
if pie.get("data"):
pie["data"][0]["hole"] = 0.4
pie["chart_type"] = "donut"
return pie
def _generate_grouped_bar_chart(
df, x_col: str, y_col: str, title: str,
group_col: Optional[str], currency_symbol: str,
colors: List[str], limit: int
) -> Dict:
"""Generate grouped bar chart for multi-category comparison."""
if not group_col or group_col not in df.columns:
return _generate_bar_chart(df, x_col, y_col, title, None, currency_symbol, colors, limit)
traces = []
groups = df[group_col].unique()[:5] # Max 5 groups
for i, group in enumerate(groups):
group_df = df[df[group_col] == group]
aggregated = group_df.groupby(x_col)[y_col].sum().reset_index()
aggregated = aggregated.sort_values(y_col, ascending=False).head(limit)
traces.append({
"type": "bar",
"name": str(group),
"x": aggregated[x_col].astype(str).tolist(),
"y": aggregated[y_col].tolist(),
"marker": {"color": colors[i % len(colors)]},
"hovertemplate": f"%{{x}}
{group}: {currency_symbol}%{{y:,.0f}}"
})
return {
"type": "plotly",
"chart_type": "grouped_bar",
"data": traces,
"layout": {
"title": {"text": title, "font": {"size": 16, "color": "#ffffff"}},
"paper_bgcolor": "rgba(0,0,0,0)",
"plot_bgcolor": "rgba(0,0,0,0)",
"barmode": "group",
"xaxis": {"title": "", "gridcolor": "rgba(255,255,255,0.1)", "color": "#ffffff"},
"yaxis": {"title": y_col, "gridcolor": "rgba(255,255,255,0.1)", "color": "#ffffff"},
"legend": {"font": {"color": "#ffffff"}},
"margin": {"l": 60, "r": 30, "t": 50, "b": 80}
}
}
def _generate_stacked_bar_chart(
df, x_col: str, y_col: str, title: str,
group_col: Optional[str], currency_symbol: str,
colors: List[str], limit: int
) -> Dict:
"""Generate stacked bar chart for composition."""
if not group_col or group_col not in df.columns:
return _generate_bar_chart(df, x_col, y_col, title, None, currency_symbol, colors, limit)
traces = []
groups = df[group_col].unique()[:8] # Max 8 stacks
for i, group in enumerate(groups):
group_df = df[df[group_col] == group]
aggregated = group_df.groupby(x_col)[y_col].sum().reset_index()
traces.append({
"type": "bar",
"name": str(group),
"x": aggregated[x_col].astype(str).tolist(),
"y": aggregated[y_col].tolist(),
"marker": {"color": colors[i % len(colors)]},
"hovertemplate": f"%{{x}}
{group}: {currency_symbol}%{{y:,.0f}}"
})
return {
"type": "plotly",
"chart_type": "stacked_bar",
"data": traces,
"layout": {
"title": {"text": title, "font": {"size": 16, "color": "#ffffff"}},
"paper_bgcolor": "rgba(0,0,0,0)",
"plot_bgcolor": "rgba(0,0,0,0)",
"barmode": "stack",
"xaxis": {"title": "", "gridcolor": "rgba(255,255,255,0.1)", "color": "#ffffff"},
"yaxis": {"title": y_col, "gridcolor": "rgba(255,255,255,0.1)", "color": "#ffffff"},
"legend": {"font": {"color": "#ffffff"}},
"margin": {"l": 60, "r": 30, "t": 50, "b": 80}
}
}
def _generate_pie_chart(
df, x_col: str, y_col: str, title: str,
group_col: Optional[str], currency_symbol: str,
colors: List[str], limit: int
) -> Dict:
"""Generate pie chart for proportions."""
if x_col and x_col in df.columns and y_col and y_col in df.columns:
aggregated = df.groupby(x_col)[y_col].sum().reset_index()
aggregated = aggregated.sort_values(y_col, ascending=False).head(8) # Max 8 slices
labels = aggregated[x_col].astype(str).tolist()
values = aggregated[y_col].tolist()
else:
labels = []
values = []
return {
"type": "plotly",
"chart_type": "pie",
"data": [{
"type": "pie",
"labels": labels,
"values": values,
"marker": {"colors": colors[:len(labels)]},
"textinfo": "label+percent",
"textposition": "inside",
"hovertemplate": f"%{{label}}
{currency_symbol}%{{value:,.0f}}
%{{percent}}"
}],
"layout": {
"title": {"text": title, "font": {"size": 16, "color": "#ffffff"}},
"paper_bgcolor": "rgba(0,0,0,0)",
"plot_bgcolor": "rgba(0,0,0,0)",
"showlegend": True,
"legend": {"font": {"color": "#ffffff"}},
"margin": {"l": 30, "r": 30, "t": 50, "b": 30}
}
}
def _generate_donut_chart(
df, x_col: str, y_col: str, title: str,
group_col: Optional[str], currency_symbol: str,
colors: List[str], limit: int
) -> Dict:
"""Generate donut chart (pie with hole)."""
chart = _generate_pie_chart(df, x_col, y_col, title, group_col, currency_symbol, colors, limit)
if chart.get("data"):
chart["data"][0]["hole"] = 0.4
chart["chart_type"] = "donut"
return chart
def _generate_area_chart(
df, x_col: str, y_col: str, title: str,
group_col: Optional[str], currency_symbol: str,
colors: List[str], limit: int
) -> Dict:
"""Generate area chart for cumulative trends."""
if x_col and x_col in df.columns and y_col and y_col in df.columns:
aggregated = df.groupby(x_col)[y_col].sum().reset_index()
aggregated = aggregated.sort_values(x_col).tail(limit)
x_values = aggregated[x_col].astype(str).tolist()
y_values = aggregated[y_col].tolist()
else:
x_values = []
y_values = []
return {
"type": "plotly",
"chart_type": "area",
"data": [{
"type": "scatter",
"mode": "lines",
"x": x_values,
"y": y_values,
"fill": "tozeroy",
"fillcolor": f"rgba(249, 115, 22, 0.3)",
"line": {"color": colors[0], "width": 2},
"hovertemplate": f"%{{x}}
{currency_symbol}%{{y:,.0f}}"
}],
"layout": {
"title": {"text": title, "font": {"size": 16, "color": "#ffffff"}},
"paper_bgcolor": "rgba(0,0,0,0)",
"plot_bgcolor": "rgba(0,0,0,0)",
"xaxis": {"title": x_col, "gridcolor": "rgba(255,255,255,0.1)", "color": "#ffffff"},
"yaxis": {"title": y_col, "gridcolor": "rgba(255,255,255,0.1)", "color": "#ffffff"},
"margin": {"l": 60, "r": 30, "t": 50, "b": 50}
}
}
def _generate_scatter_chart(
df, x_col: str, y_col: str, title: str,
group_col: Optional[str], currency_symbol: str,
colors: List[str], limit: int
) -> Dict:
"""Generate scatter plot for correlations."""
# For scatter, use first two numeric columns if not specified
numeric_cols = df.select_dtypes(include=['int64', 'float64']).columns.tolist()
if len(numeric_cols) >= 2:
x_col = x_col if x_col in numeric_cols else numeric_cols[0]
y_col = y_col if y_col in numeric_cols else numeric_cols[1]
x_values = df[x_col].tolist()[:limit*10]
y_values = df[y_col].tolist()[:limit*10]
else:
x_values = []
y_values = []
return {
"type": "plotly",
"chart_type": "scatter",
"data": [{
"type": "scatter",
"mode": "markers",
"x": x_values,
"y": y_values,
"marker": {
"color": colors[0],
"size": 10,
"opacity": 0.7,
"line": {"width": 1, "color": "#ffffff"}
},
"hovertemplate": f"{x_col}: %{{x:,.0f}}
{y_col}: %{{y:,.0f}}"
}],
"layout": {
"title": {"text": title, "font": {"size": 16, "color": "#ffffff"}},
"paper_bgcolor": "rgba(0,0,0,0)",
"plot_bgcolor": "rgba(0,0,0,0)",
"xaxis": {"title": x_col, "gridcolor": "rgba(255,255,255,0.1)", "color": "#ffffff"},
"yaxis": {"title": y_col, "gridcolor": "rgba(255,255,255,0.1)", "color": "#ffffff"},
"margin": {"l": 60, "r": 30, "t": 50, "b": 50}
}
}
def _generate_forecast_chart(
df, x_col: str, y_col: str, title: str,
group_col: Optional[str], currency_symbol: str,
colors: List[str], limit: int
) -> Dict:
"""Generate forecast chart with historical + predicted + confidence bands."""
if x_col and x_col in df.columns and y_col and y_col in df.columns:
aggregated = df.groupby(x_col)[y_col].sum().reset_index()
aggregated = aggregated.sort_values(x_col)
x_values = aggregated[x_col].astype(str).tolist()
y_values = aggregated[y_col].tolist()
# Simple linear forecast (3 periods)
if len(y_values) >= 2:
last_val = y_values[-1]
growth_rate = (y_values[-1] - y_values[0]) / len(y_values) if len(y_values) > 1 else 0
forecast_x = [f"Forecast {i+1}" for i in range(3)]
forecast_y = [last_val + growth_rate * (i + 1) for i in range(3)]
# Confidence bands (±15%)
upper_band = [v * 1.15 for v in forecast_y]
lower_band = [v * 0.85 for v in forecast_y]
else:
forecast_x = []
forecast_y = []
upper_band = []
lower_band = []
else:
x_values = []
y_values = []
forecast_x = []
forecast_y = []
upper_band = []
lower_band = []
traces = [
# Historical line
{
"type": "scatter",
"mode": "lines+markers",
"name": "Historical",
"x": x_values,
"y": y_values,
"line": {"color": colors[0], "width": 3},
"marker": {"size": 8},
"hovertemplate": f"%{{x}}
{currency_symbol}%{{y:,.0f}}"
},
# Forecast line
{
"type": "scatter",
"mode": "lines+markers",
"name": "Forecast",
"x": forecast_x,
"y": forecast_y,
"line": {"color": colors[1], "width": 3, "dash": "dash"},
"marker": {"size": 8},
"hovertemplate": f"%{{x}}
Predicted: {currency_symbol}%{{y:,.0f}}"
},
# Upper confidence band
{
"type": "scatter",
"mode": "lines",
"name": "Upper Band",
"x": forecast_x,
"y": upper_band,
"line": {"width": 0},
"showlegend": False,
"hoverinfo": "skip"
},
# Lower confidence band (with fill)
{
"type": "scatter",
"mode": "lines",
"name": "Lower Band",
"x": forecast_x,
"y": lower_band,
"line": {"width": 0},
"fill": "tonexty",
"fillcolor": "rgba(59, 130, 246, 0.2)",
"showlegend": False,
"hoverinfo": "skip"
}
]
return {
"type": "plotly",
"chart_type": "forecast",
"data": traces,
"layout": {
"title": {"text": title, "font": {"size": 16, "color": "#ffffff"}},
"paper_bgcolor": "rgba(0,0,0,0)",
"plot_bgcolor": "rgba(0,0,0,0)",
"xaxis": {"title": "", "gridcolor": "rgba(255,255,255,0.1)", "color": "#ffffff"},
"yaxis": {"title": y_col, "gridcolor": "rgba(255,255,255,0.1)", "color": "#ffffff"},
"legend": {"font": {"color": "#ffffff"}},
"margin": {"l": 60, "r": 30, "t": 50, "b": 50}
}
}
def _generate_waterfall_chart(
df, x_col: str, y_col: str, title: str,
group_col: Optional[str], currency_symbol: str,
colors: List[str], limit: int
) -> Dict:
"""Generate waterfall chart for financial breakdowns."""
if x_col and x_col in df.columns and y_col and y_col in df.columns:
aggregated = df.groupby(x_col)[y_col].sum().reset_index()
aggregated = aggregated.sort_values(y_col, ascending=False).head(limit)
labels = aggregated[x_col].astype(str).tolist()
values = aggregated[y_col].tolist()
# Add total at end
labels.append("Total")
values.append(sum(values[:-1]) if len(values) > 1 else values[0] if values else 0)
# Waterfall measure types
measures = ["relative"] * (len(labels) - 1) + ["total"]
else:
labels = []
values = []
measures = []
return {
"type": "plotly",
"chart_type": "waterfall",
"data": [{
"type": "waterfall",
"orientation": "v",
"x": labels,
"y": values,
"measure": measures,
"connector": {"line": {"color": "rgba(255,255,255,0.3)"}},
"increasing": {"marker": {"color": colors[2] if len(colors) > 2 else "#22c55e"}},
"decreasing": {"marker": {"color": colors[4] if len(colors) > 4 else "#ef4444"}},
"totals": {"marker": {"color": colors[0]}},
"hovertemplate": f"%{{x}}
{currency_symbol}%{{y:,.0f}}"
}],
"layout": {
"title": {"text": title, "font": {"size": 16, "color": "#ffffff"}},
"paper_bgcolor": "rgba(0,0,0,0)",
"plot_bgcolor": "rgba(0,0,0,0)",
"xaxis": {"title": "", "gridcolor": "rgba(255,255,255,0.1)", "color": "#ffffff"},
"yaxis": {"title": y_col, "gridcolor": "rgba(255,255,255,0.1)", "color": "#ffffff"},
"margin": {"l": 60, "r": 30, "t": 50, "b": 80}
}
}
# ============================================================================
# MAIN ENTRY POINT
# ============================================================================
def create_chart_from_decision(
decision, # VisualizationDecision from visualization_intelligence.py
df,
currency_symbol: str = "₹",
user_role: str = "analyst"
) -> Optional[Dict]:
"""
Create a chart payload from a VisualizationDecision.
This is the main entry point that bridges the decision layer
with actual chart generation.
IMPORTANT: Only uses REAL data from the DataFrame - no fabrication!
"""
if not decision.should_render:
return None
# STRICT DATA VALIDATION - Ensure we have real data
if df is None or df.empty:
return {"error": "No data available - cannot generate chart"}
# Verify columns exist in DataFrame
if decision.x_column and decision.x_column not in df.columns:
return {"error": f"Column '{decision.x_column}' not found in uploaded data"}
if decision.y_column and decision.y_column not in df.columns:
return {"error": f"Column '{decision.y_column}' not found in uploaded data"}
# Determine color scheme based on role
color_scheme = "default"
if user_role:
role_lower = user_role.lower()
if role_lower == "executive":
color_scheme = "executive"
elif role_lower == "finance":
color_scheme = "finance"
chart = generate_dynamic_chart(
df=df,
chart_type=decision.chart_type.value,
x_col=decision.x_column,
y_col=decision.y_column,
title=decision.title,
group_col=decision.group_column,
currency_symbol=currency_symbol,
color_scheme=color_scheme,
limit=decision.limit if hasattr(decision, 'limit') else 10 # Use dynamic limit from query
)
# Add data source metadata to confirm real data usage
if chart and not chart.get("error"):
chart["_data_source"] = {
"type": "uploaded_data",
"row_count": len(df),
"columns_used": [c for c in [decision.x_column, decision.y_column, decision.group_column] if c],
"verified": True
}
return chart