feat: drug timeline Gantt chart tab (Task D.3)

IMPLEMENTATION_PLAN.md

@@ -215,13 +215,17 @@ Comprehensive review and improvement of all Plotly charts in the Dash dashboard.
 - **Checkpoint**: Dose distribution visible as box/violin plots
 
 ### D.3 Drug timeline (Gantt chart)
-- [ ] Create `get_drug_timeline()` query in `pathway_queries.py`:
+- [x] Create `get_drug_timeline()` query in `pathway_queries.py`:
   - Level 3 nodes with `first_seen`, `last_seen`, `labels`, `value` per drug × directory
-- [ ] Create `create_drug_timeline_figure(data, title)` in plotly_generator.py:
+  - Aggregates across trusts: MIN(first_seen), MAX(last_seen), SUM(value), weighted avg cost_pp_pa
+  - Supports directory/trust filters
+- [x] Create `create_drug_timeline_figure(data, title)` in plotly_generator.py:
   - Gantt-style using `go.Bar` (horizontal bars from first_seen to last_seen)
-  - Grouped by directory, colored by patient count
+  - One trace per bar, grouped by directory with legend grouping
-- [ ] Add "Timeline" tab to `TAB_DEFINITIONS` in `chart_card.py`
+  - Colored by directory using `DRUG_PALETTE`, patient count as bar text
-- [ ] Add callback wiring
+  - Dynamic height (28px per bar), `_base_layout()` + `_smart_legend()`
+- [x] Add "Timeline" tab to `TAB_DEFINITIONS` in `chart_card.py` (8th tab)
+- [x] Add `_render_timeline()` helper + dispatch case in `chart.py`
 - **Checkpoint**: Timeline tab shows when each drug cohort was active
 
 ### D.4 NICE TA compliance dashboard

dash_app/callbacks/chart.py

@@ -367,6 +367,31 @@ def _render_network(app_state, title):
     return create_drug_network_figure(data, title)
 
 
+def _render_timeline(app_state, title):
+    """Build the drug timeline Gantt chart from current filter state."""
+    from dash_app.data.queries import get_drug_timeline
+    from visualization.plotly_generator import create_drug_timeline_figure
+
+    filter_id = (app_state or {}).get("date_filter_id", "all_6mo")
+    chart_type = (app_state or {}).get("chart_type", "directory")
+
+    selected_dirs = (app_state or {}).get("selected_directorates") or []
+    selected_trusts = (app_state or {}).get("selected_trusts") or []
+    directory = selected_dirs[0] if len(selected_dirs) == 1 else None
+    trust = selected_trusts[0] if len(selected_trusts) == 1 else None
+
+    try:
+        data = get_drug_timeline(filter_id, chart_type, directory, trust)
+    except Exception:
+        log.exception("Failed to load drug timeline data")
+        return _empty_figure("Failed to load drug timeline data.")
+
+    if not data:
+        return _empty_figure("No drug timeline data available.\nTry adjusting your filters.")
+
+    return create_drug_timeline_figure(data, title)
+
+
 def register_chart_callbacks(app):
     """Register tab switching, pathway data loading, and chart rendering callbacks."""
 
@@ -519,6 +544,9 @@ def register_chart_callbacks(app):
         elif active_tab == "network":
             fig = _render_network(app_state, title)
 
+        elif active_tab == "timeline":
+            fig = _render_timeline(app_state, title)
+
         else:
             # Placeholder for charts not yet implemented
             tab_label = dict(TAB_DEFINITIONS).get(active_tab, active_tab)

dash_app/components/chart_card.py

@@ -12,6 +12,7 @@ TAB_DEFINITIONS = [
     ("depth", "Depth"),
     ("scatter", "Scatter"),
     ("network", "Network"),
+    ("timeline", "Timeline"),
 ]
 
 # Full set retained for Trust Comparison dashboard (Phase 10.8)

dash_app/data/queries.py

