Andrew Charlwood
d98cd4fd69
New query functions in src/data_processing/pathway_queries.py: - get_drug_market_share: Level 3 drug nodes grouped by directory - get_pathway_costs: Level 4+ pathway nodes with cost_pp_pa - get_cost_waterfall: Directorate cost per patient from level 3 aggregation - get_drug_transitions: Sankey source/target drug transitions with ordinal line labels - get_dosing_intervals: Parsed average_spacing by trust/directory - get_drug_directory_matrix: Directory x drug pivot with patient/cost metrics - get_treatment_durations: Weighted avg_days by drug within directorates Thin wrappers added in dash_app/data/queries.py for all 7 functions.
118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
"""
|
||
Thin wrapper around shared pathway query functions.
|
||
|
||
Resolves the database path relative to this file's location and delegates
|
||
to the shared functions in src/data_processing/pathway_queries.py.
|
||
"""
|
||
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
from data_processing.pathway_queries import (
|
||
load_initial_data as _load_initial_data,
|
||
load_pathway_nodes as _load_pathway_nodes,
|
||
get_drug_market_share as _get_drug_market_share,
|
||
get_pathway_costs as _get_pathway_costs,
|
||
get_cost_waterfall as _get_cost_waterfall,
|
||
get_drug_transitions as _get_drug_transitions,
|
||
get_dosing_intervals as _get_dosing_intervals,
|
||
get_drug_directory_matrix as _get_drug_directory_matrix,
|
||
get_treatment_durations as _get_treatment_durations,
|
||
)
|
||
|
||
DB_PATH = Path(__file__).resolve().parents[2] / "data" / "pathways.db"
|
||
|
||
|
||
def load_initial_data() -> dict:
|
||
"""Load reference data (drugs, directorates, indications, refresh info)."""
|
||
return _load_initial_data(DB_PATH)
|
||
|
||
|
||
def load_pathway_data(
|
||
filter_id: str = "all_6mo",
|
||
chart_type: str = "directory",
|
||
selected_drugs: Optional[list[str]] = None,
|
||
selected_directorates: Optional[list[str]] = None,
|
||
selected_trusts: Optional[list[str]] = None,
|
||
) -> dict:
|
||
"""Load pre-computed pathway nodes with optional filters."""
|
||
return _load_pathway_nodes(
|
||
DB_PATH,
|
||
filter_id=filter_id,
|
||
chart_type=chart_type,
|
||
selected_drugs=selected_drugs,
|
||
selected_directorates=selected_directorates,
|
||
selected_trusts=selected_trusts,
|
||
)
|
||
|
||
|
||
# --- Analytics chart query wrappers (Phase 9) ---
|
||
|
||
|
||
def get_drug_market_share(
|
||
date_filter_id: str = "all_6mo",
|
||
chart_type: str = "directory",
|
||
directory: Optional[str] = None,
|
||
trust: Optional[str] = None,
|
||
) -> list[dict]:
|
||
"""Level 3 drug nodes grouped by directory with patient counts."""
|
||
return _get_drug_market_share(DB_PATH, date_filter_id, chart_type, directory, trust)
|
||
|
||
|
||
def get_pathway_costs(
|
||
date_filter_id: str = "all_6mo",
|
||
chart_type: str = "directory",
|
||
directory: Optional[str] = None,
|
||
trust: Optional[str] = None,
|
||
) -> list[dict]:
|
||
"""Level 4+ pathway nodes with annualized cost."""
|
||
return _get_pathway_costs(DB_PATH, date_filter_id, chart_type, directory, trust)
|
||
|
||
|
||
def get_cost_waterfall(
|
||
date_filter_id: str = "all_6mo",
|
||
chart_type: str = "directory",
|
||
trust: Optional[str] = None,
|
||
) -> list[dict]:
|
||
"""Level 2 directorate nodes with cost per patient."""
|
||
return _get_cost_waterfall(DB_PATH, date_filter_id, chart_type, trust)
|
||
|
||
|
||
def get_drug_transitions(
|
||
date_filter_id: str = "all_6mo",
|
||
chart_type: str = "directory",
|
||
directory: Optional[str] = None,
|
||
trust: Optional[str] = None,
|
||
) -> dict:
|
||
"""Drug transition data for Sankey diagram."""
|
||
return _get_drug_transitions(DB_PATH, date_filter_id, chart_type, directory, trust)
|
||
|
||
|
||
def get_dosing_intervals(
|
||
date_filter_id: str = "all_6mo",
|
||
chart_type: str = "directory",
|
||
drug: Optional[str] = None,
|
||
trust: Optional[str] = None,
|
||
) -> list[dict]:
|
||
"""Dosing interval data parsed from average_spacing."""
|
||
return _get_dosing_intervals(DB_PATH, date_filter_id, chart_type, drug, trust)
|
||
|
||
|
||
def get_drug_directory_matrix(
|
||
date_filter_id: str = "all_6mo",
|
||
chart_type: str = "directory",
|
||
trust: Optional[str] = None,
|
||
) -> dict:
|
||
"""Directory × drug matrix with patient counts and costs."""
|
||
return _get_drug_directory_matrix(DB_PATH, date_filter_id, chart_type, trust)
|
||
|
||
|
||
def get_treatment_durations(
|
||
date_filter_id: str = "all_6mo",
|
||
chart_type: str = "directory",
|
||
directory: Optional[str] = None,
|
||
trust: Optional[str] = None,
|
||
) -> list[dict]:
|
||
"""Treatment duration data (avg_days) by drug."""
|
||
return _get_treatment_durations(DB_PATH, date_filter_id, chart_type, directory, trust)
|