File size: 8,680 Bytes
458c8e2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 |
"""
ECG Analysis Module
Provides modular ECG visualization and analysis functions
"""
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
import io
import base64
from typing import List, Optional, Tuple, Dict, Any
class ECGAnalyzer:
"""Handles ECG data analysis and visualization"""
# Standard 12-lead ECG leads
STANDARD_LEADS = ['I', 'II', 'III', 'aVR', 'aVL', 'aVF', 'V1', 'V2', 'V3', 'V4', 'V5', 'V6']
@staticmethod
def detect_leads(df: pd.DataFrame) -> List[str]:
"""Detect available ECG leads in the dataframe"""
available_leads = []
for lead in ECGAnalyzer.STANDARD_LEADS:
if lead in df.columns:
available_leads.append(lead)
return available_leads
@staticmethod
def detect_time_column(df: pd.DataFrame) -> Optional[str]:
"""Detect the time column in the dataframe"""
time_candidates = ['time', 'Time', 'TIME', 'timestamp', 'sample']
for col in time_candidates:
if col in df.columns:
return col
# If no explicit time column, use index
return None
@staticmethod
def create_signal_plot(df: pd.DataFrame, leads: List[str], time_col: Optional[str] = None) -> str:
"""Create ECG signal waveform plot"""
fig, ax = plt.subplots(figsize=(14, 6))
if time_col and time_col in df.columns:
x_data = df[time_col]
x_label = 'Time (ms)' if 'time' in time_col.lower() else time_col
else:
x_data = df.index
x_label = 'Sample Index'
colors = plt.cm.tab10(np.linspace(0, 1, len(leads)))
for idx, lead in enumerate(leads):
if lead in df.columns:
ax.plot(x_data, df[lead], label=f'Lead {lead}',
linewidth=1.2, alpha=0.8, color=colors[idx])
ax.set_xlabel(x_label, fontsize=11)
ax.set_ylabel('Amplitude (mV)', fontsize=11)
ax.set_title('ECG Signal Waveform', fontsize=13, fontweight='bold')
ax.legend(loc='upper right', fontsize=9, ncol=min(4, len(leads)))
ax.grid(True, alpha=0.3, linestyle='--')
plt.tight_layout()
return ECGAnalyzer._fig_to_base64(fig)
@staticmethod
def create_histogram(df: pd.DataFrame, leads: List[str]) -> str:
"""Create histogram of signal amplitudes"""
fig, ax = plt.subplots(figsize=(14, 6))
colors = plt.cm.tab10(np.linspace(0, 1, len(leads)))
for idx, lead in enumerate(leads):
if lead in df.columns:
ax.hist(df[lead].dropna(), bins=50, alpha=0.6,
label=f'Lead {lead}', color=colors[idx], edgecolor='black')
ax.set_xlabel('Amplitude (mV)', fontsize=11)
ax.set_ylabel('Frequency', fontsize=11)
ax.set_title('Distribution of Signal Amplitudes', fontsize=13, fontweight='bold')
ax.legend(loc='upper right', fontsize=9)
ax.grid(True, alpha=0.3, axis='y')
plt.tight_layout()
return ECGAnalyzer._fig_to_base64(fig)
@staticmethod
def create_scatter_plot(df: pd.DataFrame, lead_x: str = 'I', lead_y: str = 'II') -> str:
"""Create scatter plot comparing two leads"""
fig, ax = plt.subplots(figsize=(8, 8))
if lead_x in df.columns and lead_y in df.columns:
ax.scatter(df[lead_x], df[lead_y], alpha=0.5, s=10, c='steelblue')
ax.set_xlabel(f'Lead {lead_x} Amplitude (mV)', fontsize=11)
ax.set_ylabel(f'Lead {lead_y} Amplitude (mV)', fontsize=11)
ax.set_title(f'Lead {lead_x} vs Lead {lead_y}', fontsize=13, fontweight='bold')
ax.grid(True, alpha=0.3)
# Add correlation coefficient
correlation = df[[lead_x, lead_y]].corr().iloc[0, 1]
ax.text(0.05, 0.95, f'Correlation: {correlation:.3f}',
transform=ax.transAxes, fontsize=10, verticalalignment='top',
bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.5))
else:
ax.text(0.5, 0.5, 'Selected leads not available',
ha='center', va='center', fontsize=12)
plt.tight_layout()
return ECGAnalyzer._fig_to_base64(fig)
@staticmethod
def create_rolling_average(df: pd.DataFrame, leads: List[str],
time_col: Optional[str] = None, window: int = 100) -> str:
"""Create rolling average plot"""
fig, ax = plt.subplots(figsize=(14, 6))
if time_col and time_col in df.columns:
x_data = df[time_col]
x_label = 'Time (ms)' if 'time' in time_col.lower() else time_col
else:
x_data = df.index
x_label = 'Sample Index'
colors = plt.cm.tab10(np.linspace(0, 1, len(leads)))
for idx, lead in enumerate(leads):
if lead in df.columns:
rolling_avg = df[lead].rolling(window=window, min_periods=1).mean()
ax.plot(x_data, rolling_avg, label=f'Lead {lead} (MA-{window})',
linewidth=1.5, alpha=0.8, color=colors[idx])
ax.set_xlabel(x_label, fontsize=11)
ax.set_ylabel('Amplitude (mV)', fontsize=11)
ax.set_title(f'Rolling Average (Window={window})', fontsize=13, fontweight='bold')
ax.legend(loc='upper right', fontsize=9, ncol=min(4, len(leads)))
ax.grid(True, alpha=0.3, linestyle='--')
plt.tight_layout()
return ECGAnalyzer._fig_to_base64(fig)
@staticmethod
def create_all_visualizations(df: pd.DataFrame, leads: List[str],
viz_types: List[str]) -> str:
"""Create multiple visualizations based on selected types"""
html_parts = []
time_col = ECGAnalyzer.detect_time_column(df)
for viz_type in viz_types:
if viz_type == "Signal Waveform":
img_base64 = ECGAnalyzer.create_signal_plot(df, leads, time_col)
html_parts.append(f'<div style="margin-bottom: 30px;"><img src="data:image/png;base64,{img_base64}" style="max-width:100%;"/></div>')
elif viz_type == "Histogram":
img_base64 = ECGAnalyzer.create_histogram(df, leads)
html_parts.append(f'<div style="margin-bottom: 30px;"><img src="data:image/png;base64,{img_base64}" style="max-width:100%;"/></div>')
elif viz_type == "Scatter Plot":
# Use first two available leads for scatter plot
lead_x = leads[0] if len(leads) > 0 else 'I'
lead_y = leads[1] if len(leads) > 1 else 'II'
img_base64 = ECGAnalyzer.create_scatter_plot(df, lead_x, lead_y)
html_parts.append(f'<div style="margin-bottom: 30px;"><img src="data:image/png;base64,{img_base64}" style="max-width:100%;"/></div>')
elif viz_type == "Rolling Average":
img_base64 = ECGAnalyzer.create_rolling_average(df, leads, time_col)
html_parts.append(f'<div style="margin-bottom: 30px;"><img src="data:image/png;base64,{img_base64}" style="max-width:100%;"/></div>')
if not html_parts:
return '<div style="text-align:center; padding:40px;"><p>No visualizations selected</p></div>'
return '<div style="text-align:center;">' + ''.join(html_parts) + '</div>'
@staticmethod
def _fig_to_base64(fig) -> str:
"""Convert matplotlib figure to base64 string"""
buf = io.BytesIO()
fig.savefig(buf, format='png', dpi=100, bbox_inches='tight')
buf.seek(0)
img_base64 = base64.b64encode(buf.read()).decode('utf-8')
plt.close(fig)
return img_base64
@staticmethod
def generate_statistics(df: pd.DataFrame, leads: List[str]) -> Dict[str, Any]:
"""Generate statistical summary for selected leads"""
stats = {}
for lead in leads:
if lead in df.columns:
lead_data = df[lead].dropna()
stats[lead] = {
'mean': float(lead_data.mean()),
'std': float(lead_data.std()),
'min': float(lead_data.min()),
'max': float(lead_data.max()),
'median': float(lead_data.median()),
'q25': float(lead_data.quantile(0.25)),
'q75': float(lead_data.quantile(0.75))
}
return stats |