File size: 11,512 Bytes
a2fd99a |
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 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 |
import numpy as np
from transformers import AutoModelForSequenceClassification, AutoTokenizer, pipeline
from typing import Dict, List, Optional
import torch
import json
class ClinicalAnalyzer:
def __init__(self):
self.initialize_models()
self.condition_patterns = self._load_condition_patterns()
def initialize_models(self):
"""Initialize transformer models for clinical analysis"""
try:
# Clinical text analysis model
self.clinical_tokenizer = AutoTokenizer.from_pretrained(
"microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext"
)
self.clinical_model = AutoModelForSequenceClassification.from_pretrained(
"microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext"
)
# Mental health assessment pipeline
self.mental_health_pipeline = pipeline(
"text-classification",
model="microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext",
tokenizer="microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext"
)
except Exception as e:
print(f"Error initializing models: {str(e)}")
# Fallback to rule-based analysis if models fail to load
self.clinical_model = None
self.clinical_tokenizer = None
self.mental_health_pipeline = None
def _load_condition_patterns(self) -> Dict:
"""Load predefined patterns for mental health conditions"""
return {
'depression': {
'eeg_patterns': {
'alpha_asymmetry': True,
'theta_increase': True,
'beta_decrease': True
},
'keywords': [
'depressed mood', 'loss of interest', 'fatigue',
'sleep disturbance', 'concentration problems'
]
},
'anxiety': {
'eeg_patterns': {
'beta_increase': True,
'alpha_decrease': True,
'high_coherence': True
},
'keywords': [
'anxiety', 'worry', 'restlessness', 'tension',
'panic', 'nervousness'
]
},
'ptsd': {
'eeg_patterns': {
'alpha_suppression': True,
'theta_increase': True,
'beta_asymmetry': True
},
'keywords': [
'trauma', 'flashbacks', 'nightmares', 'avoidance',
'hypervigilance', 'startle response'
]
}
}
def analyze(self, features: Dict, clinical_notes: str) -> Dict:
"""Perform comprehensive clinical analysis"""
analysis_results = {
'eeg_analysis': self._analyze_eeg_patterns(features),
'text_analysis': self._analyze_clinical_text(clinical_notes),
'condition_probabilities': self._calculate_condition_probabilities(
features, clinical_notes
),
'severity_assessment': self._assess_severity(features, clinical_notes),
'recommendations': self._generate_recommendations(features, clinical_notes)
}
return analysis_results
def _analyze_eeg_patterns(self, features: Dict) -> Dict:
"""Analyze EEG patterns for clinical significance"""
eeg_analysis = {}
# Analyze band power distributions
band_powers = features['band_powers']
eeg_analysis['band_power_analysis'] = {
band: {
'mean': float(np.mean(powers)),
'std': float(np.std(powers)),
'clinical_significance': self._assess_band_significance(band, powers)
}
for band, powers in band_powers.items()
}
# Analyze connectivity patterns
connectivity = features['connectivity']
eeg_analysis['connectivity_analysis'] = {
'global_connectivity': float(np.mean(connectivity['correlation'])),
'asymmetry_index': self._calculate_asymmetry_index(features)
}
return eeg_analysis
def _analyze_clinical_text(self, clinical_notes: str) -> Dict:
"""Analyze clinical notes using NLP"""
if not clinical_notes:
return {'error': 'No clinical notes provided'}
try:
if self.mental_health_pipeline:
# Use transformer model for analysis
results = self.mental_health_pipeline(clinical_notes)
text_analysis = {
'sentiment': results[0]['label'],
'confidence': float(results[0]['score'])
}
else:
# Fallback to keyword-based analysis
text_analysis = self._keyword_based_analysis(clinical_notes)
# Extract symptoms and severity
text_analysis['identified_symptoms'] = self._extract_symptoms(clinical_notes)
text_analysis['risk_factors'] = self._identify_risk_factors(clinical_notes)
return text_analysis
except Exception as e:
return {'error': f'Text analysis failed: {str(e)}'}
def _calculate_condition_probabilities(
self, features: Dict, clinical_notes: str
) -> Dict:
"""Calculate probabilities for different mental health conditions"""
probabilities = {}
for condition, patterns in self.condition_patterns.items():
# Calculate EEG pattern match score
eeg_score = self._calculate_eeg_pattern_match(
features, patterns['eeg_patterns']
)
# Calculate text pattern match score
text_score = self._calculate_text_pattern_match(
clinical_notes, patterns['keywords']
)
# Combine scores with weighted average
combined_score = 0.6 * eeg_score + 0.4 * text_score
probabilities[condition] = float(combined_score)
return probabilities
def _assess_severity(self, features: Dict, clinical_notes: str) -> Dict:
"""Assess the severity of identified conditions"""
severity = {
'overall_severity': self._calculate_overall_severity(features, clinical_notes),
'domain_severity': {
'cognitive': self._assess_cognitive_severity(features),
'emotional': self._assess_emotional_severity(features, clinical_notes),
'behavioral': self._assess_behavioral_severity(clinical_notes)
},
'risk_level': self._assess_risk_level(features, clinical_notes)
}
return severity
def _generate_recommendations(self, features: Dict, clinical_notes: str) -> List[str]:
"""Generate clinical recommendations based on analysis"""
recommendations = []
# Analyze severity and conditions
severity = self._assess_severity(features, clinical_notes)
conditions = self._calculate_condition_probabilities(features, clinical_notes)
# Generate general recommendations
if severity['overall_severity'] > 0.7:
recommendations.append("Immediate clinical intervention recommended")
elif severity['overall_severity'] > 0.4:
recommendations.append("Regular clinical monitoring recommended")
# Condition-specific recommendations
for condition, probability in conditions.items():
if probability > 0.6:
recommendations.extend(
self._get_condition_specific_recommendations(condition)
)
return recommendations
def _calculate_eeg_pattern_match(self, features: Dict, patterns: Dict) -> float:
"""Calculate how well EEG features match condition patterns"""
match_scores = []
for pattern, expected in patterns.items():
if pattern == 'alpha_asymmetry':
score = self._check_alpha_asymmetry(features)
elif pattern == 'beta_increase':
score = self._check_beta_increase(features)
elif pattern == 'theta_increase':
score = self._check_theta_increase(features)
else:
score = 0.5 # Default score for unknown patterns
match_scores.append(score if expected else 1 - score)
return np.mean(match_scores) if match_scores else 0.0
def _calculate_text_pattern_match(self, text: str, keywords: List[str]) -> float:
"""Calculate how well clinical notes match condition keywords"""
if not text:
return 0.0
text = text.lower()
matched_keywords = sum(1 for keyword in keywords if keyword.lower() in text)
return matched_keywords / len(keywords)
def _calculate_asymmetry_index(self, features: Dict) -> float:
"""Calculate brain asymmetry index from EEG features"""
try:
# Calculate alpha asymmetry between left and right hemispheres
alpha_powers = features['band_powers']['alpha']
left_channels = alpha_powers[:len(alpha_powers)//2]
right_channels = alpha_powers[len(alpha_powers)//2:]
asymmetry = np.log(np.mean(right_channels)) - np.log(np.mean(left_channels))
return float(asymmetry)
except:
return 0.0
def _assess_band_significance(self, band: str, powers: np.ndarray) -> str:
"""Assess clinical significance of frequency band power"""
mean_power = np.mean(powers)
if band == 'alpha':
if mean_power < 0.3:
return "Significantly reduced alpha power"
elif mean_power > 0.7:
return "Elevated alpha power"
elif band == 'beta':
if mean_power > 0.7:
return "Elevated beta power - possible anxiety"
elif band == 'theta':
if mean_power > 0.6:
return "Elevated theta power - possible cognitive issues"
return "Within normal range"
def _get_condition_specific_recommendations(self, condition: str) -> List[str]:
"""Get specific recommendations for identified conditions"""
recommendations = {
'depression': [
"Consider cognitive behavioral therapy",
"Evaluate need for antidepressant medication",
"Recommend regular physical activity",
"Implement sleep hygiene practices"
],
'anxiety': [
"Consider anxiety-focused psychotherapy",
"Evaluate need for anti-anxiety medication",
"Recommend relaxation techniques",
"Practice mindfulness meditation"
],
'ptsd': [
"Consider trauma-focused therapy",
"Evaluate need for PTSD-specific medication",
"Implement grounding techniques",
"Develop safety and coping plans"
]
}
return recommendations.get(condition, []) |