app.py

import os
import shutil
import torchaudio
import gradio as gr
from speechbrain.inference import SpeakerRecognition
from fastapi import HTTPException

# Initialize the speaker verification model
speaker_verification = SpeakerRecognition.from_hparams(
    source="speechbrain/spkrec-ecapa-voxceleb",
    savedir="tmp_model"
)

# Temporary folder to save uploaded files
UPLOAD_FOLDER = "uploaded_audio"
os.makedirs(UPLOAD_FOLDER, exist_ok=True)

# Function to calculate similarity score
def get_similarity(audio_path1: str, audio_path2: str):
    try:
        # Load audio files
        signal1, _ = torchaudio.load(audio_path1)
        signal2, _ = torchaudio.load(audio_path2)

        # Get similarity score and prediction
        score, prediction = speaker_verification.verify_batch(signal1, signal2)
        return float(score), "Yes" if prediction else "No"
    except Exception as e:
        return str(e), None
    finally:
        # Clean up temporary files
        if os.path.exists(audio_path1):
            os.remove(audio_path1)
        if os.path.exists(audio_path2):
            os.remove(audio_path2)

# API function to compare voices
def compare_voices(file1, file2):
    # Save uploaded files temporarily
    file1_path = os.path.join(UPLOAD_FOLDER, file1.name)
    file2_path = os.path.join(UPLOAD_FOLDER, file2.name)

    with open(file1_path, "wb") as f1:
        f1.write(file1.read())
    with open(file2_path, "wb") as f2:
        f2.write(file2.read())

    # Get similarity score
    score, is_same_user = get_similarity(file1_path, file2_path)

    if is_same_user is None:
        return "Error: " + score  # This will return the error message

    return {"Similarity Score": f"{score:.4f}", "Same User Prediction": is_same_user}

# Create Gradio Interface for the API
api = gr.Interface(
    fn=compare_voices,
    inputs=[
        gr.Audio(source="upload", type="file", label="First Audio File"),
        gr.Audio(source="upload", type="file", label="Second Audio File")
    ],
    outputs="json",  # Output results as JSON
    live=False  # No live interface, just the API
)

# Launch the API as an HTTP server
api.queue().launch(server_name="0.0.0.0", server_port=8080, share=False)