"""
Sundew Live Monitor - Enhanced "Wow" Demo
Production-quality interface showcasing neurosymbolic ECG monitoring
"""
import io
import json
import math
import os
import sys
from typing import Any, Dict, List
import gradio as gr
import matplotlib.pyplot as plt
import matplotlib.image as mpimg
import numpy as np
import pandas as pd
ROOT = os.path.dirname(os.path.abspath(__file__))
if ROOT not in sys.path:
sys.path.insert(0, ROOT)
from app.ml.gating import gate_signal
from app.ml.inference import infer_ecg, load_model
from app.rules.engine import evaluate_ecg_rules
load_model()
SCENARIOS = {
"healthy": {
"name": "Healthy Adult (60yo)",
"age": 60,
"has_prior_stroke": False,
"signal_type": "normal",
"description": "Normal sinus rhythm, no risk factors, routine monitoring",
"icon": "โ"
},
"afib_high_risk": {
"name": "AFib Suspect (85yo, Prior Stroke)",
"age": 85,
"has_prior_stroke": True,
"signal_type": "afib",
"description": "Irregular rhythm detected, high-risk patient requiring immediate review",
"icon": "โ "
},
"tachycardia": {
"name": "Tachycardia Episode (45yo)",
"age": 45,
"has_prior_stroke": False,
"signal_type": "tachy",
"description": "Elevated heart rate (120+ bpm), otherwise healthy patient",
"icon": "โ"
},
"elderly_normal": {
"name": "Elderly Patient (78yo, Normal ECG)",
"age": 78,
"has_prior_stroke": True,
"signal_type": "normal",
"description": "High-risk profile but currently stable rhythm",
"icon": "๐ค"
},
"noisy": {
"name": "Poor Signal Quality",
"age": 60,
"has_prior_stroke": False,
"signal_type": "noise",
"description": "Motion artifacts, low-quality signal requiring gating",
"icon": "~"
}
}
def generate_signal(signal_type: str, length: int = 512) -> List[float]:
if signal_type == "normal":
return [0.05 * math.sin(2 * math.pi * 2 * (i / length)) +
0.02 * math.sin(2 * math.pi * 0.5 * (i / length)) for i in range(length)]
elif signal_type == "afib":
return [
0.25 * math.sin(2 * math.pi * 6 * (i / length)) +
0.05 * math.sin(2 * math.pi * 15 * (i / length)) +
(0.15 if i % 40 == 0 else 0.0) +
0.03 * (hash(i) % 100 - 50) / 500
for i in range(length)
]
elif signal_type == "tachy":
return [0.08 * math.sin(2 * math.pi * 4.5 * (i / length)) +
0.03 * math.sin(2 * math.pi * 1 * (i / length)) for i in range(length)]
elif signal_type == "noise":
return [0.02 * math.sin(2 * math.pi * 1 * (i / length)) +
(0.01 if i % 13 == 0 else 0.0) +
0.005 * (hash(i) % 100 - 50) / 50 for i in range(length)]
return [0.0] * length
def run_pipeline(scenario_key: str):
scenario = SCENARIOS[scenario_key]
signal = generate_signal(scenario["signal_type"], length=512)
gated, gating_meta = gate_signal(signal, return_windows=True)
model_output = infer_ecg(gated, original_len=len(signal), gating_meta=gating_meta)
patient_context = {
"patient_id": scenario_key,
"age": scenario["age"],
"has_prior_stroke": scenario["has_prior_stroke"],
}
rules_result = evaluate_ecg_rules(patient_context, model_output)
# Build comprehensive results
energy_saved = (1 - gating_meta.get("ratio", 1.0)) * 100
# Summary card
summary_html = f"""
Patient: {scenario['name']}
Diagnosis
{model_output.get('label', 'Unknown').upper()}
Confidence: {model_output.get('score', 0.0):.1%}
Alert Level
{rules_result.get('alert_level', 'NONE').upper()}
HR: {model_output.get('hr')} bpm
Energy Savings: {energy_saved:.1f}% | Windows: {gating_meta.get('selected_windows', 0)}/{gating_meta.get('total_windows', 0)}
"""
# Signal visualization
fig1, axes = plt.subplots(2, 1, figsize=(12, 6))
axes[0].plot(signal, color='#3498db', linewidth=1.5, alpha=0.8)
axes[0].set_title('Original ECG Signal', fontsize=13, fontweight='bold')
axes[0].set_ylabel('Amplitude')
axes[0].grid(alpha=0.3)
axes[0].set_xlim(0, len(signal))
axes[1].plot(gated, color='#e74c3c', linewidth=1.5, alpha=0.8)
axes[1].set_title(f'Gated Signal (Compression: {gating_meta.get("ratio", 1.0):.1%})', fontsize=13, fontweight='bold')
axes[1].set_xlabel('Sample Index')
axes[1].set_ylabel('Amplitude')
axes[1].grid(alpha=0.3)
fig1.tight_layout()
buf1 = io.BytesIO()
fig1.savefig(buf1, format='png', dpi=150, bbox_inches='tight')
plt.close(fig1)
buf1.seek(0)
signal_img = mpimg.imread(buf1)
# Energy bar chart
fig2, ax = plt.subplots(figsize=(10, 4))
categories = ['Baseline\n(No Gating)', 'Sundew\n(With Gating)']
compute = [100, gating_meta.get("ratio", 1.0) * 100]
colors = ['#e74c3c', '#2ecc71']
bars = ax.barh(categories, compute, color=colors, edgecolor='black', linewidth=1.5)
ax.set_xlabel('Compute Used (%)', fontsize=12, fontweight='bold')
ax.set_xlim(0, 110)
for bar, val in zip(bars, compute):
ax.text(val + 2, bar.get_y() + bar.get_height()/2,
f'{val:.1f}%', va='center', fontsize=12, fontweight='bold')
ax.text(55, 1.6, f'Energy Savings: {energy_saved:.1f}%',
ha='center', fontsize=14, fontweight='bold',
bbox=dict(boxstyle='round,pad=0.8', facecolor='#f39c12', alpha=0.8))
ax.set_title('Computational Efficiency', fontsize=14, fontweight='bold')
ax.spines['top'].set_visible(False)
ax.spines['right'].set_visible(False)
fig2.tight_layout()
buf2 = io.BytesIO()
fig2.savefig(buf2, format='png', dpi=150, bbox_inches='tight')
plt.close(fig2)
buf2.seek(0)
energy_img = mpimg.imread(buf2)
# Rule chain
rule_md = f"""### Rule Chain Trace
**Neural Network Output:**
- Label: `{model_output.get('label')}` (Confidence: {model_output.get('score', 0.0):.3f})
- Estimated HR: `{model_output.get('hr')} bpm`
**Patient Context:**
- Age: {scenario['age']} years
- Prior Stroke: {'Yes' if scenario['has_prior_stroke'] else 'No'}
**Rules Evaluated:**
"""
for exp in rules_result.get('explanations', []):
rule_md += f"\n- {exp}"
rule_md += f"\n\n**Final Alert:** `{rules_result.get('alert_level', 'NONE').upper()}`"
return summary_html, signal_img, energy_img, rule_md
# Build Gradio Interface
with gr.Blocks(title="Sundew ECG Monitor", css="""
.gradio-container {font-family: 'Inter', sans-serif;}
.gr-button-primary {background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); border: none;}
""") as demo:
# Header
gr.HTML("""
Sundew ECG Monitor
Neurosymbolic AI for Energy-Efficient Medical Monitoring
โก 85% Energy Savings
๐ง Explainable AI
๐ฅ Clinical-Grade Rules
""")
with gr.Row():
with gr.Column(scale=1):
gr.Markdown("### Select Patient Scenario")
scenario_dropdown = gr.Radio(
choices=list(SCENARIOS.keys()),
value="afib_high_risk",
label="",
info="Choose a patient to analyze"
)
for key, val in SCENARIOS.items():
gr.Markdown(f"**{val['icon']} {val['name']}**\n{val['description']}", visible=(key=="afib_high_risk"))
run_btn = gr.Button("Run Analysis", variant="primary", size="lg")
gr.Markdown("---")
gr.Markdown("""
**Architecture:**
```
ECG Signal โ Sundew Gating โ ML Inference โ Rule Engine
(50-90% reduction) (PyTorch) (Symbolic)
```
""")
with gr.Column(scale=2):
summary_card = gr.HTML()
with gr.Tabs():
with gr.Tab("๐ Signal Analysis"):
signal_plot = gr.Image(label="ECG: Original vs Gated")
with gr.Tab("โก Energy Efficiency"):
energy_plot = gr.Image(label="Compute Savings")
with gr.Tab("๐ Rule Chain"):
rule_trace = gr.Markdown()
run_btn.click(
run_pipeline,
inputs=scenario_dropdown,
outputs=[summary_card, signal_plot, energy_plot, rule_trace]
)
# Footer
gr.HTML("""
Built with Sundew Algorithm ยท FastAPI ยท PyTorch ยท Gradio
""")
if __name__ == "__main__":
demo.launch()