app.py

"""
The Empathy Engine
A service that dynamically modulates synthesized speech based on detected text emotion.

Features:
  - Granular emotion detection (Happy, Excited, Calm, Neutral, Sad, Angry, Fearful)
  - Intensity-scaled vocal modulation (rate, pitch, volume)
  - SSML-like pause injection for natural delivery
  - Flask web interface with instant audio playback
  - Dual TTS backends: gTTS (primary/online) + pyttsx3 (offline fallback)
"""

import os
import re
import uuid
import shutil
import traceback
from flask import Flask, render_template, request, jsonify, send_from_directory
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer

# ──────────────────────────────────────────────
# Flask App Setup
# ──────────────────────────────────────────────
app = Flask(__name__)
AUDIO_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static", "audio")
os.makedirs(AUDIO_DIR, exist_ok=True)

# ──────────────────────────────────────────────
# Detect available TTS backends
# ──────────────────────────────────────────────
GTTS_AVAILABLE = False
PYTTSX3_AVAILABLE = False
PYDUB_AVAILABLE = False

try:
    from gtts import gTTS
    GTTS_AVAILABLE = True
    print("  [+] gTTS available (primary TTS engine)")
except ImportError:
    print("  [-] gTTS not installed — run: pip install gTTS")

try:
    import pyttsx3
    PYTTSX3_AVAILABLE = True
    print("  [+] pyttsx3 available (offline fallback)")
except ImportError:
    print("  [-] pyttsx3 not installed — run: pip install pyttsx3")

try:
    from pydub import AudioSegment
    if shutil.which("ffmpeg"):
        PYDUB_AVAILABLE = True
        print("  [+] pydub available (audio post-processing for rate/pitch)")
    else:
        print("  [-] pydub installed but ffmpeg not found — post-processing disabled")
        print("       Install ffmpeg: https://ffmpeg.org/download.html")
except ImportError:
    print("  [-] pydub not installed (optional) — run: pip install pydub")
    print("       Also install ffmpeg: sudo apt install ffmpeg (Linux) / brew install ffmpeg (Mac)")

if not GTTS_AVAILABLE and not PYTTSX3_AVAILABLE:
    print("\n  [!] No TTS engine available! Install at least one:")
    print("     pip install gTTS          (recommended, needs internet)")
    print("     pip install pyttsx3       (offline, needs espeak on Linux)")

# ──────────────────────────────────────────────
# Emotion Detection Module
# ──────────────────────────────────────────────

EMOTION_KEYWORDS = {
    "excited": [
        "amazing", "incredible", "fantastic", "awesome", "wonderful", "excellent",
        "outstanding", "brilliant", "thrilled", "ecstatic", "best", "love",
        "perfect", "superb", "magnificent", "unbelievable", "wow", "hooray",
        "celebrate", "congratulations", "excited", "exciting",
    ],
    "angry": [
        "angry", "furious", "outraged", "annoyed", "frustrated", "irritated",
        "hate", "terrible", "worst", "unacceptable", "ridiculous", "disgusting",
        "pathetic", "stupid", "absurd", "infuriating", "enraged", "livid",
        "demand", "refund",
    ],
    "sad": [
        "sad", "depressed", "disappointed", "heartbroken", "unfortunate",
        "sorry", "grief", "loss", "miss", "lonely", "hopeless", "miserable",
        "regret", "gloomy", "melancholy", "tragic", "unhappy", "sorrow",
    ],
    "fearful": [
        "afraid", "scared", "worried", "anxious", "terrified", "nervous",
        "panic", "dread", "horror", "frightened", "alarmed", "uneasy",
        "concerned", "apprehensive", "tense", "stress", "stressed",
    ],
    "surprised": [
        "surprised", "shocked", "astonished", "unexpected", "whoa",
        "really", "seriously", "no way", "unreal", "stunned", "speechless",
    ],
}

analyzer = SentimentIntensityAnalyzer()


