import streamlit as st
from google.oauth2 import service_account
from google.cloud import speech
import io
import torch
import numpy as np
from transformers import Wav2Vec2Processor
from transformers.models.wav2vec2 import Wav2Vec2Model
import librosa
from groq import Groq
import sounddevice as sd
import scipy.io.wavfile as wav
import os
from datetime import datetime
from pydub import AudioSegment
from pathlib import Path
from openai import OpenAI
import json
import plotly.graph_objects as go
import os
OpenAI_api_key =os.environ.get('OpenAI_api_key')
Groq_api_key = os.environ.get('Groq_api_key')
credentials_json = os.environ.get('google_creds')
google_creds = json.loads(credentials_json)
# Enhanced UI Styles
CUSTOM_CSS = """
"""
class AzanTrainerApp:
def __init__(self):
self.setup_api_clients()
self.setup_configs()
self.setup_directories()
self.initialize_models()
def setup_api_clients(self):
"""Initialize API clients"""
self.openai_client = OpenAI(api_key=OpenAI_api_key)
self.groq_client = Groq(api_key=Groq_api_key)
self.speech_client = self.init_google_speech()
def init_google_speech(self):
"""Initialize Google Speech client"""
credentials = service_account.Credentials.from_service_account_info(google_creds)
return speech.SpeechClient(credentials=credentials)
def setup_configs(self):
"""Set up configuration variables"""
self.SAMPLE_RATE = 48000
self.DURATION = 6
self.AUDIO_GAIN = 1.50
self.IDEAL_TEXT = "اللّٰهُ أَكْبَرُ، اللّٰهُ أَكْبَرُ"
self.IDEAL_TEXT_MEANING = "Allah is the Greatest, Allah is the Greatest"
def setup_directories(self):
"""Create necessary directories"""
for dir_name in ['recordings', 'feedback_audio']:
os.makedirs(dir_name, exist_ok=True)
def initialize_models(self):
"""Initialize ML models"""
self.processor = Wav2Vec2Processor.from_pretrained("models/wav2vec2-base")
self.model = Wav2Vec2Model.from_pretrained("models/wav2vec2-base")
self.ideal_embedding = torch.tensor(np.load("ideal_embedding_part_1.npy"))
def create_waveform_visualization(self, audio_path, reference_path):
"""Create waveform visualization using Plotly"""
fig = go.Figure()
# Process user audio
y_user, sr_user = librosa.load(audio_path)
times_user = np.arange(len(y_user)) / sr_user
fig.add_trace(go.Scatter(
x=times_user,
y=y_user,
name='Your Recording',
line=dict(color='#1E88E5')
))
# Process reference audio
y_ref, sr_ref = librosa.load(reference_path)
times_ref = np.arange(len(y_ref)) / sr_ref
fig.add_trace(go.Scatter(
x=times_ref,
y=y_ref,
name='Expert Recording',
line=dict(color='#4CAF50')
))
fig.update_layout(
title='Waveform Comparison',
xaxis_title='Time (s)',
yaxis_title='Amplitude',
template='plotly_white',
height=400
)
return fig
def record_audio(self):
"""Record audio from user"""
try:
audio_data = sd.rec(
int(self.DURATION * self.SAMPLE_RATE),
samplerate=self.SAMPLE_RATE,
channels=1,
dtype=np.float32
)
sd.wait()
return self.enhance_audio(audio_data)
except Exception as e:
st.error(f"Recording error: {str(e)}")
return None
def enhance_audio(self, audio_data):
"""Enhance audio quality"""
audio_data = audio_data / (np.max(np.abs(audio_data)) + 1e-10)
audio_data = audio_data * self.AUDIO_GAIN
noise_threshold = 0.01
audio_data[np.abs(audio_data) < noise_threshold] = 0
return audio_data
def save_audio(self, audio_data):
"""Save audio to file"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"recordings/audio_{timestamp}.wav"
audio_data = np.clip(audio_data * 32767, -32768, 32767).astype(np.int16)
wav.write(filename, self.SAMPLE_RATE, audio_data)
return filename
def analyze_recording(self, audio_path):
"""Analyze the recording"""
try:
# Convert to MP3 for Google Speech API
mp3_path = audio_path.replace('.wav', '.mp3')
AudioSegment.from_wav(audio_path).export(mp3_path, format="mp3")
# Transcribe audio
with open(mp3_path, 'rb') as f:
content = f.read()
audio = speech.RecognitionAudio(content=content)
config = speech.RecognitionConfig(
encoding=speech.RecognitionConfig.AudioEncoding.MP3,
sample_rate_hertz=self.SAMPLE_RATE,
language_code="ar"
)
response = self.speech_client.recognize(config=config, audio=audio)
transcription = " ".join(result.alternatives[0].transcript
for result in response.results)
# Calculate similarity
user_embedding = self.get_audio_embedding(audio_path)
similarity_score = self.calculate_similarity(user_embedding, self.ideal_embedding)
# Generate feedback
feedback = self.generate_feedback(transcription, similarity_score)
# Clean up
os.remove(mp3_path)
return transcription, similarity_score, feedback
except Exception as e:
st.error(f"Analysis error: {str(e)}")
return None, None, None
def get_audio_embedding(self, audio_path):
"""Generate audio embedding"""
audio_input, _ = librosa.load(audio_path, sr=16000)
inputs = self.processor(audio_input, sampling_rate=16000,
return_tensors="pt", padding=True)
with torch.no_grad():
embedding = self.model(inputs.input_values).last_hidden_state.mean(dim=1).squeeze()
return embedding
def calculate_similarity(self, embedding1, embedding2):
"""Calculate similarity score"""
similarity = torch.nn.functional.cosine_similarity(embedding1, embedding2, dim=0)
return similarity.item() * 100
def generate_feedback(self, transcription, similarity_score):
"""Generate feedback in natural Roman Urdu using LLM"""
prompt = f"""
Is Azan ki tilawat ka jaiza len aur natural Roman Urdu main feedback den:
Tilawat: {transcription}
Mutabiqat Score: {similarity_score:.2f}%
Feedback ko in 3 hisson main takseem karen:
1. Talaffuz (Pronunciation):
- Har lafz ka talaffuz kaisa hai
- Huroof ki tartib theek hai ya nahi
- Allah ke lafz ka talaffuz kaisa hai
- Mukammal Azan ki tarteeb kaisi hai
2. Waqt aur Lehja (Timing):
- Har hissay ka sahi dohrao
- Waqfay ki durustagi
- Aawaz ka utaar chadhao
3. Behtar Karne Ke Liye Mashwaray:
- Kahan ghaltiyan hain
- Kya behtar karna hai
- Kis cheez par zyada mehnat ki zaroorat hai
Note: Feedback zabaan-e-urdu main likhen, lekin English huroof istimal karen.
Lehja mohtaram aur madadgaar hona chahiye.
"""
response = self.groq_client.chat.completions.create(
model="llama3-70b-8192",
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=1000
)
return response.choices[0].message.content
def generate_audio_feedback(self, feedback_text):
"""Generate audio feedback"""
try:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
audio_path = f"feedback_audio/feedback_{timestamp}.mp3"
response = self.openai_client.audio.speech.create(
model="tts-1",
voice="alloy",
input=feedback_text
)
response.stream_to_file(audio_path)
return audio_path
except Exception as e:
st.error(f"Error generating audio feedback: {str(e)}")
return None
def run(self):
"""Run the enhanced Streamlit application with Persian/Masjid-inspired UI"""
st.set_page_config(
page_title="Azan Pronunciation Trainer",
layout="wide",
initial_sidebar_state="collapsed"
)
# Custom CSS with Persian/Masjid-inspired theme (Keep your existing CSS here)
st.markdown("""
""", unsafe_allow_html=True)
# Enhanced Header with Arabic Styling
st.markdown(f"""
""", unsafe_allow_html=True)
# Expert demonstration card
st.markdown("""
""", unsafe_allow_html=True)
st.video("Sequence 03_1.mp4")
st.markdown("
", unsafe_allow_html=True)
# Expert audio card
st.markdown("""
""", unsafe_allow_html=True)
st.audio("qari_part_1.mp3")
st.markdown("
", unsafe_allow_html=True)
# Recording controls card
st.markdown("""
""", unsafe_allow_html=True)
col1, col2 = st.columns(2)
with col1:
if st.button("Start Recording", help="Click to start recording (6 seconds)", key="start_rec"):
with st.spinner("Recording in progress..."):
audio_data = self.record_audio()
if audio_data is not None:
audio_path = self.save_audio(audio_data)
st.session_state['audio_file'] = audio_path
st.markdown("""
Recording completed successfully! ✅
""", unsafe_allow_html=True)
with col2:
if st.button("Clear Recording", key="clear_rec"):
if 'audio_file' in st.session_state:
if os.path.exists(st.session_state['audio_file']):
os.remove(st.session_state['audio_file'])
st.session_state['audio_file'] = None
st.markdown("""
Recording cleared! 🗑️
""", unsafe_allow_html=True)
st.markdown("
""", unsafe_allow_html=True)
st.audio(st.session_state['audio_file'])
if st.button("Analyze Recording", key="analyze"):
with st.spinner("Analyzing your recitation..."):
transcription, similarity, feedback = self.analyze_recording(
st.session_state['audio_file']
)
if all([transcription, similarity, feedback]):
# Enhanced similarity score display
st.markdown(f"""
{similarity:.1f}%
Similarity Score
""", unsafe_allow_html=True)
# Waveform visualization
fig = self.create_waveform_visualization(
st.session_state['audio_file'],
"qari_part_1.mp3"
)
st.plotly_chart(fig, use_container_width=True)
# Feedback display
st.markdown(f"""
""", unsafe_allow_html=True)
# Audio feedback
audio_feedback_path = self.generate_audio_feedback(feedback)
if audio_feedback_path:
st.markdown("""
""", unsafe_allow_html=True)
st.audio(audio_feedback_path)
st.markdown("
", unsafe_allow_html=True)
st.markdown("
1
Watch the expert demonstration video carefully
2
Listen to the reference audio to understand proper pronunciation
3
Click 'Start Recording' and recite the phrase (6 seconds)
4
Wait for the recording to complete
5
Click 'Analyze Recording' to get detailed feedback
6
Review your score and feedback to improve
7
Practice until you achieve 90% or higher similarity
""", unsafe_allow_html=True)
if __name__ == "__main__":
app = AzanTrainerApp()
app.run()