""" Engagement Score Calculator. Fuses face, speech, and text emotion data into a unified engagement metric. """ # Source weights for multimodal fusion SOURCE_WEIGHTS = { "face": 0.50, "speech": 0.30, "text": 0.20, } def calculate_engagement(face_result=None, speech_result=None, text_result=None): """ Calculate a fused engagement score from multimodal emotion data. Uses weighted combination of available modalities. Args: face_result: dict with 'engagement_score' from face model speech_result: dict with 'engagement_score' from speech model text_result: dict with 'engagement_score' from text model Returns: dict with overall engagement score, breakdown, and insights """ scores = {} weights_used = {} active_sources = [] if face_result and "engagement_score" in face_result: scores["face"] = face_result["engagement_score"] weights_used["face"] = SOURCE_WEIGHTS["face"] active_sources.append("face") if speech_result and "engagement_score" in speech_result: scores["speech"] = speech_result["engagement_score"] weights_used["speech"] = SOURCE_WEIGHTS["speech"] active_sources.append("speech") if text_result and "engagement_score" in text_result: scores["text"] = text_result["engagement_score"] weights_used["text"] = SOURCE_WEIGHTS["text"] active_sources.append("text") if not scores: return { "overall_score": 0, "breakdown": {}, "active_sources": [], "level": "Unknown", "insights": ["No data sources available"], } # Normalize weights to sum to 1.0 for active sources total_weight = sum(weights_used.values()) normalized_weights = {k: v / total_weight for k, v in weights_used.items()} # Weighted average overall = sum( scores[source] * normalized_weights[source] for source in active_sources ) # Determine engagement level level = _get_engagement_level(overall) # Generate insights insights = _generate_insights(scores, face_result, speech_result, text_result) return { "overall_score": round(overall, 2), "breakdown": { source: { "score": round(scores[source], 2), "weight": round(normalized_weights[source], 2), "weighted_contribution": round(scores[source] * normalized_weights[source], 2), } for source in active_sources }, "active_sources": active_sources, "level": level, "insights": insights, } def _get_engagement_level(score): """Map score to human-readable engagement level.""" if score >= 85: return "Highly Engaged" elif score >= 70: return "Engaged" elif score >= 55: return "Moderately Engaged" elif score >= 40: return "Disengaged" else: return "Very Disengaged" def _generate_insights(scores, face_result, speech_result, text_result): """Generate actionable insights from the analysis.""" insights = [] # Face insights if face_result: emotion = face_result.get("emotion", "Unknown") confidence = face_result.get("confidence", 0) if emotion in ("Happy", "Surprise"): insights.append(f"Student appears {emotion.lower()} ({confidence}% confidence) — positive engagement signal.") elif emotion in ("Sad", "Angry", "Fear", "Disgust"): insights.append(f"⚠️ Negative facial expression detected: {emotion} ({confidence}%) — consider changing approach.") elif emotion == "Neutral": insights.append(f"Student facial expression is neutral ({confidence}%) — may need more stimulation.") # Speech insights if speech_result: s_emotion = speech_result.get("emotion", "Unknown") if s_emotion in ("Happy", "Surprise"): insights.append(f"Voice tone indicates {s_emotion.lower()} — student is vocally engaged.") elif s_emotion in ("Sad", "Angry"): insights.append(f"⚠️ Voice indicates {s_emotion.lower()} — student may be frustrated.") # Text insights if text_result: sentiment = text_result.get("sentiment", "NEUTRAL") if sentiment == "POSITIVE": insights.append("Text sentiment is positive — student is expressing interest.") elif sentiment == "NEGATIVE": insights.append("⚠️ Negative text sentiment detected — review student comments.") # Cross-modal insights if len(scores) >= 2: values = list(scores.values()) spread = max(values) - min(values) if spread > 40: insights.append("⚠️ Large discrepancy between modalities — mixed signals detected.") if not insights: insights.append("Monitoring active. Collecting data...") return insights def calculate_session_trend(emotion_logs): """ Calculate engagement trend from a list of emotion log entries. Returns time-series data for charting. """ if not emotion_logs: return {"trend": [], "average": 0, "peak": 0, "low": 0} trend = [] engagement_values = [] for log in emotion_logs: score = log.get("engagement_score", 50) engagement_values.append(score) trend.append({ "timestamp": log.get("timestamp", ""), "score": score, "emotion": log.get("emotion", "neutral"), "source": log.get("source", "unknown"), }) return { "trend": trend, "average": round(sum(engagement_values) / len(engagement_values), 2), "peak": round(max(engagement_values), 2), "low": round(min(engagement_values), 2), }