def detect_emotion(text: str) -> dict:
    """
    Detect emotion from text using a two-stage approach:
      1. VADER sentiment scoring for valence & intensity
      2. Keyword matching for granular emotion classification
    """
    scores = analyzer.polarity_scores(text)
    compound = scores["compound"]
    text_lower = text.lower()

    # --- Stage 1: keyword-based granular classification ---
    keyword_hits = {}
    for emotion, keywords in EMOTION_KEYWORDS.items():
        count = sum(1 for kw in keywords if kw in text_lower)
        if count > 0:
            keyword_hits[emotion] = count

    # --- Stage 2: combine VADER score with keyword hits ---
    if keyword_hits:
        dominant_keyword_emotion = max(keyword_hits, key=keyword_hits.get)

        if dominant_keyword_emotion == "excited" and compound > 0.2:
            emotion = "excited"
        elif dominant_keyword_emotion == "angry" and compound < -0.1:
            emotion = "angry"
        elif dominant_keyword_emotion == "sad" and compound < -0.1:
            emotion = "sad"
        elif dominant_keyword_emotion == "fearful" and compound < 0.3:
            emotion = "fearful"
        elif dominant_keyword_emotion == "surprised":
            emotion = "surprised"
        else:
            emotion = _vader_classify(compound)
    else:
        emotion = _vader_classify(compound)

    # Intensity from VADER compound, boosted by punctuation/caps
    intensity = max(0.1, min(1.0, abs(compound)))
    exclamation_boost = min(0.3, text.count("!") * 0.08)
    caps_words = len([w for w in text.split() if w.isupper() and len(w) > 1])
    caps_boost = min(0.2, caps_words * 0.06)
    intensity = min(1.0, intensity + exclamation_boost + caps_boost)

    return {
        "emotion": emotion,
        "intensity": round(intensity, 3),
        "scores": scores,
    }


def _vader_classify(compound: float) -> str:
    if compound >= 0.5:
        return "excited"
    elif compound >= 0.15:
        return "happy"
    elif compound <= -0.5:
        return "angry"
    elif compound <= -0.15:
        return "sad"
    else:
        return "neutral"


# ──────────────────────────────────────────────
# Voice Modulation Module
# ──────────────────────────────────────────────

BASE_RATE = 175
BASE_VOLUME = 0.9

VOICE_PROFILES = {
    "excited":   {"rate": 1.25, "volume": 1.0,  "pitch_delta": 40},
    "happy":     {"rate": 1.10, "volume": 1.0,  "pitch_delta": 20},
    "neutral":   {"rate": 1.00, "volume": 0.95, "pitch_delta": 0},
    "calm":      {"rate": 0.90, "volume": 0.85, "pitch_delta": -10},
    "sad":       {"rate": 0.80, "volume": 0.75, "pitch_delta": -30},
    "angry":     {"rate": 1.15, "volume": 1.0,  "pitch_delta": 15},
    "fearful":   {"rate": 1.20, "volume": 0.80, "pitch_delta": 25},
    "surprised": {"rate": 1.30, "volume": 1.0,  "pitch_delta": 35},
}


def _interpolate_profile(emotion: str, intensity: float) -> dict:
    """Interpolate between neutral and target emotion based on intensity."""
    neutral = VOICE_PROFILES["neutral"]
    target = VOICE_PROFILES.get(emotion, neutral)
    return {
        "rate": int(BASE_RATE * (neutral["rate"] + (target["rate"] - neutral["rate"]) * intensity)),
        "volume": min(1.0, BASE_VOLUME * (neutral["volume"] + (target["volume"] - neutral["volume"]) * intensity)),
        "pitch_delta": int(target["pitch_delta"] * intensity),
    }


def inject_pauses(text: str, emotion: str) -> str:
    """SSML-like pause injection for natural delivery."""
    text = re.sub(r"([,;])\s*", r"\1 ... ", text)
    text = re.sub(r"([.!?])\s+", r"\1 .... ", text)
    if emotion in ("sad", "fearful"):
        text = re.sub(r"\.\.\.", ".....", text)
    return text


# ──────────────────────────────────────────────
# TTS Synthesis — Dual Backend
# ──────────────────────────────────────────────

def synthesize_speech(text: str, emotion: str, intensity: float) -> str:
    """
    Generate an audio file with emotionally modulated TTS.
    Tries gTTS first (reliable), falls back to pyttsx3 (offline).
    """
    profile = _interpolate_profile(emotion, intensity)
    processed_text = inject_pauses(text, emotion)

    # Try gTTS first (much more reliable in web server context)
    if GTTS_AVAILABLE:
        try:
            return _synthesize_gtts(processed_text, profile, emotion)
        except Exception as e:
            print(f"  [-] gTTS failed: {e}")

    # Fallback to pyttsx3
    if PYTTSX3_AVAILABLE:
        try:
            return _synthesize_pyttsx3(processed_text, profile)
        except Exception as e:
            print(f"  [!] pyttsx3 also failed: {e}")
            traceback.print_exc()

    raise RuntimeError("No TTS engine available. Install gTTS: pip install gTTS")


