evaluation/live_stream_test.py

# -*- coding: utf-8 -*-
"""
Live Stream Test - Test real-time emotion transitions

Tests that emoji changes correctly when user types new text
Simulates live typing with emotion transitions
"""

import time
from typing import Dict, List, Tuple, Any
from dataclasses import dataclass


@dataclass
class TransitionResult:
    """Result of an emotion transition test"""
    from_text: str
    from_emotion: str
    from_emoji: str
    to_text: str
    to_emotion: str
    to_emoji: str
    combined_text: str
    final_detected_emotion: str
    final_emoji: str
    transition_correct: bool
    transition_time_ms: float


class LiveStreamTest:
    """
    Test live streaming emotion detection
    
    Simulates real-time typing with emotion transitions
    """
    
    # Emotion transition test cases
    # Format: [(from_text, from_emotion), (to_text, to_emotion)]
    TRANSITION_TESTS: List[Tuple[Tuple[str, str], Tuple[str, str]]] = [
        # Positive to Negative
        (("I love this!", "positive"), ("But now I'm angry", "negative")),
        (("This is amazing!", "positive"), ("Wait, this is terrible", "negative")),
        (("I'm so happy!", "positive"), ("Now I feel sad", "negative")),
        (("Great work!", "positive"), ("Actually, this is frustrating", "negative")),
        (("I'm excited!", "positive"), ("Now I'm disappointed", "negative")),
        
        # Negative to Positive
        (("I'm sad", "negative"), ("But now I'm happy!", "positive")),
        (("This is terrible", "negative"), ("Actually, it's great!", "positive")),
        (("I'm angry", "negative"), ("Now I feel better", "positive")),
        (("I'm frustrated", "negative"), ("But I'm grateful now", "positive")),
        (("I'm scared", "negative"), ("Now I'm excited!", "positive")),
        
        # Neutral transitions
        (("The weather is okay", "neutral"), ("I love sunny days!", "positive")),
        (("It's just normal", "neutral"), ("This is terrible news", "negative")),
        
        # Complex transitions
        (("I was worried", "negative"), ("But everything worked out great!", "positive")),
        (("Started feeling anxious", "negative"), ("Now I'm relieved and happy", "positive")),
        (("I'm confused", "negative"), ("But this explanation is amazing!", "positive")),
    ]
    
    def __init__(self, analyzer, emoji_mapper):
        """
        Initialize with analyzer and mapper
        
        Args:
            analyzer: SentimentAnalyzer instance
            emoji_mapper: EmojiMapper instance
        """
        self.analyzer = analyzer
        self.emoji_mapper = emoji_mapper
    
    def _get_polarity(self, label: str) -> str:
        """Map label to polarity"""
        positive = ["happiness", "joy", "positive", "love", "excitement"]
        negative = ["sadness", "anger", "negative", "fear", "frustration"]
        
        label_lower = label.lower()
        if label_lower in positive or "happiness" in label_lower:
            return "positive"
        elif label_lower in negative or "sadness" in label_lower:
            return "negative"
        return "neutral"
    
    def run_single_transition(
        self,
        from_text: str,
        from_expected: str,
        to_text: str,
        to_expected: str
    ) -> TransitionResult:
        """Run a single transition test"""
        # Analyze first text
        result1 = self.analyzer.analyze(from_text)
        from_emotion = result1.get("label", "neutral")
        from_emoji = self.emoji_mapper.get_emoji(from_emotion)
        
        # Combine texts (simulating continued typing)
        combined = f"{from_text} {to_text}"
        
        # Analyze combined (should detect LAST sentence emotion)
        start_time = time.perf_counter()
        result2 = self.analyzer.analyze(combined)
        end_time = time.perf_counter()
        
        transition_time_ms = (end_time - start_time) * 1000
        
        final_emotion = result2.get("label", "neutral")
        final_emoji = self.emoji_mapper.get_emoji(final_emotion)
        
        # Check if transition is correct (final emotion matches to_expected polarity)
        final_polarity = self._get_polarity(final_emotion)
        transition_correct = final_polarity == to_expected
        
        return TransitionResult(
            from_text=from_text,
            from_emotion=from_emotion,
            from_emoji=from_emoji,
            to_text=to_text,
            to_emotion=to_expected,
            to_emoji="",  # Expected emoji
            combined_text=combined,
            final_detected_emotion=final_emotion,
            final_emoji=final_emoji,
            transition_correct=transition_correct,
            transition_time_ms=transition_time_ms
        )
    
    def run_all_transitions(self) -> List[TransitionResult]:
        """Run all transition tests"""
        results = []
        
        for (from_text, from_exp), (to_text, to_exp) in self.TRANSITION_TESTS:
            result = self.run_single_transition(from_text, from_exp, to_text, to_exp)
            results.append(result)
        
        return results
    
    def get_transition_report(self, results: List[TransitionResult]) -> str:
        """Generate human-readable transition report"""
        correct = sum(1 for r in results if r.transition_correct)
        total = len(results)
        accuracy = correct / total if total > 0 else 0
        avg_time = sum(r.transition_time_ms for r in results) / total if total > 0 else 0
        
        lines = [
            "=" * 80,
            "LIVE STREAM EMOTION TRANSITION TEST",
            "=" * 80,
            "",
            f"Total Transitions: {total}",
            f"Correct Transitions: {correct}",
            f"Transition Accuracy: {accuracy:.1%}",
            f"Avg Transition Time: {avg_time:.2f} ms",
            "",
            "-" * 80,
            "TRANSITION DETAILS",
            "-" * 80,
        ]
        
        for i, r in enumerate(results, 1):
            status = "✓" if r.transition_correct else "✗"
            lines.extend([
                f"\n{status} Test {i}:",
                f"   From: \"{r.from_text[:40]}...\" → {r.from_emoji} ({r.from_emotion})",
                f"   To:   \"{r.to_text[:40]}...\" → Expected: {r.to_emotion}",
                f"   Combined: \"{r.combined_text[:50]}...\"",
                f"   Detected: {r.final_emoji} ({r.final_detected_emotion})",
                f"   Time: {r.transition_time_ms:.2f} ms",
            ])
        
        # Summary of failures
        failures = [r for r in results if not r.transition_correct]
        if failures:
            lines.extend([
                "",
                "-" * 80,
                f"FAILED TRANSITIONS: {len(failures)}",
                "-" * 80,
            ])
            for r in failures:
                lines.append(
                    f"  ✗ \"{r.combined_text[:60]}...\" "
                    f"→ Expected {r.to_emotion}, got {r.final_detected_emotion}"
                )
        
        lines.append("=" * 80)
        
        return "\n".join(lines)


if __name__ == "__main__":
    from avatar import SentimentAnalyzer, EmojiMapper
    
    analyzer = SentimentAnalyzer()
    mapper = EmojiMapper()
    test = LiveStreamTest(analyzer, mapper)
    
    results = test.run_all_transitions()
    print(test.get_transition_report(results))