Update app.py

app.py

@@ -1,23 +1,16 @@
-# Advanced Real-Time Call Sentiment Analyzer
-# app.py - 
+# Call Sentiment Analyzer
+# app.py
 # Features:
-# - Live microphone recording (real-time streaming)
-# - Live transcription (streaming Whisper)
-# - Live multilingual sentiment analysis (5-star)
-# - Live translation (to English for best sentiment accuracy)
+# - Record via microphone OR upload an audio file
+# - Transcription via Whisper
+# - 5-star multilingual sentiment analysis
 # - Call Quality Indicator (noise/volume estimation)
 # - Keyword Extraction (top relevant words)
-# - Animated live sentiment indicator (changes in real-time)
-# - Conversation between Agent & Caller simulation (alternating turns)
 
 import gradio as gr
 import torch
 import numpy as np
 import librosa
-import threading
-import queue
-import time
-from collections import deque
 from transformers import (
     AutoProcessor,
     AutoModelForSpeechSeq2Seq,
@@ -27,11 +20,11 @@ from transformers import (
 )
 import torch.nn.functional as F
 
-print("Loading advanced models... This may take 1-2 minutes.")
+print("Loading models... This may take 1-2 minutes.")
 
 device = "cuda" if torch.cuda.is_available() else "cpu"
 
-# === Whisper for streaming transcription ===
+# === Whisper for transcription ===
 processor = AutoProcessor.from_pretrained("openai/whisper-base.en")
 whisper_model = AutoModelForSpeechSeq2Seq.from_pretrained("openai/whisper-base.en")
 whisper_model.eval()
@@ -43,35 +36,58 @@ sentiment_model = AutoModelForSequenceClassification.from_pretrained('nlptown/be
 sentiment_model.eval()
 sentiment_model.to(device)
 
-# === Keyword Extraction (simple but effective) ===
+# === Keyword Extraction ===
 keyword_extractor = pipeline("ner", aggregation_strategy="simple", device=0 if device == "cuda" else -1)
 
-print("All models ready! Starting live call analyzer...")
+print("All models ready!")
 
-# Global state for live conversation
-conversation_history = deque(maxlen=20)  # Last 20 lines
-current_speaker = "Agent"  # Alternates between Agent and Caller
-sentiment_history = deque(maxlen=10)  # For smoothing animation
 
-# Queues for threading
-audio_queue = queue.Queue()
-transcription_queue = queue.Queue()
+# Helper: Transcribe audio
+def transcribe(audio_path):
+    import soundfile as sf
+    audio, sr = sf.read(audio_path)
+
+    # Convert stereo to mono if needed
+    if audio.ndim > 1:
+        audio = audio.mean(axis=1)
+
+    # Resample to 16kHz if needed
+    if sr != 16000:
+        audio = librosa.resample(audio.astype(np.float32), orig_sr=sr, target_sr=16000)
+
+    audio = audio.astype(np.float32)
+
+    # Whisper processes in 30s chunks
+    chunk_size = 16000 * 30
+    texts = []
+    for i in range(0, len(audio), chunk_size):
+        chunk = audio[i:i + chunk_size]
+        input_features = processor(chunk, sampling_rate=16000, return_tensors="pt").input_features.to(device)
+        with torch.no_grad():
+            predicted_ids = whisper_model.generate(input_features)
+        text = processor.batch_decode(predicted_ids, skip_special_tokens=True)[0].strip()
+        if text:
+            texts.append(text)
+
+    return " ".join(texts)
+
 
 # Helper: Sentiment analysis
 def get_sentiment(text):
     if not text.strip():
-        return 3, 50.0
-    
+        return "—", "—", "—"
+
     inputs = sentiment_tokenizer(text[:512], return_tensors="pt", truncation=True).to(device)
     with torch.no_grad():
         logits = sentiment_model(**inputs).logits
     probs = F.softmax(logits, dim=-1)[0]
     pred = torch.argmax(probs).item() + 1
-    conf = probs[pred-1].item() * 100
-    
+    conf = probs[pred - 1].item() * 100
+
     stars = "⭐" * pred
-    label = ["Very Negative", "Negative", "Neutral", "Positive", "Very Positive"][pred-1]
-    return pred, conf, f"{stars} {label}"
+    label = ["Very Negative", "Negative", "Neutral", "Positive", "Very Positive"][pred - 1]
+    return f"{stars} {label}", f"{conf:.1f}%", pred
+
 
 # Helper: Keyword extraction
 def extract_keywords(text):
@@ -82,179 +98,111 @@ def extract_keywords(text):
     except:
         return "None detected"
 
-# Helper: Estimate call quality (volume + noise)
-def estimate_call_quality(audio_chunk, sr=16000):
-    if len(audio_chunk) == 0:
-        return "Poor", 0
-    
-    # Volume (RMS)
-    rms = np.sqrt(np.mean(audio_chunk**2))
-    volume_db = 20 * np.log10(rms + 1e-8)
-    volume_level = "Low" if volume_db < -40 else "Good" if volume_db < -20 else "Loud"
-    
-    # Simple noise estimate (spectral flatness)
-    S = np.abs(librosa.stft(audio_chunk))
-    flatness = np.mean(librosa.feature.spectral_flatness(S=S))
-    noise_level = "High Noise" if flatness > 0.3 else "Clear"
-    
-    quality = "Good" if volume_level in ["Good", "Loud"] and noise_level == "Clear" else "Fair" if volume_level == "Good" else "Poor"
-    return quality, round(volume_db + 60)  # Normalize to 0-100
-
-# Live transcription processor (background thread)
-def transcription_worker():
-    buffer = np.array([], dtype=np.float32)
-    while True:
-        try:
-            chunk = audio_queue.get(timeout=1)
-            if chunk is None:
-                break
-            buffer = np.concatenate([buffer, chunk])
-            
-            # Process every ~3 seconds of audio
-            if len(buffer) >= 48000:  # 3 sec at 16kHz
-                speech = buffer[:48000]
-                buffer = buffer[48000:]
-                
-                input_features = processor(speech, sampling_rate=16000, return_tensors="pt").input_features.to(device)
-                with torch.no_grad():
-                    predicted_ids = whisper_model.generate(input_features)
-                text = processor.batch_decode(predicted_ids, skip_special_tokens=True)[0].strip()
-                
-                if text:
-                    transcription_queue.put(text)
-        except:
-            continue
-
-# Start background thread
-threading.Thread(target=transcription_worker, daemon=True).start()
-
-# Main live processing function
-def live_stream(audio_chunk):
-    global current_speaker, conversation_history, sentiment_history
-    
-    if audio_chunk is None:
-        return "", "", "", "", "🔴 Not Recording", "No audio", "Neutral", ""
-    
-    sr, audio = audio_chunk
-    audio = audio.astype(np.float32) / 32768.0  # Normalize
-    
-    # Put into queue for transcription
-    audio_queue.put(audio)
-    
-    # Estimate call quality
-    quality, volume = estimate_call_quality(audio, sr)
-    quality_text = f"{quality} (Volume: {volume}%)"
-    
-    # Try to get new transcription
-    new_text = ""
-    while not transcription_queue.empty():
-        new_text = transcription_queue.get()
-    
-    if new_text:
-        # Alternate speaker (simple simulation)
-        speaker = current_speaker
-        current_speaker = "Caller" if current_speaker == "Agent" else "Agent"
-        
-        # Analyze sentiment
-        _, conf, sentiment_label = get_sentiment(new_text)
-        sentiment_history.append(conf)
-        
-        # Keywords
-        keywords = extract_keywords(new_text)
-        
-        # Update conversation
-        line = f"**{speaker}:** {new_text}"
-        conversation_history.append(line)
-        
-        # Animated sentiment (emoji based on average recent confidence)
-        avg_conf = np.mean(sentiment_history) if sentiment_history else 50
-        if avg_conf > 80:
-            anim = "🟢 Excellent"
-        elif avg_conf > 60:
-            anim = "🟡 Good"
-        elif avg_conf > 40:
-            anim = "🟠 Fair"
-        else:
-            anim = "🔴 Needs Attention"
-        
-        full_convo = "\n".join(list(conversation_history)[-10:])  # Last 10 lines
-        
-        return (
-            full_convo,
-            sentiment_label,
-            f"{conf:.1f}% Confidence",
-            keywords,
-            "🟢 Live Analysis Active",
-            quality_text,
-            anim,
-            f"Current Speaker: {speaker}"
+
+# Helper: Estimate call quality
+def estimate_call_quality(audio_path):
+    try:
+        import soundfile as sf
+        audio, sr = sf.read(audio_path)
+        if audio.ndim > 1:
+            audio = audio.mean(axis=1)
+        audio = audio.astype(np.float32)
+
+        rms = np.sqrt(np.mean(audio ** 2))
+        volume_db = 20 * np.log10(rms + 1e-8)
+        volume_level = "Low" if volume_db < -40 else "Good" if volume_db < -20 else "Loud"
+
+        S = np.abs(librosa.stft(audio))
+        flatness = np.mean(librosa.feature.spectral_flatness(S=S))
+        noise_level = "High Noise" if flatness > 0.3 else "Clear"
+
+        quality = (
+            "Good 🟢" if volume_level in ["Good", "Loud"] and noise_level == "Clear"
+            else "Fair 🟡" if volume_level == "Good"
+            else "Poor 🔴"
         )
-    
-    # No new transcription yet
+        volume_pct = max(0, min(100, round(volume_db + 60)))
+        return quality, volume_pct, noise_level
+    except:
+        return "Unknown", 0, "Unknown"
+
+
+# Main analysis function
+def analyze_audio(audio_path):
+    if audio_path is None:
+        return "No audio provided.", "—", "—", "—", "—", "—", "—"
+
+    # Transcribe
+    transcript = transcribe(audio_path)
+    if not transcript:
+        return "Could not transcribe audio. Please try again.", "—", "—", "—", "—", "—", "—"
+
+    # Sentiment
+    sentiment_label, confidence, stars = get_sentiment(transcript)
+
+    # Keywords
+    keywords = extract_keywords(transcript)
+
+    # Call quality
+    quality, volume_pct, noise = estimate_call_quality(audio_path)
+
+    # Overall health summary
+    if stars == 5 or stars == 4:
+        health = "🟢 Positive Call"
+    elif stars == 3:
+        health = "🟡 Neutral Call"
+    else:
+        health = "🔴 Needs Attention"
+
     return (
-        "\n".join(list(conversation_history)[-10:]) or "Speak to begin conversation...",
-        "Waiting for speech...",
-        "",
-        "Listening...",
-        "🟢 Live Analysis Active",
-        quality_text,
-        "🔄 Processing...",
-        f"Current Speaker: {current_speaker}"
+        transcript,
+        sentiment_label,
+        confidence,
+        keywords,
+        quality,
+        f"{volume_pct}%",
+        f"{health} | Noise: {noise}"
     )
 
-# Gradio Interface with Live Features
-with gr.Blocks(title="Live Call Sentiment Analyzer", theme=gr.themes.Soft()) as demo:
-    gr.Markdown("# 🎤 Live Call Sentiment & Translation Analyzer")
+
+# Gradio Interface
+with gr.Blocks(title="Call Sentiment Analyzer", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 🎤 Call Sentiment Analyzer")
     gr.Markdown("""
-    **Real-time Features:**
-    - Live streaming transcription (English)
-    - Live 5-star sentiment analysis
-    - Simulated Agent ↔ Caller conversation
+    **Record or upload an audio file to analyze:**
+    - Transcription via Whisper
+    - 5-star sentiment analysis (multilingual)
     - Call quality monitoring
     - Keyword extraction
-    - Animated sentiment feedback
     """)
-    
+
     with gr.Row():
-        with gr.Column(scale=2):
-            live_audio = gr.Audio(
-                sources=["microphone"],
-                type="numpy",
-                streaming=True,
-                label="🔴 Live Microphone (Start speaking!)"
-            )
-    
+        audio_input = gr.Audio(
+            sources=["microphone", "upload"],
+            type="filepath",
+            label="🎙️ Record or Upload Audio"
+        )
+
+    analyze_btn = gr.Button("🔍 Analyze", variant="primary")
+
     with gr.Row():
         with gr.Column():
-            convo_box = gr.Markdown(label="Live Conversation", value="Ready. Start speaking...")
-        
+            transcript_box = gr.Textbox(label="📝 Transcript", lines=6, interactive=False)
         with gr.Column():
-            sentiment_box = gr.Textbox(label="Current Sentiment", interactive=False)
-            conf_box = gr.Textbox(label="Confidence", interactive=False)
-            keyword_box = gr.Textbox(label="Detected Keywords", interactive=False)
-    
+            sentiment_box = gr.Textbox(label="😊 Sentiment", interactive=False)
+            conf_box = gr.Textbox(label="📊 Confidence", interactive=False)
+            keyword_box = gr.Textbox(label="🔑 Detected Keywords", interactive=False)
+
     with gr.Row():
-        status = gr.Textbox(label="System Status", value="🔴 Not Recording", interactive=False)
-        quality_box = gr.Textbox(label="Call Quality", interactive=False)
-        anim_sentiment = gr.Textbox(label="Live Sentiment Health", interactive=False)
-        speaker_box = gr.Textbox(label="Current Speaker", interactive=False)
-    
-    live_audio.stream(
-        fn=live_stream,
-        inputs=live_audio,
-        outputs=[convo_box, sentiment_box, conf_box, keyword_box, status, quality_box, anim_sentiment, speaker_box],
-        time_limit=300  # 5 minutes max
+        quality_box = gr.Textbox(label="📶 Call Quality", interactive=False)
+        volume_box = gr.Textbox(label="🔊 Volume Level", interactive=False)
+        health_box = gr.Textbox(label="❤️ Call Health", interactive=False)
+
+    analyze_btn.click(
+        fn=analyze_audio,
+        inputs=audio_input,
+        outputs=[transcript_box, sentiment_box, conf_box, keyword_box, quality_box, volume_box, health_box]
     )
-    
-    # gr.Markdown("""
-    # ### How to Use
-    # - Click the microphone and **start speaking naturally**
-    # - The app simulates a call: alternates between **Agent** and **Caller**
-    # - Watch sentiment, quality, and keywords update **live**
-    # - Green animation = positive/good call, Red = needs attention
-    # - Perfect for training, monitoring, or analyzing live customer calls
-    # """)
-
-# Run app
+
 if __name__ == "__main__":
     demo.launch()