WESAD Quality Metrics
Adjustable Thresholds Edition
""", unsafe_allow_html=True)
# Report information
st.sidebar.subheader("📁 Report Information")
if metadata:
st.sidebar.info(f"""
**File:** {metadata['filename']}
**Modified:** {metadata['modified_time'][:19]}
**Size:** {metadata['file_size_mb']:.1f} MB
**Users:** {metadata['user_count']}
""")
# Refresh button
if st.sidebar.button("🔄 Refresh Data", help="Reload the latest quality assessment"):
st.cache_data.clear()
st.rerun()
st.sidebar.divider()
# User selection
st.sidebar.subheader("👥 User Selection")
all_users = [s['user_id'] for s in summaries]
# Quick select buttons
col1, col2 = st.sidebar.columns(2)
with col1:
if st.button("Select All"):
st.session_state['selected_users'] = all_users
with col2:
if st.button("Clear All"):
st.session_state['selected_users'] = []
selected_users = st.sidebar.multiselect(
"Select users to analyze:",
options=all_users,
default=st.session_state.get('selected_users', all_users),
key='user_selector'
)
st.session_state['selected_users'] = selected_users
# Get adjustable thresholds
thresholds = render_threshold_controls()
# Display settings
st.sidebar.divider()
st.sidebar.subheader("⚙️ Display Settings")
show_raw_data = st.sidebar.checkbox("Show raw data tables", value=False)
expand_all = st.sidebar.checkbox("Expand all user sections", value=False)
show_threshold_lines = st.sidebar.checkbox("Show threshold lines in charts", value=True)
return selected_users, thresholds, show_raw_data, expand_all, show_threshold_lines
def render_overview_with_thresholds(summaries: List[Dict], selected_users: List[str], thresholds: Dict, show_lines: bool):
"""Overview tab with threshold visualization"""
st.header("📈 Quality Metrics Overview")
# Filter summaries based on dynamic thresholds
min_quality = thresholds['global']['overall_score_min']
filtered_summaries = [
s for s in summaries
if s['user_id'] in selected_users
and (s['overall_mean_score'] or 0) >= min_quality
]
# Display current threshold
st.info(f"🎚️ Current Overall Quality Threshold: **{min_quality:.2f}**")
if not filtered_summaries:
st.warning(f"No users meet the current threshold ({min_quality:.2f})")
st.write(f"Users below threshold: {len([s for s in summaries if s['user_id'] in selected_users])} total")
else:
# KPI Cards
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric(
label="Meeting Threshold",
value=len(filtered_summaries),
delta=f"of {len([s for s in summaries if s['user_id'] in selected_users])} selected"
)
with col2:
avg_score = sum(s['overall_mean_score'] for s in filtered_summaries) / len(filtered_summaries)
st.metric(
label="Mean Score (Above Threshold)",
value=f"{avg_score:.3f}",
delta=f"Threshold: {min_quality:.2f}"
)
with col3:
below_threshold = len([s for s in summaries if s['user_id'] in selected_users and (s['overall_mean_score'] or 0) < min_quality])
st.metric(
label="Below Threshold",
value=below_threshold,
delta="Need attention" if below_threshold > 0 else "All pass",
delta_color="inverse" if below_threshold > 0 else "normal"
)
with col4:
# Count alerts based on current thresholds
total_alerts = 0
for s in summaries:
if s['user_id'] in selected_users:
if s['overall_mean_score'] and s['overall_mean_score'] < min_quality:
total_alerts += 1
st.metric(
label="Active Alerts",
value=total_alerts,
delta="Based on current thresholds"
)
st.divider()
# Visualization with threshold lines
if len([s for s in summaries if s['user_id'] in selected_users]) > 0:
col1, col2 = st.columns(2)
with col1:
st.subheader("📊 Score Distribution vs Threshold")
# Prepare data for all selected users (not just filtered)
all_selected = [s for s in summaries if s['user_id'] in selected_users]
scores = [s['overall_mean_score'] for s in all_selected if s['overall_mean_score']]
fig = go.Figure()
# Add histogram
fig.add_trace(go.Histogram(
x=scores,
nbinsx=20,
name='Score Distribution',
marker_color='lightblue',
opacity=0.7
))
# Add threshold line
if show_lines:
fig.add_vline(
x=min_quality,
line_dash="dash",
line_color="red",
annotation_text=f"Threshold ({min_quality:.2f})"
)
fig.update_layout(
xaxis_title="Overall Quality Score",
yaxis_title="Count",
showlegend=True,
height=400
)
st.plotly_chart(fig, use_container_width=True, key="overview_histogram_adjustable")
with col2:
st.subheader("📈 Layer Scores vs Thresholds")
# Create box plot with threshold lines
layer_data = []
for s in all_selected:
if s['layer1_score']:
layer_data.append({'Layer': 'Layer 1', 'Score': s['layer1_score']})
if s['layer2_score']:
layer_data.append({'Layer': 'Layer 2', 'Score': s['layer2_score']})
if s['layer3_score']:
layer_data.append({'Layer': 'Layer 3', 'Score': s['layer3_score']})
if layer_data:
df_box = pd.DataFrame(layer_data)
fig = px.box(df_box, x='Layer', y='Score', color='Layer')
# Add threshold lines for each layer
if show_lines:
fig.add_hline(
y=thresholds['layer1']['any_validator_score_min'],
line_dash="dash",
line_color="gray",
annotation_text="L1 Threshold"
)
# For L2, we use an approximate threshold (inverse of DLR/MDR)
l2_approx = 1 - thresholds['layer2']['dlr_max']
fig.add_hline(
y=l2_approx,
line_dash="dot",
line_color="gray",
annotation_text="L2 Approx"
)
fig.add_hline(
y=thresholds['layer3']['ppg_quality_min'],
line_dash="dashdot",
line_color="gray",
annotation_text="L3 Threshold"
)
fig.update_layout(showlegend=False, height=400)
st.plotly_chart(fig, use_container_width=True, key="overview_box_adjustable")
# User table with threshold indicators
st.subheader("👥 User Quality Summary")
if len([s for s in summaries if s['user_id'] in selected_users]) > 0:
df_display = pd.DataFrame([{
'User': s['user_id'],
'Layer 1': f"{s['layer1_score']:.3f}" if s['layer1_score'] else 'N/A',
'Layer 2': f"{s['layer2_score']:.3f}" if s['layer2_score'] else 'N/A',
'Layer 3': f"{s['layer3_score']:.3f}" if s['layer3_score'] else 'N/A',
'Overall': f"{s['overall_mean_score']:.3f}" if s['overall_mean_score'] else 'N/A',
'Status': '✅ Pass' if s['overall_mean_score'] and s['overall_mean_score'] >= min_quality else '❌ Fail',
'Margin': f"{(s['overall_mean_score'] - min_quality):.3f}" if s['overall_mean_score'] else 'N/A'
} for s in summaries if s['user_id'] in selected_users])
# Color code based on threshold
st.dataframe(
df_display,
use_container_width=True,
hide_index=True,
column_config={
"Status": st.column_config.TextColumn(
"Threshold Status",
help=f"Based on threshold {min_quality:.2f}"
),
"Margin": st.column_config.TextColumn(
"Margin",
help="Distance from threshold (positive = pass)"
)
}
)
def render_layer1_with_thresholds(results: Dict, selected_users: List[str], thresholds: Dict, expand_all: bool):
"""Layer 1 with threshold indicators"""
st.header("🔍 Layer 1: Data Integrity")
threshold = thresholds['layer1']['any_validator_score_min']
st.info(f"🎚️ Current Layer 1 Threshold: **{threshold:.2f}**")
if 'layer1' not in results:
st.warning("No Layer 1 data available")
return
layer1_data = results['layer1']
# Overall statistics
col1, col2, col3, col4 = st.columns(4)
with col1:
avg_score = sum(layer1_data[u].get('overall_score', 0) for u in selected_users if u in layer1_data) / len(selected_users)
st.metric("Average Layer 1 Score", f"{avg_score:.3f}")
with col2:
perfect_count = sum(1 for u in selected_users if u in layer1_data and layer1_data[u].get('overall_score', 0) == 1.0)
st.metric("Perfect Scores", f"{perfect_count}/{len(selected_users)}")
with col3:
above_threshold = sum(1 for u in selected_users if u in layer1_data and layer1_data[u].get('overall_score', 0) >= threshold)
st.metric("Above Threshold", f"{above_threshold}/{len(selected_users)}")
with col4:
issues_count = sum(len(layer1_data[u].get('issues', [])) for u in selected_users if u in layer1_data)
st.metric("Total Issues", issues_count)
st.divider()
# Detailed per-user sections
for user_id in selected_users:
if user_id not in layer1_data:
continue
user_data = layer1_data[user_id]
overall_score = user_data.get('overall_score', 0)
# Status based on threshold
if overall_score >= threshold:
status_icon = "✅"
status_text = "Pass"
else:
status_icon = "❌"
status_text = f"Below threshold ({threshold:.2f})"
with st.expander(f"{status_icon} User: {user_id} - Score: {overall_score:.3f} - {status_text}", expanded=expand_all):
col1, col2 = st.columns([1, 2])
with col1:
st.metric("Overall Score", f"{overall_score:.3f}")
st.metric("Distance from Threshold", f"{overall_score - threshold:+.3f}")
# Validator scores
st.write("**📋 Validator Scores:**")
validators = ['readme_validator', 'quest_validator', 'pkl_metadata_validator',
'file_format_validator', 'structural_validator']
for validator in validators:
if validator in user_data:
v_data = user_data[validator]
score = v_data.get('score', 0)
passed = v_data.get('passed', False)
status = "✅" if score >= threshold else "⚠️"
validator_name = validator.replace('_', ' ').title()
st.write(f"{status} {validator_name}: {score:.2f}")
with col2:
st.write("**🔍 Validation Details:**")
total_issues = 0
for validator in validators:
if validator in user_data:
issues = user_data[validator].get('issues', [])
if issues:
total_issues += len(issues)
st.warning(f"{validator}: {len(issues)} issues")
for issue in issues[:3]:
st.text(f" • {issue}")
if len(issues) > 3:
st.text(f" ... and {len(issues)-3} more")
if total_issues == 0:
st.success("✅ No validation issues found!")
else:
st.error(f"⚠️ Total issues: {total_issues}")
def render_layer2_with_thresholds(results: Dict, selected_users: List[str], thresholds: Dict, expand_all: bool):
"""Layer 2 with threshold indicators"""
st.header("📊 Layer 2: Data Completeness")
# Show current thresholds
st.info(f"""🎚️ **Current Layer 2 Thresholds:**
• DLR Max: {thresholds['layer2']['dlr_max']:.0%} | MDR Max: {thresholds['layer2']['mdr_max']:.0%}
• SCR Min: {thresholds['layer2']['scr_min']:.0%} | Wear Time Min: {thresholds['layer2']['wear_time_pct_min']:.0f}%
• Plausibility Min: {thresholds['layer2']['plausibility_pct_min']:.0f}%""")
if 'layer2' not in results:
st.warning("No Layer 2 data available")
return
layer2_data = results['layer2']
# Overall statistics
col1, col2, col3, col4 = st.columns(4)
with col1:
avg_score = sum(layer2_data[u].get('overall_score', 0) for u in selected_users if u in layer2_data) / len(selected_users)
st.metric("Average Layer 2 Score", f"{avg_score:.3f}")
with col2:
avg_dlr = sum(layer2_data[u].get('data_loss_ratio', {}).get('overall_dlr', 0) for u in selected_users if u in layer2_data) / len(selected_users)
color = "normal" if avg_dlr <= thresholds['layer2']['dlr_max'] else "inverse"
st.metric("Avg Data Loss Ratio", f"{avg_dlr:.2%}", delta="Pass" if avg_dlr <= thresholds['layer2']['dlr_max'] else "Fail", delta_color=color)
with col3:
avg_mdr = sum(layer2_data[u].get('missing_data_ratio', {}).get('overall_mdr', 0) for u in selected_users if u in layer2_data) / len(selected_users)
color = "normal" if avg_mdr <= thresholds['layer2']['mdr_max'] else "inverse"
st.metric("Avg Missing Data Ratio", f"{avg_mdr:.2%}", delta="Pass" if avg_mdr <= thresholds['layer2']['mdr_max'] else "Fail", delta_color=color)
with col4:
avg_wear = sum(layer2_data[u].get('wear_time', {}).get('wear_time_percentage', 0) for u in selected_users if u in layer2_data) / len(selected_users)
color = "normal" if avg_wear >= thresholds['layer2']['wear_time_pct_min'] else "inverse"
st.metric("Avg Wear Time", f"{avg_wear:.1f}%", delta="Pass" if avg_wear >= thresholds['layer2']['wear_time_pct_min'] else "Fail", delta_color=color)
st.divider()
# Detailed per-user sections
for user_id in selected_users:
if user_id not in layer2_data:
continue
user_data = layer2_data[user_id]
overall_score = user_data.get('overall_score', 0)
# Check threshold violations
violations = []
dlr = user_data.get('data_loss_ratio', {}).get('overall_dlr', 0)
mdr = user_data.get('missing_data_ratio', {}).get('overall_mdr', 0)
scr = user_data.get('sensor_channel_ratio', {}).get('overall_scr', 1.0)
wear_time = user_data.get('wear_time', {}).get('wear_time_percentage', 100)
plausibility = user_data.get('plausibility', {}).get('average_plausible_percentage', 100)
if dlr > thresholds['layer2']['dlr_max']:
violations.append(f"DLR ({dlr:.0%} > {thresholds['layer2']['dlr_max']:.0%})")
if mdr > thresholds['layer2']['mdr_max']:
violations.append(f"MDR ({mdr:.0%} > {thresholds['layer2']['mdr_max']:.0%})")
if scr < thresholds['layer2']['scr_min']:
violations.append(f"SCR ({scr:.0%} < {thresholds['layer2']['scr_min']:.0%})")
if wear_time < thresholds['layer2']['wear_time_pct_min']:
violations.append(f"Wear ({wear_time:.0f}% < {thresholds['layer2']['wear_time_pct_min']:.0f}%)")
if plausibility < thresholds['layer2']['plausibility_pct_min']:
violations.append(f"Plaus ({plausibility:.0f}% < {thresholds['layer2']['plausibility_pct_min']:.0f}%)")
status_icon = "✅" if not violations else "❌"
status_text = "All Pass" if not violations else f"{len(violations)} violations"
with st.expander(f"{status_icon} User: {user_id} - Score: {overall_score:.3f} - {status_text}", expanded=expand_all):
if violations:
st.error(f"**Threshold Violations:** {', '.join(violations)}")
# Metrics in columns
col1, col2, col3 = st.columns(3)
with col1:
st.metric(
"Data Loss Ratio",
f"{dlr:.2%}",
delta=f"Threshold: {thresholds['layer2']['dlr_max']:.0%}",
delta_color="normal" if dlr <= thresholds['layer2']['dlr_max'] else "inverse"
)
st.metric(
"Missing Data Ratio",
f"{mdr:.2%}",
delta=f"Threshold: {thresholds['layer2']['mdr_max']:.0%}",
delta_color="normal" if mdr <= thresholds['layer2']['mdr_max'] else "inverse"
)
with col2:
st.metric(
"Sensor Channel Ratio",
f"{scr:.2f}",
delta=f"Min: {thresholds['layer2']['scr_min']:.2f}",
delta_color="normal" if scr >= thresholds['layer2']['scr_min'] else "inverse"
)
st.metric(
"Wear Time",
f"{wear_time:.1f}%",
delta=f"Min: {thresholds['layer2']['wear_time_pct_min']:.0f}%",
delta_color="normal" if wear_time >= thresholds['layer2']['wear_time_pct_min'] else "inverse"
)
with col3:
st.metric(
"Data Plausibility",
f"{plausibility:.1f}%",
delta=f"Min: {thresholds['layer2']['plausibility_pct_min']:.0f}%",
delta_color="normal" if plausibility >= thresholds['layer2']['plausibility_pct_min'] else "inverse"
)
st.metric(
"Overall Layer 2",
f"{overall_score:.3f}",
delta="Pass" if not violations else f"{len(violations)} issues"
)
def render_layer3_with_thresholds(results: Dict, selected_users: List[str], thresholds: Dict, expand_all: bool):
"""Layer 3 with threshold indicators and FIXED extraction"""
st.header("📡 Layer 3: Signal Quality")
# Show current thresholds
st.info(f"""🎚️ **Current Layer 3 Thresholds:**
• PPG: {thresholds['layer3']['ppg_quality_min']:.2f} | ACC: {thresholds['layer3']['acc_quality_min']:.2f}
• EDA Wrist: {thresholds['layer3']['eda_wrist_quality_min']:.2f} | EDA Chest: {thresholds['layer3']['eda_chest_quality_min']:.2f}
• Temp Wrist: {thresholds['layer3']['temp_wrist_quality_min']:.2f} | Temp Chest: {thresholds['layer3']['temp_chest_quality_min']:.2f}""")
if 'layer3' not in results or 'users' not in results['layer3']:
st.warning("No Layer 3 data available")
return
layer3_users = results['layer3']['users']
# Overall statistics
col1, col2, col3 = st.columns(3)
valid_scores = []
for user_id in selected_users:
if user_id in layer3_users:
score = layer3_users[user_id].get('overall_score', None)
if score:
valid_scores.append(score)
with col1:
if valid_scores:
avg_score = sum(valid_scores) / len(valid_scores)
st.metric("Average Layer 3 Score", f"{avg_score:.3f}")
with col2:
# Count users above minimum threshold (using PPG as reference)
above_threshold = sum(1 for s in valid_scores if s > thresholds['layer3']['ppg_quality_min'])
st.metric("Above Threshold", f"{above_threshold}/{len(valid_scores)}")
with col3:
st.metric("Modalities Analyzed", "6")
st.divider()
# Detailed per-user sections
for user_id in selected_users:
if user_id not in layer3_users:
continue
user_data = layer3_users[user_id]
overall_score = user_data.get('overall_score', 0)
# Get signal quality data - USING CORRECT STRUCTURE
signal_quality = user_data.get('signal_quality', {})
# Check threshold violations for each modality
violations = []
scores = []
labels = []
details = []
# PPG
if 'ppg' in signal_quality and 'summary' in signal_quality['ppg']:
summary = signal_quality['ppg']['summary']
score = summary.get('overall_quality_score', None)
if score is not None:
scores.append(score)
labels.append('PPG')
if score < thresholds['layer3']['ppg_quality_min']:
violations.append(f"PPG ({score:.2f} < {thresholds['layer3']['ppg_quality_min']:.2f})")
details.append({
'Windows': summary.get('total_windows_analyzed', 0),
'SNR': summary.get('mean_snr', 0),
'Threshold': thresholds['layer3']['ppg_quality_min']
})
# ACC - uses 'quality_score' not 'overall_quality_score'!
if 'acc' in signal_quality and 'summary' in signal_quality['acc']:
summary = signal_quality['acc']['summary']
score = summary.get('quality_score', None) # Different field!
if score is not None:
scores.append(score)
labels.append('ACC')
if score < thresholds['layer3']['acc_quality_min']:
violations.append(f"ACC ({score:.2f} < {thresholds['layer3']['acc_quality_min']:.2f})")
details.append({
'Issues': summary.get('total_issues_detected', 0),
'MAI': summary.get('mai_assessment', 'N/A'),
'Threshold': thresholds['layer3']['acc_quality_min']
})
# EDA - split into wrist and chest
if 'eda' in signal_quality:
if 'wrist' in signal_quality['eda'] and 'summary' in signal_quality['eda']['wrist']:
summary = signal_quality['eda']['wrist']['summary']
score = summary.get('overall_quality_score', None)
if score is not None:
scores.append(score)
labels.append('EDA WRIST')
if score < thresholds['layer3']['eda_wrist_quality_min']:
violations.append(f"EDA Wrist ({score:.2f} < {thresholds['layer3']['eda_wrist_quality_min']:.2f})")
details.append({
'SCR Peaks': summary.get('total_scr_peaks', 0),
'Threshold': thresholds['layer3']['eda_wrist_quality_min']
})
if 'chest' in signal_quality['eda'] and 'summary' in signal_quality['eda']['chest']:
summary = signal_quality['eda']['chest']['summary']
score = summary.get('overall_quality_score', None)
if score is not None:
scores.append(score)
labels.append('EDA CHEST')
if score < thresholds['layer3']['eda_chest_quality_min']:
violations.append(f"EDA Chest ({score:.2f} < {thresholds['layer3']['eda_chest_quality_min']:.2f})")
details.append({
'SCR Peaks': summary.get('total_scr_peaks', 0),
'Threshold': thresholds['layer3']['eda_chest_quality_min']
})
# Temperature - split into wrist and chest
if 'temperature' in signal_quality:
if 'wrist' in signal_quality['temperature'] and 'summary' in signal_quality['temperature']['wrist']:
summary = signal_quality['temperature']['wrist']['summary']
score = summary.get('overall_quality_score', None)
if score is not None:
scores.append(score)
labels.append('TEMP WRIST')
if score < thresholds['layer3']['temp_wrist_quality_min']:
violations.append(f"Temp Wrist ({score:.2f} < {thresholds['layer3']['temp_wrist_quality_min']:.2f})")
details.append({
'Mean Temp': f"{summary.get('mean_temperature', 0):.1f}°C",
'Threshold': thresholds['layer3']['temp_wrist_quality_min']
})
if 'chest' in signal_quality['temperature'] and 'summary' in signal_quality['temperature']['chest']:
summary = signal_quality['temperature']['chest']['summary']
score = summary.get('overall_quality_score', None)
if score is not None:
scores.append(score)
labels.append('TEMP CHEST')
if score < thresholds['layer3']['temp_chest_quality_min']:
violations.append(f"Temp Chest ({score:.2f} < {thresholds['layer3']['temp_chest_quality_min']:.2f})")
mean_temp = summary.get('mean_temperature', 'N/A')
if isinstance(mean_temp, str):
temp_str = mean_temp
else:
temp_str = f"{float(mean_temp):.1f}°C"
details.append({
'Mean Temp': temp_str,
'Threshold': thresholds['layer3']['temp_chest_quality_min']
})
status_icon = "✅" if not violations else "❌"
status_text = "All Pass" if not violations else f"{len(violations)} below threshold"
with st.expander(f"{status_icon} User: {user_id} - Score: {overall_score:.3f} - {status_text}", expanded=expand_all):
st.metric("Overall Signal Quality", f"{overall_score:.3f}")
if violations:
st.error(f"**Threshold Violations:** {', '.join(violations[:3])}")
if len(violations) > 3:
st.error(f"... and {len(violations)-3} more")
st.divider()
if scores:
# Create bar chart with threshold lines
fig = go.Figure()
# Add bars
colors = []
for i, (label, score) in enumerate(zip(labels, scores)):
# Get the appropriate threshold for this modality
if 'PPG' in label:
thresh = thresholds['layer3']['ppg_quality_min']
elif 'ACC' in label:
thresh = thresholds['layer3']['acc_quality_min']
elif 'EDA WRIST' in label:
thresh = thresholds['layer3']['eda_wrist_quality_min']
elif 'EDA CHEST' in label:
thresh = thresholds['layer3']['eda_chest_quality_min']
elif 'TEMP WRIST' in label:
thresh = thresholds['layer3']['temp_wrist_quality_min']
elif 'TEMP CHEST' in label:
thresh = thresholds['layer3']['temp_chest_quality_min']
else:
thresh = 0.8
color = 'green' if score >= thresh else 'red'
colors.append(color)
fig.add_trace(go.Bar(
x=labels,
y=scores,
marker_color=colors,
text=[f"{s:.3f}" for s in scores],
textposition='outside'
))
# Add individual threshold lines for each modality
for i, (label, detail) in enumerate(zip(labels, details)):
if 'Threshold' in detail:
fig.add_shape(
type="line",
x0=i-0.4, x1=i+0.4,
y0=detail['Threshold'], y1=detail['Threshold'],
line=dict(color="gray", width=2, dash="dash")
)
fig.update_layout(
title=f"Signal Quality vs Thresholds - {user_id}",
yaxis=dict(range=[0, 1.1], title="Quality Score"),
xaxis=dict(title="Modality"),
showlegend=False,
height=350
)
st.plotly_chart(fig, use_container_width=True, key=f"layer3_bar_thresh_{user_id}")
# Detailed metrics
st.subheader("📊 Detailed Metrics")
cols = st.columns(3)
for idx, (label, score, detail) in enumerate(zip(labels, scores, details)):
col_idx = idx % 3
with cols[col_idx]:
thresh = detail.get('Threshold', 0.8)
status = "✅" if score >= thresh else "❌"
st.write(f"**{status} {label}**")
st.write(f"Score: {score:.3f}")
st.write(f"Threshold: {thresh:.2f}")
st.write(f"Margin: {score - thresh:+.3f}")
for key, value in detail.items():
if key != 'Threshold':
st.write(f"• {key}: {value}")
else:
st.info("No signal quality data available for this user")
def render_alerts_with_thresholds(summaries: List[Dict], results: Dict, selected_users: List[str], thresholds: Dict):
"""Alerts tab showing violations based on current thresholds"""
st.header("⚠️ Quality Alerts - Dynamic Thresholds")
# Show current threshold summary
with st.expander("📋 Current Threshold Settings", expanded=True):
col1, col2, col3 = st.columns(3)
with col1:
st.write("**Global:**")
st.write(f"• Overall Min: {thresholds['global']['overall_score_min']:.2f}")
st.write("**Layer 1:**")
st.write(f"• Validator Min: {thresholds['layer1']['any_validator_score_min']:.2f}")
with col2:
st.write("**Layer 2:**")
st.write(f"• DLR Max: {thresholds['layer2']['dlr_max']:.0%}")
st.write(f"• MDR Max: {thresholds['layer2']['mdr_max']:.0%}")
st.write(f"• Wear Time Min: {thresholds['layer2']['wear_time_pct_min']:.0f}%")
with col3:
st.write("**Layer 3:**")
st.write(f"• PPG Min: {thresholds['layer3']['ppg_quality_min']:.2f}")
st.write(f"• ACC Min: {thresholds['layer3']['acc_quality_min']:.2f}")
st.write(f"• EDA Min: {thresholds['layer3']['eda_wrist_quality_min']:.2f}")
st.divider()
# Evaluate alerts with current thresholds
all_alerts = []
for summary in summaries:
if summary['user_id'] not in selected_users:
continue
user_alerts = evaluate_alerts_with_custom_thresholds(summary, results, thresholds)
all_alerts.extend(user_alerts)
# Display summary metrics
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric("Total Alerts", len(all_alerts))
with col2:
high_alerts = sum(1 for a in all_alerts if a['severity'] == 'HIGH')
st.metric("High Severity", high_alerts, delta_color="inverse" if high_alerts > 0 else "off")
with col3:
medium_alerts = sum(1 for a in all_alerts if a['severity'] == 'MEDIUM')
st.metric("Medium Severity", medium_alerts)
with col4:
users_with_alerts = len(set(a['user_id'] for a in all_alerts))
st.metric("Users Affected", f"{users_with_alerts}/{len(selected_users)}")
st.divider()
if all_alerts:
st.subheader("🚨 Active Alerts")
# Create DataFrame for better display
alerts_df = pd.DataFrame(all_alerts)
# Display by severity
for severity in ['HIGH', 'MEDIUM']:
severity_alerts = alerts_df[alerts_df['severity'] == severity]
if not severity_alerts.empty:
if severity == 'HIGH':
st.error(f"**High Severity Alerts ({len(severity_alerts)})**")
else:
st.warning(f"**Medium Severity Alerts ({len(severity_alerts)})**")
for _, alert in severity_alerts.iterrows():
col1, col2, col3 = st.columns([2, 3, 2])
with col1:
st.write(f"**{alert['user_id']}** - {alert['layer']}")
with col2:
st.write(alert['message'])
with col3:
st.write(f"Value: {alert['value']:.3f}")
# Summary statistics
st.divider()
st.subheader("📊 Alert Statistics")
col1, col2 = st.columns(2)
with col1:
# Alerts by layer
layer_counts = alerts_df['layer'].value_counts()
fig = px.pie(values=layer_counts.values, names=layer_counts.index, title="Alerts by Layer")
st.plotly_chart(fig, use_container_width=True, key="alerts_by_layer")
with col2:
# Alerts by metric
metric_counts = alerts_df['metric'].value_counts().head(5)
fig = px.bar(x=metric_counts.values, y=metric_counts.index, orientation='h', title="Top 5 Metrics with Alerts")
st.plotly_chart(fig, use_container_width=True, key="alerts_by_metric")
else:
st.success("✅ No alerts triggered with current thresholds!")
st.balloons()
# Show what would trigger alerts
st.info("💡 Adjust thresholds in the sidebar to explore different quality criteria")
def main():
"""Main dashboard application with adjustable thresholds"""
# Initialize session state
initialize_threshold_state()
st.title("🎚️ WESAD Quality Metrics Dashboard - Adjustable Thresholds")
st.markdown("**Interactive threshold adjustment for dynamic quality assessment**")
# Load data
with st.spinner("Loading quality assessment data..."):
results, summaries, metadata = load_dashboard_data()
if not results:
st.error("❌ No quality assessment report found!")
st.info("Please run: `python3 run_full.py --save` to generate a report")
return
# Enhanced sidebar with threshold controls
selected_users, thresholds, show_raw_data, expand_all, show_threshold_lines = render_enhanced_sidebar(metadata, summaries)
# Create tabs
tab1, tab2, tab3, tab4, tab5, tab6 = st.tabs([
"📈 Overview",
"🔍 Layer 1: Integrity",
"📊 Layer 2: Completeness",
"📡 Layer 3: Signal Quality",
"⚠️ Dynamic Alerts",
"📊 Threshold Analysis"
])
with tab1:
render_overview_with_thresholds(summaries, selected_users, thresholds, show_threshold_lines)
with tab2:
render_layer1_with_thresholds(results, selected_users, thresholds, expand_all)
with tab3:
render_layer2_with_thresholds(results, selected_users, thresholds, expand_all)
with tab4:
render_layer3_with_thresholds(results, selected_users, thresholds, expand_all)
with tab5:
render_alerts_with_thresholds(summaries, results, selected_users, thresholds)
with tab6:
st.header("📊 Threshold Sensitivity Analysis")
# Threshold impact analysis
st.subheader("Impact of Current Thresholds")
# Calculate pass rates for different thresholds
threshold_ranges = {
'Strict (0.90)': 0.90,
'High (0.85)': 0.85,
'Current': thresholds['global']['overall_score_min'],
'Standard (0.80)': 0.80,
'Moderate (0.70)': 0.70,
'Lenient (0.60)': 0.60
}
pass_rates = []
for name, thresh in threshold_ranges.items():
passing = sum(1 for s in summaries if s['user_id'] in selected_users and s.get('overall_mean_score', 0) >= thresh)
total = len([s for s in summaries if s['user_id'] in selected_users])
pass_rates.append({
'Threshold': f"{name} ({thresh:.2f})",
'Pass Rate': (passing / total * 100) if total > 0 else 0,
'Passing': passing,
'Failing': total - passing
})
df_pass = pd.DataFrame(pass_rates)
col1, col2 = st.columns(2)
with col1:
fig = px.bar(df_pass, x='Threshold', y='Pass Rate', title="Pass Rate by Threshold Level")
fig.add_hline(y=80, line_dash="dash", annotation_text="80% target")
st.plotly_chart(fig, use_container_width=True, key="pass_rate_analysis")
with col2:
fig = px.bar(df_pass, x='Threshold', y=['Passing', 'Failing'],
title="Users Passing vs Failing", barmode='stack')
st.plotly_chart(fig, use_container_width=True, key="pass_fail_stack")
# Detailed threshold impact table
st.subheader("Detailed Threshold Impact")
st.dataframe(df_pass, use_container_width=True, hide_index=True)
# Footer
st.divider()
st.caption(f"Adjustable Thresholds Dashboard | Data: {metadata['filename']} | Updated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
if __name__ == "__main__":
main()