"""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