app.py

import streamlit as st
import os
import tempfile
import requests
import random
import matplotlib.pyplot as plt
import torchaudio
import torch
import ffmpeg

# Load SpeechBrain
try:
    from speechbrain.inference import EncoderClassifier
    speechbrain_classifier = EncoderClassifier.from_hparams(
        source="speechbrain/lang-id-commonlanguage_ecapa",
        savedir="pretrained_models/lang-id-commonlanguage_ecapa"
    )
    SPEECHBRAIN_LOADED = True
except Exception as e:
    st.warning(f"Could not load SpeechBrain model: {e}. Using simulation.")
    SPEECHBRAIN_LOADED = False

class AccentAnalyzer:
    def __init__(self):
        self.accent_profiles = {
            "American": {"features": ["rhotic", "flapped_t", "cot_caught_merger"]},
            "British": {"features": ["non_rhotic", "t_glottalization", "trap_bath_split"]},
            "Australian": {"features": ["non_rhotic", "flat_a", "high_rising_terminal"]},
            "Canadian": {"features": ["rhotic", "canadian_raising", "eh_tag"]},
            "Indian": {"features": ["retroflex_consonants", "monophthongization", "syllable_timing"]},
            "Irish": {"features": ["dental_fricatives", "alveolar_l", "soft_consonants"]},
            "Scottish": {"features": ["rolled_r", "monophthongs", "glottal_stops"]},
            "South African": {"features": ["non_rhotic", "kit_split", "kw_hw_distinction"]}
        }
        self.accent_data = self._simulate_profiles()

    def _simulate_profiles(self):
        all_features = set(f for p in self.accent_profiles.values() for f in p["features"])
        data = {}
        for name, profile in self.accent_profiles.items():
            data[name] = {
                "primary_features": profile["features"],
                "feature_probabilities": {
                    f: random.uniform(0.7, 0.9) if f in profile["features"] else random.uniform(0.1, 0.4)
                    for f in all_features
                }
            }
        return data

    def _simulate_accent_classification(self, audio_path):
        all_features = {f for p in self.accent_profiles.values() for f in p["features"]}
        detected = {f: random.uniform(0.1, 0.9) for f in all_features}
        scores = {}
        for accent, data in self.accent_data.items():
            score = sum(
                detected[f] * data["feature_probabilities"][f] * (3.0 if f in data["primary_features"] else 1.0)
                for f in all_features
            )
            scores[accent] = score
        top = max(scores, key=scores.get)
        conf = (scores[top] / max(scores.values())) * 100
        return {
            "accent_type": top,
            "confidence": conf,
            "explanation": f"Detected **{top}** accent with {conf:.1f}% confidence.",
            "all_scores": scores
        }

    def analyze_accent(self, audio_path):
        if not SPEECHBRAIN_LOADED:
            return self._simulate_accent_classification(audio_path)

        try:
            signal, sr = torchaudio.load(audio_path)
            duration = signal.shape[1] / sr
            if duration < 1.0:
                raise ValueError("Audio too short to analyze.")

            if signal.shape[0] > 1:
                signal = signal.mean(dim=0, keepdim=True)
            if sr != 16000:
                signal = torchaudio.transforms.Resample(sr, 16000)(signal)
            signal = signal.unsqueeze(0)  # [1, 1, time]

            pred = speechbrain_classifier.classify_batch(signal)
            probs = pred[0].squeeze(0).tolist()
            labels = pred[1][0]
            scores = {speechbrain_classifier.hparams.label_encoder.ind2lab[i]: p * 100 for i, p in enumerate(probs)}

            if labels[0] == 'en':
                result = self._simulate_accent_classification(audio_path)
                result["all_scores"] = scores
                return result
            return {
                "accent_type": labels[0],
                "confidence": max(probs) * 100,
                "explanation": f"Detected language: **{labels[0]}** ({max(probs)*100:.1f}%)",
                "all_scores": scores
            }
        except Exception as e:
            st.warning(f"Fallback to simulation: {e}")
            return self._simulate_accent_classification(audio_path)

def download_and_extract_audio(url_or_path, is_upload=False):
    temp_dir = tempfile.mkdtemp()
    video_path = os.path.join(temp_dir, "video.mp4")
    audio_path = os.path.join(temp_dir, "audio.wav")

    if is_upload:
        with open(video_path, "wb") as f:
            f.write(url_or_path.read())
    else:
        with requests.get(url_or_path, stream=True) as r:
            r.raise_for_status()
            with open(video_path, 'wb') as f:
                for chunk in r.iter_content(chunk_size=8192):
                    f.write(chunk)

    (
        ffmpeg
        .input(video_path)
        .output(audio_path, ar=16000, ac=1, format='wav')
        .run(quiet=True, overwrite_output=True)
    )
    return audio_path

# --- Streamlit App ---
st.set_page_config(page_title="Accent Analyzer", layout="wide")
st.title("🗣️ English Accent or Language Analyzer")

st.markdown("Upload a video/audio file or provide a direct `.mp4` or `.wav` URL:")

url = st.text_input("🔗 Enter Direct MP4/WAV URL:")
uploaded_file = st.file_uploader("📁 Or upload a file (MP4/WAV)", type=["mp4", "wav"])

if st.button("Analyze"):
    if not url and not uploaded_file:
        st.error("Please enter a valid URL or upload a file.")
    else:
        try:
            with st.spinner("Processing audio..."):
                audio_path = download_and_extract_audio(uploaded_file if uploaded_file else url, is_upload=bool(uploaded_file))
                analyzer = AccentAnalyzer()
                results = analyzer.analyze_accent(audio_path)

            st.success(results["explanation"])

            labels, values = zip(*results["all_scores"].items())
            fig, ax = plt.subplots()
            ax.bar(labels, values, color='skyblue')
            ax.set_ylabel('Confidence (%)')
            ax.set_title('Accent/Language Confidence')
            plt.xticks(rotation=45)
            st.pyplot(fig)

        except Exception as e:
            st.error(f"Failed to analyze: {e}")