"""
visualization/gantt.py
-----------------------
Interactive Plotly Gantt chart: planned vs actual vs forecasted timelines.
"""
import pandas as pd
import plotly.graph_objects as go
from datetime import datetime
from typing import Optional
from data_loader import DataLoader
STATUS_COLORS = {
"completed": "#22c55e", # green
"in_progress": "#f59e0b", # amber
"not_started": "#94a3b8", # slate
"critical": "#ef4444", # red (critical path)
}
TIMELINE_COLORS = {
"planned": "rgba(99,102,241,0.6)", # indigo
"actual": "rgba(34,197,94,0.7)", # green
"forecasted": "rgba(245,158,11,0.7)", # amber
}
def build_gantt(project_id: str,
loader: Optional[DataLoader] = None,
predictions_df: Optional[pd.DataFrame] = None,
critical_path_ids: Optional[list] = None,
today: Optional[datetime] = None) -> go.Figure:
"""
Build a Plotly Gantt chart for the given project.
Bars per activity (3 tracks):
- Top: Planned (planned_start → planned_end)
- Mid: Actual (actual_start → actual_end or today)
- Bot: Forecasted (today → ensemble_end from predictions)
"""
if loader is None:
loader = DataLoader()
if today is None:
today = datetime(2024, 6, 1)
if critical_path_ids is None:
critical_path_ids = []
today_ts = pd.Timestamp(today)
acts = loader.get_project_activities(project_id)
if acts.empty:
return go.Figure().add_annotation(text="No activities found", showarrow=False)
# Merge predictions
pred_map = {}
if predictions_df is not None and not predictions_df.empty:
for _, row in predictions_df.iterrows():
pred_map[row.get("activity_id", "")] = row.to_dict()
# Sort by planned start
acts = acts.sort_values("planned_start_date", na_position="last")
fig = go.Figure()
y_labels = []
legend_added = set()
def add_bar(name, x0, x1, y_pos, color, showlegend=False):
if pd.isna(x0) or pd.isna(x1) or x0 >= x1:
return
leg = name not in legend_added
if leg:
legend_added.add(name)
fig.add_trace(go.Bar(
x=[(x1 - x0).total_seconds() * 1000],
base=[x0],
y=[y_pos],
orientation="h",
name=name,
marker_color=color,
showlegend=showlegend or leg,
hovertemplate=(
f"{name}
"
f"Start: {x0.strftime('%Y-%m-%d')}
"
f"End: {x1.strftime('%Y-%m-%d')}
"
f"Duration: {(x1-x0).days} days"
),
width=0.25,
))
for idx, (_, row) in enumerate(acts.iterrows()):
act_id = str(row["id"])
act_name = row.get("name", act_id)
status = str(row.get("status", "not_started"))
is_critical = act_id in critical_path_ids
# Y position: 3 tracks per activity
base_y = idx * 3
y_label = f"{'🔴 ' if is_critical else ''}{act_name[:35]}"
y_labels.extend([
f"{y_label} (planned)",
f"{y_label} (actual)",
f"{y_label} (forecast)",
])
planned_s = pd.to_datetime(row.get("planned_start_date"), errors="coerce")
planned_e = pd.to_datetime(row.get("planned_end_date"), errors="coerce")
actual_s = pd.to_datetime(row.get("actual_start_date"), errors="coerce")
actual_e = pd.to_datetime(row.get("actual_end_date"), errors="coerce")
# Planned bar
if not pd.isna(planned_s) and not pd.isna(planned_e):
color = TIMELINE_COLORS["planned"]
add_bar("Planned", planned_s, planned_e, base_y, color, showlegend=True)
# Actual bar
if not pd.isna(actual_s):
act_end_display = actual_e if not pd.isna(actual_e) else today_ts
color = STATUS_COLORS.get(status, STATUS_COLORS["not_started"])
add_bar("Actual", actual_s, act_end_display, base_y + 1, color, showlegend=True)
# Forecasted bar (from predictions)
pred = pred_map.get(act_id, {})
ens_end = pred.get("ensemble_end")
if ens_end and status in ("in_progress", "not_started"):
ens_ts = pd.Timestamp(ens_end)
fc_start = today_ts
if not pd.isna(actual_s):
fc_start = actual_s if pd.isna(actual_e) else actual_e
add_bar("Forecasted", fc_start, ens_ts, base_y + 2,
TIMELINE_COLORS["forecasted"], showlegend=True)
# Today line
fig.add_vline(
x=today_ts.timestamp() * 1000,
line_color="red", line_dash="dash", line_width=2,
annotation_text="Today",
annotation_position="top right",
)
n_acts = len(acts)
fig.update_layout(
title=dict(text=f"📅 Project Schedule — {project_id}", font=dict(size=18)),
xaxis=dict(title="Date", type="date"),
yaxis=dict(
title="",
tickvals=list(range(0, n_acts * 3, 3)),
ticktext=[acts.iloc[i].get("name", "")[:35]
for i in range(min(n_acts, len(acts)))],
autorange="reversed",
),
height=max(400, n_acts * 55),
barmode="overlay",
legend=dict(orientation="h", y=1.05),
plot_bgcolor="#0f172a",
paper_bgcolor="#0f172a",
font=dict(color="#e2e8f0"),
)
return fig