import streamlit as st import torch import numpy as np from pydub import AudioSegment from transformers import pipeline, WhisperProcessor, AutoModelForSpeechSeq2Seq # AutoModelForSpeechSeq2Seq əlavə edildi import webrtcvad from phonemizer.backend import EspeakBackend from fastdtw import fastdtw from scipy.signal import find_peaks import librosa import io import os # Hugging Face Spaces-də `src` qovluğu strukturunu nəzərə alaraq BASE_DIR = os.path.dirname(os.path.abspath(__file__)) # src qovluğunun yolu PROJECT_ROOT = os.path.join(BASE_DIR, "..") # layihənin kök qovluğu model_path = os.path.join(PROJECT_ROOT, "model_whisper-large-v3-turbo-aze-60hours(part15_aug)-lab") tokenizer_path = os.path.join(PROJECT_ROOT, "tokenizer_whisper-large-v3-turbo-aze-60hours(part15_aug)-lab") # Model və tokenizer klasörlərinin varlığını kontrol et if not os.path.exists(model_path) or not os.path.exists(tokenizer_path): st.error(f"Hata: Model veya tokenizer klasörleri bulunamadı. Lütfen '{model_path}' ve '{tokenizer_path}' yollarını ve Hugging Face Space-deki dosya yapısını kontrol edin. Model fayllarının ({model_path}/pytorch_model.bin or model.safetensors) mövcud olduğundan əmin olun.") st.stop() # Uygulamayı durdur @st.cache_resource def load_asr_components(): """ASR modelini ve işlemcisini önbelleğe alır.""" try: # Önce işlemciyi yükle processor = WhisperProcessor.from_pretrained(tokenizer_path, language='aze', task='transcribe') # Modeli doğrudan Whisper model sınıfı olarak yükle # Bu, AutoModelForCTC hatasının önüne geçmek için kritikdir. # Eğer modeliniz safetensors formatındaysa, from_pretrained otomatik olarak onu tanıyacaktır. model = AutoModelForSpeechSeq2Seq.from_pretrained( model_path, torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32, device_map="auto" if torch.cuda.is_available() else None # GPU varsa otomatik olaraq cihazı seç ) # ASR pipeline-ı qur asr_pipeline = pipeline( "automatic-speech-recognition", model=model, # Yüklenen model obyektini ötürürük tokenizer=processor.tokenizer, feature_extractor=processor.feature_extractor, device=0 if torch.cuda.is_available() else -1, torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32, generate_kwargs={ "task": "transcribe", "language": "azerbaijani", "num_beams": 2, "condition_on_prev_tokens": False, "compression_ratio_threshold": 2.4, "temperature": (0.0, 0.2, 0.4), "logprob_threshold": -1.0, "return_timestamps": True, } ) return processor, asr_pipeline except Exception as e: st.error(f"ASR bileşenleri yüklenirken hata oluştu: {e}. Lütfen model ve tokenizer yollarını kontrol edin. Modelinizde `pytorch_model.bin` veya `model.safetensors` faylının olduğundan əmin olun.") return None, None processor, asr_pipeline = load_asr_components() class DictionAnalyzer: def __init__(self): self.phonemizer = EspeakBackend("az") def analyze_prosody(self, y, sr): try: # pyin fasilələr səbəbindən NaN dəyərlər qaytara bilər, bunları idarə edin pitch, _, _ = librosa.pyin(y, fmin=librosa.note_to_hz('C2'), fmax=librosa.note_to_hz('C7'), sr=sr, frame_length=2048, hop_length=512) # NaN dəyərləri sıfırla dolduraraq və ya qulaq ardına vuraraq orta dəyəri hesablayın pitch = pitch[~np.isnan(pitch)] # NaN dəyərləri çıxar if pitch.size == 0: pitch = np.array([0.0]) # Boş qalarsa sıfır ver except Exception: pitch = np.array([0.0]) # Hata durumunda da sıfır ver energy = librosa.feature.rms(y=y)[0] # Enerji pikleri boş enerji dizileri için handle peaks, _ = find_peaks(energy, height=np.mean(energy) if energy.size > 0 else 0) return { "average_pitch": np.mean(pitch), "pitch_variance": np.var(pitch), "energy_peaks": len(peaks) } def analyze_speed_and_fluency(self, transcript, audio_duration_seconds): words = len(transcript.split()) if audio_duration_seconds > 0: words_per_second = words / audio_duration_seconds else: words_per_second = 0 return {"words_per_second": words_per_second, "word_count": words} def compare_pronunciation(self, transcript, reference_text): phonetic_transcript = self.phonemizer.phonemize([transcript], strip=True) phonetic_reference = self.phonemizer.phonemize([reference_text], strip=True) def text_to_numeric(phonetic_text): if not phonetic_text or not phonetic_text[0]: return np.array([]) return np.array([ord(char) for char in phonetic_text[0]]) numeric_transcript = text_to_numeric(phonetic_transcript) numeric_reference = text_to_numeric(phonetic_reference) if numeric_transcript.size == 0 or numeric_reference.size == 0: return {"phonetic_distance": float('inf')} try: # DTW üçün numpy.ndarray obyektlərini istifadə edin dtw_distance, _ = fastdtw(numeric_transcript, numeric_reference, dist=lambda x, y: np.linalg.norm(x - y)) except ValueError: # Boş ardıcıllıqlar üçün DTW xətasını idarə et dtw_distance = float('inf') return {"phonetic_distance": dtw_distance} def evaluate_score(self, value, good_range, medium_range, is_lower_better=False): """ Dəyəri yaxşı, orta, pis kateqoriyasına uyğun olaraq 1-10 arası bal verir. is_lower_better: Əgər dəyər nə qədər aşağı olarsa, o qədər yaxşıdırsa True qeyd edin. """ if value is None or np.isnan(value) or value == float('inf'): # Geçərsiz dəyərləri idarə et return "Hesaplanamadı", 0 if is_lower_better: if value <= good_range[1]: return "Yaxşı", 10 elif good_range[1] < value <= medium_range[1]: return "Orta", 6 else: return "Pis", 2 else: if value >= good_range[0]: return "Yaxşı", 10 elif medium_range[0] <= value < good_range[0]: return "Orta", 6 else: return "Pis", 2 def full_analysis(self, audio_data, reference_text): # audio_data dictionary olaraq gəlməlidir: {"raw": np.array, "sampling_rate": int} y = audio_data["raw"] sr = audio_data["sampling_rate"] audio_duration = librosa.get_duration(y=y, sr=sr) if asr_pipeline is None: st.error("ASR pipeline yüklənmədi. Analiz yapılamıyor.") return None try: transcript_result = asr_pipeline({"raw": y, "sampling_rate": sr}) transcript = transcript_result['text'] except Exception as e: st.error(f"Transkripsiyon sırasında hata oluştu: {e}") transcript = "" # Boş transkript ile devam et prosody = self.analyze_prosody(y, sr) fluency = self.analyze_speed_and_fluency(transcript, audio_duration) pronunciation = self.compare_pronunciation(transcript, reference_text) # Dəyərləndirmə aralıqları: Bu aralıqlar, modelinizin performansı və istədiyiniz metriklərə görə ayarlanmalıdır. # Bunlar sadəcə nümunələrdir. prosody_avg_pitch_eval = self.evaluate_score(prosody["average_pitch"], (100, 250), (70, 300), is_lower_better=False) prosody_pitch_variance_eval = self.evaluate_score(prosody["pitch_variance"], (500, 3000), (200, 5000), is_lower_better=False) # energy_peaks üçün audio_duration'a görə bir oran düşünə bilərsiniz. Örneğin: 0.1-0.5 peak/saniyə yaxşı ola bilər. prosody_energy_peaks_eval = self.evaluate_score(prosody["energy_peaks"] / audio_duration if audio_duration > 0 else 0, (0.1, 0.5), (0.05, 0.7), is_lower_better=False) fluency_wps_eval = self.evaluate_score(fluency["words_per_second"], (2.0, 3.0), (1.5, 3.5), is_lower_better=False) pronunciation_phonetic_distance_eval = self.evaluate_score(pronunciation["phonetic_distance"], (0, 200), (201, 1000), is_lower_better=True) return { "transcript": transcript, "prosody": { "average_pitch": prosody["average_pitch"], "average_pitch_score": prosody_avg_pitch_eval, "pitch_variance": prosody["pitch_variance"], "pitch_variance_score": prosody_pitch_variance_eval, "energy_peaks": prosody["energy_peaks"], "energy_peaks_score": prosody_energy_peaks_eval }, "fluency": { "words_per_second": fluency["words_per_second"], "words_per_second_score": fluency_wps_eval, "word_count": fluency["word_count"] }, "pronunciation": { "phonetic_distance": pronunciation["phonetic_distance"], "phonetic_distance_score": pronunciation_phonetic_distance_eval } } # Streamlit UI st.title("Diksiyon Analiz Uygulaması") st.write("Ses dosyanızı yükləyin və referans metin girərək diksiyonunuzu analiz edin.") uploaded_file = st.file_uploader("Ses Dosyası Yükləyin (.wav, .mp3)", type=["wav", "mp3"]) reference_text_input = st.text_area("Referans Metin Girin", "Salam, hər vaxtınız xeyr, mənim adım Əlidir. Salam, Əli bəy. Sizə necə kömək edə bilərəm? Kartımın şifrəsini dəyişmək istəyirəm və şifrəmi unutmuşam. Şifrənizi dəyişdirmək üçün sizdən bəzi məlumatlar istəyəcəm. Zəhmət olmasa, adınızı, soyadınızı və bir də vəsigənin fin kodunu deyə bilərsiniz. Əlbəttə, deyə bilərəm. Mənim adım Ələl Əkbəlli. Fin kodum isə 7CRH7UE. Zəhmət olmasa, yenə təkrar bilərsiniz fin kodu? 7-C-R-H-7-U-E Təşəkkür edirəm, Əli bəy. Şifrənizi sıfırlamaq üçün sizə indi kod gəlməlidir telefona, gəldi, hal-hazırda? Bəli, bəli, kod gəldi. İndi parolu dəyişə bilərəm. Bəli, dəyişə bilərsiniz? Oldu, hazırda şifrəmi dəyişdirdim. Aha, hal-hazırda əməliyyatınız uğurla tamamlanıb və şifrəniz dəyişilib. Yeni şifrədən istifadə edə bilərsiniz, kart hesabınıza daxil olmaq üçün. Başqa sualınız varmı? Xeyr, çox təşəkkür edəm sizə. Buyurun, gününüz xoş geçsin. Sağ olun.") if uploaded_file is not None and reference_text_input: st.audio(uploaded_file, format='audio/wav') # Kullanıcının yüklədiyi səsi oynat with st.spinner('Ses dosyası işlənir...'): try: audio_bytes = uploaded_file.read() audio_segment = AudioSegment.from_file(io.BytesIO(audio_bytes)) audio_array = np.array(audio_segment.get_array_of_samples(), dtype=np.float32) # Ses stereo ise mono'ya çevir if audio_segment.channels == 2: audio_array = audio_array.reshape((-1, 2)).mean(axis=1) # Normalizasyon audio_array /= np.iinfo(audio_segment.array_type).max # Eğer sample rate 16000 değilse, yeniden örnekle if audio_segment.frame_rate != 16000: st.info(f"Ses dosyası {audio_segment.frame_rate} Hz. 16000 Hz'e dönüştürülür...") audio_array = librosa.resample(y=audio_array, orig_sr=audio_segment.frame_rate, target_sr=16000) audio_input_for_analysis = {"raw": audio_array, "sampling_rate": 16000} except Exception as e: st.error(f"Ses dosyası oxunarkən və ya işlənərkən hata oluştu: {e}") audio_input_for_analysis = None if audio_input_for_analysis: analyzer = DictionAnalyzer() st.write("Analiz edilir, lütfən gözləyin...") with st.spinner('Analiz tamamlanır...'): report = analyzer.full_analysis(audio_input_for_analysis, reference_text_input) if report: st.subheader("Analiz Sonuçları") st.write(f"**Transkript:** {report['transcript']}") st.subheader("Prosoya Analizi") st.write(f"Ortalama Ton: {report['prosody']['average_pitch']:.2f} Hz - **Kategori: {report['prosody']['average_pitch_score'][0]} ({report['prosody']['average_pitch_score'][1]}/10)**") st.write(f"Ton Farkı: {report['prosody']['pitch_variance']:.2f} - **Kategori: {report['prosody']['pitch_variance_score'][0]} ({report['prosody']['pitch_variance_score'][1]}/10)**") st.write(f"Enerji Pikleri: {report['prosody']['energy_peaks']} - **Kategori: {report['prosody']['energy_peaks_score'][0]} ({report['prosody']['energy_peaks_score'][1]}/10)** (Saniyedeki tepe noktası oranı)") st.subheader("Sürat ve Akıcılık Analizi") st.write(f"Saniyedeki Kelime Sayısı: {report['fluency']['words_per_second']:.2f} - **Kategori: {report['fluency']['words_per_second_score'][0]} ({report['fluency']['words_per_second_score'][1]}/10)**") st.write(f"Toplam Kelime Sayısı: {report['fluency']['word_count']}") st.subheader("Tələffüz Müqayisəsi") st.write(f"Fonetik Məsafə (DTW): {report['pronunciation']['phonetic_distance']:.2f} - **Kategori: {report['pronunciation']['phonetic_distance_score'][0]} ({report['pronunciation']['phonetic_distance_score'][1]}/10)**") else: st.error("Analiz tamamlanamadı.") elif uploaded_file is None and st.button("Analiz Et"): st.warning("Lütfen bir ses dosyası yükləyin və referans metin girin.")