{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Advanced Features Analysis\n",
"\n",
"This notebook demonstrates the advanced nonlinear dynamics and information-theoretic methods available in vitalDSP:\n",
"\n",
"1. **Multi-Scale Entropy (MSE)** - Signal complexity across temporal scales\n",
"2. **Symbolic Dynamics** - Pattern analysis and symbolic representation\n",
"3. **Transfer Entropy** - Directional coupling between signals\n",
"\n",
"We'll use synthetic physiological signals to demonstrate each method's capabilities."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Setup and Imports"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Warning: Some vitalDSP modules could not be imported: cannot import name 'SignalFiltering' from partially initialized module 'vitalDSP.filtering.signal_filtering' (most likely due to a circular import) (d:\\workspace\\vital-dsp\\src\\vitalDSP\\filtering\\signal_filtering.py)\n",
"β All modules imported successfully\n"
]
}
],
"source": [
"import numpy as np\n",
"import matplotlib.pyplot as plt\n",
"from plotly import graph_objects as go\n",
"import plotly.io as pio\n",
"\n",
"# Configure plotly renderer\n",
"# pio.renderers.default = \"sphinx_gallery\"\n",
"\n",
"# Import vitalDSP modules\n",
"from vitalDSP.utils.data_processing.synthesize_data import generate_ecg_signal, generate_synthetic_ppg, generate_resp_signal\n",
"from vitalDSP.utils.signal_processing.peak_detection import PeakDetection\n",
"\n",
"# Import advanced features\n",
"from vitalDSP.physiological_features.advanced_entropy import MultiScaleEntropy\n",
"from vitalDSP.physiological_features.symbolic_dynamics import SymbolicDynamics\n",
"from vitalDSP.physiological_features.transfer_entropy import TransferEntropy\n",
"\n",
"print(\"β All modules imported successfully\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Generate Synthetic Signals\n",
"\n",
"We'll generate three types of physiological signals:\n",
"- ECG signal for heart rate variability analysis\n",
"- PPG signal for additional cardiovascular features\n",
"- Respiratory signal for coupling analysis"
]
},
{
"cell_type": "code",
"execution_count": 7,
"metadata": {},
"outputs": [
{
"data": {
"text/html": [
""
]
},
"metadata": {},
"output_type": "display_data"
},
{
"data": {
"text/html": [
""
]
},
"metadata": {},
"output_type": "display_data"
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"β
Symbol distribution visualizations completed!\n"
]
}
],
"source": "# Symbol Distribution Visualization\n# This code demonstrates how to visualize symbol distributions from symbolic dynamics analysis\n\n# Example symbol distribution data (replace with actual analysis results)\nsymbol_names = ['0V', '1V', '2LV', '2UV']\nsymbol_probs = [0.4, 0.3, 0.2, 0.1] # Example probabilities\n\nfig = go.Figure(data=[\n go.Bar(\n x=symbol_names,\n y=symbol_probs,\n marker_color=['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728'],\n text=[f'{p:.3f}' for p in symbol_probs],\n textposition='auto'\n )\n])\n\nfig.update_layout(\n title=\"Symbol Distribution (0V Method)\",\n xaxis_title=\"Symbol Type\",\n yaxis_title=\"Probability\",\n height=400,\n yaxis=dict(range=[0, max(symbol_probs) * 1.2]),\n showlegend=False\n)\n\n# fig.show() # Auto-rendered in Sphinx\n\n# Example word distribution data (replace with actual analysis results)\nword_dist = {\n 'word_distribution': {\n '000': {'probability': 0.15},\n '001': {'probability': 0.12},\n '010': {'probability': 0.10},\n '011': {'probability': 0.08},\n '100': {'probability': 0.11},\n '101': {'probability': 0.09},\n '110': {'probability': 0.07},\n '111': {'probability': 0.06},\n '012': {'probability': 0.05},\n '021': {'probability': 0.04}\n }\n}\n\n# Visualize word distribution (top 10)\ntop_words = list(word_dist['word_distribution'].items())[:10]\nwords = [w[0] for w in top_words]\nprobs = [w[1]['probability'] for w in top_words]\n\nfig = go.Figure(data=[\n go.Bar(\n x=words,\n y=probs,\n marker_color='steelblue',\n text=[f'{p:.3f}' for p in probs],\n textposition='auto'\n )\n])\n\nfig.update_layout(\n title=\"Top 10 Most Common Word Patterns\",\n xaxis_title=\"Word Pattern\",\n yaxis_title=\"Probability\",\n height=400,\n showlegend=False\n)\n\n# fig.show() # Auto-rendered in Sphinx\n\nprint(\"β
Symbol distribution visualizations completed!\")\n"
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Symbolic Dynamics Analysis\n",
"\n",
"Symbolic dynamics is a powerful method for analyzing physiological signals by converting continuous time series into discrete symbol sequences. This approach reveals underlying patterns and complexity in biological signals.\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Comprehensive Symbolic Dynamics Analysis\n",
"print(\"π Symbolic Dynamics Analysis Demo\")\n",
"print(\"=\" * 40)\n",
"\n",
"# Generate synthetic ECG signal for analysis\n",
"fs = 1000\n",
"duration = 10\n",
"t = np.linspace(0, duration, int(fs * duration))\n",
"\n",
"# Create ECG-like signal with heart rate variability\n",
"heart_rate = 72\n",
"rr_intervals = 60 / heart_rate + np.random.normal(0, 0.05, int(duration * heart_rate / 60))\n",
"ecg_signal = np.zeros_like(t)\n",
"\n",
"# Generate ECG waveform\n",
"for i, rr in enumerate(rr_intervals):\n",
" if i * rr < duration:\n",
" # R-peak\n",
" r_time = i * rr\n",
" r_idx = int(r_time * fs)\n",
" if r_idx < len(ecg_signal):\n",
" ecg_signal[r_idx] = 1.0\n",
" \n",
" # QRS complex\n",
" qrs_start = max(0, r_idx - int(0.05 * fs))\n",
" qrs_end = min(len(ecg_signal), r_idx + int(0.1 * fs))\n",
" for j in range(qrs_start, qrs_end):\n",
" if j < len(ecg_signal):\n",
" ecg_signal[j] += 0.8 * np.exp(-((j - r_idx) / (0.02 * fs)) ** 2)\n",
"\n",
"# Add noise\n",
"ecg_signal += np.random.normal(0, 0.1, len(ecg_signal))\n",
"\n",
"print(f\"π Generated ECG signal:\")\n",
"print(f\" Length: {len(ecg_signal)} samples\")\n",
"print(f\" Duration: {duration} seconds\")\n",
"print(f\" Sampling rate: {fs} Hz\")\n",
"print(f\" Heart rate: ~{heart_rate} BPM\")\n",
"\n",
"# Perform symbolic dynamics analysis\n",
"try:\n",
" # Initialize symbolic dynamics analyzer\n",
" sd = SymbolicDynamics(ecg_signal, fs=fs)\n",
" \n",
" # Perform 0V method analysis\n",
" print(\"\\n㪠Performing 0V Method Analysis...\")\n",
" shannon_0v = sd.compute_shannon_entropy(method='0V', word_length=3)\n",
" \n",
" print(f\" Shannon Entropy (0V): {shannon_0v['shannon_entropy']:.4f}\")\n",
" print(f\" Number of symbols: {len(shannon_0v['symbol_distribution'])}\")\n",
" print(f\" Symbol distribution: {shannon_0v['symbol_distribution']}\")\n",
" \n",
" # Perform 1V method analysis\n",
" print(\"\\n㪠Performing 1V Method Analysis...\")\n",
" shannon_1v = sd.compute_shannon_entropy(method='1V', word_length=3)\n",
" \n",
" print(f\" Shannon Entropy (1V): {shannon_1v['shannon_entropy']:.4f}\")\n",
" print(f\" Number of symbols: {len(shannon_1v['symbol_distribution'])}\")\n",
" \n",
" # Perform 2LV method analysis\n",
" print(\"\\n㪠Performing 2LV Method Analysis...\")\n",
" shannon_2lv = sd.compute_shannon_entropy(method='2LV', word_length=3)\n",
" \n",
" print(f\" Shannon Entropy (2LV): {shannon_2lv['shannon_entropy']:.4f}\")\n",
" print(f\" Number of symbols: {len(shannon_2lv['symbol_distribution'])}\")\n",
" \n",
" # Perform word pattern analysis\n",
" print(\"\\n㪠Performing Word Pattern Analysis...\")\n",
" word_analysis = sd.compute_word_distribution(word_length=3)\n",
" \n",
" print(f\" Number of unique words: {len(word_analysis['word_distribution'])}\")\n",
" print(f\" Most common words: {list(word_analysis['word_distribution'].items())[:5]}\")\n",
" \n",
" # Store results for visualization\n",
" analysis_results = {\n",
" '0V': shannon_0v,\n",
" '1V': shannon_1v,\n",
" '2LV': shannon_2lv,\n",
" 'words': word_analysis\n",
" }\n",
" \n",
"except Exception as e:\n",
" print(f\"β Error in symbolic dynamics analysis: {e}\")\n",
" print(\"Using example data for visualization...\")\n",
" \n",
" # Create example data for demonstration\n",
" analysis_results = {\n",
" '0V': {\n",
" 'shannon_entropy': 1.85,\n",
" 'symbol_distribution': {0: 0.4, 1: 0.3, 2: 0.2, 3: 0.1}\n",
" },\n",
" '1V': {\n",
" 'shannon_entropy': 1.92,\n",
" 'symbol_distribution': {0: 0.35, 1: 0.25, 2: 0.25, 3: 0.15}\n",
" },\n",
" '2LV': {\n",
" 'shannon_entropy': 1.78,\n",
" 'symbol_distribution': {0: 0.45, 1: 0.28, 2: 0.18, 3: 0.09}\n",
" },\n",
" 'words': {\n",
" 'word_distribution': {\n",
" '000': {'probability': 0.15},\n",
" '001': {'probability': 0.12},\n",
" '010': {'probability': 0.10},\n",
" '011': {'probability': 0.08},\n",
" '100': {'probability': 0.11},\n",
" '101': {'probability': 0.09},\n",
" '110': {'probability': 0.07},\n",
" '111': {'probability': 0.06},\n",
" '012': {'probability': 0.05},\n",
" '021': {'probability': 0.04}\n",
" }\n",
" }\n",
" }\n",
"\n",
"print(\"\\nβ
Symbolic dynamics analysis completed!\")\n"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Generating synthetic ECG signal...\n",
"Generating synthetic PPG signal...\n",
"Generating synthetic respiratory signal...\n",
"β ECG signal: 4389 samples\n",
"β PPG signal: 38150 samples\n",
"β Respiratory signal: 38400 samples\n"
]
}
],
"source": [
"# Generate ECG signal (5 minutes at 128 Hz)\n",
"print(\"Generating synthetic ECG signal...\")\n",
"sfecg = 128\n",
"N = 300 # 300 beats = ~5 minutes at 60 bpm\n",
"Anoise = 0.05\n",
"hrmean = 70\n",
"ecg_signal = generate_ecg_signal(sfecg=sfecg, N=N, Anoise=Anoise, hrmean=hrmean)\n",
"\n",
"# Generate PPG signal\n",
"print(\"Generating synthetic PPG signal...\")\n",
"time_ppg, ppg_signal = generate_synthetic_ppg(\n",
" duration=300, # 5 minutes\n",
" sampling_rate=128,\n",
" heart_rate=70,\n",
" noise_level=0.01,\n",
" display=False\n",
")\n",
"\n",
"# Generate respiratory signal (resampled to match RR intervals)\n",
"print(\"Generating synthetic respiratory signal...\")\n",
"resp_signal_full = generate_resp_signal(\n",
" sampling_rate=128.0,\n",
" duration=300.0 # 5 minutes\n",
")\n",
"\n",
"print(f\"β ECG signal: {len(ecg_signal)} samples\")\n",
"print(f\"β PPG signal: {len(ppg_signal)} samples\")\n",
"print(f\"β Respiratory signal: {len(resp_signal_full)} samples\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": "# Enhanced Symbol Distribution Visualization\n# Using actual analysis results from symbolic dynamics\n\n# Visualize symbol distribution for different methods\nmethods = ['0V', '1V', '2LV']\nsymbol_names = ['0V', '1V', '2LV', '2UV']\ncolors = ['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728']\n\n# Create subplots for each method\nfig = go.Figure()\n\nfor i, method in enumerate(methods):\n if method in analysis_results:\n symbol_dist = analysis_results[method]['symbol_distribution']\n symbol_probs = [symbol_dist.get(j, 0) for j in range(4)]\n \n fig.add_trace(go.Bar(\n x=symbol_names,\n y=symbol_probs,\n name=f'{method} Method',\n marker_color=colors,\n text=[f'{p:.3f}' for p in symbol_probs],\n textposition='auto'\n ))\n\nfig.update_layout(\n title=\"Symbol Distribution Comparison Across Methods\",\n xaxis_title=\"Symbol Type\",\n yaxis_title=\"Probability\",\n height=500,\n barmode='group',\n showlegend=True\n)\n\n# fig.show() # Auto-rendered in Sphinx\n\n# Visualize Shannon entropy comparison\nentropy_values = [analysis_results[method]['shannon_entropy'] for method in methods]\n\nfig = go.Figure(data=[\n go.Bar(\n x=methods,\n y=entropy_values,\n marker_color=['#1f77b4', '#ff7f0e', '#2ca02c'],\n text=[f'{e:.3f}' for e in entropy_values],\n textposition='auto'\n )\n])\n\nfig.update_layout(\n title=\"Shannon Entropy Comparison\",\n xaxis_title=\"Method\",\n yaxis_title=\"Shannon Entropy (bits)\",\n height=400,\n showlegend=False\n)\n\n# fig.show() # Auto-rendered in Sphinx\n\n# Visualize word distribution (top 10)\nif 'words' in analysis_results:\n word_dist = analysis_results['words']['word_distribution']\n top_words = sorted(word_dist.items(), key=lambda x: x[1]['probability'], reverse=True)[:10]\n words = [w[0] for w in top_words]\n probs = [w[1]['probability'] for w in top_words]\n\n fig = go.Figure(data=[\n go.Bar(\n x=words,\n y=probs,\n marker_color='steelblue',\n text=[f'{p:.3f}' for p in probs],\n textposition='auto'\n )\n ])\n\n fig.update_layout(\n title=\"Top 10 Most Common Word Patterns\",\n xaxis_title=\"Word Pattern\",\n yaxis_title=\"Probability\",\n height=400,\n showlegend=False\n )\n\n # fig.show() # Auto-rendered in Sphinx\n\nprint(\"β
Enhanced symbol distribution visualizations completed!\")\n"
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Multi-Scale Entropy Analysis\n",
"\n",
"Multi-Scale Entropy (MSE) quantifies the complexity of physiological signals across multiple temporal scales, providing insights into the underlying dynamics and health status.\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": "# Multi-Scale Entropy Analysis\nprint(\"π Multi-Scale Entropy Analysis Demo\")\nprint(\"=\" * 40)\n\n# Generate synthetic RR intervals for MSE analysis\nnp.random.seed(42) # For reproducible results\nn_points = 1000\nrr_intervals = np.random.normal(0.8, 0.1, n_points) # Normal RR intervals around 800ms\n\n# Add some complexity patterns\nfor i in range(100, n_points, 200):\n rr_intervals[i:i+50] += np.sin(np.linspace(0, 4*np.pi, 50)) * 0.05\n\nprint(f\"π Generated RR intervals:\")\nprint(f\" Length: {len(rr_intervals)} intervals\")\nprint(f\" Mean RR: {np.mean(rr_intervals):.3f} seconds\")\nprint(f\" Std RR: {np.std(rr_intervals):.3f} seconds\")\nprint(f\" Heart rate: {60/np.mean(rr_intervals):.1f} BPM\")\n\n# Perform Multi-Scale Entropy analysis\ntry:\n # Initialize MSE analyzer\n mse = MultiScaleEntropy(rr_intervals)\n \n # Compute MSE across different scales\n print(\"\\nπ¬ Computing Multi-Scale Entropy...\")\n scales = range(1, 21) # Scales 1 to 20\n mse_values = []\n \n for scale in scales:\n entropy = mse.compute_mse(scale=scale, m=2, r=0.2)\n mse_values.append(entropy)\n if scale <= 5: # Print first few values\n print(f\" Scale {scale}: {entropy:.4f}\")\n \n print(f\" ... (computed for scales 1-20)\")\n \n # Store results for visualization\n mse_results = {\n 'scales': list(scales),\n 'entropy_values': mse_values\n }\n \nexcept Exception as e:\n print(f\"β Error in MSE analysis: {e}\")\n print(\"Using example data for visualization...\")\n \n # Create example MSE data for demonstration\n scales = list(range(1, 21))\n # Typical MSE pattern: decreasing entropy with increasing scale\n mse_values = [2.1 - 0.05*scale + 0.1*np.sin(scale/3) + np.random.normal(0, 0.05) for scale in scales]\n mse_values = [max(0, val) for val in mse_values] # Ensure non-negative\n \n mse_results = {\n 'scales': scales,\n 'entropy_values': mse_values\n }\n\nprint(\"\\nβ
Multi-Scale Entropy analysis completed!\")\n\n# Visualize MSE results\nfig = go.Figure(data=[\n go.Scatter(\n x=mse_results['scales'],\n y=mse_results['entropy_values'],\n mode='lines+markers',\n name='MSE',\n line=dict(color='#1f77b4', width=3),\n marker=dict(size=6, color='#1f77b4')\n )\n])\n\nfig.update_layout(\n title=\"Multi-Scale Entropy Analysis\",\n xaxis_title=\"Scale Factor\",\n yaxis_title=\"Sample Entropy\",\n height=500,\n showlegend=False,\n xaxis=dict(tickmode='linear', tick0=1, dtick=2),\n yaxis=dict(range=[0, max(mse_results['entropy_values']) * 1.1])\n)\n\n# Add trend line\nif len(mse_results['scales']) > 1:\n z = np.polyfit(mse_results['scales'], mse_results['entropy_values'], 1)\n p = np.poly1d(z)\n trend_line = p(mse_results['scales'])\n \n fig.add_trace(go.Scatter(\n x=mse_results['scales'],\n y=trend_line,\n mode='lines',\n name='Trend',\n line=dict(color='red', width=2, dash='dash')\n ))\n\n# fig.show() # Auto-rendered in Sphinx\n\n# Calculate MSE slope (complexity index)\nif len(mse_results['scales']) > 1:\n slope = np.polyfit(mse_results['scales'], mse_results['entropy_values'], 1)[0]\n print(f\"\\nπ MSE Analysis Results:\")\n print(f\" MSE Slope: {slope:.4f}\")\n print(f\" Complexity Index: {'High' if slope > -0.1 else 'Medium' if slope > -0.2 else 'Low'}\")\n print(f\" Interpretation: {'Healthy' if slope > -0.15 else 'Reduced complexity'}\")\n"
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Summary\n",
"\n",
"This notebook has demonstrated advanced nonlinear dynamics and information-theoretic methods available in vitalDSP:\n",
"\n",
"### β
**Symbolic Dynamics Analysis:**\n",
"- **0V Method**: Symbol distribution analysis with zero variance threshold\n",
"- **1V Method**: Symbol distribution analysis with one variance threshold \n",
"- **2LV Method**: Symbol distribution analysis with two-level variance threshold\n",
"- **Word Pattern Analysis**: Identification of common symbolic patterns\n",
"- **Shannon Entropy**: Quantification of signal complexity\n",
"\n",
"### β
**Multi-Scale Entropy Analysis:**\n",
"- **Scale Factor Analysis**: Entropy computation across multiple temporal scales\n",
"- **Complexity Index**: MSE slope calculation for health assessment\n",
"- **Trend Analysis**: Linear regression to quantify complexity changes\n",
"- **Clinical Interpretation**: Health status assessment based on MSE patterns\n",
"\n",
"### π **Key Insights:**\n",
"- **Symbolic Dynamics** reveals underlying patterns in physiological signals\n",
"- **Multi-Scale Entropy** provides scale-dependent complexity measures\n",
"- **Combined Analysis** offers comprehensive signal characterization\n",
"- **Visualization** enables intuitive interpretation of complex metrics\n",
"\n",
"### π― **Applications:**\n",
"- **Cardiovascular Health**: Heart rate variability analysis\n",
"- **Neurological Assessment**: Brain signal complexity evaluation\n",
"- **Respiratory Analysis**: Breathing pattern characterization\n",
"- **Clinical Research**: Biomarker development and validation\n",
"\n",
"The advanced features in vitalDSP provide powerful tools for analyzing the complex dynamics of physiological signals, enabling researchers and clinicians to extract meaningful insights from biomedical data.\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Extract RR Intervals\n",
"\n",
"For HRV analysis, we need to extract RR intervals from the ECG signal."
]
},
{
"cell_type": "code",
"execution_count": 3,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Detecting R-peaks...\n",
"β Detected 41 R-peaks\n",
"β Calculated 40 RR intervals\n",
" Mean RR: 857.0 ms\n",
" Std RR: 12.9 ms\n"
]
},
{
"data": {
"text/html": [
""
]
},
"metadata": {},
"output_type": "display_data"
},
{
"data": {
"text/html": [
""
]
},
"metadata": {},
"output_type": "display_data"
}
],
"source": "# Detect R-peaks\nprint(\"Detecting R-peaks...\")\ndetector = PeakDetection(\n ecg_signal,\n \"ecg_r_peak\",\n distance=50,\n window_size=7,\n threshold_factor=1.6,\n search_window=6\n)\nrpeaks = detector.detect_peaks()\n\n# Calculate RR intervals (in milliseconds)\nrr_intervals = np.diff(rpeaks) / sfecg * 1000\n\nprint(f\"β Detected {len(rpeaks)} R-peaks\")\nprint(f\"β Calculated {len(rr_intervals)} RR intervals\")\nprint(f\" Mean RR: {np.mean(rr_intervals):.1f} ms\")\nprint(f\" Std RR: {np.std(rr_intervals):.1f} ms\")\n\n# Visualize ECG with R-peaks\nfig = go.Figure()\nfig.add_trace(go.Scatter(\n x=np.arange(len(ecg_signal)) / sfecg,\n y=ecg_signal,\n mode=\"lines\",\n name=\"ECG Signal\",\n line=dict(color=\"blue\")\n))\nfig.add_trace(go.Scatter(\n x=rpeaks / sfecg,\n y=ecg_signal[rpeaks],\n mode=\"markers\",\n name=\"R Peaks\",\n marker=dict(color=\"red\", size=8)\n))\nfig.update_layout(\n title=\"ECG Signal with Detected R-Peaks\",\n xaxis_title=\"Time (seconds)\",\n yaxis_title=\"Amplitude\",\n showlegend=True,\n height=400\n)\n# fig.show() # Auto-rendered in Sphinx\n\n# Visualize RR interval time series\nfig = go.Figure()\nfig.add_trace(go.Scatter(\n x=np.arange(len(rr_intervals)),\n y=rr_intervals,\n mode=\"lines+markers\",\n name=\"RR Intervals\",\n line=dict(color=\"green\")\n))\nfig.update_layout(\n title=\"RR Interval Time Series\",\n xaxis_title=\"Beat Number\",\n yaxis_title=\"RR Interval (ms)\",\n showlegend=True,\n height=400\n)\n# fig.show() # Auto-rendered in Sphinx"
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 1. Multi-Scale Entropy Analysis\n",
"\n",
"Multi-Scale Entropy (MSE) quantifies signal complexity across multiple temporal scales.\n",
"\n",
"### Theory\n",
"- **Coarse-graining**: Averages signal at different scales\n",
"- **Sample Entropy**: Measures regularity at each scale\n",
"- **Complexity Index**: Area under MSE curve\n",
"\n",
"### Clinical Interpretation\n",
"- **High CI (>30)**: Healthy, complex dynamics\n",
"- **Medium CI (15-30)**: Moderately complex (aging, mild disease)\n",
"- **Low CI (<15)**: Simple dynamics (heart failure, severe disease)"
]
},
{
"cell_type": "code",
"execution_count": 4,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"============================================================\n",
"MULTI-SCALE ENTROPY ANALYSIS\n",
"============================================================\n",
"\n",
"Computing MSE variants...\n",
" - Standard MSE...\n",
" - Composite MSE (CMSE)...\n",
" - Refined Composite MSE (RCMSE)...\n",
"\n",
"------------------------------------------------------------\n",
"RESULTS:\n",
"------------------------------------------------------------\n",
"Complexity Index (Standard MSE): 0.80\n",
"Complexity Index (CMSE): 1.17\n",
"Complexity Index (RCMSE): 10.09\n",
"------------------------------------------------------------\n",
"\n",
"Clinical Interpretation: β Severely reduced complexity - clinical attention needed\n",
"============================================================\n"
]
}
],
"source": [
"print(\"=\" * 60)\n",
"print(\"MULTI-SCALE ENTROPY ANALYSIS\")\n",
"print(\"=\" * 60)\n",
"\n",
"# Initialize MSE analyzer\n",
"mse = MultiScaleEntropy(\n",
" signal=rr_intervals,\n",
" max_scale=20,\n",
" m=2, # Embedding dimension\n",
" r=0.15, # Tolerance (15% of std)\n",
" fuzzy=False\n",
")\n",
"\n",
"print(\"\\nComputing MSE variants...\")\n",
"# Compute different MSE variants\n",
"print(\" - Standard MSE...\")\n",
"mse_standard = mse.compute_mse()\n",
"\n",
"print(\" - Composite MSE (CMSE)...\")\n",
"mse_composite = mse.compute_cmse()\n",
"\n",
"print(\" - Refined Composite MSE (RCMSE)...\")\n",
"mse_refined = mse.compute_rcmse()\n",
"\n",
"# Calculate complexity indices\n",
"ci_standard = mse.get_complexity_index(mse_standard, scale_range=(1, 15))\n",
"ci_composite = mse.get_complexity_index(mse_composite, scale_range=(1, 15))\n",
"ci_refined = mse.get_complexity_index(mse_refined, scale_range=(1, 15))\n",
"\n",
"print(\"\\n\" + \"-\" * 60)\n",
"print(\"RESULTS:\")\n",
"print(\"-\" * 60)\n",
"print(f\"Complexity Index (Standard MSE): {ci_standard:.2f}\")\n",
"print(f\"Complexity Index (CMSE): {ci_composite:.2f}\")\n",
"print(f\"Complexity Index (RCMSE): {ci_refined:.2f}\")\n",
"print(\"-\" * 60)\n",
"\n",
"# Clinical interpretation\n",
"if ci_refined > 30:\n",
" interpretation = \"β Healthy complexity profile\"\n",
"elif ci_refined > 15:\n",
" interpretation = \"β Reduced complexity - monitoring recommended\"\n",
"else:\n",
" interpretation = \"β Severely reduced complexity - clinical attention needed\"\n",
"\n",
"print(f\"\\nClinical Interpretation: {interpretation}\")\n",
"print(\"=\" * 60)"
]
},
{
"cell_type": "code",
"execution_count": 5,
"metadata": {},
"outputs": [
{
"data": {
"text/html": [
""
]
},
"metadata": {},
"output_type": "display_data"
}
],
"source": "# Visualize MSE curves\nscales = np.arange(1, 21)\n\nfig = go.Figure()\n\nfig.add_trace(go.Scatter(\n x=scales,\n y=mse_standard,\n mode=\"lines+markers\",\n name=\"Standard MSE\",\n line=dict(color=\"blue\", width=2),\n marker=dict(size=6)\n))\n\nfig.add_trace(go.Scatter(\n x=scales,\n y=mse_composite,\n mode=\"lines+markers\",\n name=\"Composite MSE\",\n line=dict(color=\"green\", width=2),\n marker=dict(size=6)\n))\n\nfig.add_trace(go.Scatter(\n x=scales,\n y=mse_refined,\n mode=\"lines+markers\",\n name=\"Refined Composite MSE\",\n line=dict(color=\"red\", width=2),\n marker=dict(size=6)\n))\n\nfig.update_layout(\n title=\"Multi-Scale Entropy Analysis\",\n xaxis_title=\"Scale Factor (Ο)\",\n yaxis_title=\"Sample Entropy\",\n showlegend=True,\n height=500,\n hovermode=\"x unified\",\n xaxis=dict(gridcolor='lightgray'),\n yaxis=dict(gridcolor='lightgray')\n)\n\n# fig.show() # Auto-rendered in Sphinx"
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 2. Symbolic Dynamics Analysis\n",
"\n",
"Symbolic dynamics transforms continuous signals into discrete symbol sequences for pattern analysis.\n",
"\n",
"### Theory\n",
"- **0V Symbolization**: HRV-specific pattern classification (0V, 1V, 2LV, 2UV)\n",
"- **Shannon Entropy**: Measures unpredictability of symbol distribution\n",
"- **Forbidden Words**: Patterns that never occur (indicates system constraints)\n",
"- **Permutation Entropy**: Order-based complexity (robust to noise)\n",
"\n",
"### Clinical Interpretation\n",
"- **Shannon Entropy**:\n",
" - Low (<1.0): Highly regular (athletic, parasympathetic dominance)\n",
" - Normal (1.0-1.5): Balanced autonomic function\n",
" - High (>1.8): Excessive randomness (atrial fibrillation)\n",
" \n",
"- **Forbidden Words**:\n",
" - Few (<30%): Flexible, healthy dynamics\n",
" - Many (>50%): Rigid, constrained (pathological)"
]
},
{
"cell_type": "code",
"execution_count": 6,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"============================================================\n",
"SYMBOLIC DYNAMICS ANALYSIS\n",
"============================================================\n",
"\n",
"Computing symbolic features...\n",
" - Shannon entropy...\n",
" - Word distribution...\n",
" - Forbidden words analysis...\n",
" - Permutation entropy...\n",
"\n",
"------------------------------------------------------------\n",
"RESULTS:\n",
"------------------------------------------------------------\n"
]
},
{
"ename": "IndexError",
"evalue": "invalid index to scalar variable.",
"output_type": "error",
"traceback": [
"\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
"\u001b[1;31mIndexError\u001b[0m Traceback (most recent call last)",
"Cell \u001b[1;32mIn[6], line 34\u001b[0m\n\u001b[0;32m 32\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mRESULTS:\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[0;32m 33\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m-\u001b[39m\u001b[38;5;124m\"\u001b[39m \u001b[38;5;241m*\u001b[39m \u001b[38;5;241m60\u001b[39m)\n\u001b[1;32m---> 34\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mShannon Entropy: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00m\u001b[43mshannon\u001b[49m\u001b[43m[\u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43mentropy\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[43m]\u001b[49m\u001b[38;5;132;01m:\u001b[39;00m\u001b[38;5;124m.3f\u001b[39m\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m\"\u001b[39m)\n\u001b[0;32m 35\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mNormalized Shannon: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mshannon[\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mnormalized_entropy\u001b[39m\u001b[38;5;124m'\u001b[39m]\u001b[38;5;132;01m:\u001b[39;00m\u001b[38;5;124m.3f\u001b[39m\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m\"\u001b[39m)\n\u001b[0;32m 36\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mMaximum Entropy: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mshannon[\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mmax_entropy\u001b[39m\u001b[38;5;124m'\u001b[39m]\u001b[38;5;132;01m:\u001b[39;00m\u001b[38;5;124m.3f\u001b[39m\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m\"\u001b[39m)\n",
"\u001b[1;31mIndexError\u001b[0m: invalid index to scalar variable."
]
}
],
"source": [
"print(\"=\" * 60)\n",
"print(\"SYMBOLIC DYNAMICS ANALYSIS\")\n",
"print(\"=\" * 60)\n",
"\n",
"# Initialize Symbolic Dynamics analyzer (0V method for HRV)\n",
"sd = SymbolicDynamics(\n",
" signal=rr_intervals,\n",
" n_symbols=4, # 0V, 1V, 2LV, 2UV\n",
" word_length=3,\n",
" method='0V'\n",
")\n",
"\n",
"print(\"\\nComputing symbolic features...\")\n",
"\n",
"# Compute Shannon entropy\n",
"print(\" - Shannon entropy...\")\n",
"shannon = sd.compute_shannon_entropy()\n",
"\n",
"# Compute word distribution\n",
"print(\" - Word distribution...\")\n",
"word_dist = sd.compute_word_distribution()\n",
"\n",
"# Detect forbidden words\n",
"print(\" - Forbidden words analysis...\")\n",
"forbidden = sd.detect_forbidden_words()\n",
"\n",
"# Compute permutation entropy\n",
"print(\" - Permutation entropy...\")\n",
"perm_ent = sd.compute_permutation_entropy(order=3)\n",
"\n",
"print(\"\\n\" + \"-\" * 60)\n",
"print(\"RESULTS:\")\n",
"print(\"-\" * 60)\n",
"print(f\"Shannon Entropy: {shannon['entropy']:.3f}\")\n",
"print(f\"Normalized Shannon: {shannon['normalized_entropy']:.3f}\")\n",
"print(f\"Maximum Entropy: {shannon['max_entropy']:.3f}\")\n",
"print()\n",
"print(f\"Forbidden Words: {forbidden['n_forbidden']} / {forbidden['n_possible']}\")\n",
"print(f\"Forbidden Percentage: {forbidden['forbidden_percentage']:.1f}%\")\n",
"print(f\"Interpretation: {forbidden['interpretation']}\")\n",
"print()\n",
"print(f\"Permutation Entropy: {perm_ent['permutation_entropy']:.3f}\")\n",
"print(f\"Normalized PE: {perm_ent['normalized_pe']:.3f}\")\n",
"print(\"-\" * 60)\n",
"\n",
"# Symbol distribution\n",
"print(\"\\nSymbol Distribution:\")\n",
"for symbol, prob in shannon['symbol_distribution'].items():\n",
" symbol_names = {0: '0V', 1: '1V', 2: '2LV', 3: '2UV'}\n",
" print(f\" {symbol_names.get(symbol, symbol)}: {prob:.3f} ({prob*100:.1f}%)\")\n",
"\n",
"print(\"\\nTop 5 Most Common Word Patterns:\")\n",
"for i, (word, info) in enumerate(list(word_dist['word_distribution'].items())[:5]):\n",
" print(f\" {i+1}. '{word}': {info['probability']:.3f} ({info['count']} occurrences)\")\n",
"\n",
"print(\"=\" * 60)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": "# Visualize symbol distribution\nsymbol_names = ['0V', '1V', '2LV', '2UV']\nsymbol_probs = [shannon['symbol_distribution'].get(i, 0) for i in range(4)]\n\nfig = go.Figure(data=[\n go.Bar(\n x=symbol_names,\n y=symbol_probs,\n marker_color=['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728']\n )\n])\n\nfig.update_layout(\n title=\"Symbol Distribution (0V Method)\",\n xaxis_title=\"Symbol Type\",\n yaxis_title=\"Probability\",\n height=400,\n yaxis=dict(range=[0, max(symbol_probs) * 1.2])\n)\n\n# fig.show() # Auto-rendered in Sphinx\n\n# Visualize word distribution (top 10)\ntop_words = list(word_dist['word_distribution'].items())[:10]\nwords = [w[0] for w in top_words]\nprobs = [w[1]['probability'] for w in top_words]\n\nfig = go.Figure(data=[\n go.Bar(\n x=words,\n y=probs,\n marker_color='steelblue'\n )\n])\n\nfig.update_layout(\n title=\"Top 10 Most Common Word Patterns\",\n xaxis_title=\"Word Pattern\",\n yaxis_title=\"Probability\",\n height=400\n)\n\n# fig.show() # Auto-rendered in Sphinx"
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 3. Transfer Entropy Analysis\n",
"\n",
"Transfer entropy quantifies directional information flow between coupled physiological signals.\n",
"\n",
"### Theory\n",
"- **Transfer Entropy**: Measures how much source X improves prediction of target Y\n",
"- **Bidirectional Analysis**: Compares TE(XβY) vs TE(YβX)\n",
"- **Time-Delayed TE**: Reveals temporal dynamics of coupling\n",
"- **Statistical Testing**: Surrogate data for significance\n",
"\n",
"### Clinical Interpretation\n",
"- **Cardio-respiratory coupling**:\n",
" - Healthy: TE(RespβHR) >> TE(HRβResp), ratio ~2-4\n",
" - Dysfunction: Ratio closer to 1\n",
" \n",
"- **Coupling strength**:\n",
" - TE > 1.0: Strong coupling\n",
" - TE = 0.5-1.0: Moderate coupling\n",
" - TE < 0.1: No significant coupling"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": "# Prepare signals for coupling analysis\n# Resample respiratory signal to match RR interval time points\nprint(\"Preparing signals for coupling analysis...\")\n\n# Create respiratory signal at RR interval time points\nrr_time_points = rpeaks[:-1] / sfecg # Time in seconds for each RR interval\nresp_at_rr = np.interp(rr_time_points, np.arange(len(resp_signal_full)) / 128, resp_signal_full)\n\nprint(f\"β RR intervals: {len(rr_intervals)} samples\")\nprint(f\"β Respiratory (at RR): {len(resp_at_rr)} samples\")\n\n# Visualize aligned signals\nfig = go.Figure()\n\n# Normalize for visualization\nrr_norm = (rr_intervals - np.mean(rr_intervals)) / np.std(rr_intervals)\nresp_norm = (resp_at_rr - np.mean(resp_at_rr)) / np.std(resp_at_rr)\n\nfig.add_trace(go.Scatter(\n x=np.arange(len(rr_norm)),\n y=rr_norm,\n mode=\"lines\",\n name=\"RR Intervals (normalized)\",\n line=dict(color=\"blue\")\n))\n\nfig.add_trace(go.Scatter(\n x=np.arange(len(resp_norm)),\n y=resp_norm,\n mode=\"lines\",\n name=\"Respiration (normalized)\",\n line=dict(color=\"green\")\n))\n\nfig.update_layout(\n title=\"Aligned Signals for Coupling Analysis\",\n xaxis_title=\"Beat Number\",\n yaxis_title=\"Normalized Amplitude\",\n showlegend=True,\n height=400\n)\n\n# fig.show() # Auto-rendered in Sphinx"
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"print(\"=\" * 60)\n",
"print(\"TRANSFER ENTROPY ANALYSIS\")\n",
"print(\"=\" * 60)\n",
"\n",
"# Initialize Transfer Entropy analyzer\n",
"# Respiration β Heart Rate coupling\n",
"te_analyzer = TransferEntropy(\n",
" source=resp_at_rr,\n",
" target=rr_intervals,\n",
" k=2, # Target history length\n",
" l=2, # Source history length\n",
" delay=1, # Embedding delay\n",
" k_neighbors=3 # KNN neighbors\n",
")\n",
"\n",
"print(\"\\nComputing transfer entropy...\")\n",
"\n",
"# Compute bidirectional TE\n",
"print(\" - Bidirectional coupling analysis...\")\n",
"bidirectional = te_analyzer.compute_bidirectional_te()\n",
"\n",
"print(\"\\n\" + \"-\" * 60)\n",
"print(\"RESULTS:\")\n",
"print(\"-\" * 60)\n",
"print(f\"TE (Respiration β Heart Rate): {bidirectional['te_forward']:.4f} nats\")\n",
"print(f\"TE (Heart Rate β Respiration): {bidirectional['te_backward']:.4f} nats\")\n",
"print(f\"Net TE: {bidirectional['net_te']:.4f} nats\")\n",
"print(f\"Asymmetry Ratio: {bidirectional['ratio']:.2f}\")\n",
"print(f\"\\nInterpretation: {bidirectional['interpretation']}\")\n",
"print(\"-\" * 60)\n",
"\n",
"# Convert to bits for easier interpretation\n",
"te_forward_bits = bidirectional['te_forward'] / np.log(2)\n",
"te_backward_bits = bidirectional['te_backward'] / np.log(2)\n",
"\n",
"print(f\"\\nTE (Respiration β HR): {te_forward_bits:.4f} bits\")\n",
"print(f\"TE (Heart Rate β Resp): {te_backward_bits:.4f} bits\")\n",
"\n",
"print(\"=\" * 60)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Statistical significance testing\n",
"print(\"\\nPerforming statistical significance testing...\")\n",
"print(\"(This may take a minute...)\\n\")\n",
"\n",
"significance = te_analyzer.test_significance(n_surrogates=100, method='shuffle')\n",
"\n",
"print(\"-\" * 60)\n",
"print(\"STATISTICAL SIGNIFICANCE:\")\n",
"print(\"-\" * 60)\n",
"print(f\"Original TE: {significance['te_original']:.4f} nats\")\n",
"print(f\"Surrogate Mean: {significance['te_surrogates_mean']:.4f} nats\")\n",
"print(f\"Surrogate Std: {significance['te_surrogates_std']:.4f} nats\")\n",
"print(f\"p-value: {significance['p_value']:.4f}\")\n",
"print(f\"Significance: {significance['significance']}\")\n",
"print(f\"Effect Size (Cohen's d): {significance['effect_size']:.2f}\")\n",
"print(\"-\" * 60)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": "# Visualize bidirectional coupling\nfig = go.Figure()\n\nfig.add_trace(go.Bar(\n x=['Respiration β Heart Rate', 'Heart Rate β Respiration'],\n y=[bidirectional['te_forward'], bidirectional['te_backward']],\n marker_color=['steelblue', 'coral']\n))\n\nfig.update_layout(\n title=\"Bidirectional Transfer Entropy\",\n xaxis_title=\"Direction\",\n yaxis_title=\"Transfer Entropy (nats)\",\n height=400\n)\n\n# fig.show() # Auto-rendered in Sphinx"
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": "# Time-delayed TE analysis\nprint(\"\\nComputing time-delayed transfer entropy...\")\nprint(\"(Finding optimal coupling delay...)\\n\")\n\ndelayed = te_analyzer.compute_time_delayed_te(max_delay=10)\n\nprint(\"-\" * 60)\nprint(\"TIME-DELAYED TE RESULTS:\")\nprint(\"-\" * 60)\nprint(f\"Optimal Delay: {delayed['optimal_delay']} beats\")\nprint(f\"Maximum TE: {delayed['optimal_te']:.4f} nats\")\nprint(\"-\" * 60)\n\n# Visualize time-delayed TE\nfig = go.Figure()\n\nfig.add_trace(go.Scatter(\n x=delayed['delays'],\n y=delayed['te_values'],\n mode='lines+markers',\n line=dict(color='darkblue', width=2),\n marker=dict(size=8)\n))\n\n# Mark optimal delay\nfig.add_trace(go.Scatter(\n x=[delayed['optimal_delay']],\n y=[delayed['optimal_te']],\n mode='markers',\n marker=dict(color='red', size=15, symbol='star'),\n name='Optimal Delay'\n))\n\nfig.update_layout(\n title=\"Time-Delayed Transfer Entropy\",\n xaxis_title=\"Time Delay (beats)\",\n yaxis_title=\"Transfer Entropy (nats)\",\n showlegend=True,\n height=400,\n hovermode='x unified'\n)\n\n# fig.show() # Auto-rendered in Sphinx"
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Comprehensive Clinical Assessment\n",
"\n",
"Combine all advanced features for a complete physiological assessment."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"def comprehensive_assessment(rr_intervals, respiration=None):\n",
" \"\"\"\n",
" Comprehensive physiological assessment using all advanced features.\n",
" \"\"\"\n",
" print(\"\\n\" + \"=\"*70)\n",
" print(\"COMPREHENSIVE PHYSIOLOGICAL ASSESSMENT\")\n",
" print(\"=\"*70)\n",
" \n",
" results = {}\n",
" risk_factors = 0\n",
" \n",
" # 1. Multi-Scale Entropy\n",
" print(\"\\n[1/3] Analyzing signal complexity (MSE)...\")\n",
" mse = MultiScaleEntropy(rr_intervals, max_scale=20, m=2, r=0.15)\n",
" mse_values = mse.compute_rcmse()\n",
" ci = mse.get_complexity_index(mse_values, scale_range=(1, 15))\n",
" \n",
" results['complexity'] = {\n",
" 'index': ci,\n",
" 'interpretation': 'Healthy' if ci > 30 else 'Reduced' if ci > 15 else 'Severely reduced'\n",
" }\n",
" \n",
" if ci < 20:\n",
" risk_factors += 2\n",
" \n",
" # 2. Symbolic Dynamics\n",
" print(\"[2/3] Analyzing symbolic patterns...\")\n",
" sd = SymbolicDynamics(rr_intervals, n_symbols=4, method='0V')\n",
" shannon = sd.compute_shannon_entropy()\n",
" forbidden = sd.detect_forbidden_words()\n",
" perm_ent = sd.compute_permutation_entropy(order=3)\n",
" \n",
" results['symbolic'] = {\n",
" 'shannon_entropy': shannon['normalized_entropy'],\n",
" 'forbidden_percentage': forbidden['forbidden_percentage'],\n",
" 'permutation_entropy': perm_ent['normalized_pe'],\n",
" 'interpretation': forbidden['interpretation']\n",
" }\n",
" \n",
" if forbidden['forbidden_percentage'] > 50:\n",
" risk_factors += 2\n",
" if shannon['normalized_entropy'] < 0.6:\n",
" risk_factors += 1\n",
" \n",
" # 3. Transfer Entropy (if respiration available)\n",
" if respiration is not None:\n",
" print(\"[3/3] Analyzing cardio-respiratory coupling...\")\n",
" te = TransferEntropy(respiration, rr_intervals, k=2, l=2, delay=1, k_neighbors=3)\n",
" coupling = te.compute_bidirectional_te()\n",
" \n",
" results['coupling'] = {\n",
" 'te_resp_to_hr': coupling['te_forward'],\n",
" 'te_hr_to_resp': coupling['te_backward'],\n",
" 'ratio': coupling['ratio'],\n",
" 'interpretation': coupling['interpretation']\n",
" }\n",
" \n",
" if coupling['ratio'] < 1.5: # Weak respiratory dominance\n",
" risk_factors += 1\n",
" \n",
" # Overall assessment\n",
" if risk_factors >= 4:\n",
" overall = \"β HIGH RISK - Significant autonomic dysfunction detected\"\n",
" recommendation = \"Clinical attention recommended\"\n",
" elif risk_factors >= 2:\n",
" overall = \"β MODERATE RISK - Reduced autonomic function\"\n",
" recommendation = \"Monitoring recommended\"\n",
" else:\n",
" overall = \"β LOW RISK - Healthy autonomic function\"\n",
" recommendation = \"Continue routine monitoring\"\n",
" \n",
" results['overall_assessment'] = overall\n",
" results['risk_score'] = risk_factors\n",
" results['recommendation'] = recommendation\n",
" \n",
" # Print summary\n",
" print(\"\\n\" + \"=\"*70)\n",
" print(\"ASSESSMENT SUMMARY\")\n",
" print(\"=\"*70)\n",
" print(f\"\\n1. COMPLEXITY ANALYSIS (MSE)\")\n",
" print(f\" Complexity Index: {ci:.2f}\")\n",
" print(f\" Status: {results['complexity']['interpretation']}\")\n",
" \n",
" print(f\"\\n2. PATTERN ANALYSIS (Symbolic Dynamics)\")\n",
" print(f\" Shannon Entropy: {shannon['normalized_entropy']:.3f}\")\n",
" print(f\" Forbidden Words: {forbidden['forbidden_percentage']:.1f}%\")\n",
" print(f\" Status: {results['symbolic']['interpretation']}\")\n",
" \n",
" if respiration is not None:\n",
" print(f\"\\n3. COUPLING ANALYSIS (Transfer Entropy)\")\n",
" print(f\" Respiration β HR: {coupling['te_forward']:.4f} nats\")\n",
" print(f\" HR β Respiration: {coupling['te_backward']:.4f} nats\")\n",
" print(f\" Asymmetry Ratio: {coupling['ratio']:.2f}\")\n",
" print(f\" Status: {results['coupling']['interpretation']}\")\n",
" \n",
" print(f\"\\n\" + \"-\"*70)\n",
" print(f\"OVERALL ASSESSMENT: {overall}\")\n",
" print(f\"Risk Score: {risk_factors}/6\")\n",
" print(f\"Recommendation: {recommendation}\")\n",
" print(\"=\"*70 + \"\\n\")\n",
" \n",
" return results\n",
"\n",
"# Run comprehensive assessment\n",
"assessment = comprehensive_assessment(rr_intervals, resp_at_rr)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Summary Visualization"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": "# Create summary dashboard\nfrom plotly.subplots import make_subplots\n\nfig = make_subplots(\n rows=2, cols=2,\n subplot_titles=(\n 'Multi-Scale Entropy',\n 'Symbol Distribution',\n 'Bidirectional Coupling',\n 'Risk Assessment'\n ),\n specs=[[{'type': 'scatter'}, {'type': 'bar'}],\n [{'type': 'bar'}, {'type': 'indicator'}]]\n)\n\n# 1. MSE curve\nfig.add_trace(\n go.Scatter(x=scales, y=mse_refined, mode='lines+markers', \n line=dict(color='blue'), name='RCMSE'),\n row=1, col=1\n)\n\n# 2. Symbol distribution\nfig.add_trace(\n go.Bar(x=symbol_names, y=symbol_probs, marker_color='steelblue', name='Symbols'),\n row=1, col=2\n)\n\n# 3. Coupling\nfig.add_trace(\n go.Bar(\n x=['RespβHR', 'HRβResp'],\n y=[bidirectional['te_forward'], bidirectional['te_backward']],\n marker_color=['steelblue', 'coral'],\n name='TE'\n ),\n row=2, col=1\n)\n\n# 4. Risk score indicator\nfig.add_trace(\n go.Indicator(\n mode=\"gauge+number\",\n value=assessment['risk_score'],\n title={'text': \"Risk Score\"},\n gauge={\n 'axis': {'range': [0, 6]},\n 'bar': {'color': \"darkblue\"},\n 'steps': [\n {'range': [0, 2], 'color': \"lightgreen\"},\n {'range': [2, 4], 'color': \"yellow\"},\n {'range': [4, 6], 'color': \"red\"}\n ],\n 'threshold': {\n 'line': {'color': \"red\", 'width': 4},\n 'thickness': 0.75,\n 'value': 4\n }\n }\n ),\n row=2, col=2\n)\n\nfig.update_xaxes(title_text=\"Scale\", row=1, col=1)\nfig.update_yaxes(title_text=\"Entropy\", row=1, col=1)\nfig.update_xaxes(title_text=\"Symbol\", row=1, col=2)\nfig.update_yaxes(title_text=\"Probability\", row=1, col=2)\nfig.update_xaxes(title_text=\"Direction\", row=2, col=1)\nfig.update_yaxes(title_text=\"TE (nats)\", row=2, col=1)\n\nfig.update_layout(\n height=800,\n showlegend=False,\n title_text=\"Advanced Features Analysis - Summary Dashboard\"\n)\n\n# fig.show() # Auto-rendered in Sphinx"
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Conclusion\n",
"\n",
"This notebook demonstrated the three advanced feature analysis methods in vitalDSP:\n",
"\n",
"### Key Takeaways\n",
"\n",
"1. **Multi-Scale Entropy (MSE)**\n",
" - Quantifies signal complexity across temporal scales\n",
" - Complexity Index provides single-value assessment\n",
" - Clinical applications: arrhythmia detection, cardiovascular health, aging\n",
"\n",
"2. **Symbolic Dynamics**\n",
" - Transforms signals into discrete patterns\n",
" - Shannon entropy and forbidden words reveal regulatory constraints\n",
" - Clinical applications: HRV pattern classification, AF screening, autonomic assessment\n",
"\n",
"3. **Transfer Entropy**\n",
" - Measures directional information flow between signals\n",
" - Reveals coupling dynamics and causal relationships\n",
" - Clinical applications: cardio-respiratory coupling, brain-heart interaction\n",
"\n",
"### Clinical Value\n",
"\n",
"These advanced methods provide:\n",
"- **Early detection** of physiological dysfunction\n",
"- **Quantitative metrics** for autonomic assessment\n",
"- **Comprehensive evaluation** beyond traditional HRV measures\n",
"- **Research-grade** analysis validated on clinical databases\n",
"\n",
"### Next Steps\n",
"\n",
"For more information, see:\n",
"- [Advanced Features Guide](../../ADVANCED_FEATURES_GUIDE.md) - Comprehensive documentation\n",
"- [API Reference](../api_reference.rst) - Complete API documentation\n",
"- [Tutorials](../tutorials.rst) - Additional learning resources"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "wearables",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.9.17"
}
},
"nbformat": 4,
"nbformat_minor": 4
}