def _synthesize_gtts(text: str, profile: dict, emotion: str) -> str:
    """
    Generate speech using Google Text-to-Speech.

    gTTS itself doesn't support rate/pitch, but when pydub is installed,
    we post-process the audio to adjust speed (which shifts pitch too)
    and volume — achieving real vocal modulation.
    """
    # Use slow=True for sad/calm emotions as a coarse rate control
    slow = emotion in ("sad", "calm")
    tts = gTTS(text=text, lang="en", slow=slow)

    filename = f"empathy_{uuid.uuid4().hex[:10]}.mp3"
    filepath = os.path.join(AUDIO_DIR, filename)
    tts.save(filepath)

    # Post-process with pydub for finer rate/pitch/volume control
    if PYDUB_AVAILABLE:
        try:
            audio = AudioSegment.from_mp3(filepath)

            # Adjust speed (also shifts pitch — this is intentional and desirable)
            rate_multiplier = profile["rate"] / BASE_RATE
            if abs(rate_multiplier - 1.0) > 0.05:
                new_frame_rate = int(audio.frame_rate * rate_multiplier)
                audio = audio._spawn(audio.raw_data, overrides={"frame_rate": new_frame_rate})
                audio = audio.set_frame_rate(44100)

            # Adjust volume in dB
            if profile["volume"] < 0.9:
                audio = audio + (-6 * (1 - profile["volume"]))
            elif profile["volume"] > 0.95:
                audio = audio + 2

            audio.export(filepath, format="mp3")
        except Exception as e:
            print(f"  [-] Post-processing skipped: {e}")

    return filename


def _synthesize_pyttsx3(text: str, profile: dict) -> str:
    """Generate speech using pyttsx3 (offline)."""
    engine = pyttsx3.init()

    engine.setProperty("rate", profile["rate"])
    engine.setProperty("volume", profile["volume"])

    try:
        base_pitch = 50
        new_pitch = max(0, min(99, base_pitch + profile["pitch_delta"]))
        engine.setProperty("pitch", new_pitch)
    except Exception:
        pass

    try:
        voices = engine.getProperty("voices")
        if voices:
            engine.setProperty("voice", voices[0].id)
    except Exception:
        pass

    filename = f"empathy_{uuid.uuid4().hex[:10]}.wav"
    filepath = os.path.join(AUDIO_DIR, filename)
    engine.save_to_file(text, filepath)
    engine.runAndWait()
    engine.stop()

    if not os.path.exists(filepath) or os.path.getsize(filepath) == 0:
        raise RuntimeError("pyttsx3 produced empty audio file")

    return filename


# ──────────────────────────────────────────────
# Flask Routes
# ──────────────────────────────────────────────

@app.route("/")
def index():
    return render_template("index.html")


@app.route("/synthesize", methods=["POST"])
def synthesize():
    data = request.get_json(force=True)
    text = data.get("text", "").strip()

    if not text:
        return jsonify({"error": "No text provided"}), 400

    emotion_result = detect_emotion(text)
    emotion = emotion_result["emotion"]
    intensity = emotion_result["intensity"]
    profile = _interpolate_profile(emotion, intensity)

    try:
        filename = synthesize_speech(text, emotion, intensity)
    except Exception as e:
        traceback.print_exc()
        return jsonify({"error": f"TTS failed: {str(e)}"}), 500

    return jsonify({
        "emotion": emotion,
        "intensity": intensity,
        "vader_scores": emotion_result["scores"],
        "voice_profile": profile,
        "audio_url": f"/audio/{filename}",
    })


@app.route("/audio/<filename>")
def serve_audio(filename):
    return send_from_directory(AUDIO_DIR, filename)


# ──────────────────────────────────────────────
# CLI Mode
# ──────────────────────────────────────────────

def cli_mode():
    print("\n[*] The Empathy Engine — CLI Mode")
    print("=" * 45)
    print("Type a sentence and hear it spoken with emotion.")
    print("Type 'quit' to exit.\n")

    while True:
        text = input("[?] Enter text: ").strip()
        if text.lower() in ("quit", "exit", "q"):
            print("Goodbye!")
            break

        result = detect_emotion(text)
        emotion = result["emotion"]
        intensity = result["intensity"]
        profile = _interpolate_profile(emotion, intensity)

        print(f"\n   [emotion] Emotion   : {emotion.upper()}")
        print(f"   [stats]   Intensity : {intensity}")
        print(f"   [volume]  Rate={profile['rate']}wpm  Volume={profile['volume']:.2f}  Pitch Δ={profile['pitch_delta']}Hz")

        try:
            filename = synthesize_speech(text, emotion, intensity)
            print(f"   [saved]   Audio saved: static/audio/{filename}\n")
        except Exception as e:
            print(f"   ❌ Error: {e}\n")


# ──────────────────────────────────────────────
# Entry Point
# ──────────────────────────────────────────────

if __name__ == "__main__":
    import sys

    if "--cli" in sys.argv:
        cli_mode()
    else:
        print("\n[*] The Empathy Engine")
        print("   Starting web server at http://localhost:5000")
        print(f"   TTS: gTTS={'[OK]' if GTTS_AVAILABLE else '[NO]'}  pyttsx3={'[OK]' if PYTTSX3_AVAILABLE else '[NO]'}  pydub={'[OK]' if PYDUB_AVAILABLE else '[NO]'}\n")
        app.run(debug=True, port=5000, threaded=False)