"""Visualization components for campaign simulation results.""" import pandas as pd import plotly.express as px import plotly.graph_objects as go from plotly.subplots import make_subplots from src.models import SimulationResult def create_gantt_chart(result: SimulationResult) -> go.Figure: """ Create Gantt chart showing quota allocation per campaign per minute. Args: result: Simulation result Returns: Plotly figure """ data: list[dict[str, str | int]] = [] for campaign in result.campaigns: for minute, sends in campaign.sends_by_minute.items(): data.append( { "Campaign": campaign.name, "Start": minute, "Finish": minute + 1, "Emails": sends, "Quota": result.quota_per_minute, } ) if not data: # Return empty figure if no data fig = go.Figure() fig.update_layout( title="No data to display", xaxis_title="Minute", yaxis_title="Campaign", ) return fig df = pd.DataFrame(data) # Create custom hover template hover_template = ( "%{y}
" + "Minute: %{customdata[0]}
" + "Emails Sent: %{customdata[1]}
" + "Quota: %{customdata[2]}
" + "" ) fig = go.Figure() # Get unique campaigns and assign colors campaigns = df["Campaign"].unique() colors = px.colors.qualitative.Plotly for i, campaign in enumerate(campaigns): campaign_data = df[df["Campaign"] == campaign] fig.add_trace( go.Bar( name=campaign, x=campaign_data["Finish"] - campaign_data["Start"], y=campaign_data["Campaign"], base=campaign_data["Start"], orientation="h", marker=dict( color=colors[i % len(colors)], line=dict(color="white", width=0.5), ), customdata=campaign_data[["Start", "Emails", "Quota"]], hovertemplate=hover_template, ) ) fig.update_layout( title=f"Campaign Execution Timeline - {result.strategy_name}", xaxis_title="Minute", yaxis_title="Campaign", barmode="overlay", height=max(400, len(campaigns) * 50), showlegend=True, hovermode="closest", ) return fig def create_quota_split_chart(result: SimulationResult) -> go.Figure: """ Create stacked bar chart showing quota split by minute. Args: result: Simulation result Returns: Plotly figure """ data: list[dict[str, str | int]] = [] # Get all minutes with activity all_minutes = set() for campaign in result.campaigns: all_minutes.update(campaign.sends_by_minute.keys()) sorted_minutes = sorted(all_minutes) # Build data for stacked bar chart for minute in sorted_minutes: for campaign in result.campaigns: sends = campaign.sends_by_minute.get(minute, 0) if sends > 0: data.append( { "Minute": minute, "Campaign": campaign.name, "Emails": sends, } ) if not data: fig = go.Figure() fig.update_layout( title="No data to display", xaxis_title="Minute", yaxis_title="Emails Sent", ) return fig df = pd.DataFrame(data) fig = px.bar( df, x="Minute", y="Emails", color="Campaign", title=f"Quota Split by Minute - {result.strategy_name}", labels={"Emails": "Emails Sent", "Minute": "Minute"}, color_discrete_sequence=px.colors.qualitative.Plotly, ) # Add quota limit line fig.add_hline( y=result.quota_per_minute, line_dash="dash", line_color="red", annotation_text=f"Quota Limit ({result.quota_per_minute})", annotation_position="right", ) fig.update_layout( xaxis=dict(dtick=1), height=500, hovermode="x unified", ) return fig def create_completion_time_comparison( results: list[SimulationResult], ) -> go.Figure: """ Create bar chart comparing completion times across strategies. Args: results: List of simulation results from different strategies Returns: Plotly figure """ data: list[dict[str, str | int]] = [] for result in results: for campaign in result.campaigns: completion_time = ( campaign.completion_minute - campaign.start_minute if campaign.completion_minute is not None else result.total_minutes - campaign.start_minute ) data.append( { "Strategy": result.strategy_name, "Campaign": campaign.name, "Completion Time (minutes)": completion_time, } ) if not data: fig = go.Figure() fig.update_layout(title="No data to display") return fig df = pd.DataFrame(data) fig = px.bar( df, x="Campaign", y="Completion Time (minutes)", color="Strategy", barmode="group", title="Completion Time Comparison by Strategy", color_discrete_sequence=px.colors.qualitative.Set2, ) fig.update_layout( height=500, xaxis_title="Campaign", yaxis_title="Completion Time (minutes)", legend_title="Strategy", ) return fig def create_slowdown_comparison( results: list[SimulationResult], baseline_times: dict[str, int], ) -> go.Figure: """ Create bar chart showing slowdown factors compared to baseline. Args: results: List of simulation results from different strategies baseline_times: Dict of campaign name to baseline completion time Returns: Plotly figure """ data: list[dict[str, str | float]] = [] for result in results: slowdowns = result.calculate_slowdown_factors(baseline_times) for campaign_name, slowdown in slowdowns.items(): data.append( { "Strategy": result.strategy_name, "Campaign": campaign_name, "Slowdown Factor": slowdown, } ) if not data: fig = go.Figure() fig.update_layout(title="No data to display") return fig df = pd.DataFrame(data) fig = px.bar( df, x="Campaign", y="Slowdown Factor", color="Strategy", barmode="group", title="Slowdown Factor Comparison (vs. Running Alone)", color_discrete_sequence=px.colors.qualitative.Set2, ) # Add baseline reference line at y=1.0 fig.add_hline( y=1.0, line_dash="dash", line_color="gray", annotation_text="Baseline (no slowdown)", annotation_position="right", ) fig.update_layout( height=500, xaxis_title="Campaign", yaxis_title="Slowdown Factor (higher = slower)", legend_title="Strategy", ) return fig def create_metrics_table(results: list[SimulationResult]) -> pd.DataFrame: """ Create summary metrics table for all strategies. Args: results: List of simulation results from different strategies Returns: Pandas DataFrame with metrics """ data: list[dict[str, str | float | int]] = [] for result in results: data.append( { "Strategy": result.strategy_name, "Total Minutes": result.total_minutes, "Avg Completion Time": round(result.avg_completion_time, 2), "Max Completion Time": result.max_completion_time, "Quota Utilization (%)": round(result.quota_utilization, 2), } ) return pd.DataFrame(data) def create_timeline_view(result: SimulationResult) -> go.Figure: """ Create timeline view showing when each campaign is sending emails. Args: result: Simulation result Returns: Plotly figure """ fig = go.Figure() colors = px.colors.qualitative.Plotly for i, campaign in enumerate(result.campaigns): if not campaign.sends_by_minute: continue minutes = sorted(campaign.sends_by_minute.keys()) sends = [campaign.sends_by_minute[m] for m in minutes] fig.add_trace( go.Scatter( x=minutes, y=sends, name=campaign.name, mode="lines+markers", line=dict(color=colors[i % len(colors)], width=2), marker=dict(size=6), hovertemplate=( f"{campaign.name}
" + "Minute: %{x}
" + "Emails Sent: %{y}
" + "" ), ) ) fig.update_layout( title=f"Email Send Timeline - {result.strategy_name}", xaxis_title="Minute", yaxis_title="Emails Sent", height=500, hovermode="x unified", showlegend=True, ) # Add quota limit reference line fig.add_hline( y=result.quota_per_minute, line_dash="dash", line_color="red", opacity=0.5, annotation_text=f"Quota Limit ({result.quota_per_minute})", annotation_position="top right", ) return fig def create_strategy_comparison_dashboard( results: list[SimulationResult], baseline_times: dict[str, int], ) -> go.Figure: """ Create comprehensive dashboard comparing all strategies. Args: results: List of simulation results from different strategies baseline_times: Dict of campaign name to baseline completion time Returns: Plotly figure with subplots """ # Create subplots fig = make_subplots( rows=2, cols=2, subplot_titles=( "Completion Times", "Slowdown Factors", "Quota Utilization", "Average Completion Time", ), specs=[ [{"type": "bar"}, {"type": "bar"}], [{"type": "bar"}, {"type": "bar"}], ], ) # Prepare data completion_data: list[dict[str, str | int]] = [] slowdown_data: list[dict[str, str | float]] = [] utilization_data: list[dict[str, str | float]] = [] avg_time_data: list[dict[str, str | float]] = [] for result in results: # Completion times for campaign in result.campaigns: completion_time = ( campaign.completion_minute - campaign.start_minute if campaign.completion_minute is not None else result.total_minutes - campaign.start_minute ) completion_data.append( { "Strategy": result.strategy_name, "Campaign": campaign.name, "Time": completion_time, } ) # Slowdown factors slowdowns = result.calculate_slowdown_factors(baseline_times) for campaign_name, slowdown in slowdowns.items(): slowdown_data.append( { "Strategy": result.strategy_name, "Campaign": campaign_name, "Slowdown": slowdown, } ) # Utilization utilization_data.append( { "Strategy": result.strategy_name, "Utilization": result.quota_utilization, } ) # Average completion time avg_time_data.append( { "Strategy": result.strategy_name, "Avg Time": result.avg_completion_time, } ) # Create dataframes completion_df = pd.DataFrame(completion_data) slowdown_df = pd.DataFrame(slowdown_data) utilization_df = pd.DataFrame(utilization_data) avg_time_df = pd.DataFrame(avg_time_data) colors = px.colors.qualitative.Set2 # Plot 1: Completion times (grouped by strategy) for i, strategy in enumerate(completion_df["Strategy"].unique()): strategy_data = completion_df[completion_df["Strategy"] == strategy] fig.add_trace( go.Bar( x=strategy_data["Campaign"], y=strategy_data["Time"], name=strategy, marker_color=colors[i % len(colors)], showlegend=True, legendgroup=strategy, ), row=1, col=1, ) # Plot 2: Slowdown factors for i, strategy in enumerate(slowdown_df["Strategy"].unique()): strategy_data = slowdown_df[slowdown_df["Strategy"] == strategy] fig.add_trace( go.Bar( x=strategy_data["Campaign"], y=strategy_data["Slowdown"], name=strategy, marker_color=colors[i % len(colors)], showlegend=False, legendgroup=strategy, ), row=1, col=2, ) # Plot 3: Quota utilization fig.add_trace( go.Bar( x=utilization_df["Strategy"], y=utilization_df["Utilization"], marker_color=colors, showlegend=False, ), row=2, col=1, ) # Plot 4: Average completion time fig.add_trace( go.Bar( x=avg_time_df["Strategy"], y=avg_time_df["Avg Time"], marker_color=colors, showlegend=False, ), row=2, col=2, ) # Update layout fig.update_xaxes(title_text="Campaign", row=1, col=1) fig.update_xaxes(title_text="Campaign", row=1, col=2) fig.update_xaxes(title_text="Strategy", row=2, col=1) fig.update_xaxes(title_text="Strategy", row=2, col=2) fig.update_yaxes(title_text="Minutes", row=1, col=1) fig.update_yaxes(title_text="Slowdown Factor", row=1, col=2) fig.update_yaxes(title_text="Utilization (%)", row=2, col=1) fig.update_yaxes(title_text="Minutes", row=2, col=2) fig.update_layout( height=800, title_text="Strategy Comparison Dashboard", showlegend=True, barmode="group", ) return fig