Andrew Charlwood
2185 lines
80 KiB
Python
2185 lines
80 KiB
Python
"""
|
|
NHS High-Cost Drug Patient Pathway Analysis Tool - Reflex Application.
|
|
|
|
This is the main Reflex application module containing state management
|
|
and page components for the pathway analysis tool.
|
|
"""
|
|
|
|
import reflex as rx
|
|
from datetime import date, timedelta
|
|
from typing import Optional
|
|
import pandas as pd
|
|
import numpy as np
|
|
from pathlib import Path
|
|
import plotly.graph_objects as go
|
|
import traceback
|
|
import os
|
|
|
|
from rxconfig import config
|
|
from pathways_app.components.layout import main_layout, content_area
|
|
|
|
|
|
# NHS Color constants
|
|
NHS_BLUE = "rgb(0, 94, 184)"
|
|
NHS_DARK_BLUE = "rgb(0, 48, 135)"
|
|
|
|
# Supported file extensions
|
|
SUPPORTED_EXTENSIONS = [".csv", ".parquet", ".pq"]
|
|
|
|
|
|
class State(rx.State):
|
|
"""
|
|
Application state for the NHS High-Cost Drug Patient Pathway Analysis Tool.
|
|
|
|
Manages all filter variables, reference data, and analysis state.
|
|
This corresponds to the AnalysisFilters dataclass in core/models.py
|
|
but is adapted for Reflex's reactive state system.
|
|
"""
|
|
|
|
# Date filter state
|
|
start_date: str = "" # ISO format YYYY-MM-DD
|
|
end_date: str = ""
|
|
last_seen_date: str = ""
|
|
|
|
# Selection filters (list of selected items)
|
|
selected_trusts: list[str] = []
|
|
selected_drugs: list[str] = []
|
|
selected_directories: list[str] = []
|
|
|
|
# Analysis parameters
|
|
minimum_patients: int = 0
|
|
custom_title: str = ""
|
|
|
|
# Reference data (available options loaded from CSV/SQLite)
|
|
available_trusts: list[str] = []
|
|
available_drugs: list[str] = []
|
|
available_directories: list[str] = []
|
|
|
|
# Drug default selections (Include=1 in include.csv)
|
|
default_drugs: list[str] = []
|
|
|
|
# Analysis state
|
|
analysis_running: bool = False
|
|
status_message: str = ""
|
|
error_message: str = ""
|
|
|
|
# Chart state - the Plotly figure
|
|
chart_data: go.Figure = go.Figure()
|
|
has_chart: bool = False
|
|
|
|
# Data source state
|
|
data_file_path: str = ""
|
|
data_source: str = "file" # "file", "sqlite", "snowflake"
|
|
data_loaded: bool = False
|
|
data_row_count: int = 0
|
|
|
|
# Snowflake connection state
|
|
snowflake_available: bool = False
|
|
snowflake_configured: bool = False
|
|
snowflake_connected: bool = False
|
|
|
|
# File upload state
|
|
uploaded_file_name: str = ""
|
|
uploaded_file_size: int = 0 # bytes
|
|
file_upload_error: str = ""
|
|
file_upload_success: bool = False
|
|
file_processing: bool = False
|
|
|
|
# SQLite database state
|
|
sqlite_available: bool = False
|
|
sqlite_row_count: int = 0
|
|
sqlite_patient_count: int = 0
|
|
|
|
# Search/filter state for selection pages
|
|
drug_search: str = ""
|
|
trust_search: str = ""
|
|
directory_search: str = ""
|
|
|
|
# Export state
|
|
last_export_path: str = ""
|
|
export_message: str = ""
|
|
export_error: str = ""
|
|
|
|
# Indication validation state
|
|
indication_validation_enabled: bool = True
|
|
indication_validation_running: bool = False
|
|
indication_validation_results: dict = {} # drug_name -> {total, matched, rate}
|
|
indication_validation_summary: str = ""
|
|
|
|
# Store the underlying data for export
|
|
_analysis_data: pd.DataFrame = pd.DataFrame()
|
|
|
|
def _set_default_dates(self):
|
|
"""Set default date values based on typical analysis period."""
|
|
today = date.today()
|
|
one_year_ago = today - timedelta(days=365)
|
|
|
|
self.start_date = one_year_ago.isoformat()
|
|
self.end_date = today.isoformat()
|
|
self.last_seen_date = one_year_ago.isoformat()
|
|
|
|
def load_reference_data(self):
|
|
"""
|
|
Load reference data from CSV files.
|
|
|
|
This loads the available drugs, trusts, and directories
|
|
that can be selected in the filters.
|
|
"""
|
|
data_dir = Path("data")
|
|
|
|
# Load drugs from include.csv
|
|
try:
|
|
drugs_df = pd.read_csv(data_dir / "include.csv")
|
|
self.available_drugs = sorted(drugs_df.iloc[:, 0].astype(str).tolist())
|
|
# Get default selections (Include=1)
|
|
if "Include" in drugs_df.columns:
|
|
self.default_drugs = drugs_df[drugs_df["Include"] == 1].iloc[:, 0].astype(str).tolist()
|
|
self.selected_drugs = self.default_drugs.copy()
|
|
self.status_message = f"Loaded {len(self.available_drugs)} drugs"
|
|
except Exception as e:
|
|
self.error_message = f"Failed to load drugs: {e}"
|
|
|
|
# Load trusts from defaultTrusts.csv
|
|
try:
|
|
trusts_df = pd.read_csv(data_dir / "defaultTrusts.csv")
|
|
self.available_trusts = sorted(trusts_df.iloc[:, 0].astype(str).tolist())
|
|
# By default, no trusts selected (include all)
|
|
self.selected_trusts = []
|
|
except Exception as e:
|
|
self.error_message = f"Failed to load trusts: {e}"
|
|
|
|
# Load directories from directory_list.csv
|
|
try:
|
|
dirs_df = pd.read_csv(data_dir / "directory_list.csv")
|
|
self.available_directories = sorted(dirs_df.iloc[:, 0].astype(str).tolist())
|
|
# By default, no directories selected (include all)
|
|
self.selected_directories = []
|
|
except Exception as e:
|
|
self.error_message = f"Failed to load directories: {e}"
|
|
|
|
# Set default dates
|
|
self._set_default_dates()
|
|
|
|
# Check Snowflake availability
|
|
try:
|
|
from data_processing.snowflake_connector import is_snowflake_available, is_snowflake_configured
|
|
self.snowflake_available = is_snowflake_available()
|
|
self.snowflake_configured = is_snowflake_configured()
|
|
except ImportError:
|
|
self.snowflake_available = False
|
|
self.snowflake_configured = False
|
|
|
|
# Check SQLite database status
|
|
self.check_sqlite_status()
|
|
|
|
# Auto-select best data source
|
|
if self.sqlite_available and self.sqlite_row_count > 0:
|
|
self.data_source = "sqlite"
|
|
elif self.snowflake_configured:
|
|
self.data_source = "snowflake"
|
|
else:
|
|
self.data_source = "file"
|
|
|
|
# Date setters
|
|
def set_start_date(self, value: str):
|
|
"""Set the start date for analysis."""
|
|
self.start_date = value
|
|
|
|
def set_end_date(self, value: str):
|
|
"""Set the end date for analysis."""
|
|
self.end_date = value
|
|
|
|
def set_last_seen_date(self, value: str):
|
|
"""Set the last seen date filter."""
|
|
self.last_seen_date = value
|
|
|
|
# Selection setters
|
|
def set_selected_trusts(self, trusts: list[str]):
|
|
"""Set the selected NHS trusts."""
|
|
self.selected_trusts = trusts
|
|
|
|
def toggle_trust(self, trust: str):
|
|
"""Toggle a trust selection."""
|
|
if trust in self.selected_trusts:
|
|
self.selected_trusts = [t for t in self.selected_trusts if t != trust]
|
|
else:
|
|
self.selected_trusts = self.selected_trusts + [trust]
|
|
|
|
def select_all_trusts(self):
|
|
"""Select all available trusts."""
|
|
self.selected_trusts = self.available_trusts.copy()
|
|
|
|
def clear_trusts(self):
|
|
"""Clear all trust selections."""
|
|
self.selected_trusts = []
|
|
|
|
def set_selected_drugs(self, drugs: list[str]):
|
|
"""Set the selected drugs."""
|
|
self.selected_drugs = drugs
|
|
|
|
def toggle_drug(self, drug: str):
|
|
"""Toggle a drug selection."""
|
|
if drug in self.selected_drugs:
|
|
self.selected_drugs = [d for d in self.selected_drugs if d != drug]
|
|
else:
|
|
self.selected_drugs = self.selected_drugs + [drug]
|
|
|
|
def select_all_drugs(self):
|
|
"""Select all available drugs."""
|
|
self.selected_drugs = self.available_drugs.copy()
|
|
|
|
def select_default_drugs(self):
|
|
"""Select only the default drugs (Include=1)."""
|
|
self.selected_drugs = self.default_drugs.copy()
|
|
|
|
def clear_drugs(self):
|
|
"""Clear all drug selections."""
|
|
self.selected_drugs = []
|
|
|
|
def set_selected_directories(self, directories: list[str]):
|
|
"""Set the selected directories."""
|
|
self.selected_directories = directories
|
|
|
|
def toggle_directory(self, directory: str):
|
|
"""Toggle a directory selection."""
|
|
if directory in self.selected_directories:
|
|
self.selected_directories = [d for d in self.selected_directories if d != directory]
|
|
else:
|
|
self.selected_directories = self.selected_directories + [directory]
|
|
|
|
def select_all_directories(self):
|
|
"""Select all available directories."""
|
|
self.selected_directories = self.available_directories.copy()
|
|
|
|
def clear_directories(self):
|
|
"""Clear all directory selections."""
|
|
self.selected_directories = []
|
|
|
|
# Analysis parameter setters
|
|
def set_minimum_patients(self, value: int):
|
|
"""Set the minimum patients threshold."""
|
|
self.minimum_patients = max(0, value)
|
|
|
|
def set_minimum_patients_from_input(self, value: str):
|
|
"""Set minimum patients threshold from string input."""
|
|
try:
|
|
self.minimum_patients = max(0, int(value)) if value else 0
|
|
except ValueError:
|
|
pass # Ignore invalid input
|
|
|
|
def set_minimum_patients_from_slider(self, values: list[float]):
|
|
"""Set minimum patients threshold from slider value (list)."""
|
|
if values:
|
|
self.minimum_patients = max(0, int(values[0]))
|
|
|
|
def set_custom_title(self, value: str):
|
|
"""Set a custom title for the analysis."""
|
|
self.custom_title = value
|
|
|
|
# Data source methods
|
|
def set_data_file_path(self, path: str):
|
|
"""Set the data file path for analysis."""
|
|
self.data_file_path = path
|
|
|
|
def set_data_source(self, source: str):
|
|
"""Set the data source type (file, sqlite, snowflake)."""
|
|
if source in ("file", "sqlite", "snowflake"):
|
|
self.data_source = source
|
|
|
|
# Status methods
|
|
def set_status(self, message: str):
|
|
"""Update the status message."""
|
|
self.status_message = message
|
|
|
|
def set_error(self, message: str):
|
|
"""Set an error message."""
|
|
self.error_message = message
|
|
|
|
def clear_error(self):
|
|
"""Clear the error message."""
|
|
self.error_message = ""
|
|
|
|
# File handling methods
|
|
async def handle_file_upload(self, files: list[rx.UploadFile]):
|
|
"""
|
|
Handle file upload for CSV/Parquet data files.
|
|
|
|
This accepts uploaded files and processes them for analysis.
|
|
"""
|
|
self.file_upload_error = ""
|
|
self.file_upload_success = False
|
|
|
|
if not files:
|
|
self.file_upload_error = "No file selected"
|
|
return
|
|
|
|
file = files[0] # Take first file only
|
|
file_name = file.filename
|
|
file_ext = Path(file_name).suffix.lower()
|
|
|
|
# Validate file extension
|
|
if file_ext not in SUPPORTED_EXTENSIONS:
|
|
self.file_upload_error = f"Unsupported file type: {file_ext}. Please upload CSV or Parquet files."
|
|
return
|
|
|
|
self.file_processing = True
|
|
self.status_message = f"Processing {file_name}..."
|
|
yield # Update UI
|
|
|
|
try:
|
|
# Read file content
|
|
file_content = await file.read()
|
|
file_size = len(file_content)
|
|
self.uploaded_file_size = file_size
|
|
|
|
# Save to uploads directory
|
|
upload_dir = Path("data/uploads")
|
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
upload_path = upload_dir / file_name
|
|
with open(upload_path, "wb") as f:
|
|
f.write(file_content)
|
|
|
|
self.uploaded_file_name = file_name
|
|
self.data_file_path = str(upload_path)
|
|
self.data_source = "file"
|
|
self.file_upload_success = True
|
|
|
|
# Format file size for display
|
|
if file_size < 1024:
|
|
size_str = f"{file_size} bytes"
|
|
elif file_size < 1024 * 1024:
|
|
size_str = f"{file_size / 1024:.1f} KB"
|
|
else:
|
|
size_str = f"{file_size / (1024 * 1024):.1f} MB"
|
|
|
|
self.status_message = f"Uploaded {file_name} ({size_str})"
|
|
|
|
except Exception as e:
|
|
self.file_upload_error = f"Upload failed: {str(e)}"
|
|
self.file_upload_success = False
|
|
|
|
finally:
|
|
self.file_processing = False
|
|
|
|
def clear_uploaded_file(self):
|
|
"""Clear the uploaded file and reset file state."""
|
|
self.uploaded_file_name = ""
|
|
self.uploaded_file_size = 0
|
|
self.data_file_path = ""
|
|
self.file_upload_success = False
|
|
self.file_upload_error = ""
|
|
self.status_message = "File cleared"
|
|
|
|
def check_sqlite_status(self):
|
|
"""Check if SQLite database is available and get statistics."""
|
|
try:
|
|
from data_processing.database import default_db_manager
|
|
from data_processing.patient_data import get_patient_data_stats
|
|
|
|
if default_db_manager.exists:
|
|
stats = get_patient_data_stats(default_db_manager)
|
|
self.sqlite_available = stats.get("total_rows", 0) > 0
|
|
self.sqlite_row_count = stats.get("total_rows", 0)
|
|
self.sqlite_patient_count = stats.get("unique_patients", 0)
|
|
|
|
if self.sqlite_available:
|
|
self.status_message = f"SQLite database: {self.sqlite_row_count:,} rows, {self.sqlite_patient_count:,} patients"
|
|
else:
|
|
self.status_message = "SQLite database exists but has no data"
|
|
else:
|
|
self.sqlite_available = False
|
|
self.sqlite_row_count = 0
|
|
self.sqlite_patient_count = 0
|
|
self.status_message = "SQLite database not found"
|
|
except ImportError:
|
|
self.sqlite_available = False
|
|
self.status_message = "Data processing module not available"
|
|
except Exception as e:
|
|
self.sqlite_available = False
|
|
self.status_message = f"Error checking SQLite: {str(e)}"
|
|
|
|
def use_sqlite_source(self):
|
|
"""Set data source to SQLite database."""
|
|
self.data_source = "sqlite"
|
|
self.data_file_path = ""
|
|
self.status_message = "Using SQLite database as data source"
|
|
|
|
def use_file_source(self):
|
|
"""Set data source to uploaded file."""
|
|
if self.uploaded_file_name:
|
|
self.data_source = "file"
|
|
self.status_message = f"Using uploaded file: {self.uploaded_file_name}"
|
|
else:
|
|
self.status_message = "No file uploaded. Please upload a file first."
|
|
|
|
def use_snowflake_source(self):
|
|
"""Set data source to Snowflake (if available)."""
|
|
if self.snowflake_configured:
|
|
self.data_source = "snowflake"
|
|
self.status_message = "Using Snowflake as data source"
|
|
else:
|
|
self.status_message = "Snowflake is not configured. Check config/snowflake.toml"
|
|
|
|
@rx.var
|
|
def data_source_display(self) -> str:
|
|
"""Human-readable data source description."""
|
|
if self.data_source == "file":
|
|
if self.uploaded_file_name:
|
|
return f"File: {self.uploaded_file_name}"
|
|
return "File: No file selected"
|
|
elif self.data_source == "sqlite":
|
|
if self.sqlite_available:
|
|
return f"SQLite: {self.sqlite_row_count:,} rows"
|
|
return "SQLite: Not available"
|
|
elif self.data_source == "snowflake":
|
|
if self.snowflake_configured:
|
|
return "Snowflake: Ready"
|
|
return "Snowflake: Not configured"
|
|
return "Unknown"
|
|
|
|
@rx.var
|
|
def file_size_display(self) -> str:
|
|
"""Human-readable file size."""
|
|
if self.uploaded_file_size == 0:
|
|
return ""
|
|
if self.uploaded_file_size < 1024:
|
|
return f"{self.uploaded_file_size} bytes"
|
|
elif self.uploaded_file_size < 1024 * 1024:
|
|
return f"{self.uploaded_file_size / 1024:.1f} KB"
|
|
else:
|
|
return f"{self.uploaded_file_size / (1024 * 1024):.1f} MB"
|
|
|
|
# Validation
|
|
def validate_filters(self) -> list[str]:
|
|
"""
|
|
Validate the current filter configuration.
|
|
|
|
Returns a list of error messages (empty if valid).
|
|
"""
|
|
errors = []
|
|
|
|
# Check dates are set
|
|
if not self.start_date:
|
|
errors.append("Start date is required")
|
|
if not self.end_date:
|
|
errors.append("End date is required")
|
|
if not self.last_seen_date:
|
|
errors.append("Last seen date is required")
|
|
|
|
# Check date order
|
|
if self.start_date and self.end_date:
|
|
if self.end_date < self.start_date:
|
|
errors.append("End date cannot be before start date")
|
|
|
|
if self.last_seen_date and self.end_date:
|
|
if self.last_seen_date > self.end_date:
|
|
errors.append("Last seen date is after end date (would exclude all patients)")
|
|
|
|
# Check minimum patients
|
|
if self.minimum_patients < 0:
|
|
errors.append("Minimum patients cannot be negative")
|
|
|
|
# Check at least some drugs are selected (warning, not error)
|
|
# Empty selection means "include all"
|
|
|
|
return errors
|
|
|
|
@rx.var
|
|
def filter_summary(self) -> str:
|
|
"""Generate a summary of current filter settings."""
|
|
lines = []
|
|
|
|
if self.start_date and self.end_date:
|
|
lines.append(f"Date range: {self.start_date} to {self.end_date}")
|
|
if self.last_seen_date:
|
|
lines.append(f"Last seen after: {self.last_seen_date}")
|
|
lines.append(f"Minimum patients: {self.minimum_patients}")
|
|
|
|
if self.selected_trusts:
|
|
lines.append(f"Trusts: {len(self.selected_trusts)} selected")
|
|
else:
|
|
lines.append("Trusts: All")
|
|
|
|
if self.selected_drugs:
|
|
lines.append(f"Drugs: {len(self.selected_drugs)} selected")
|
|
else:
|
|
lines.append("Drugs: All")
|
|
|
|
if self.selected_directories:
|
|
lines.append(f"Directories: {len(self.selected_directories)} selected")
|
|
else:
|
|
lines.append("Directories: All")
|
|
|
|
return "\n".join(lines)
|
|
|
|
@rx.var
|
|
def display_title(self) -> str:
|
|
"""Generate the display title for the analysis."""
|
|
if self.custom_title:
|
|
return self.custom_title
|
|
if self.start_date and self.end_date:
|
|
return f"Patients initiated from {self.start_date} to {self.end_date}"
|
|
return "Patient Pathway Analysis"
|
|
|
|
@rx.var
|
|
def drug_selection_count(self) -> str:
|
|
"""Display count of selected drugs."""
|
|
return f"{len(self.selected_drugs)} of {len(self.available_drugs)} drugs selected"
|
|
|
|
@rx.var
|
|
def trust_selection_count(self) -> str:
|
|
"""Display count of selected trusts."""
|
|
if not self.selected_trusts:
|
|
return f"All {len(self.available_trusts)} trusts (none selected)"
|
|
return f"{len(self.selected_trusts)} of {len(self.available_trusts)} trusts selected"
|
|
|
|
@rx.var
|
|
def directory_selection_count(self) -> str:
|
|
"""Display count of selected directories."""
|
|
if not self.selected_directories:
|
|
return f"All {len(self.available_directories)} directories (none selected)"
|
|
return f"{len(self.selected_directories)} of {len(self.available_directories)} directories selected"
|
|
|
|
# Search setters
|
|
def set_drug_search(self, value: str):
|
|
"""Set the drug search filter text."""
|
|
self.drug_search = value
|
|
|
|
def set_trust_search(self, value: str):
|
|
"""Set the trust search filter text."""
|
|
self.trust_search = value
|
|
|
|
def set_directory_search(self, value: str):
|
|
"""Set the directory search filter text."""
|
|
self.directory_search = value
|
|
|
|
def clear_drug_search(self):
|
|
"""Clear the drug search filter."""
|
|
self.drug_search = ""
|
|
|
|
def clear_trust_search(self):
|
|
"""Clear the trust search filter."""
|
|
self.trust_search = ""
|
|
|
|
def clear_directory_search(self):
|
|
"""Clear the directory search filter."""
|
|
self.directory_search = ""
|
|
|
|
@rx.var
|
|
def filtered_drugs(self) -> list[str]:
|
|
"""Get the list of drugs filtered by search text."""
|
|
if not self.drug_search:
|
|
return self.available_drugs
|
|
search_lower = self.drug_search.lower()
|
|
return [d for d in self.available_drugs if search_lower in d.lower()]
|
|
|
|
@rx.var
|
|
def filtered_trusts(self) -> list[str]:
|
|
"""Get the list of trusts filtered by search text."""
|
|
if not self.trust_search:
|
|
return self.available_trusts
|
|
search_lower = self.trust_search.lower()
|
|
return [t for t in self.available_trusts if search_lower in t.lower()]
|
|
|
|
@rx.var
|
|
def filtered_directories(self) -> list[str]:
|
|
"""Get the list of directories filtered by search text."""
|
|
if not self.directory_search:
|
|
return self.available_directories
|
|
search_lower = self.directory_search.lower()
|
|
return [d for d in self.available_directories if search_lower in d.lower()]
|
|
|
|
@rx.var
|
|
def drug_search_result_count(self) -> str:
|
|
"""Display count of drugs matching search."""
|
|
total = len(self.available_drugs)
|
|
filtered = len(self.filtered_drugs)
|
|
if not self.drug_search:
|
|
return f"{total} drugs"
|
|
return f"Showing {filtered} of {total} drugs"
|
|
|
|
@rx.var
|
|
def trust_search_result_count(self) -> str:
|
|
"""Display count of trusts matching search."""
|
|
total = len(self.available_trusts)
|
|
filtered = len(self.filtered_trusts)
|
|
if not self.trust_search:
|
|
return f"{total} trusts"
|
|
return f"Showing {filtered} of {total} trusts"
|
|
|
|
@rx.var
|
|
def directory_search_result_count(self) -> str:
|
|
"""Display count of directories matching search."""
|
|
total = len(self.available_directories)
|
|
filtered = len(self.filtered_directories)
|
|
if not self.directory_search:
|
|
return f"{total} directories"
|
|
return f"Showing {filtered} of {total} directories"
|
|
|
|
# Analysis methods
|
|
def run_analysis(self):
|
|
"""
|
|
Run the patient pathway analysis with current filter settings.
|
|
|
|
This is an async generator that yields state updates for progress indication.
|
|
Uses the existing analysis pipeline from tools/dashboard_gui.py.
|
|
"""
|
|
# Validate filters first
|
|
errors = self.validate_filters()
|
|
if errors:
|
|
self.error_message = "Validation errors:\n" + "\n".join(errors)
|
|
return
|
|
|
|
self.analysis_running = True
|
|
self.error_message = ""
|
|
self.status_message = "Starting analysis..."
|
|
self.has_chart = False
|
|
yield # Update UI to show running state
|
|
|
|
try:
|
|
# Import analysis modules
|
|
from core import AnalysisFilters, PathConfig, default_paths
|
|
from data_processing.data_source import get_data
|
|
from tools.dashboard_gui import generate_graph
|
|
|
|
# Get the data using fallback chain (cache -> Snowflake -> SQLite -> file)
|
|
self.status_message = "Loading patient data..."
|
|
yield
|
|
|
|
# Build filter parameters
|
|
trusts = self.selected_trusts if self.selected_trusts else self.available_trusts
|
|
drugs = self.selected_drugs if self.selected_drugs else self.available_drugs
|
|
directories = self.selected_directories if self.selected_directories else self.available_directories
|
|
|
|
# Get data from the data source manager
|
|
result = get_data(
|
|
start_date=self.start_date,
|
|
end_date=self.end_date,
|
|
trusts=trusts,
|
|
drugs=drugs,
|
|
directories=directories,
|
|
)
|
|
|
|
if result.df is None or len(result.df) == 0:
|
|
self.error_message = "No data available. Please check your data source configuration."
|
|
self.analysis_running = False
|
|
return
|
|
|
|
self.data_source = result.source_type.value
|
|
self.data_row_count = len(result.df)
|
|
self.status_message = f"Loaded {self.data_row_count:,} rows from {self.data_source}"
|
|
yield
|
|
|
|
# Create AnalysisFilters object for generate_graph
|
|
self.status_message = "Processing pathways..."
|
|
yield
|
|
|
|
# Generate the chart data (without writing to file)
|
|
# We'll create the figure data directly instead of calling generate_graph
|
|
# which writes to file and opens browser
|
|
fig_data = self._generate_chart_data(
|
|
df=result.df,
|
|
trusts=trusts,
|
|
drugs=drugs,
|
|
directories=directories,
|
|
)
|
|
|
|
if fig_data is not None:
|
|
self.chart_data = fig_data
|
|
self.has_chart = True
|
|
self.status_message = f"Analysis complete! Showing {self.data_row_count:,} interventions."
|
|
else:
|
|
self.error_message = "No data found matching the selected filters."
|
|
self.has_chart = False
|
|
|
|
except Exception as e:
|
|
self.error_message = f"Analysis failed: {str(e)}\n\n{traceback.format_exc()}"
|
|
self.has_chart = False
|
|
|
|
finally:
|
|
self.analysis_running = False
|
|
|
|
yield # Final UI update
|
|
|
|
def _generate_chart_data(
|
|
self,
|
|
df: pd.DataFrame,
|
|
trusts: list[str],
|
|
drugs: list[str],
|
|
directories: list[str],
|
|
) -> Optional[go.Figure]:
|
|
"""
|
|
Generate Plotly chart data from processed DataFrame.
|
|
|
|
This replicates the core logic of generate_graph() and figure() but
|
|
returns the figure dict instead of writing to file and opening browser.
|
|
This is a workaround to avoid modifying generate_graph() internals
|
|
(which is deferred to Phase 5).
|
|
"""
|
|
from core import default_paths
|
|
|
|
# Use the org_codes mapping
|
|
org_codes = pd.read_csv(default_paths.org_codes_csv, index_col=1)
|
|
|
|
# Make a copy to avoid modifying original
|
|
df1 = df.copy()
|
|
|
|
# Create UPID + Treatment column for deduplication
|
|
df1["UPIDTreatment"] = df1["UPID"] + df1["Drug Name"]
|
|
|
|
# Map provider codes to names
|
|
df1["Provider Code"] = df1["Provider Code"].map(org_codes["Name"])
|
|
|
|
# Apply filters
|
|
df1 = df1[
|
|
(df1["Provider Code"].isin(trusts)) &
|
|
(df1["Drug Name"].isin(drugs)) &
|
|
(df1["Directory"].isin(directories))
|
|
]
|
|
|
|
if len(df1) == 0:
|
|
return None
|
|
|
|
# Apply date filters
|
|
df1 = df1[
|
|
(df1["Intervention Date"] >= self.start_date) &
|
|
(df1["Intervention Date"] <= self.end_date)
|
|
]
|
|
|
|
if len(df1) == 0:
|
|
return None
|
|
|
|
# Add indication validation columns (if enabled and Snowflake available)
|
|
df1 = self._add_indication_validation(df1)
|
|
|
|
# Store filtered data for CSV export (now includes indication columns)
|
|
self._analysis_data = df1.copy()
|
|
|
|
# Build a simplified hierarchy for the icicle chart
|
|
# Group by Trust -> Directory -> Drug to get patient counts
|
|
hierarchy_data = self._build_hierarchy(df1, org_codes)
|
|
|
|
if hierarchy_data.empty:
|
|
return None
|
|
|
|
# Apply minimum patients filter
|
|
hierarchy_data = hierarchy_data[hierarchy_data['value'] >= self.minimum_patients]
|
|
|
|
if hierarchy_data.empty:
|
|
return None
|
|
|
|
# Create the Plotly icicle figure
|
|
fig = go.Figure(go.Icicle(
|
|
labels=hierarchy_data['labels'].tolist(),
|
|
ids=hierarchy_data['ids'].tolist(),
|
|
parents=hierarchy_data['parents'].tolist(),
|
|
values=hierarchy_data['value'].tolist(),
|
|
branchvalues="total",
|
|
marker=dict(
|
|
colors=hierarchy_data['colour'].tolist() if 'colour' in hierarchy_data.columns else None,
|
|
colorscale='Viridis',
|
|
),
|
|
maxdepth=3,
|
|
texttemplate='<b>%{label}</b><br>Patients: %{value}',
|
|
hovertemplate='<b>%{label}</b><br>Patients: %{value}<extra></extra>',
|
|
))
|
|
|
|
# Set chart title
|
|
title_text = self.custom_title if self.custom_title else f"Patients initiated {self.start_date} to {self.end_date}"
|
|
|
|
fig.update_layout(
|
|
margin=dict(t=60, l=1, r=1, b=60),
|
|
title=f"Norfolk & Waveney ICS High-Cost Drug Patient Pathways - {title_text}",
|
|
title_x=0.5,
|
|
hoverlabel=dict(font_size=16),
|
|
)
|
|
|
|
# Return figure for rx.plotly()
|
|
return fig
|
|
|
|
def _build_hierarchy(self, df: pd.DataFrame, org_codes: pd.DataFrame) -> pd.DataFrame:
|
|
"""
|
|
Build a hierarchical dataframe for icicle chart.
|
|
|
|
Creates Trust -> Directory -> Drug hierarchy with patient counts.
|
|
"""
|
|
# Create directory mapping from UPID
|
|
directory_df = df[["UPID", "Directory"]].drop_duplicates("UPID").set_index("UPID")
|
|
|
|
# Get unique patients per drug
|
|
patient_drugs = df[["UPID", "Drug Name", "Provider Code", "Directory"]].drop_duplicates(subset=["UPID", "Drug Name"])
|
|
|
|
# Build hierarchy: Trust -> Directory -> Drug
|
|
rows = []
|
|
|
|
# Root node
|
|
total_patients = patient_drugs["UPID"].nunique()
|
|
rows.append({
|
|
'parents': '',
|
|
'ids': 'N&WICS',
|
|
'labels': 'N&WICS',
|
|
'value': total_patients,
|
|
'colour': 1.0,
|
|
})
|
|
|
|
# Trust level
|
|
trust_counts = patient_drugs.groupby("Provider Code")["UPID"].nunique().reset_index()
|
|
trust_counts.columns = ["trust", "count"]
|
|
|
|
for _, row in trust_counts.iterrows():
|
|
trust = row["trust"]
|
|
if pd.isna(trust):
|
|
continue
|
|
rows.append({
|
|
'parents': 'N&WICS',
|
|
'ids': f'N&WICS - {trust}',
|
|
'labels': trust,
|
|
'value': row["count"],
|
|
'colour': row["count"] / total_patients,
|
|
})
|
|
|
|
# Directory level (under each trust)
|
|
trust_dir_counts = patient_drugs.groupby(["Provider Code", "Directory"])["UPID"].nunique().reset_index()
|
|
trust_dir_counts.columns = ["trust", "directory", "count"]
|
|
|
|
for _, row in trust_dir_counts.iterrows():
|
|
trust = row["trust"]
|
|
directory = row["directory"]
|
|
if pd.isna(trust) or pd.isna(directory):
|
|
continue
|
|
trust_total = trust_counts[trust_counts["trust"] == trust]["count"].values
|
|
trust_total = trust_total[0] if len(trust_total) > 0 else 1
|
|
rows.append({
|
|
'parents': f'N&WICS - {trust}',
|
|
'ids': f'N&WICS - {trust} - {directory}',
|
|
'labels': directory,
|
|
'value': row["count"],
|
|
'colour': row["count"] / trust_total,
|
|
})
|
|
|
|
# Drug level (under each trust-directory)
|
|
trust_dir_drug_counts = patient_drugs.groupby(["Provider Code", "Directory", "Drug Name"])["UPID"].nunique().reset_index()
|
|
trust_dir_drug_counts.columns = ["trust", "directory", "drug", "count"]
|
|
|
|
for _, row in trust_dir_drug_counts.iterrows():
|
|
trust = row["trust"]
|
|
directory = row["directory"]
|
|
drug = row["drug"]
|
|
if pd.isna(trust) or pd.isna(directory) or pd.isna(drug):
|
|
continue
|
|
dir_total = trust_dir_counts[
|
|
(trust_dir_counts["trust"] == trust) &
|
|
(trust_dir_counts["directory"] == directory)
|
|
]["count"].values
|
|
dir_total = dir_total[0] if len(dir_total) > 0 else 1
|
|
rows.append({
|
|
'parents': f'N&WICS - {trust} - {directory}',
|
|
'ids': f'N&WICS - {trust} - {directory} - {drug}',
|
|
'labels': drug,
|
|
'value': row["count"],
|
|
'colour': row["count"] / dir_total,
|
|
})
|
|
|
|
return pd.DataFrame(rows)
|
|
|
|
def _add_indication_validation(self, df: pd.DataFrame) -> pd.DataFrame:
|
|
"""
|
|
Add indication validation columns to the DataFrame.
|
|
|
|
Adds columns:
|
|
- Indication_Valid: Boolean indicating if patient has valid GP diagnosis
|
|
- Indication_Source: "GP_SNOMED" | "NONE" | "NOT_CHECKED"
|
|
- Indication_Cluster: The matched SNOMED cluster ID (if any)
|
|
|
|
This requires Snowflake connectivity for GP record lookups.
|
|
If Snowflake is not available, columns are added with "NOT_CHECKED" status.
|
|
"""
|
|
# Initialize columns with default values
|
|
df = df.copy()
|
|
df["Indication_Valid"] = False
|
|
df["Indication_Source"] = "NOT_CHECKED"
|
|
df["Indication_Cluster"] = ""
|
|
|
|
# Check if indication validation is enabled and Snowflake is available
|
|
if not self.indication_validation_enabled:
|
|
return df
|
|
|
|
try:
|
|
from data_processing.snowflake_connector import (
|
|
is_snowflake_available,
|
|
is_snowflake_configured,
|
|
get_connector,
|
|
)
|
|
from data_processing.diagnosis_lookup import (
|
|
get_drug_cluster_ids,
|
|
patient_has_indication,
|
|
)
|
|
|
|
if not is_snowflake_available() or not is_snowflake_configured():
|
|
# Snowflake not available - can't validate indications
|
|
self.indication_validation_summary = "Indication validation skipped (Snowflake not configured)"
|
|
return df
|
|
|
|
self.indication_validation_running = True
|
|
|
|
# Get unique patient-drug pairs
|
|
patient_drug_pairs = df[["UPID", "Drug Name"]].drop_duplicates()
|
|
total_pairs = len(patient_drug_pairs)
|
|
|
|
# Cache drug clusters to avoid repeated lookups
|
|
drug_clusters_cache = {}
|
|
|
|
# Track results for summary
|
|
validation_results = {} # drug -> {total, matched}
|
|
connector = get_connector()
|
|
|
|
for idx, (_, row) in enumerate(patient_drug_pairs.iterrows()):
|
|
upid = row["UPID"]
|
|
drug_name = row["Drug Name"]
|
|
|
|
# Get drug clusters (cached)
|
|
drug_upper = drug_name.upper() if drug_name else ""
|
|
if drug_upper not in drug_clusters_cache:
|
|
drug_clusters_cache[drug_upper] = get_drug_cluster_ids(drug_name)
|
|
|
|
cluster_ids = drug_clusters_cache[drug_upper]
|
|
|
|
# Initialize drug in results tracking
|
|
if drug_upper not in validation_results:
|
|
validation_results[drug_upper] = {"total": 0, "matched": 0, "name": drug_name}
|
|
|
|
validation_results[drug_upper]["total"] += 1
|
|
|
|
if not cluster_ids:
|
|
# No cluster mapping for this drug - mark as NONE
|
|
mask = (df["UPID"] == upid) & (df["Drug Name"] == drug_name)
|
|
df.loc[mask, "Indication_Source"] = "NONE"
|
|
continue
|
|
|
|
# Check patient indication in GP records
|
|
# Note: We use the UPID as patient identifier - this may need mapping to pseudonymised NHS number
|
|
# For now, assume UPID can be used directly or is already the pseudonymised ID
|
|
has_indication, matched_cluster, _, _ = patient_has_indication(
|
|
patient_pseudonym=upid,
|
|
cluster_ids=cluster_ids,
|
|
connector=connector,
|
|
)
|
|
|
|
# Update dataframe for this patient-drug combination
|
|
mask = (df["UPID"] == upid) & (df["Drug Name"] == drug_name)
|
|
df.loc[mask, "Indication_Valid"] = has_indication
|
|
df.loc[mask, "Indication_Source"] = "GP_SNOMED" if has_indication else "NONE"
|
|
if matched_cluster:
|
|
df.loc[mask, "Indication_Cluster"] = matched_cluster
|
|
|
|
if has_indication:
|
|
validation_results[drug_upper]["matched"] += 1
|
|
|
|
# Store validation results and create summary
|
|
self.indication_validation_results = {
|
|
drug: {
|
|
"drug_name": data["name"],
|
|
"total_patients": data["total"],
|
|
"patients_with_indication": data["matched"],
|
|
"match_rate": round(data["matched"] / data["total"] * 100, 1) if data["total"] > 0 else 0,
|
|
}
|
|
for drug, data in validation_results.items()
|
|
}
|
|
|
|
# Create summary text
|
|
total_patients = sum(d["total"] for d in validation_results.values())
|
|
matched_patients = sum(d["matched"] for d in validation_results.values())
|
|
overall_rate = round(matched_patients / total_patients * 100, 1) if total_patients > 0 else 0
|
|
|
|
self.indication_validation_summary = (
|
|
f"GP Indication Validation: {matched_patients}/{total_patients} "
|
|
f"({overall_rate}%) patients have valid GP diagnosis"
|
|
)
|
|
|
|
except Exception as e:
|
|
self.indication_validation_summary = f"Indication validation error: {str(e)}"
|
|
# Don't fail the whole analysis - just leave columns as NOT_CHECKED
|
|
|
|
finally:
|
|
self.indication_validation_running = False
|
|
|
|
return df
|
|
|
|
def toggle_indication_validation(self):
|
|
"""Toggle indication validation on/off."""
|
|
self.indication_validation_enabled = not self.indication_validation_enabled
|
|
|
|
@rx.var
|
|
def indication_validation_status(self) -> str:
|
|
"""Get human-readable indication validation status."""
|
|
if self.indication_validation_running:
|
|
return "Validating patient indications..."
|
|
if self.indication_validation_summary:
|
|
return self.indication_validation_summary
|
|
if self.indication_validation_enabled:
|
|
return "Enabled (will check GP records)"
|
|
return "Disabled"
|
|
|
|
@rx.var
|
|
def indication_results_list(self) -> list[dict]:
|
|
"""
|
|
Get indication validation results as a list for display.
|
|
|
|
Returns list of dicts with: drug_name, total_patients, patients_with_indication, match_rate
|
|
Sorted by match rate ascending (worst first) for easy identification of issues.
|
|
"""
|
|
if not self.indication_validation_results:
|
|
return []
|
|
|
|
results = []
|
|
for drug_key, data in self.indication_validation_results.items():
|
|
results.append({
|
|
"drug_name": data.get("drug_name", drug_key),
|
|
"total_patients": data.get("total_patients", 0),
|
|
"patients_with_indication": data.get("patients_with_indication", 0),
|
|
"match_rate": data.get("match_rate", 0),
|
|
})
|
|
|
|
# Sort by match rate ascending (lowest first to highlight issues)
|
|
results.sort(key=lambda x: x["match_rate"])
|
|
return results
|
|
|
|
@rx.var
|
|
def has_indication_results(self) -> bool:
|
|
"""Check if there are indication validation results to display."""
|
|
return len(self.indication_validation_results) > 0
|
|
|
|
def export_chart_html(self):
|
|
"""
|
|
Export the current chart as an interactive HTML file.
|
|
|
|
The file is saved to data/exports/ directory with a timestamped filename.
|
|
"""
|
|
if not self.has_chart:
|
|
self.export_error = "No chart to export. Please run analysis first."
|
|
return
|
|
|
|
self.export_error = ""
|
|
self.export_message = ""
|
|
|
|
try:
|
|
from datetime import datetime
|
|
|
|
# Create exports directory
|
|
export_dir = Path("data/exports")
|
|
export_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Generate filename with timestamp
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
filename = f"pathway_chart_{timestamp}.html"
|
|
filepath = export_dir / filename
|
|
|
|
# Export the chart to HTML
|
|
self.chart_data.write_html(
|
|
str(filepath),
|
|
include_plotlyjs=True,
|
|
full_html=True,
|
|
)
|
|
|
|
self.last_export_path = str(filepath)
|
|
self.export_message = f"Chart exported to {filename}"
|
|
|
|
except Exception as e:
|
|
self.export_error = f"Export failed: {str(e)}"
|
|
|
|
def export_data_csv(self):
|
|
"""
|
|
Export the underlying analysis data as a CSV file.
|
|
|
|
The file is saved to data/exports/ directory with a timestamped filename.
|
|
"""
|
|
if self._analysis_data is None or len(self._analysis_data) == 0:
|
|
self.export_error = "No data to export. Please run analysis first."
|
|
return
|
|
|
|
self.export_error = ""
|
|
self.export_message = ""
|
|
|
|
try:
|
|
from datetime import datetime
|
|
|
|
# Create exports directory
|
|
export_dir = Path("data/exports")
|
|
export_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Generate filename with timestamp
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
filename = f"pathway_data_{timestamp}.csv"
|
|
filepath = export_dir / filename
|
|
|
|
# Export the data to CSV
|
|
self._analysis_data.to_csv(filepath, index=False)
|
|
|
|
self.last_export_path = str(filepath)
|
|
self.export_message = f"Data exported to {filename}"
|
|
|
|
except Exception as e:
|
|
self.export_error = f"Export failed: {str(e)}"
|
|
|
|
def clear_export_messages(self):
|
|
"""Clear export status messages."""
|
|
self.export_message = ""
|
|
self.export_error = ""
|
|
|
|
|
|
# =============================================================================
|
|
# Page Components
|
|
# =============================================================================
|
|
|
|
def info_card(title: str, value: str, icon: str) -> rx.Component:
|
|
"""Create an info card showing a statistic."""
|
|
return rx.box(
|
|
rx.vstack(
|
|
rx.hstack(
|
|
rx.icon(icon, size=20, color=NHS_BLUE),
|
|
rx.text(title, size="2", color="gray"),
|
|
spacing="2",
|
|
align="center",
|
|
),
|
|
rx.text(value, size="5", weight="bold"),
|
|
spacing="1",
|
|
align="start",
|
|
),
|
|
padding="16px",
|
|
background="white",
|
|
border_radius="8px",
|
|
border="1px solid rgb(229, 231, 235)",
|
|
width="100%",
|
|
)
|
|
|
|
|
|
def date_input(label: str, value: rx.Var, on_change, help_text: str = "", input_id: str = "") -> rx.Component:
|
|
"""Create a labeled date input component with accessibility support."""
|
|
# Generate a unique ID if not provided
|
|
label_id = f"{input_id}-label" if input_id else ""
|
|
help_id = f"{input_id}-help" if input_id else ""
|
|
|
|
return rx.vstack(
|
|
rx.el.label(
|
|
label,
|
|
html_for=input_id,
|
|
font_size="14px",
|
|
font_weight="500",
|
|
color=NHS_DARK_BLUE,
|
|
),
|
|
rx.input(
|
|
type="date",
|
|
value=value,
|
|
on_change=on_change,
|
|
width="100%",
|
|
id=input_id,
|
|
aria_describedby=help_id if help_text else "",
|
|
),
|
|
rx.cond(
|
|
help_text != "",
|
|
rx.text(help_text, size="1", color="gray", id=help_id),
|
|
),
|
|
spacing="1",
|
|
align="start",
|
|
width="100%",
|
|
)
|
|
|
|
|
|
def data_source_selector() -> rx.Component:
|
|
"""Data source selector with file upload, SQLite, and Snowflake options."""
|
|
return rx.box(
|
|
rx.vstack(
|
|
rx.heading("Data Source", size="5", color=NHS_DARK_BLUE),
|
|
rx.text(
|
|
"Select where to load patient data from",
|
|
size="2",
|
|
color="gray",
|
|
),
|
|
rx.divider(margin_y="8px"),
|
|
# Current data source display
|
|
rx.hstack(
|
|
rx.text("Current source:", weight="medium"),
|
|
rx.badge(
|
|
State.data_source_display,
|
|
color_scheme=rx.cond(
|
|
State.data_source == "sqlite",
|
|
"green",
|
|
rx.cond(
|
|
State.data_source == "snowflake",
|
|
"blue",
|
|
"gray",
|
|
),
|
|
),
|
|
size="2",
|
|
),
|
|
spacing="2",
|
|
align="center",
|
|
),
|
|
rx.divider(margin_y="8px"),
|
|
# Data source options
|
|
rx.vstack(
|
|
# SQLite option
|
|
rx.box(
|
|
rx.hstack(
|
|
rx.icon("database", size=20, color=NHS_BLUE),
|
|
rx.vstack(
|
|
rx.hstack(
|
|
rx.text("SQLite Database", weight="medium"),
|
|
rx.cond(
|
|
State.sqlite_available,
|
|
rx.badge("Available", color_scheme="green", size="1"),
|
|
rx.badge("No data", color_scheme="gray", size="1"),
|
|
),
|
|
spacing="2",
|
|
),
|
|
rx.cond(
|
|
State.sqlite_available,
|
|
rx.text(
|
|
f"Contains pre-loaded patient data",
|
|
size="1",
|
|
color="gray",
|
|
),
|
|
rx.text(
|
|
"Run data migration to populate",
|
|
size="1",
|
|
color="gray",
|
|
),
|
|
),
|
|
spacing="1",
|
|
align="start",
|
|
),
|
|
rx.spacer(),
|
|
rx.button(
|
|
"Use SQLite",
|
|
on_click=State.use_sqlite_source,
|
|
variant=rx.cond(State.data_source == "sqlite", "solid", "outline"),
|
|
color_scheme="green",
|
|
size="2",
|
|
disabled=~State.sqlite_available,
|
|
),
|
|
spacing="3",
|
|
align="center",
|
|
width="100%",
|
|
),
|
|
padding="12px",
|
|
background=rx.cond(
|
|
State.data_source == "sqlite",
|
|
"rgba(0, 94, 184, 0.05)",
|
|
"transparent",
|
|
),
|
|
border_radius="6px",
|
|
border=rx.cond(
|
|
State.data_source == "sqlite",
|
|
"1px solid rgb(0, 94, 184)",
|
|
"1px solid transparent",
|
|
),
|
|
width="100%",
|
|
),
|
|
# File upload option
|
|
rx.box(
|
|
rx.vstack(
|
|
rx.hstack(
|
|
rx.icon("upload", size=20, color=NHS_BLUE),
|
|
rx.vstack(
|
|
rx.hstack(
|
|
rx.text("Upload File", weight="medium"),
|
|
rx.cond(
|
|
State.file_upload_success,
|
|
rx.badge(State.file_size_display, color_scheme="green", size="1"),
|
|
),
|
|
spacing="2",
|
|
),
|
|
rx.text(
|
|
"Upload CSV or Parquet file",
|
|
size="1",
|
|
color="gray",
|
|
),
|
|
spacing="1",
|
|
align="start",
|
|
),
|
|
rx.spacer(),
|
|
rx.cond(
|
|
State.file_upload_success,
|
|
rx.hstack(
|
|
rx.button(
|
|
"Use File",
|
|
on_click=State.use_file_source,
|
|
variant=rx.cond(State.data_source == "file", "solid", "outline"),
|
|
color_scheme="blue",
|
|
size="2",
|
|
),
|
|
rx.button(
|
|
rx.icon("x", size=14),
|
|
on_click=State.clear_uploaded_file,
|
|
variant="ghost",
|
|
color_scheme="red",
|
|
size="1",
|
|
),
|
|
spacing="1",
|
|
),
|
|
),
|
|
spacing="3",
|
|
align="center",
|
|
width="100%",
|
|
),
|
|
rx.cond(
|
|
State.file_upload_success,
|
|
rx.text(
|
|
State.uploaded_file_name,
|
|
size="2",
|
|
color=NHS_BLUE,
|
|
font_family="monospace",
|
|
),
|
|
rx.upload(
|
|
rx.vstack(
|
|
rx.cond(
|
|
State.file_processing,
|
|
rx.spinner(size="2"),
|
|
rx.icon("file-up", size=24, color="gray"),
|
|
),
|
|
rx.text(
|
|
"Drag & drop or click to browse",
|
|
size="2",
|
|
color="gray",
|
|
),
|
|
rx.text(
|
|
"Supports CSV, Parquet",
|
|
size="1",
|
|
color="gray",
|
|
),
|
|
spacing="2",
|
|
align="center",
|
|
padding="16px",
|
|
),
|
|
id="file_upload",
|
|
accept={
|
|
"text/csv": [".csv"],
|
|
"application/octet-stream": [".parquet", ".pq"],
|
|
},
|
|
max_files=1,
|
|
border="1px dashed rgb(200, 200, 200)",
|
|
border_radius="6px",
|
|
padding="4px",
|
|
width="100%",
|
|
on_drop=State.handle_file_upload(rx.upload_files(upload_id="file_upload")),
|
|
),
|
|
),
|
|
rx.cond(
|
|
State.file_upload_error != "",
|
|
rx.text(
|
|
State.file_upload_error,
|
|
size="2",
|
|
color="red",
|
|
),
|
|
),
|
|
spacing="2",
|
|
width="100%",
|
|
),
|
|
padding="12px",
|
|
background=rx.cond(
|
|
(State.data_source == "file") & State.file_upload_success,
|
|
"rgba(0, 94, 184, 0.05)",
|
|
"transparent",
|
|
),
|
|
border_radius="6px",
|
|
border=rx.cond(
|
|
(State.data_source == "file") & State.file_upload_success,
|
|
"1px solid rgb(0, 94, 184)",
|
|
"1px solid transparent",
|
|
),
|
|
width="100%",
|
|
),
|
|
# Snowflake option
|
|
rx.box(
|
|
rx.hstack(
|
|
rx.icon("cloud", size=20, color=NHS_BLUE),
|
|
rx.vstack(
|
|
rx.hstack(
|
|
rx.text("Snowflake", weight="medium"),
|
|
rx.cond(
|
|
State.snowflake_configured,
|
|
rx.badge("Configured", color_scheme="blue", size="1"),
|
|
rx.badge("Not configured", color_scheme="gray", size="1"),
|
|
),
|
|
spacing="2",
|
|
),
|
|
rx.text(
|
|
"Query live data from Snowflake",
|
|
size="1",
|
|
color="gray",
|
|
),
|
|
spacing="1",
|
|
align="start",
|
|
),
|
|
rx.spacer(),
|
|
rx.button(
|
|
"Use Snowflake",
|
|
on_click=State.use_snowflake_source,
|
|
variant=rx.cond(State.data_source == "snowflake", "solid", "outline"),
|
|
color_scheme="blue",
|
|
size="2",
|
|
disabled=~State.snowflake_configured,
|
|
),
|
|
spacing="3",
|
|
align="center",
|
|
width="100%",
|
|
),
|
|
padding="12px",
|
|
background=rx.cond(
|
|
State.data_source == "snowflake",
|
|
"rgba(0, 94, 184, 0.05)",
|
|
"transparent",
|
|
),
|
|
border_radius="6px",
|
|
border=rx.cond(
|
|
State.data_source == "snowflake",
|
|
"1px solid rgb(0, 94, 184)",
|
|
"1px solid transparent",
|
|
),
|
|
width="100%",
|
|
),
|
|
spacing="2",
|
|
width="100%",
|
|
),
|
|
spacing="3",
|
|
align="start",
|
|
width="100%",
|
|
),
|
|
padding="20px",
|
|
background="white",
|
|
border_radius="8px",
|
|
border="1px solid rgb(229, 231, 235)",
|
|
width="100%",
|
|
)
|
|
|
|
|
|
def filter_controls() -> rx.Component:
|
|
"""Filter controls section with date pickers, minimum patients, and custom title."""
|
|
return rx.box(
|
|
rx.vstack(
|
|
rx.heading("Analysis Settings", size="5", color=NHS_DARK_BLUE, id="analysis-settings-heading"),
|
|
# Date range row
|
|
rx.hstack(
|
|
date_input(
|
|
"Start Date",
|
|
State.start_date,
|
|
State.set_start_date,
|
|
"Include patients initiated from this date",
|
|
input_id="start-date",
|
|
),
|
|
date_input(
|
|
"End Date",
|
|
State.end_date,
|
|
State.set_end_date,
|
|
"Include patients initiated until this date",
|
|
input_id="end-date",
|
|
),
|
|
date_input(
|
|
"Last Seen After",
|
|
State.last_seen_date,
|
|
State.set_last_seen_date,
|
|
"Only include patients seen after this date",
|
|
input_id="last-seen-date",
|
|
),
|
|
spacing="4",
|
|
width="100%",
|
|
flex_wrap="wrap",
|
|
role="group",
|
|
aria_label="Date range filters",
|
|
),
|
|
rx.divider(margin_y="12px"),
|
|
# Additional settings row
|
|
rx.hstack(
|
|
# Minimum patients
|
|
rx.vstack(
|
|
rx.el.label(
|
|
"Minimum Patients",
|
|
html_for="min-patients",
|
|
font_size="14px",
|
|
font_weight="500",
|
|
color=NHS_DARK_BLUE,
|
|
),
|
|
rx.hstack(
|
|
rx.input(
|
|
type="number",
|
|
value=State.minimum_patients.to_string(),
|
|
on_change=State.set_minimum_patients_from_input,
|
|
min="0",
|
|
max="1000",
|
|
width="100px",
|
|
id="min-patients",
|
|
aria_describedby="min-patients-help",
|
|
),
|
|
rx.slider(
|
|
value=[State.minimum_patients],
|
|
on_change=State.set_minimum_patients_from_slider,
|
|
min=0,
|
|
max=100,
|
|
step=1,
|
|
width="150px",
|
|
aria_label="Minimum patients slider",
|
|
),
|
|
spacing="3",
|
|
align="center",
|
|
),
|
|
rx.text(
|
|
"Hide pathways with fewer patients",
|
|
size="1",
|
|
color="gray",
|
|
id="min-patients-help",
|
|
),
|
|
spacing="1",
|
|
align="start",
|
|
),
|
|
# Custom title
|
|
rx.vstack(
|
|
rx.el.label(
|
|
"Custom Title (Optional)",
|
|
html_for="custom-title",
|
|
font_size="14px",
|
|
font_weight="500",
|
|
color=NHS_DARK_BLUE,
|
|
),
|
|
rx.input(
|
|
placeholder="Leave empty for auto-generated title",
|
|
value=State.custom_title,
|
|
on_change=State.set_custom_title,
|
|
width="300px",
|
|
id="custom-title",
|
|
aria_describedby="custom-title-help",
|
|
),
|
|
rx.text(
|
|
"Override the default chart title",
|
|
size="1",
|
|
color="gray",
|
|
id="custom-title-help",
|
|
),
|
|
spacing="1",
|
|
align="start",
|
|
),
|
|
spacing="6",
|
|
width="100%",
|
|
flex_wrap="wrap",
|
|
align="start",
|
|
),
|
|
spacing="4",
|
|
align="start",
|
|
width="100%",
|
|
),
|
|
padding="20px",
|
|
background="white",
|
|
border_radius="8px",
|
|
border="1px solid rgb(229, 231, 235)",
|
|
width="100%",
|
|
role="region",
|
|
aria_labelledby="analysis-settings-heading",
|
|
)
|
|
|
|
|
|
def indication_result_row(result: dict) -> rx.Component:
|
|
"""Render a single row in the indication validation results table."""
|
|
match_rate = result["match_rate"]
|
|
# Color code: green for high match rates, amber for moderate, red for low
|
|
# Use .to(int) to cast Reflex Var for comparison (rx.foreach items are Vars)
|
|
rate_color = rx.cond(
|
|
match_rate.to(int) >= 80,
|
|
"green",
|
|
rx.cond(match_rate.to(int) >= 50, "orange", "red"),
|
|
)
|
|
return rx.table.row(
|
|
rx.table.cell(rx.text(result["drug_name"], weight="medium")),
|
|
rx.table.cell(result["total_patients"].to_string()),
|
|
rx.table.cell(result["patients_with_indication"].to_string()),
|
|
rx.table.cell(
|
|
rx.hstack(
|
|
rx.progress(
|
|
value=match_rate,
|
|
max=100,
|
|
width="60px",
|
|
height="8px",
|
|
color_scheme=rate_color,
|
|
),
|
|
rx.text(
|
|
match_rate.to_string() + "%",
|
|
size="2",
|
|
color=rate_color,
|
|
weight="medium",
|
|
),
|
|
spacing="2",
|
|
align="center",
|
|
)
|
|
),
|
|
)
|
|
|
|
|
|
def indication_validation_summary() -> rx.Component:
|
|
"""
|
|
Component to display indication validation results per drug.
|
|
|
|
Shows a collapsible section with a table of per-drug match rates,
|
|
helping users identify which drugs have good vs poor GP diagnosis coverage.
|
|
"""
|
|
return rx.cond(
|
|
State.has_indication_results,
|
|
rx.el.section(
|
|
rx.vstack(
|
|
# Header with overall summary
|
|
rx.hstack(
|
|
rx.hstack(
|
|
rx.icon("clipboard-check", size=20, color=NHS_DARK_BLUE, aria_hidden="true"),
|
|
rx.heading(
|
|
"GP Indication Validation Results",
|
|
size="5",
|
|
color=NHS_DARK_BLUE,
|
|
id="indication-results-heading",
|
|
),
|
|
spacing="2",
|
|
align="center",
|
|
),
|
|
rx.spacer(),
|
|
rx.badge(
|
|
State.indication_validation_summary,
|
|
color_scheme="blue",
|
|
size="2",
|
|
),
|
|
width="100%",
|
|
align="center",
|
|
),
|
|
rx.text(
|
|
"Shows the percentage of patients with valid GP diagnoses matching their prescribed drug's indication. "
|
|
"Lower rates may indicate prescribing for off-label use, data quality issues, or patients treated across multiple providers.",
|
|
size="2",
|
|
color="gray",
|
|
),
|
|
# Results table
|
|
rx.table.root(
|
|
rx.table.header(
|
|
rx.table.row(
|
|
rx.table.column_header_cell("Drug Name"),
|
|
rx.table.column_header_cell("Total Patients"),
|
|
rx.table.column_header_cell("With GP Indication"),
|
|
rx.table.column_header_cell("Match Rate"),
|
|
),
|
|
),
|
|
rx.table.body(
|
|
rx.foreach(State.indication_results_list, indication_result_row)
|
|
),
|
|
width="100%",
|
|
size="2",
|
|
),
|
|
# Legend
|
|
rx.hstack(
|
|
rx.text("Legend:", size="1", color="gray", weight="medium"),
|
|
rx.hstack(
|
|
rx.badge("80%+", color_scheme="green", size="1"),
|
|
rx.text("Good coverage", size="1", color="gray"),
|
|
spacing="1",
|
|
align="center",
|
|
),
|
|
rx.hstack(
|
|
rx.badge("50-79%", color_scheme="orange", size="1"),
|
|
rx.text("Moderate", size="1", color="gray"),
|
|
spacing="1",
|
|
align="center",
|
|
),
|
|
rx.hstack(
|
|
rx.badge("<50%", color_scheme="red", size="1"),
|
|
rx.text("Low coverage", size="1", color="gray"),
|
|
spacing="1",
|
|
align="center",
|
|
),
|
|
spacing="4",
|
|
flex_wrap="wrap",
|
|
),
|
|
spacing="3",
|
|
width="100%",
|
|
align="start",
|
|
),
|
|
padding="20px",
|
|
background="white",
|
|
border_radius="8px",
|
|
border="1px solid rgb(229, 231, 235)",
|
|
width="100%",
|
|
aria_labelledby="indication-results-heading",
|
|
),
|
|
)
|
|
|
|
|
|
def home_content() -> rx.Component:
|
|
"""Home page content with filter configuration and analysis controls."""
|
|
return rx.vstack(
|
|
# Hero section
|
|
rx.box(
|
|
rx.vstack(
|
|
rx.image(
|
|
src="/logo.png",
|
|
height="60px",
|
|
alt="NHS Logo",
|
|
),
|
|
rx.heading(
|
|
"Patient Pathway Analysis",
|
|
size="8",
|
|
color=NHS_DARK_BLUE,
|
|
),
|
|
rx.text(
|
|
"Analyze secondary care treatment pathways for high-cost drugs",
|
|
size="4",
|
|
color="gray",
|
|
),
|
|
spacing="3",
|
|
align="center",
|
|
),
|
|
padding="32px",
|
|
background="white",
|
|
border_radius="12px",
|
|
border="1px solid rgb(229, 231, 235)",
|
|
width="100%",
|
|
text_align="center",
|
|
),
|
|
# Status cards
|
|
rx.hstack(
|
|
info_card("Drugs Loaded", State.drug_selection_count, "pill"),
|
|
info_card("Trusts", State.trust_selection_count, "building"),
|
|
info_card("Directories", State.directory_selection_count, "folder"),
|
|
spacing="4",
|
|
width="100%",
|
|
flex_wrap="wrap",
|
|
),
|
|
# Data source selector
|
|
data_source_selector(),
|
|
# Filter controls (date pickers, minimum patients, custom title)
|
|
filter_controls(),
|
|
# Filter summary
|
|
rx.box(
|
|
rx.vstack(
|
|
rx.heading("Current Filter Settings", size="4", color=NHS_DARK_BLUE),
|
|
rx.text(
|
|
State.filter_summary,
|
|
white_space="pre-wrap",
|
|
font_family="monospace",
|
|
font_size="13px",
|
|
color="gray",
|
|
),
|
|
spacing="2",
|
|
align="start",
|
|
width="100%",
|
|
),
|
|
padding="20px",
|
|
background="white",
|
|
border_radius="8px",
|
|
border="1px solid rgb(229, 231, 235)",
|
|
width="100%",
|
|
),
|
|
# Action buttons
|
|
rx.hstack(
|
|
rx.button(
|
|
rx.icon("database", size=16, aria_hidden="true"),
|
|
"Load Reference Data",
|
|
on_click=State.load_reference_data,
|
|
color_scheme="blue",
|
|
size="3",
|
|
disabled=State.analysis_running,
|
|
aria_label="Load reference data from CSV files",
|
|
),
|
|
rx.button(
|
|
rx.cond(
|
|
State.analysis_running,
|
|
rx.hstack(
|
|
rx.spinner(size="1"),
|
|
rx.text("Running..."),
|
|
spacing="2",
|
|
align="center",
|
|
),
|
|
rx.hstack(
|
|
rx.icon("play", size=16, aria_hidden="true"),
|
|
rx.text("Run Analysis"),
|
|
spacing="2",
|
|
align="center",
|
|
),
|
|
),
|
|
on_click=State.run_analysis,
|
|
color_scheme="green",
|
|
size="3",
|
|
disabled=State.analysis_running,
|
|
aria_label="Run patient pathway analysis",
|
|
aria_busy=State.analysis_running,
|
|
),
|
|
spacing="3",
|
|
role="toolbar",
|
|
aria_label="Analysis actions",
|
|
),
|
|
# Messages with live regions for screen readers
|
|
rx.cond(
|
|
State.status_message != "",
|
|
rx.callout(
|
|
State.status_message,
|
|
icon="info",
|
|
color="blue",
|
|
role="status",
|
|
aria_live="polite",
|
|
),
|
|
),
|
|
rx.cond(
|
|
State.error_message != "",
|
|
rx.callout(
|
|
State.error_message,
|
|
icon="triangle-alert",
|
|
color="red",
|
|
role="alert",
|
|
aria_live="assertive",
|
|
),
|
|
),
|
|
# Chart display
|
|
rx.cond(
|
|
State.has_chart,
|
|
rx.el.section(
|
|
rx.vstack(
|
|
rx.hstack(
|
|
rx.heading("Patient Pathway Chart", size="5", color=NHS_DARK_BLUE, id="chart-heading"),
|
|
rx.spacer(),
|
|
rx.hstack(
|
|
rx.button(
|
|
rx.icon("download", size=14, aria_hidden="true"),
|
|
"Export HTML",
|
|
on_click=State.export_chart_html,
|
|
variant="outline",
|
|
size="2",
|
|
aria_label="Export chart as interactive HTML file",
|
|
),
|
|
rx.button(
|
|
rx.icon("file-spreadsheet", size=14, aria_hidden="true"),
|
|
"Export CSV",
|
|
on_click=State.export_data_csv,
|
|
variant="outline",
|
|
size="2",
|
|
aria_label="Export data as CSV spreadsheet",
|
|
),
|
|
spacing="2",
|
|
role="toolbar",
|
|
aria_label="Export options",
|
|
),
|
|
width="100%",
|
|
align="center",
|
|
),
|
|
rx.text(
|
|
"Click on sections to zoom in. Use the toolbar for additional options.",
|
|
size="2",
|
|
color="gray",
|
|
),
|
|
# Export messages
|
|
rx.cond(
|
|
State.export_message != "",
|
|
rx.callout(
|
|
State.export_message,
|
|
icon="check",
|
|
color="green",
|
|
role="status",
|
|
aria_live="polite",
|
|
),
|
|
),
|
|
rx.cond(
|
|
State.export_error != "",
|
|
rx.callout(
|
|
State.export_error,
|
|
icon="triangle-alert",
|
|
color="red",
|
|
role="alert",
|
|
),
|
|
),
|
|
rx.el.figure(
|
|
rx.plotly(data=State.chart_data),
|
|
aria_label="Interactive patient pathway icicle chart showing treatment hierarchy",
|
|
),
|
|
spacing="3",
|
|
width="100%",
|
|
),
|
|
padding="20px",
|
|
background="white",
|
|
border_radius="8px",
|
|
border="1px solid rgb(229, 231, 235)",
|
|
width="100%",
|
|
aria_labelledby="chart-heading",
|
|
),
|
|
),
|
|
# Indication validation results (shown after chart)
|
|
indication_validation_summary(),
|
|
spacing="5",
|
|
width="100%",
|
|
align="start",
|
|
)
|
|
|
|
|
|
def selection_page_content(
|
|
title: str,
|
|
description: str,
|
|
items: rx.Var,
|
|
selected_items: rx.Var,
|
|
toggle_handler,
|
|
select_all_handler,
|
|
clear_handler,
|
|
count_text: rx.Var,
|
|
search_value: rx.Var,
|
|
search_handler,
|
|
clear_search_handler,
|
|
search_result_text: rx.Var,
|
|
extra_buttons: list[rx.Component] = None,
|
|
page_id: str = "selection",
|
|
) -> rx.Component:
|
|
"""Generic selection page content for drugs, trusts, directories with search and accessibility."""
|
|
heading_id = f"{page_id}-heading"
|
|
search_id = f"{page_id}-search"
|
|
list_id = f"{page_id}-list"
|
|
|
|
buttons = [
|
|
rx.button(
|
|
"Select All",
|
|
on_click=select_all_handler,
|
|
variant="outline",
|
|
size="2",
|
|
aria_label=f"Select all {title.lower()}",
|
|
),
|
|
rx.button(
|
|
"Clear All",
|
|
on_click=clear_handler,
|
|
variant="outline",
|
|
size="2",
|
|
aria_label=f"Clear all {title.lower()} selections",
|
|
),
|
|
]
|
|
if extra_buttons:
|
|
buttons.extend(extra_buttons)
|
|
|
|
return rx.vstack(
|
|
# Header
|
|
rx.el.header(
|
|
rx.vstack(
|
|
rx.heading(title, size="6", color=NHS_DARK_BLUE, id=heading_id),
|
|
rx.text(description, color="gray"),
|
|
rx.el.div(
|
|
count_text,
|
|
font_weight="500",
|
|
color=NHS_BLUE,
|
|
aria_live="polite",
|
|
aria_atomic="true",
|
|
),
|
|
spacing="2",
|
|
align="start",
|
|
),
|
|
padding="20px",
|
|
background="white",
|
|
border_radius="8px",
|
|
border="1px solid rgb(229, 231, 235)",
|
|
width="100%",
|
|
),
|
|
# Search input
|
|
rx.box(
|
|
rx.hstack(
|
|
rx.icon("search", size=16, color="gray", aria_hidden="true"),
|
|
rx.input(
|
|
placeholder=f"Search {title.lower()}...",
|
|
value=search_value,
|
|
on_change=search_handler,
|
|
width="100%",
|
|
id=search_id,
|
|
aria_label=f"Search {title.lower()}",
|
|
aria_controls=list_id,
|
|
),
|
|
rx.cond(
|
|
search_value != "",
|
|
rx.button(
|
|
rx.icon("x", size=14, aria_hidden="true"),
|
|
on_click=clear_search_handler,
|
|
variant="ghost",
|
|
color_scheme="gray",
|
|
size="1",
|
|
aria_label="Clear search",
|
|
),
|
|
),
|
|
spacing="2",
|
|
align="center",
|
|
width="100%",
|
|
),
|
|
padding="12px 16px",
|
|
background="white",
|
|
border_radius="8px",
|
|
border="1px solid rgb(229, 231, 235)",
|
|
width="100%",
|
|
role="search",
|
|
),
|
|
# Action buttons and search result count
|
|
rx.hstack(
|
|
rx.hstack(*buttons, spacing="2", role="toolbar", aria_label="Selection actions"),
|
|
rx.spacer(),
|
|
rx.el.div(
|
|
search_result_text,
|
|
font_size="14px",
|
|
color="gray",
|
|
aria_live="polite",
|
|
),
|
|
spacing="3",
|
|
width="100%",
|
|
align="center",
|
|
),
|
|
# Selection grid
|
|
rx.box(
|
|
rx.vstack(
|
|
rx.foreach(
|
|
items,
|
|
lambda item: rx.box(
|
|
rx.checkbox(
|
|
item,
|
|
checked=selected_items.contains(item),
|
|
on_change=lambda: toggle_handler(item),
|
|
size="2",
|
|
),
|
|
padding="8px 12px",
|
|
background=rx.cond(
|
|
selected_items.contains(item),
|
|
"rgba(0, 94, 184, 0.1)",
|
|
"transparent",
|
|
),
|
|
border_radius="4px",
|
|
width="100%",
|
|
),
|
|
),
|
|
spacing="1",
|
|
width="100%",
|
|
max_height="500px",
|
|
overflow_y="auto",
|
|
id=list_id,
|
|
role="group",
|
|
aria_labelledby=heading_id,
|
|
),
|
|
padding="16px",
|
|
background="white",
|
|
border_radius="8px",
|
|
border="1px solid rgb(229, 231, 235)",
|
|
width="100%",
|
|
),
|
|
spacing="4",
|
|
width="100%",
|
|
align="start",
|
|
)
|
|
|
|
|
|
def drugs_content() -> rx.Component:
|
|
"""Drug selection page content."""
|
|
return selection_page_content(
|
|
title="Drug Selection",
|
|
description="Select which high-cost drugs to include in the analysis",
|
|
items=State.filtered_drugs,
|
|
selected_items=State.selected_drugs,
|
|
toggle_handler=State.toggle_drug,
|
|
select_all_handler=State.select_all_drugs,
|
|
clear_handler=State.clear_drugs,
|
|
count_text=State.drug_selection_count,
|
|
search_value=State.drug_search,
|
|
search_handler=State.set_drug_search,
|
|
clear_search_handler=State.clear_drug_search,
|
|
search_result_text=State.drug_search_result_count,
|
|
extra_buttons=[
|
|
rx.button(
|
|
"Select Defaults",
|
|
on_click=State.select_default_drugs,
|
|
variant="outline",
|
|
size="2",
|
|
aria_label="Select default drugs (Include=1)",
|
|
),
|
|
],
|
|
page_id="drugs",
|
|
)
|
|
|
|
|
|
def trusts_content() -> rx.Component:
|
|
"""Trust selection page content."""
|
|
return selection_page_content(
|
|
title="Trust Selection",
|
|
description="Select NHS trusts to include (leave empty for all trusts)",
|
|
items=State.filtered_trusts,
|
|
selected_items=State.selected_trusts,
|
|
toggle_handler=State.toggle_trust,
|
|
select_all_handler=State.select_all_trusts,
|
|
clear_handler=State.clear_trusts,
|
|
count_text=State.trust_selection_count,
|
|
search_value=State.trust_search,
|
|
search_handler=State.set_trust_search,
|
|
clear_search_handler=State.clear_trust_search,
|
|
search_result_text=State.trust_search_result_count,
|
|
page_id="trusts",
|
|
)
|
|
|
|
|
|
def directories_content() -> rx.Component:
|
|
"""Directory selection page content."""
|
|
return selection_page_content(
|
|
title="Directory Selection",
|
|
description="Select medical directories/specialties to include (leave empty for all)",
|
|
items=State.filtered_directories,
|
|
selected_items=State.selected_directories,
|
|
toggle_handler=State.toggle_directory,
|
|
select_all_handler=State.select_all_directories,
|
|
clear_handler=State.clear_directories,
|
|
count_text=State.directory_selection_count,
|
|
search_value=State.directory_search,
|
|
search_handler=State.set_directory_search,
|
|
clear_search_handler=State.clear_directory_search,
|
|
search_result_text=State.directory_search_result_count,
|
|
page_id="directories",
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Page Definitions
|
|
# =============================================================================
|
|
|
|
def index() -> rx.Component:
|
|
"""Home page."""
|
|
return main_layout(
|
|
content_area(home_content(), page_title="Home"),
|
|
current_page="home",
|
|
)
|
|
|
|
|
|
def drugs_page() -> rx.Component:
|
|
"""Drug selection page."""
|
|
return main_layout(
|
|
content_area(drugs_content(), page_title=""),
|
|
current_page="drugs",
|
|
)
|
|
|
|
|
|
def trusts_page() -> rx.Component:
|
|
"""Trust selection page."""
|
|
return main_layout(
|
|
content_area(trusts_content(), page_title=""),
|
|
current_page="trusts",
|
|
)
|
|
|
|
|
|
def directories_page() -> rx.Component:
|
|
"""Directory selection page."""
|
|
return main_layout(
|
|
content_area(directories_content(), page_title=""),
|
|
current_page="directories",
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# App Configuration
|
|
# =============================================================================
|
|
|
|
app = rx.App(
|
|
theme=rx.theme(
|
|
accent_color="blue",
|
|
gray_color="slate",
|
|
radius="medium",
|
|
),
|
|
)
|
|
|
|
# Add pages
|
|
app.add_page(index, route="/", title="Home | NHS HCD Analysis")
|
|
app.add_page(drugs_page, route="/drugs", title="Drug Selection | NHS HCD Analysis")
|
|
app.add_page(trusts_page, route="/trusts", title="Trust Selection | NHS HCD Analysis")
|
|
app.add_page(directories_page, route="/directories", title="Directory Selection | NHS HCD Analysis")
|