# Project export: CogniSync
This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.
## Project metadata
- Hackathon: TreeHacks 2025
- Tagline: Transforming elderly care through voice rhythm and stability analysis, detecting cognitive decline before visible symptoms appear and enabling proactive medical intervention to preserve brain health.
- Devpost: https://devpost.com/software/cognisync
- GitHub: https://github.com/Sanskriti-Slngh/CognitiveApp
- Video: https://www.youtube.com/embed/8fIqQKHXviI?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 1 GitHub contributor(s) — Sanskriti Singh (15 commits)
## Devpost submission (written by the team)
### Inspiration
We are faced by a stark reality: 1 in 10 adults aged 45 and older experience worsening memory loss or confusion - early signs that often go unnoticed. What's most concerning is that these subtle changes can appear up to 10 years before becoming obvious symptoms. This gap between early signs and diagnosis is what inspired us to create CogniSync. Core Technology We developed a wearable device that serves as the foundation of our system. Using an integrated IMU sensor, it captures two critical types of data that provide early insights into cognitive health. Movement Analysis The device continuously tracks movement patterns through its integrated sensors. By processing acceleration, gyroscope, and magnetometer readings, we calculate stability indices from acceleration variance and measure movement efficiency through jerk analysis. These measurements allow us to identify early concerns in tremor patterns, gait stability, and fall risk assessment, providing early warnings of potential cognitive decline. Voice Analysis The same wearable device captures speech patterns, which research shows can be one of the earliest indicators of cognitive changes. Our analysis examines voice stability, where we look for readings above 65% for normal function. We also track attention patterns through speech rhythm, aiming for scores above 80%, and monitor memory indicators through processing speed, targeting above 90%. Key metrics like mean pause duration are measured against clinical thresholds, with 1.2 seconds being a crucial marker, while beta power measurements are compared to a 5.465 threshold for memory function assessment. Cognitive Assessment To complement the wearable, we developed engaging cognitive games that transform traditional assessments into interactive experiences. Our suite includes memory recall exercises that dynamically adjust memorization times, reaction-based challenges that introduce random variations, and pattern recognition tasks that scale in complexity. These games continuously adapt to player performance, ensuring both engagement and accurate assessment. By combining game performance with our wearable data, we aim to create a comprehensive picture of cognitive health that can detect subtle changes before they become apparent symptoms. The results from all three components - movement, voice, and cognitive games - are displayed through a dashboard that transforms these metrics into clear, actionable insights, enabling earlier intervention when it matters most. How we Built It We integrated our wearable device's data streaming through TCP/IP sockets with a Python backend. For data analysis, we used NumPy for statistical computations of movement metrics like stability indices and jerk calculations. The voice analysis pipeline utilized pre-trained transformer models from Hugging Face for pattern recognition and emotion classification. We built our interactive dashboard using Streamlit and Plotly, creating real-time visualizations of sensor data and health metrics. The system stores user interaction data and cognitive game results in JSON format for persistent tracking and analysis. Challenges We Ran Into Our biggest challenges came from processing multiple data streams simultaneously and extracting meaningful data from voice patterns. Managing continuous sensor input while running voice analysis models required significant performance optimization. We also dealt with calibrating accurate health metric thresholds, especially aligning our voice stability and movement analysis with clinical standards. In overcoming these challenges, we were able to transform raw sensor and voice data into meaningful health insights through our monitoring platform.
### Accomplishments we're proud of
Our biggest achievement was successfully integrating complex health monitoring into a single, intuitive interface that makes data accessible and meaningful. We're also particularly proud of our selection of cognitive games, intended to balance assessment and engagement.
### What we learned
The development process taught us crucial lessons about balancing real-time data processing with system performance, while maintaining accuracy. We discovered the importance of translating complex clinical metrics into user-friendly insights that anyone can understand. Most importantly, we learned how continuous sensor data, when properly analyzed, can reveal subtle patterns that indicate significant health trends.
### What's next
We plan to expand our system by incorporating more sophisticated health metrics, developing additional cognitive games, and refining our threshold detection algorithms based on clinical research. We're also exploring ways to make our platform more accessible through mobile development and improved data visualization techniques.
## README (from the GitHub repository)
## Install Dependencies:
This repository contains a web application for analyzing cognitive health using multiple modalities including speech, movement (IMU sensor data), and interactive cognitive games. The app is built with [Streamlit](https://streamlit.io/) and includes data analysis using machine learning pipelines, sensor data visualization, and a performance tracker for cognitive games.
## Repository Structure
├── static
│ ├── script.js
│ └── style.css
├── imu_client │
├── imu_client.h
│ └── imu_client.ino
├── imu_server
│ ├── imu_server.py
│ └── model_server.py
├── analyze_sensor_data.py
├── cognitive_files.html
├── demo.py
├── diagnostic_model.py
├── feature_extractor.py
├── index.html
├── sensor_metrics.json
└── README.md
## Features
- **Voice Analysis:**
Analyze speech features (e.g., voice stability, pause count, frequency, etc.) and generate advanced neural metrics and clinical insights.
- **Movement Analysis:**
Real-time IMU sensor data collection and visualization including movement stability and smoothness metrics.
- **Sensor Data Analysis Report:**
Process static sensor data from JSON, visualize technical summaries and structured sensor data, and highlight detected situations with visual components.
- **Cognitive Games & Performance Tracker:**
Play cognitive games and track user performance across multiple games.
## Setup and Installation
1. **Clone the Repository:**
```bash
git clone https://github.com/your-username/your-repository.git
cd your-repository
```
2. **Install Dependencies:**
Ensure you have Python 3.7 or later installed. Install the required packages using pip:
```bash
pip install streamlit plotly transformers torch numpy
```
Optionally, if you plan to expose the app publicly (e.g., running in Google Colab), install [pyngrok](https://pypi.org/project/pyngrok/):
```bash
pip install pyngrok
```
3. **(Optional) Set Up Ngrok:**
If you wish to expose the app publicly, configure Ngrok by obtaining a token from [ngrok.com](https://ngrok.com/) and following their documentation.
## Running the Application
### Running the Streamlit App
To start the main application, run:
```bash
streamlit run demo.py
```
This command launches the Streamlit app locally (by default on port `8501`). To expose it publicly using Ngrok, you can use the following snippet:
```python
from pyngrok import ngrok
import os
# Run Streamlit in the background
os.system("streamlit run demo.py &")
# Connect Ngrok tunnel to the default Streamlit port (8501)
public_url = ngrok.connect(8501)
print("Streamlit Public URL:", public_url)
```
### Running the Cognitive Games (Optional)
The cognitive games interface is integrated into the Streamlit app in the "Cognitive Games" tab. If you prefer to run a standalone version (e.g., using `index.html`), you can serve it using Python's HTTP server:
```bash
python -m http.server 8000
```
Then navigate to `http://localhost:8000` in your browser.
## File Descriptions
- **analyze_sensor_data.py:**
Contains the `SensorAnalyzer` class for processing raw sensor data, generating technical summaries, performing emotion analysis, detecting movement situations (e.g., tremors, falls), and visualizing sensor data.
- **diagnostic_model.py:**
Implements a diagnostic model that uses extracted speech features to generate clinical insights and risk assessments.
- **feature_extractor.py:**
Extracts various features from a provided audio file for voice analysis.
- **demo.py:**
The main Streamlit application that integrates voice analysis, movement analysis, sensor data reports, and cognitive games into a unified web interface.
- **sensor_metrics.json:**
A sample JSON file containing sensor metrics data used for sensor data analysis.
- **static/**
Contains static assets (CSS and JS) used by the cognitive games interface.
- **cognitive_files.html:**
HTML for the cognitive games interface that is inlined within the Streamlit app.
- **index.html:**
A standalone HTML page (if needed) for displaying the cognitive games interface.
## Troubleshooting
- **Ngrok Issues:**
If you experience issues with Ngrok (e.g., `ERR_NGROK_3200`), ensure that your Ngrok token is properly configured and that no conflicting tunnels are active.
- **Localhost Conflicts:**
If you receive errors regarding ports already in use, ensure that no other applications are running on the specified ports (`8501` for Streamlit and `8000` for the HTTP server).
## Contact
For any questions or feedback, please open an issue or contact [ssingh7@mit.edu](mailto:ssingh7@mit.edu).
## Detected evidence (automated analysis)
Indexed codebase: 13 recognized source files, 118 KB.
- C (language) — detected in the code
- CSS (language) — detected in the code
- HTML (language) — detected in the code
- JavaScript (language) — detected in the code
- Python (language) — detected in the code
- Streamlit (technology) — claimed on Devpost, not found in the code
## Codebase structure (from repository index)
### Files (15 of 15)
```
analyze_sensor_data.py
cognitive_files.html
cognitve_app.py
demo.py
diagnostic_model.py
feature_extractor.py
imu_client/IMU_client.h
imu_client/IMU_Client.ino
imu_server/IMU_server.py
imu_server/model_server.py
index.html
README.md
sensor_metrics.json
static/script.js
static/style.css
```
### Dependencies
No dependency index available.
### Recent commits (newest first)
- Add files via upload
- Update README.md
- Add files via upload
- Create IMU_client.h
- Create model_server.py
- Create IMU_server.py
- Update README.md
- Update README.md
- Update README.md
- Delete script.js
- Delete style.css
- Create style.css
- Create script.js
- Add files via upload
- Initial commit
## Key source files (fetched from GitHub, selected and truncated for size)
### cognitve_app.py
```python
# -*- coding: utf-8 -*-
"""cognitve app.ipynb
Automatically generated by Colab.
Original file is located at
https://colab.research.google.com/drive/1LTQwBy6bheXgwfLNQLUwEKzI28dOvEWf
"""
from google.colab import drive
drive.mount('/content/drive')
import os
os.chdir('/content/drive/MyDrive/CognitiveApp')
!pwd # To verify your current directory
!pip install streamlit plotly numpy joblib scikit-learn spacy torch torchaudio whisper transformers librosa pandas textblob fastdtw scipy pyngrok
!pip install pyngrok
!brew install ngrok
!pip uninstall whisper
!pip install git+https://github.com/openai/whisper.git
!ngrok config add-authtoken 2t5xRPoMFuzy4iMynoHMNBjPUJ2_3dBm2iTa8UTB6MStym4xP
from pyngrok import ngrok
import os
import time
# --- Kill any existing ngrok tunnels ---
ngrok.kill()
# --- Step 2: Start the HTTP server for cognitive games on port 8000 ---
# This will serve cognitive_games.html and its related files.
# (Make sure cognitive_games.html is in the current directory.)
os.system("python -m http.server 8000 &")
time.sleep(5) # Wait a few seconds for the server to start
# Use the local URL for cognitive games
# If you prefer to expose it via ngrok, you can uncomment the next two lines:
# games_tunnel = ngrok.connect(8000)
# cognitive_games_url = games_tunnel.public_url
cognitive_games_url = "http://localhost:8000/index.html"
print("Cognitive Games URL:", cognitive_games_url)
# Set an environment variable so your Streamlit app can pick up the cognitive games URL
os.environ["COGNITIVE_GAMES_URL"] = cognitive_games_url
# --- Step 3: Start your Streamlit app (demo.py) ---
os.system("streamlit run demo.py &")
time.sleep(5) # Wait for Streamlit to start
# --- Step 4: Expose the Streamlit app via ngrok (port 8501) ---
streamlit_tunnel = ngrok.connect(8501)
streamlit_url = streamlit_tunnel.public_url
print("Streamlit Public URL:", streamlit_url)
# The script will print both URLs.
ngrok.kill()
from pyngrok import ngrok
import os
os.system("streamlit run demo.py &")
public_url = ngrok.connect(8501) # Streamlit default port
print("Streamlit Public URL:", public_url)
```
### diagnostic_model.py
```python
# diagnostic_model.py (updated)
import shap
import numpy as np
import joblib
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
from typing import Dict
from typing import Dict
class DiagnosticModel:
def __init__(self):
# Define thresholds based on research papers
self.thresholds = {
"pause_count": (5, "Excessive pauses may indicate working memory deficits (Smith et al. 2023)"),
"mean_pause_duration": (1.2, "Long pauses correlate with lexical retrieval difficulties"),
"theta_power": (20.003, "Elevated theta waves suggest attention regulation issues"),
"beta_power": (5.465, "Reduced beta activity links to working memory impairment")
}
def diagnose(self, features: Dict) -> Dict:
clinical_notes = []
risk_indicators = {}
for k, (threshold, explanation) in self.thresholds.items():
if features.get(k, 0) > threshold:
risk_indicators[k] = {
"value": features[k],
"threshold": threshold,
"clinical_significance": explanation
}
clinical_notes.append(f"{k.replace('_', ' ').title()} exceeds clinical thresholds")
probability = min(0.99, len(risk_indicators)/len(self.thresholds))
return {
"probability": {
"healthy": 1 - probability,
"cognitive_decline": probability
},
"risk_indicators": risk_indicators,
"clinical_notes": clinical_notes
}
def load_dementiabank_data(self, processor):
"""Load and process DementiaBank data"""
print("Loading DementiaBank data...")
data = processor.process_dataset()
# Filter for age and relevant features
data = data[data['age'] > self.age_threshold]
X = data[self.feature_names].values
y = data['diagnosis'].map({'Control': 0, 'Dementia': 1}).values
# Split dataset
self.X_train, self.X_test, self.y_train, self.y_test = train_test_split(
X, y, test_size=0.2, stratify=y, random_state=42
)
def train(self):
"""Train the model on loaded data"""
print(f"Training on {len(self.X_train)} samples...")
self.model.fit(self.X_train, self.y_train)
# Create explainer
self.explainer = shap.TreeExplainer(self.model)
# Evaluate
train_acc = accuracy_score(self.y_train, self.model.predict(self.X_train))
test_acc = accuracy_score(self.y_test, self.model.predict(self.X_test))
print(f"Training accuracy: {train_acc:.2f}, Test accuracy: {test_acc:.2f}")
def save_model(self, path="dementia_model.pkl"):
joblib.dump(self.model, path)
def load_model(self, path="dementia_model.pkl"):
self.model = joblib.load(path)
self.explainer = shap.TreeExplainer(self.model)
```
### index.html
```html
Cognitive Odyssey
Cognitive Odyssey
Embark on your cognitive journey.
Home
Choose a game to challenge your mind. Your overall cognitive performance and risk assessments update automatically.
Cognitive Performance
Risk Assessment
Memory Game
Memorize the words shown below, then type them back (separated by commas).
Reaction Time
When the button appears (at a random spot), click it as quickly as you can.
Wait...
Stroop Test
Select the color of the text (ignore the word).
Sequence Memory
Watch the flashing sequence, then click the tiles in the same order.
Math Challenge
Solve the math problem. Both accuracy and speed count!
Word Scramble
Unscramble the letters to form the correct word.
Cognitive Odyssey
Cognitive Odyssey
Embark on your cognitive journey.
Home
Choose a game to challenge your mind. Your overall cognitive performance and risk assessments update automatically.
Cognitive Performance
Risk Assessment
Advanced Analysis
This section provides a detailed diagnostic assessment.
Memory Game
Memorize the words shown below, then type them back (separated by commas).
Reaction Time
When the button appears (at a random spot), click it as quickly as you can.
Wait...
Stroop Test
Select the color of the text (ignore the word).
Sequence Memory
Watch the flashing sequence, then click the tiles in the same order.
Math Challenge
Solve the math problem. Both accuracy and speed count!
Word Scramble
Unscramble the letters to form the correct word.
[truncated — 3162 more characters]
```
### analyze_sensor_data.py
```python
import json
import numpy as np
import matplotlib.pyplot as plt # Import matplotlib for plotting
from transformers import pipeline
import torch
# Disable MPS and force CPU
torch.backends.mps.is_available = lambda: False
torch.backends.mps.is_built = lambda: False
class SensorAnalyzer:
def __init__(self):
# Initialize the emotion classifier and summarizer on CPU
self.classifier = pipeline("text-classification", model="j-hartmann/emotion-english-distilroberta-base", device="cpu")
self.summarizer = pipeline("summarization", model="Falconsai/text_summarization", device="cpu")
def load_data(self, json_file):
"""Load sensor data from a JSON file."""
data = []
with open(json_file, 'r') as f:
for line in f:
data.append(json.loads(line))
return data
def process_data(self, data):
"""Process sensor data into structured format."""
structured_data = {
'acc_x': [], 'acc_y': [], 'acc_z': [],
'gyro_x': [], 'gyro_y': [], 'gyro_z': [],
'mag_x': [], 'mag_y': [], 'mag_z': [],
'timestamps': []
}
for entry in data:
if entry['type'] == 'acceleration':
structured_data['acc_x'].append(entry['data']['x'])
structured_data['acc_y'].append(entry['data']['y'])
structured_data['acc_z'].append(entry['data']['z'])
elif entry['type'] == 'gyroscope':
structured_data['gyro_x'].append(entry['data']['x'])
structured_data['gyro_y'].append(entry['data']['y'])
structured_data['gyro_z'].append(entry['data']['z'])
elif entry['type'] == 'magnetometer':
structured_data['mag_x'].append(entry['data']['x'])
structured_data['mag_y'].append(entry['data']['y'])
structured_data['mag_z'].append(entry['data']['z'])
structured_data['timestamps'].append(entry['timestamp'])
return structured_data
def analyze_movement_patterns(self, data):
"""Analyze movement patterns and detect specific situations."""
# Combine gyroscope data for analysis
gyro_data = data['gyro_x'] + data['gyro_y'] + data['gyro_z']
# Convert sensor data to textual description
sensor_summary = f"""
Movement patterns show:
- Average acceleration: X={np.mean(data['acc_x']):.2f}, Y={np.mean(data['acc_y']):.2f}, Z={np.mean(data['acc_z']):.2f}
- Maximum gyroscope variation: {np.max(gyro_data):.2f} rad/s
- Magnetometer stability: {np.std(data['mag_x'] + data['mag_y'] + data['mag_z']):.2f} μT
"""
# Detect specific situations
situation = self._detect_situation(data)
# Generate emotional tone analysis
emotion = self.classifier(sensor_summary[:512])[0]
# Create health assessment
analysis = f"""
Movement Profile Analysis:
{sensor_summary}
Detected Situation:
{situation}
Detected Pattern Characteristics:
{self._generate_pattern_description(data)}
"""
# Summarize the analysis
input_text = analysis[:1024]
summary = self.summarizer(input_text, max_length=50, min_length=10, do_sample=False)[0]['summary_text']
return {
"technical_summary": sensor_summary,
"emotional_tone": emotion,
"health_insights": summary,
"detected_situation": situation
}
def _detect_situation(self, data):
"""Detect specific situations based on sensor data."""
# Fall detection: sudden spike in acceleration followed by stillness
if self._detect_fall(data):
return "Fall detected: Sudden spike in acceleration followed by stillness."
# Tremor detection: small, rapid oscillations in gyroscope and acceleration data
if self._detect_tremors(data):
return "Tremors detected: Small, rapid oscillations in movement data."
# Irregular gait detection: uneven patterns in acceleration and gyroscope data
if self._detect_irregular_gait(data):
return "Irregular gait detected: Uneven walking patterns."
return "No specific situation detected."
def _detect_fall(self, data):
"""Detect a fall based on sudden spike in acceleration and stillness."""
acc_magnitude = np.sqrt(np.array(data['acc_x'])**2 + np.array(data['acc_y'])**2 + np.array(data['acc_z'])**2)
# Check for sudden spike (fall) followed by stillness
spike_threshold = 4.0 # Threshold for fall detection
stillness_threshold = 0.5 # Threshold for stillness
# Detect spike
if np.max(acc_magnitude) > spike_threshold:
# Check for stillness after spike
spike_index = np.argmax(acc_magnitude)
if spike_index < len(acc_magnitude) - 10: # Ensure there's enough data after the spike
post_spike_magnitude = np.mean(acc_magnitude[spike_index + 1:spike_index + 10])
if post_spike_magnitude < stillness_threshold:
return True
return False
def _detect_tremors(self, data):
"""Detect tremors based on small, rapid oscillations in gyroscope and acceleration data."""
# Calculate variance in gyroscope and acceleration data
gyro_variance = np.var(data['gyro_x'] + data['gyro_y'] + data['gyro_z'])
acc_variance = np.var(data['acc_x'] + data['acc_y'] + data['acc_z'])
# Tremors are characterized by high variance in small movements
tremor_threshold = 0.1 # Threshold for tremor detection
if gyro_variance > tremor_threshold and acc_variance > tremor_threshold:
return True
return False
def _detect_irregular_gait(self, dat
[truncated — 4539 more characters]
```
### feature_extractor.py
```python
import torch
import torchaudio
import whisper
import numpy as np
from transformers import Pipeline, AutoTokenizer, AutoModelForSequenceClassification
import librosa
import pandas as pd
from typing import Dict, List, Tuple
import spacy
from scipy.stats import entropy
from textblob import TextBlob
from fastdtw import fastdtw
from scipy import signal
from transformers import pipeline
from typing import Dict, List, Tuple
import os
class SpeechFeatureExtractor:
def __init__(self):
self.whisper_model = whisper.load_model("base", device="cpu")
self.nlp = spacy.load("en_core_web_sm")
self.theta_threshold = 20.003 # aMCI threshold from 2022 study
self.beta_threshold = 5.465 # aMCI threshold from 2022 study
self.llm = pipeline("text-generation", model="gpt2", framework="pt")
def generate_clinical_insights(self, features: Dict) -> str:
"""
Generate clinical insights based on extracted features.
References recent research for clinical significance.
"""
insights = []
# Check each feature against clinical thresholds
if features.get("pause_count", 0) > 5:
insights.append("Increased pause count may indicate working memory challenges.")
if features.get("mean_pause_duration", 0) > 1.2:
insights.append("Longer pauses may suggest lexical retrieval difficulties.")
if features.get("theta_power", 0) > self.theta_threshold:
insights.append("Elevated theta power could indicate attention regulation issues.")
if features.get("beta_power", 0) < self.beta_threshold:
insights.append("Reduced beta power may be associated with working memory impairment.")
# Use LLM to enhance insights with more natural language
if insights:
context = "\n".join(insights)
prompt = f"Based on speech analysis, the following patterns were observed:\n{context}\n\nProvide a brief, patient-friendly summary:"
try:
llm_response = self.llm(prompt, max_length=200, num_return_sequences=1)
enhanced_insights = llm_response[0]['generated_text']
return enhanced_insights
except Exception as e:
print(f"LLM generation error: {e}")
return "\n".join(insights) # Fallback to basic insights
return "No significant cognitive risk indicators detected in the speech patterns."
def _extract_llm_features(self, text: str):
prompt = f"Analyze cognitive state from: {text}\nFeatures:"
response = self.llm(prompt, max_length=100)
return {
"comprehension_score": self._parse_llm_output(response),
"memory_issues": ...
}
def _extract_longitudinal_features(self, audio_clips: List[str]):
trends = {}
for clip in audio_clips:
features = self.extract_features(clip)
# Track feature drift over time (e.g., increasing pause duration)
for key in features:
trends[key] = trends.get(key, []) + [features[key]]
return trends
def _extract_parkinsons_features(self, audio: np.ndarray, sr: int):
# Use existing _extract_parkinsons_markers method
return self._extract_parkinsons_markers(audio, sr)
def _extract_acoustic_features(self, audio: np.ndarray, sr: int) -> Dict:
try:
# Get f0 and handle array operations correctly
f0, voiced_flag, _ = librosa.pyin(audio,
fmin=librosa.note_to_hz('C2'),
fmax=librosa.note_to_hz('C7'),
sr=sr)
# Ensure we get single values
voice_stability = float(np.nanstd(f0[voiced_flag])) if np.any(voiced_flag) else 0.0
mean_f0 = float(np.nanmean(f0[voiced_flag])) if np.any(voiced_flag) else 0.0
# Spectral features
spec_cent = librosa.feature.spectral_centroid(y=audio, sr=sr)[0]
spec_cent_mean = float(np.nanmean(spec_cent))
return {
"voice_stability": voice_stability,
"mean_f0": mean_f0,
"spectral_centroid": spec_cent_mean
}
except Exception as e:
print(f"Acoustic feature error: {e}")
return {"voice_stability": 0.0, "mean_f0": 0.0, "spectral_centroid": 0.0}
def _extract_prosodic_features(self, audio: np.ndarray, sr: int) -> Dict:
"""Extract prosodic features with proper array handling"""
try:
# Energy and RMS
rms = librosa.feature.rms(y=audio)[0]
# Silence detection with proper array handling
silence_intervals = librosa.effects.split(audio, top_db=20)
# Handle pauses calculation safely
if len(silence_intervals) > 1:
pauses = np.diff(silence_intervals.flatten()) / sr
pause_count = len(pauses)
mean_pause_duration = float(np.mean(pauses))
else:
pause_count = 0
mean_pause_duration = 0.0
# Safely calculate energy variance
energy_variance = float(np.var(rms)) if len(rms) > 0 else 0.0
return {
"energy_variance": energy_variance,
"pause_count": pause_count,
"mean_pause_duration": mean_pause_duration
}
except Exception as e:
print(f"Prosodic feature error: {str(e)}")
return {
"energy_variance": 0.0,
"pause_count": 0,
"mean_pause_duration": 0.0
}
def extract_features(self, audio_path: str) -> Dict:
"""Generate synthetic features for demo"
[truncated — 5826 more characters]
```
### demo.py
```python
import numpy as np
import socket
import threading
import time
import json
import struct
import os
from queue import Queue
import streamlit as st
import streamlit.components.v1 as components
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from feature_extractor import SpeechFeatureExtractor
from diagnostic_model import DiagnosticModel
from analyze_sensor_data import SensorAnalyzer # Import the analyzer class
# Global queue for IMU data sharing
imu_data_queue = Queue()
# -------------------- IMU Data Processor --------------------
class IMUDataProcessor:
def __init__(self):
self.acceleration_data = []
self.gyroscope_data = []
self.magnetometer_data = []
def update_data(self, data_type, x, y, z):
if data_type == "acceleration":
self.acceleration_data.append({"x": x, "y": y, "z": z})
if len(self.acceleration_data) > 100:
self.acceleration_data.pop(0)
elif data_type == "gyroscope":
self.gyroscope_data.append({"x": x, "y": y, "z": z})
if len(self.gyroscope_data) > 100:
self.gyroscope_data.pop(0)
elif data_type == "magnetometer":
self.magnetometer_data.append({"x": x, "y": y, "z": z})
if len(self.magnetometer_data) > 100:
self.magnetometer_data.pop(0)
# -------------------- Voice Analysis Helpers --------------------
def calculate_cognitive_score(features):
weights = {
'voice_stability': 0.2,
'pause_count': -0.15,
'mean_pause_duration': -0.15,
'theta_power': -0.25,
'beta_power': 0.25
}
total = sum(weights[k] * features[k] for k in weights if k in features)
return min(99.99, max(0.01, 50 + total * 10))
def format_metric(value):
return f"{value:.2f}"
def display_advanced_health_assessment(features):
st.subheader("🧠 Advanced Health Assessment")
cognitive_score = calculate_cognitive_score(features)
processing_score = 100 - (features['mean_pause_duration'] * 20)
attention_score = 100 - (features['theta_power'] * 2)
memory_score = features['beta_power'] * 10
st.markdown("#### Detailed Neural Metrics")
c1, c2, c3 = st.columns(3)
with c1:
st.metric("Voice Stability", f"{features['voice_stability']*100:.2f}%")
with c2:
st.metric("Attention Index", f"{attention_score:.2f}%")
with c3:
st.metric("Memory Score", f"{memory_score:.2f}%")
st.markdown("#### 🔍 Risk Analysis")
rc1, rc2 = st.columns(2)
with rc1:
risk_score = 100 - cognitive_score
st.metric("Cognitive Decline Risk", f"{risk_score:.2f}%", delta=f"{cognitive_score - 50:+.2f}% from baseline", delta_color="inverse")
with rc2:
proc_risk = 100 - processing_score
st.metric("Processing Speed Risk", f"{proc_risk:.2f}%", delta=f"{processing_score - 50:+.2f}% from baseline", delta_color="inverse")
# Radar chart visualization
categories = ['Attention', 'Memory', 'Language', 'Processing', 'Stability']
values = [
attention_score,
memory_score,
cognitive_score,
processing_score,
features['voice_stability'] * 100
]
fig = go.Figure()
fig.add_trace(go.Scatterpolar(r=values, theta=categories, fill='toself', name="Neural Metrics"))
fig.update_layout(polar=dict(radialaxis=dict(visible=True, range=[0, 100])), showlegend=False)
st.plotly_chart(fig, use_container_width=True)
# -------------------- IMU & Movement Analysis --------------------
def calculate_stability(data):
if not data:
return 100.0
x = [d['x'] for d in data]
y = [d['y'] for d in data]
z = [d['z'] for d in data]
var_avg = np.mean([np.var(x), np.var(y), np.var(z)])
return max(0, min(100, 100 * (1 - var_avg)))
def calculate_smoothness(data):
if not data:
return 0.0
x = [d['x'] for d in data]
y = [d['y'] for d in data]
z = [d['z'] for d in data]
jerk_x = np.diff(x, 2) if len(x) > 2 else [0]
jerk_y = np.diff(y, 2) if len(y) > 2 else [0]
jerk_z = np.diff(z, 2) if len(z) > 2 else [0]
jerk_score = np.mean([np.mean(np.abs(jerk_x)), np.mean(np.abs(jerk_y)), np.mean(np.abs(jerk_z))])
return max(0, min(100, 100 * (1 - jerk_score)))
def display_imu_data(imu_processor):
st.subheader("🔄 Real-time Movement Analysis")
while not imu_data_queue.empty():
try:
pkt = imu_data_queue.get_nowait()
imu_processor.update_data("acceleration", pkt["acceleration"]["x"], pkt["acceleration"]["y"], pkt["acceleration"]["z"])
imu_processor.update_data("gyroscope", pkt["gyroscope"]["x"], pkt["gyroscope"]["y"], pkt["gyroscope"]["z"])
imu_processor.update_data("magnetometer", pkt["magnetometer"]["x"], pkt["magnetometer"]["y"], pkt["magnetometer"]["z"])
except Exception as e:
st.write("Error processing IMU data:", e)
c1, c2 = st.columns(2)
with c1:
st.metric("Movement Stability", f"{calculate_stability(imu_processor.acceleration_data):.2f}%")
with c2:
st.metric("Movement Smoothness", f"{calculate_smoothness(imu_processor.gyroscope_data):.2f}%")
if imu_processor.acceleration_data:
times = list(range(len(imu_processor.acceleration_data)))
fig = go.Figure()
fig.add_trace(go.Scatter(x=times, y=[d['x'] for d in imu_processor.acceleration_data],
name='X-axis', line=dict(color='#4281A4')))
fig.add_trace(go.Scatter(x=times, y=[d['y'] for d in imu_processor.acceleration_data],
name='Y-axis', line=dict(color='#48A9A6')))
fig.add_trace(go.Scatter(x=times, y=[d['z'] for d in imu_processor.acceleration_data],
name='Z-axis', line=dict(color='#C1666B')))
fig.update_layout(title="Movement Patterns Over Time",
xaxis
[truncated — 9305 more characters]
```
[5 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]