""" VårdSignal — Care Workflow Intelligence Platform Operator dashboard demo (Gradio frontend) Course demonstration only. Not for clinical use. Synthetic data; per-patient Isolation Forest + clinical rule triggers. """ import gradio as gr import pandas as pd import numpy as np import matplotlib.pyplot as plt import matplotlib.dates as mdates import pickle from datetime import datetime, timedelta from io import BytesIO from PIL import Image # ─── Load artifacts ───────────────────────────────────────────── patients = pd.read_csv("patients.csv") streams = pd.read_csv("synthetic_streams.csv") streams["date"] = pd.to_datetime(streams["date"]) with open("models.pkl", "rb") as f: models = pickle.load(f) with open("baselines.pkl", "rb") as f: baselines = pickle.load(f) with open("scenario_assignments.pkl", "rb") as f: scenario_assignments = pickle.load(f) # ─── Configuration ────────────────────────────────────────────── DAYS = 90 TRAIN_WINDOW = DAYS - 14 EVAL_WINDOW = 14 FEATURE_COLS = ["resting_hr", "steps", "sleep_hours", "adherence"] # Tier visual styling — light, calm, professional palette TIER_STYLE = { "High risk": {"bg": "#FEF2F2", "fg": "#991B1B", "border": "#DC2626", "icon": "▲"}, "Watch": {"bg": "#FFFBEB", "fg": "#92400E", "border": "#D97706", "icon": "●"}, "Stable": {"bg": "#F0FDF4", "fg": "#166534", "border": "#16A34A", "icon": "✓"}, } TRIGGER_LABEL = { "rhr_elevation_3d": "Sustained heart rate elevation (3 days)", "activity_drop": "Significant activity drop (3-day window)", "sleep_disturbance": "Sleep disturbance pattern (7-day window)", "adherence_drop": "Medication adherence collapse", "possible_fall": "Possible fall pattern detected", "declining_trend": "Declining activity trend (14 days)", } # Colour palette for plots — matches dashboard theme COL_HR = "#DC2626" COL_STEPS = "#2563EB" COL_SLEEP = "#16A34A" COL_BASE = "#94A3B8" COL_ALERT = "#F59E0B" COL_GRID = "#E2E8F0" plt.rcParams.update({ "font.family": "sans-serif", "font.size": 9, "axes.spines.top": False, "axes.spines.right": False, "axes.grid": True, "grid.alpha": 0.3, "grid.color": COL_GRID, "axes.edgecolor": "#CBD5E1", "axes.labelcolor": "#475569", "xtick.color": "#64748B", "ytick.color": "#64748B", }) # ─── Inference engine ─────────────────────────────────────────── def evaluate_patient(pid): df = streams[streams.patient_id == pid].sort_values("date").reset_index(drop=True) eval_df = df.iloc[TRAIN_WINDOW:].copy() iforest, scaler = models[pid] Xs = scaler.transform(eval_df[FEATURE_COLS].values.astype(float)) scores = iforest.decision_function(Xs) iforest_min = float(scores.min()) iforest_flagged = int((scores < -0.02).sum()) base = baselines[pid] rhr_mean, rhr_std = base["resting_hr"] steps_mean, steps_std = base["steps"] sleep_mean, sleep_std = base["sleep_hours"] adh_baseline = base["adherence_rate"] triggers = [] recent3 = eval_df.tail(3) recent7 = eval_df.tail(7) if (recent3["resting_hr"] > rhr_mean + 2 * rhr_std).all(): triggers.append("rhr_elevation_3d") if recent3["steps"].mean() < steps_mean - 1.5 * steps_std: triggers.append("activity_drop") if recent7["sleep_hours"].mean() < sleep_mean - 1.5 * sleep_std: triggers.append("sleep_disturbance") if recent7["adherence"].mean() < adh_baseline - 0.25: triggers.append("adherence_drop") for _, row in recent3.iterrows(): if (row["steps"] < steps_mean - 3 * steps_std and row["resting_hr"] > rhr_mean + 2.5 * rhr_std): triggers.append("possible_fall") break eval_steps = eval_df["steps"].values if len(eval_steps) >= 14: first_half = eval_steps[:7].mean() second_half = eval_steps[7:].mean() if first_half > 0 and (first_half - second_half) / first_half > 0.20: triggers.append("declining_trend") if "possible_fall" in triggers: tier = "High risk" elif "rhr_elevation_3d" in triggers and "activity_drop" in triggers: tier = "High risk" elif iforest_min < -0.10 or iforest_flagged >= 5: tier = "High risk" elif len(triggers) >= 1 or iforest_flagged >= 2 or iforest_min < -0.04: tier = "Watch" else: tier = "Stable" return { "patient_id": pid, "tier": tier, "iforest_min_score": iforest_min, "iforest_flagged_days": iforest_flagged, "triggers": triggers, "scores_per_day": scores, "eval_df": eval_df, "train_df": df.iloc[:TRAIN_WINDOW], "full_df": df, } # ─── Pre-compute all evaluations ───────────────────────────────── print("Evaluating all patients...") ALL_EVALS = {pid: evaluate_patient(pid) for pid in patients.patient_id} print(f"Done. {len(ALL_EVALS)} patients scored.") # ─── Cohort overview table ────────────────────────────────────── def get_cohort_table(filter_tier="All"): rows = [] for pid in patients.patient_id: ev = ALL_EVALS[pid] p = patients[patients.patient_id == pid].iloc[0] if filter_tier != "All" and ev["tier"] != filter_tier: continue rows.append({ "Tier": f'{TIER_STYLE[ev["tier"]]["icon"]} {ev["tier"]}', "Patient ID": pid, "Name": p["name"], "Age": p["age"], "Municipality": p["municipality"], "Conditions": p["conditions"], "Triggers": len(ev["triggers"]), "Anomaly score": f'{ev["iforest_min_score"]:.3f}', "Flagged days": ev["iforest_flagged_days"], "Devices": p["devices"], }) df = pd.DataFrame(rows) if df.empty: return df tier_order = {"▲ High risk": 0, "● Watch": 1, "✓ Stable": 2} df["_o"] = df["Tier"].map(tier_order) df = df.sort_values("_o").drop("_o", axis=1).reset_index(drop=True) return df # ─── Definitions / glossary card ──────────────────────────────── def get_definitions_card(): return """
▲ High risk — Possible-fall pattern detected, sustained heart-rate elevation paired with activity drop, or anomaly score below −0.10. Recommended action: district nurse review within hours.
● Watch — One or more clinical rule triggers fired, or 2+ flagged days in the evaluation window. Recommended action: hemtjänst coordinator notes for next visit.
✓ Stable — Patient's recent measurements remain within their personalised baseline. No action required.
Anomaly score — Output of the Isolation Forest model (range approx. −0.2 to +0.3). Lower is more anomalous. The displayed value is the lowest score across the 14-day evaluation window.
Flagged days — Number of days within the 14-day evaluation window where the anomaly score fell below the watch threshold (−0.02).
Triggers — Count of clinical rule triggers that fired (rate elevation, activity drop, sleep disturbance, possible fall, declining trend, adherence drop). Listed individually in the Patient Detail tab.
Devices — Hardware partners supplying signals for this patient. VårdSignal aggregates across vendors; the operator chooses devices per patient based on clinical profile.
""" # ─── Patient detail plot ──────────────────────────────────────── def render_patient_plot(pid): ev = ALL_EVALS[pid] p = patients[patients.patient_id == pid].iloc[0] full = ev["full_df"] base = baselines[pid] rhr_mean, rhr_std = base["resting_hr"] steps_mean, steps_std = base["steps"] sleep_mean, sleep_std = base["sleep_hours"] fig, axes = plt.subplots(4, 1, figsize=(11, 9), sharex=True) fig.patch.set_facecolor("#FFFFFF") # eval window highlight (last 14 days) eval_start = full["date"].iloc[TRAIN_WINDOW] eval_end = full["date"].iloc[-1] def add_eval_band(ax, show_label=False): ax.axvspan(eval_start, eval_end, color="#FEF3C7", alpha=0.55, label="Evaluation window" if show_label else None) # 1) Resting HR ax = axes[0] add_eval_band(ax, show_label=True) ax.plot(full["date"], full["resting_hr"], color=COL_HR, lw=1.4) ax.axhline(rhr_mean, color=COL_BASE, lw=0.9, ls="--", label=f"Baseline μ ({rhr_mean:.1f} bpm)") ax.fill_between(full["date"], rhr_mean - 2 * rhr_std, rhr_mean + 2 * rhr_std, color=COL_BASE, alpha=0.10, label="±2σ band") ax.set_ylabel("Resting HR (bpm)", fontsize=9, color="#475569") ax.legend(loc="upper left", fontsize=7.5, framealpha=0.85, edgecolor="#E2E8F0") ax.set_title(f"{p['name']} ({pid}) — {p['age']}-year-old {p['sex']}", fontsize=11, color="#0F172A", loc="left", weight="bold") # 2) Steps ax = axes[1] add_eval_band(ax) ax.bar(full["date"], full["steps"], color=COL_STEPS, alpha=0.65, width=0.85) ax.axhline(steps_mean, color=COL_BASE, lw=0.9, ls="--", label=f"Baseline μ ({steps_mean:.0f})") ax.set_ylabel("Daily steps", fontsize=9, color="#475569") ax.legend(loc="upper left", fontsize=7.5, framealpha=0.85, edgecolor="#E2E8F0") # 3) Sleep ax = axes[2] add_eval_band(ax) ax.plot(full["date"], full["sleep_hours"], color=COL_SLEEP, lw=1.4) ax.axhline(sleep_mean, color=COL_BASE, lw=0.9, ls="--", label=f"Baseline μ ({sleep_mean:.2f} h)") ax.set_ylabel("Sleep (hours)", fontsize=9, color="#475569") ax.legend(loc="upper left", fontsize=7.5, framealpha=0.85, edgecolor="#E2E8F0") # 4) Anomaly score (eval window only) ax = axes[3] add_eval_band(ax) eval_dates = ev["eval_df"]["date"].values scores = ev["scores_per_day"] colors = ["#DC2626" if s < -0.04 else "#F59E0B" if s < -0.02 else "#94A3B8" for s in scores] ax.bar(eval_dates, scores, color=colors, alpha=0.85, width=0.85) ax.axhline(0, color="#94A3B8", lw=0.6) ax.axhline(-0.04, color="#DC2626", lw=0.7, ls=":", label="High-risk threshold") ax.axhline(-0.02, color="#F59E0B", lw=0.7, ls=":", label="Watch threshold") ax.set_ylabel("Anomaly score", fontsize=9, color="#475569") ax.set_xlabel("Date", fontsize=9, color="#475569") ax.legend(loc="lower left", fontsize=7.5, framealpha=0.85, edgecolor="#E2E8F0") # Explanatory annotation: scores only computed during evaluation window train_mid = full["date"].iloc[TRAIN_WINDOW // 2] ymin_ax, ymax_ax = ax.get_ylim() ax.text(train_mid, (ymin_ax + ymax_ax) / 2, "Scores are computed only on the 14-day\n" "evaluation window (yellow band).\n" "Earlier days were used to learn the\n" "personalised baseline.", fontsize=8, color="#64748B", style="italic", ha="center", va="center", bbox=dict(boxstyle="round,pad=0.4", fc="#F8FAFC", ec="#CBD5E1", lw=0.6)) for ax in axes: ax.xaxis.set_major_locator(mdates.WeekdayLocator(interval=2)) ax.xaxis.set_major_formatter(mdates.DateFormatter("%d %b")) ax.tick_params(axis="x", labelsize=8) ax.tick_params(axis="y", labelsize=8) ax.set_facecolor("#FAFAFA") plt.tight_layout() buf = BytesIO() plt.savefig(buf, format="png", dpi=110, bbox_inches="tight", facecolor="#FFFFFF") plt.close(fig) buf.seek(0) return Image.open(buf) # ─── Patient detail panel HTML ────────────────────────────────── def render_patient_panel(pid): ev = ALL_EVALS[pid] p = patients[patients.patient_id == pid].iloc[0] style = TIER_STYLE[ev["tier"]] base = baselines[pid] eval_df = ev["eval_df"] rhr_mean, rhr_std = base["resting_hr"] steps_mean, _ = base["steps"] sleep_mean, _ = base["sleep_hours"] adh_base = base["adherence_rate"] recent_rhr = eval_df["resting_hr"].tail(3).mean() recent_steps = eval_df["steps"].tail(3).mean() recent_sleep = eval_df["sleep_hours"].tail(7).mean() recent_adh = eval_df["adherence"].tail(7).mean() def delta(recent, base, unit="", fmt="{:.1f}"): d = recent - base sign = "+" if d > 0 else "" color = "#991B1B" if abs(d) > 0.5 * base * 0.15 else "#475569" return (f'' f'{fmt.format(recent)}{unit} ({sign}{fmt.format(d)})' f'') triggers_html = "" if ev["triggers"]: triggers_html = "" else: triggers_html = ('
No clinical rule triggers fired.
') routing_role = "" routing_action = "" if ev["tier"] == "High risk": routing_role = "District Nurse (HSL)" routing_action = ("Immediate review; consider home visit within " "4 hours; document escalation in care record.") elif ev["tier"] == "Watch": routing_role = "Hemtjänst Coordinator (SoL)" routing_action = ("Note in next scheduled visit; flag for review at " "weekly care meeting; monitor over next 72 hours.") else: routing_role = "—" routing_action = ("No action required; baseline pattern within expected " "individual variation.") html = f"""
Patient {pid}
{p['name']}
{p['age']} years · {p['sex']} · {p['municipality']} · Activity: {p['activity_class']}
Risk tier
{style['icon']} {ev['tier']}
Conditions: {p['conditions']}
Devices: {p['devices']}
Recent measurements vs personalised baseline
Resting HR (3-day avg) {recent_rhr:.1f} bpm · baseline {rhr_mean:.1f} bpm
Daily steps (3-day avg) {recent_steps:.0f} · baseline {steps_mean:.0f}
Sleep (7-day avg) {recent_sleep:.2f} h · baseline {sleep_mean:.2f} h
Adherence (7-day) {recent_adh*100:.0f}% · baseline {adh_base*100:.0f}%
Detection signals
Isolation Forest: minimum score {ev['iforest_min_score']:.3f} · {ev['iforest_flagged_days']} flagged day(s) in 14-day window
Clinical rule triggers: {triggers_html}
Suggested workflow routing
→ {routing_role}
{routing_action}
""" return html # ─── KPI summary cards ────────────────────────────────────────── def get_kpis(): counts = {"High risk": 0, "Watch": 0, "Stable": 0} for pid in patients.patient_id: counts[ALL_EVALS[pid]["tier"]] += 1 cards = [] for tier, n in counts.items(): s = TIER_STYLE[tier] cards.append(f"""
{s['icon']} {tier}
{n}
of {len(patients)} patients
""") return f'
{"".join(cards)}
' # ─── Build Gradio interface ───────────────────────────────────── LIGHT_THEME = gr.themes.Soft( primary_hue="blue", secondary_hue="slate", neutral_hue="slate", radius_size="md", font=[gr.themes.Font("Segoe UI"), gr.themes.Font("SF Pro Text"), gr.themes.Font("system-ui"), gr.themes.Font("sans-serif")], ).set( body_background_fill="#FFFFFF", background_fill_primary="#FFFFFF", background_fill_secondary="#F8FAFC", block_background_fill="#FFFFFF", block_border_width="1px", block_border_color="#E2E8F0", block_label_background_fill="#F8FAFC", block_label_text_color="#475569", button_primary_background_fill="#1E40AF", button_primary_background_fill_hover="#1E3A8A", button_primary_text_color="#FFFFFF", ) CSS = """ .gradio-container { max-width: 1280px !important; } .dashboard-header { background: linear-gradient(90deg, #F8FAFC 0%, #FFFFFF 100%); border-bottom: 1px solid #E2E8F0; padding: 20px 24px 16px; margin: -16px -16px 16px -16px; } .dashboard-title { font-size: 22px; font-weight: 700; color: #0F172A; margin: 0 0 4px 0; font-family: 'Segoe UI', 'SF Pro Display', -apple-system, sans-serif; } .dashboard-subtitle { font-size: 13px; color: #64748B; margin: 0; font-family: 'Segoe UI', 'SF Pro Text', -apple-system, sans-serif; } .demo-disclaimer { background: #FEF3C7; border-left: 3px solid #D97706; padding: 8px 12px; border-radius: 4px; font-size: 12px; color: #92400E; margin-top: 10px; line-height: 1.4; font-family: 'Segoe UI', sans-serif; } table { font-size: 13px !important; } .tab-nav button { font-size: 14px !important; } """ with gr.Blocks(title="VårdSignal Demo") as demo: gr.HTML("""
VårdSignal — Care Workflow Intelligence
Operator dashboard · Vendor-neutral RPM aggregation · Personalised AI baseline learning
Demonstration only. Synthetic data for 50 simulated patients. Per-patient Isolation Forest with clinical rule triggers. Not a medical device; not for clinical use.
""") with gr.Tabs(): # ============= TAB 1: Cohort Overview ============= with gr.Tab("📋 Cohort Overview"): with gr.Accordion("ℹ️ How to read this dashboard", open=False): gr.HTML(get_definitions_card()) kpi_html = gr.HTML(get_kpis()) with gr.Row(): tier_filter = gr.Radio( choices=["All", "High risk", "Watch", "Stable"], value="All", label="Filter by tier", interactive=True, ) cohort_table = gr.Dataframe( value=get_cohort_table("All"), interactive=False, wrap=True, column_widths=["100px","75px","145px","55px","100px","190px","70px","90px","75px","160px"], ) tier_filter.change( fn=lambda t: get_cohort_table(t), inputs=tier_filter, outputs=cohort_table, ) # ============= TAB 2: Patient Detail ============= with gr.Tab("🩺 Patient Detail"): sorted_pids = sorted( patients.patient_id.tolist(), key=lambda p: ( {"High risk": 0, "Watch": 1, "Stable": 2}[ALL_EVALS[p]["tier"]], p, ), ) patient_choices = [ f'{TIER_STYLE[ALL_EVALS[p]["tier"]]["icon"]} {p} — ' f'{patients[patients.patient_id == p].iloc[0]["name"]}' for p in sorted_pids ] patient_dropdown = gr.Dropdown( choices=patient_choices, value=patient_choices[0], label="Select patient (sorted by tier)", interactive=True, ) with gr.Row(): with gr.Column(scale=2): patient_panel = gr.HTML(render_patient_panel(sorted_pids[0])) with gr.Column(scale=3): patient_plot = gr.Image( value=render_patient_plot(sorted_pids[0]), label="14-day evaluation window vs personalised baseline", height=600, ) def update_patient(choice): pid = choice.split()[1] return render_patient_panel(pid), render_patient_plot(pid) patient_dropdown.change( fn=update_patient, inputs=patient_dropdown, outputs=[patient_panel, patient_plot], ) # ============= TAB 3: About ============= with gr.Tab("ℹ️ About this demo"): gr.HTML("""

What this demo shows

A simulation of the VårdSignal platform's AI-driven care workflow intelligence layer. The system aggregates physiological and behavioural signals from multiple CE-marked clinical-grade RPM devices, learns each patient's individual baseline over a multi-week onboarding window, and routes prioritised alerts to the appropriate caregiver role under Sweden's split SoL/HSL care framework.

Methodology

Each of the 50 simulated patients has individualised baseline parameters drawn from clinically plausible ranges for the 70–95 year-old demographic. Ninety days of daily aggregate signals (resting heart rate, daily step count, sleep duration, medication adherence) are generated per patient. Around 35 per cent of patients have a clinical anomaly scenario injected into the final 14 days (UTI onset, fall event, gradual decline, sundowning, adherence collapse).

For each patient the platform trains an Isolation Forest on the first 76 days of normal baseline data, with a StandardScaler for feature normalisation. The remaining 14 days are scored against the trained baseline. A complementary set of clinical rule triggers (sustained HR elevation, large activity drop, sleep disturbance, possible-fall pattern, declining trend, adherence collapse) runs in parallel. The risk tier is derived from the combined output.

Why this architecture

What this demo does not claim

ICT Startups and High-Tech Entrepreneurship · Blekinge Institute of Technology · Synthetic data only · Demo not for clinical use.

""") if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860, share=False, theme=LIGHT_THEME, css=CSS)