Spaces:
Running on Zero
Running on Zero
File size: 9,376 Bytes
90fa9aa | 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 | import sys
import numpy as np
from pathlib import Path
sys.path.append(str(Path(__file__).resolve().parent.parent.parent))
from src.config import DEFAULT_AUDIO_WEIGHT, DEFAULT_TEXT_WEIGHT, MENTAL_HEALTH_CATEGORIES
from src.models.audio_classifier import AudioEnsemblePipeline
from src.models.text_classifier import LinguisticStressClassifier
class LateDecisionFusion:
"""
Multimodal Late Decision Fusion & Risk Calibration Engine.
Synthesizes predictions from the Speech/Acoustic pipeline (AudioEnsemblePipeline)
and the Linguistic/Text pipeline (LinguisticStressClassifier) using dynamic confidence calibration.
"""
def __init__(self, text_pipeline=None, audio_pipeline=None):
self.text_pipeline = text_pipeline or LinguisticStressClassifier()
self.audio_pipeline = audio_pipeline or AudioEnsemblePipeline()
def analyze_multimodal(self, text_input=None, audio_features_195=None):
"""
Performs dual-modality analysis. If one modality is missing, gracefully falls back to single-modality assessment.
"""
text_result = None
audio_result = None
if text_input and str(text_input).strip():
try:
text_result = self.text_pipeline.predict(str(text_input))
except Exception as e:
print(f"[Fusion] Text prediction error: {e}")
if audio_features_195 is not None and len(audio_features_195) > 0:
try:
audio_result = self.audio_pipeline.predict(audio_features_195)
except Exception as e:
print(f"[Fusion] Audio prediction error: {e}")
# 1. Handle missing modalities
if text_result is None and audio_result is None:
return self._default_result()
elif text_result is None:
return self._single_modality_audio_result(audio_result)
elif audio_result is None:
return self._single_modality_text_result(text_result)
# 2. Both modalities available -> Perform Late Decision Fusion
t_conf = text_result.get("confidence", 0.8)
a_conf = audio_result.get("confidence", 0.8)
# Dynamic confidence weighting: give more weight to the more confident modality
total_conf = t_conf + a_conf
if total_conf > 0:
w_text = (t_conf / total_conf) * 0.7 + (DEFAULT_TEXT_WEIGHT * 0.3)
w_audio = 1.0 - w_text
else:
w_text = DEFAULT_TEXT_WEIGHT
w_audio = DEFAULT_AUDIO_WEIGHT
t_stress = text_result.get("linguistic_stress_score", 0.0)
a_stress = audio_result.get("acoustic_stress_score", 0.0)
combined_stress_score = round(float(w_text * t_stress + w_audio * a_stress), 2)
# Determine primary stress origin
t_cat = text_result.get("predicted_category", "Normal")
a_cat = audio_result.get("predicted_emotion", "Neutral")
if combined_stress_score < 40.0:
final_category = "Normal"
else:
if t_cat != "Normal":
final_category = t_cat
elif a_cat not in ["Normal", "Neutral", "Stress"]:
final_category = a_cat
else:
final_category = "Stress"
# Determine clinical risk tier
if combined_stress_score < 40.0:
risk_tier = "Normal"
color_code = "green"
action_summary = "No significant psychological stress detected. Emotional tone is balanced."
elif combined_stress_score < 60.0:
risk_tier = final_category if final_category not in ["Calm / Normal"] else "Stress"
color_code = "blue"
action_summary = f"Mild symptoms of {final_category.lower()} observed. Recommended: short breaks and time-management strategies."
elif combined_stress_score < 80.0:
risk_tier = final_category if final_category not in ["Calm / Normal"] else "Stress"
color_code = "orange"
action_summary = f"Significant signs of {final_category.lower()} and vocal tension detected. Recommended: structured counseling or grounding CBT exercises."
else:
risk_tier = final_category if final_category not in ["Calm / Normal"] else "Stress"
color_code = "red"
action_summary = f"High emotional distress and acute {final_category.lower()} detected across both speech and text modalities. Immediate psychological check-in and support advised."
return {
"modality_status": "Dual-Modality (Text + Speech)",
"combined_stress_score": combined_stress_score,
"final_stress_category": final_category,
"risk_tier": risk_tier,
"color_code": color_code,
"action_summary": action_summary,
"fusion_weights": {
"text_weight": round(float(w_text), 3),
"audio_weight": round(float(w_audio), 3)
},
"text_analysis": text_result,
"audio_analysis": audio_result
}
def _single_modality_text_result(self, text_result):
score = text_result.get("linguistic_stress_score", 0.0)
cat = text_result.get("predicted_category", "Calm / Normal")
if score < 30.0:
tier, color, summary = "Normal", "green", "No significant text stress markers identified."
elif score < 55.0:
tier = cat if cat not in ["Calm / Normal"] else "Stress"
color, summary = "blue", f"Mild linguistic markers of {cat.lower()} identified."
elif score < 80.0:
tier = cat if cat not in ["Calm / Normal"] else "Stress"
color, summary = "orange", f"High frequency of {cat.lower()} vocabulary identified."
else:
tier = cat if cat not in ["Calm / Normal"] else "Stress"
color, summary = "red", f"Critical linguistic stress markers for {cat.lower()} detected."
return {
"modality_status": "Single-Modality (Text Only)",
"combined_stress_score": score,
"final_stress_category": cat,
"risk_tier": tier,
"color_code": color,
"action_summary": summary,
"fusion_weights": {"text_weight": 1.0, "audio_weight": 0.0},
"text_analysis": text_result,
"audio_analysis": None
}
def _single_modality_audio_result(self, audio_result):
score = audio_result.get("acoustic_stress_score", 0.0)
emotion = audio_result.get("predicted_emotion", "Neutral")
# Determine category based on emotion or score
if emotion in ["Angry", "Fearful", "Sad", "Disgust", "Stress", "Anxiety", "Depression", "Emotional Distress"]:
cat = "Non-Academic / Vocal Stress"
elif score >= 40.0:
cat = "Non-Academic / Vocal Stress"
else:
cat = "Calm / Normal"
emotion_str = emotion if emotion not in ["Neutral"] else "Normal"
if score < 40.0:
tier, color, summary = "Normal", "green", "Vocal tone is calm and stable."
elif score < 60.0:
tier = emotion_str if emotion_str not in ["Normal"] else "Stress"
color, summary = "blue", f"Slight vocal tension ({emotion.lower()}) observed."
elif score < 80.0:
tier = emotion_str if emotion_str not in ["Normal"] else "Stress"
color, summary = "orange", f"High vocal agitation ({emotion.lower()}) and spectral energy detected."
else:
tier = emotion_str if emotion_str not in ["Normal"] else "Stress"
color, summary = "red", f"Severe acoustic stress and emotional agitation ({emotion.lower()}) recorded."
return {
"modality_status": "Single-Modality (Speech Only)",
"combined_stress_score": score,
"final_stress_category": cat,
"risk_tier": tier,
"color_code": color,
"action_summary": summary,
"fusion_weights": {"text_weight": 0.0, "audio_weight": 1.0},
"text_analysis": None,
"audio_analysis": audio_result
}
def _default_result(self):
return {
"modality_status": "No Input Provided",
"combined_stress_score": 0.0,
"final_stress_category": "Calm / Normal",
"risk_tier": "Minimal / Normal",
"color_code": "green",
"action_summary": "No data analyzed yet. Please provide text or audio recording.",
"fusion_weights": {"text_weight": 0.5, "audio_weight": 0.5},
"text_analysis": None,
"audio_analysis": None
}
if __name__ == "__main__":
fusion = LateDecisionFusion()
res = fusion.analyze_multimodal(
text_input="I have three university exams next week and the assignment deadline is overwhelming me completely.",
audio_features_195=np.random.normal(0.5, 0.2, size=195)
)
print("Test Multimodal Output:", res)
|