@@ -28,6 +28,7 @@ from data_processing.pathway_queries import (
     get_pathway_depth_distribution as _get_pathway_depth_distribution,
     get_duration_cost_scatter as _get_duration_cost_scatter,
     get_drug_network as _get_drug_network,
+    get_drug_timeline as _get_drug_timeline,
 )
 
 DB_PATH = Path(__file__).resolve().parents[2] / "data" / "pathways.db"
@@ -227,3 +228,13 @@ def get_drug_network(
 ) -> dict:
     """Undirected drug co-occurrence network for network graph."""
     return _get_drug_network(DB_PATH, date_filter_id, chart_type, directory, trust)
+
+
+def get_drug_timeline(
+    date_filter_id: str = "all_6mo",
+    chart_type: str = "directory",
+    directory: Optional[str] = None,
+    trust: Optional[str] = None,
+) -> list[dict]:
+    """Drug timeline data (first_seen, last_seen) for Gantt chart."""
+    return _get_drug_timeline(DB_PATH, date_filter_id, chart_type, directory, trust)

src/data_processing/pathway_queries.py

@@ -1373,6 +1373,76 @@ def get_drug_network(
         conn.close()
 
 
+def get_drug_timeline(
+    db_path: Path,
+    date_filter_id: str,
+    chart_type: str,
+    directory: Optional[str] = None,
+    trust: Optional[str] = None,
+) -> list[dict]:
+    """Get drug timeline data for Gantt-style chart.
+
+    Queries level 3 nodes and aggregates across trusts to get the earliest
+    first_seen and latest last_seen per drug × directory.
+
+    Returns list of dicts sorted by directory then first_seen:
+        [{"drug": "ADALIMUMAB", "directory": "RHEUMATOLOGY",
+          "first_seen": "2019-04-04T00:00:00", "last_seen": "2025-12-30T00:00:00",
+          "patients": 2053, "cost_pp_pa": 2170.5}, ...]
+    """
+    conn = sqlite3.connect(str(db_path))
+    conn.row_factory = sqlite3.Row
+    try:
+        conditions = ["date_filter_id = ?", "chart_type = ?", "level = 3"]
+        params: list = [date_filter_id, chart_type]
+
+        if directory:
+            conditions.append("directory = ?")
+            params.append(directory)
+        if trust:
+            conditions.append("trust_name = ?")
+            params.append(trust)
+
+        where = " AND ".join(conditions)
+        query = f"""
+            SELECT
+                labels AS drug,
+                directory,
+                MIN(first_seen) AS first_seen,
+                MAX(last_seen) AS last_seen,
+                SUM(value) AS patients,
+                CASE
+                    WHEN SUM(value) > 0
+                    THEN ROUND(SUM(CAST(cost_pp_pa AS REAL) * value) / SUM(value), 2)
+                    ELSE 0
+                END AS cost_pp_pa
+            FROM pathway_nodes
+            WHERE {where}
+                AND first_seen IS NOT NULL AND first_seen != ''
+                AND last_seen IS NOT NULL AND last_seen != ''
+            GROUP BY labels, directory
+            ORDER BY directory, first_seen
+        """
+
+        rows = conn.execute(query, params).fetchall()
+        return [
+            {
+                "drug": r["drug"],
+                "directory": r["directory"],
+                "first_seen": r["first_seen"],
+                "last_seen": r["last_seen"],
+                "patients": r["patients"] or 0,
+                "cost_pp_pa": r["cost_pp_pa"] or 0,
+            }
+            for r in rows
+            if r["patients"] and r["patients"] > 0
+        ]
+    except sqlite3.Error:
+        return []
+    finally:
+        conn.close()
+
+
 def get_directorate_summary(
     db_path: Path,
     date_filter_id: str,

src/visualization/plotly_generator.py

@@ -2085,3 +2085,124 @@ def create_drug_network_figure(data: dict, title: str = "") -> go.Figure:
     fig.update_layout(**layout)
 
     return fig
+
+
+def create_drug_timeline_figure(data: list[dict], title: str = "") -> go.Figure:
+    """Create a Gantt-style timeline showing when each drug cohort was active.
+
+    Each horizontal bar spans from first_seen to last_seen for a drug,
+    grouped by directory, with color indicating directory and text showing
+    patient count.
+
+    Args:
+        data: List of dicts with keys: drug, directory, first_seen, last_seen,
+              patients, cost_pp_pa.
+        title: Chart title.
+
+    Returns:
+        Plotly Figure with horizontal bars.
+    """
+    if not data:
+        return go.Figure()
+
+    from datetime import datetime
+
+    display_title = title or "Drug Timeline"
+
+    # Parse dates and sort by directory then first_seen
+    for d in data:
+        d["_fs"] = datetime.fromisoformat(d["first_seen"])
+        d["_ls"] = datetime.fromisoformat(d["last_seen"])
+        d["_duration_days"] = max(1, (d["_ls"] - d["_fs"]).days)
+
+    # Sort: by directory alphabetically, then by first_seen ascending
+    data.sort(key=lambda d: (d["directory"], d["_fs"]))
+
+    # Assign colors by directory
+    directories = list(dict.fromkeys(d["directory"] for d in data))
+    dir_colors = {
+        d: DRUG_PALETTE[i % len(DRUG_PALETTE)]
+        for i, d in enumerate(directories)
+    }
+
+    # Build y-axis labels: "Drug (Directory)" for multi-directory views, just "Drug" for single
+    single_directory = len(directories) == 1
+    y_labels = []
+    for d in data:
+        if single_directory:
+            y_labels.append(d["drug"])
+        else:
+            y_labels.append(f"{d['drug']} ({d['directory']})")
+
+    # Build one trace per directory for legend grouping
+    fig = go.Figure()
+    dir_legend_shown = set()
+
+    for i, d in enumerate(data):
+        show_legend = d["directory"] not in dir_legend_shown
+        dir_legend_shown.add(d["directory"])
+
+        duration_ms = d["_duration_days"] * 86_400_000  # days → milliseconds
+        patients = d["patients"]
+        cost = d["cost_pp_pa"]
+
+        fig.add_trace(
+            go.Bar(
+                y=[y_labels[i]],
+                x=[duration_ms],
+                base=[d["_fs"]],
+                orientation="h",
+                marker=dict(
+                    color=dir_colors[d["directory"]],
+                    line=dict(width=0),
+                ),
+                name=d["directory"],
+                legendgroup=d["directory"],
+                showlegend=show_legend,
+                text=f"{patients:,}",
+                textposition="inside",
+                textfont=dict(color="white", size=10),
+                hovertemplate=(
+                    f"<b>{d['drug']}</b><br>"
+                    f"Directory: {d['directory']}<br>"
+                    f"First seen: {d['_fs'].strftime('%b %Y')}<br>"
+                    f"Last seen: {d['_ls'].strftime('%b %Y')}<br>"
+                    f"Duration: {d['_duration_days']:,} days<br>"
+                    f"Patients: {patients:,}<br>"
+                    f"Cost p.a.: £{cost:,.0f}"
+                    "<extra></extra>"
+                ),
+            )
+        )
+
+    # Layout
+    n_bars = len(data)
+    bar_height = 28
+    dynamic_height = max(400, n_bars * bar_height + 120)
+
+    n_dirs = len(directories)
+    legend_margins = _smart_legend_margin(n_dirs)
+    legend = _smart_legend(n_dirs, legend_title="Directory")
+
+    layout = _base_layout(display_title)
+    layout.update(
+        xaxis=dict(
+            type="date",
+            gridcolor=GRID_COLOR,
+            dtick="M6",
+            tickformat="%b\n%Y",
+        ),
+        yaxis=dict(
+            automargin=True,
+            autorange="reversed",
+            tickfont=dict(size=11),
+        ),
+        barmode="overlay",
+        height=dynamic_height,
+        margin=dict(t=60, l=8, **legend_margins),
+        legend=legend,
+        bargap=0.3,
+    )
+    fig.update_layout(**layout)
+
+    return fig