Update app.py

app.py

@@ -4,7 +4,6 @@ import sys
 import logging
 import traceback
 
-
 # Configure logging
 logging.basicConfig(
     level=logging.INFO,
@@ -47,7 +46,7 @@ try:
     from transformers import Wav2Vec2ForCTC, AutoProcessor, VitsModel, AutoTokenizer
     from transformers import MarianMTModel, MarianTokenizer
     from werkzeug.utils import secure_filename
-    
+
     logger.info("✅ All required libraries imported successfully")
 except ImportError as e:
     logger.critical(f"❌ Failed to import necessary libraries: {str(e)}")
@@ -77,7 +76,7 @@ try:
         cache_dir=cache_dirs["TRANSFORMERS_CACHE"]
     )
     logger.info("✅ ASR processor loaded successfully")
-    
+
     asr_model = Wav2Vec2ForCTC.from_pretrained(
         ASR_MODEL_ID,
         cache_dir=cache_dirs["TRANSFORMERS_CACHE"]
@@ -113,13 +112,13 @@ for lang, model_id in TTS_MODELS.items():
     logger.info(f"🔄 Loading TTS model for {lang}: {model_id}")
     try:
         tts_processors[lang] = AutoTokenizer.from_pretrained(
-            model_id, 
+            model_id,
             cache_dir=cache_dirs["TRANSFORMERS_CACHE"]
         )
         logger.info(f"✅ {lang} TTS processor loaded")
-        
+
         tts_models[lang] = VitsModel.from_pretrained(
-            model_id, 
+            model_id,
             cache_dir=cache_dirs["TRANSFORMERS_CACHE"]
         )
         tts_models[lang].to(device)
@@ -135,7 +134,7 @@ TRANSLATION_MODELS = {
     "eng-pam": "Coco-18/opus-mt-en-pam",
     "tgl-eng": "Helsinki-NLP/opus-mt-tl-en",
     "eng-tgl": "Helsinki-NLP/opus-mt-en-tl",
-    "phi": "Coco-18/opus-mt-phi"  
+    "phi": "Coco-18/opus-mt-phi"
 }
 
 logger.info(f"🔄 Loading Translation model: {TRANSLATION_MODELS}")
@@ -146,14 +145,14 @@ translation_tokenizers = {}
 
 for model_key, model_id in TRANSLATION_MODELS.items():
     logger.info(f"🔄 Loading Translation model: {model_id}")
-    
+
     try:
         translation_tokenizers[model_key] = MarianTokenizer.from_pretrained(
             model_id,
             cache_dir=cache_dirs["TRANSFORMERS_CACHE"]
         )
         logger.info(f"✅ Translation tokenizer loaded successfully for {model_key}")
-        
+
         translation_models[model_key] = MarianMTModel.from_pretrained(
             model_id,
             cache_dir=cache_dirs["TRANSFORMERS_CACHE"]
@@ -169,7 +168,7 @@ for model_key, model_id in TRANSLATION_MODELS.items():
 # Constants
 SAMPLE_RATE = 16000
 OUTPUT_DIR = "/tmp/audio_outputs"
-REFERENCE_AUDIO_DIR = "./reference_audios" 
+REFERENCE_AUDIO_DIR = "./reference_audios"
 
 try:
     os.makedirs(OUTPUT_DIR, exist_ok=True)
@@ -177,43 +176,47 @@ try:
 except Exception as e:
     logger.error(f"❌ Failed to create output directory: {str(e)}")
 
+
 @app.route("/", methods=["GET"])
 def home():
     return jsonify({"message": "Speech API is running", "status": "active"})
 
+
 @app.route("/health", methods=["GET"])
 def health_check():
     # Initialize direct language pair statuses based on loaded models
     translation_status = {}
-    
+
     # Add status for direct model pairs
     for lang_pair in ["pam-eng", "eng-pam", "tgl-eng", "eng-tgl"]:
-        translation_status[lang_pair] = "loaded" if lang_pair in translation_models and translation_models[lang_pair] is not None else "failed"
-    
+        translation_status[lang_pair] = "loaded" if lang_pair in translation_models and translation_models[
+            lang_pair] is not None else "failed"
+
     # Add special phi model status
     phi_status = "loaded" if "phi" in translation_models and translation_models["phi"] is not None else "failed"
     translation_status["pam-fil"] = phi_status
     translation_status["fil-pam"] = phi_status
     translation_status["pam-tgl"] = phi_status  # Using phi model but replacing tgl with fil
     translation_status["tgl-pam"] = phi_status  # Using phi model but replacing tgl with fil
-    
+
     health_status = {
         "api_status": "online",
         "asr_model": "loaded" if asr_model is not None else "failed",
-        "tts_models": {lang: "loaded" if model is not None else "failed" 
-                      for lang, model in tts_models.items()},
+        "tts_models": {lang: "loaded" if model is not None else "failed"
+                       for lang, model in tts_models.items()},
         "translation_models": translation_status,
         "device": device
     }
     return jsonify(health_status)
 
+
 @app.route("/check_references", methods=["GET"])
 def check_references():
     """Endpoint to check if reference files exist and are accessible"""
-    ref_patterns = ["mayap_a_abak", "mayap_a_ugtu", "mayap_a_gatpanapun", 
+    ref_patterns = ["mayap_a_abak", "mayap_a_ugtu", "mayap_a_gatpanapun",
                     "mayap_a_bengi", "komusta_ka"]
     results = {}
-    
+
     for pattern in ref_patterns:
         pattern_dir = os.path.join(REFERENCE_AUDIO_DIR, pattern)
         if os.path.exists(pattern_dir):
@@ -229,19 +232,20 @@ def check_references():
                 "exists": False,
                 "path": pattern_dir
             }
-    
+
     return jsonify({
         "reference_audio_dir": REFERENCE_AUDIO_DIR,
         "directory_exists": os.path.exists(REFERENCE_AUDIO_DIR),
         "patterns": results
     })
 
+
 @app.route("/asr", methods=["POST"])
 def transcribe_audio():
     if asr_model is None or asr_processor is None:
         logger.error("❌ ASR endpoint called but models aren't loaded")
         return jsonify({"error": "ASR model not available"}), 503
-        
+
     try:
         if "audio" not in request.files:
             logger.warning("⚠️ ASR request missing audio file")
@@ -252,7 +256,8 @@ def transcribe_audio():
 
         if language not in LANGUAGE_CODES:
             logger.warning(f"⚠️ Unsupported language requested: {language}")
-            return jsonify({"error": f"Unsupported language: {language}. Available: {list(LANGUAGE_CODES.keys())}"}), 400
+            return jsonify(
+                {"error": f"Unsupported language: {language}. Available: {list(LANGUAGE_CODES.keys())}"}), 400
 
         lang_code = LANGUAGE_CODES[language]
         logger.info(f"🔄 Processing {language} audio for ASR")
@@ -310,9 +315,9 @@ def transcribe_audio():
                 logits = asr_model(**inputs).logits
             ids = torch.argmax(logits, dim=-1)[0]
             transcription = asr_processor.decode(ids)
-            
+
             logger.info(f"✅ Transcription ({language}): {transcription}")
-            
+
             # Clean up temp files
             try:
                 os.unlink(temp_audio_path)
@@ -320,7 +325,7 @@ def transcribe_audio():
                     os.unlink(wav_path)
             except Exception as e:
                 logger.warning(f"⚠️ Failed to clean up temp files: {str(e)}")
-                
+
             return jsonify({
                 "transcription": transcription,
                 "language": language,
@@ -344,24 +349,24 @@ def generate_tts():
         if not data:
             logger.warning("⚠️ TTS endpoint called with no JSON data")
             return jsonify({"error": "No JSON data provided"}), 400
-            
+
         text_input = data.get("text", "").strip()
         language = data.get("language", "kapampangan").lower()
 
         if not text_input:
             logger.warning("⚠️ TTS request with empty text")
             return jsonify({"error": "No text provided"}), 400
-            
+
         if language not in TTS_MODELS:
             logger.warning(f"⚠️ TTS requested for unsupported language: {language}")
             return jsonify({"error": f"Invalid language. Available options: {list(TTS_MODELS.keys())}"}), 400
-            
+
         if tts_models[language] is None:
             logger.error(f"❌ TTS model for {language} not loaded")
             return jsonify({"error": f"TTS model for {language} not available"}), 503
 
         logger.info(f"🔄 Generating TTS for language: {language}, text: '{text_input}'")
-        
+
         try:
             processor = tts_processors[language]
             model = tts_models[language]
@@ -409,10 +414,11 @@ def download_audio(filename):
     if os.path.exists(file_path):
         logger.info(f"📤 Serving audio file: {file_path}")
         return send_file(file_path, mimetype="audio/wav", as_attachment=True)
-    
+
     logger.warning(f"⚠️ Requested file not found: {file_path}")
     return jsonify({"error": "File not found"}), 404
 
+
 @app.route("/translate", methods=["POST"])
 def translate_text():
     try:
@@ -420,7 +426,7 @@ def translate_text():
         if not data:
             logger.warning("⚠️ Translation endpoint called with no JSON data")
             return jsonify({"error": "No JSON data provided"}), 400
-            
+
         source_text = data.get("text", "").strip()
         source_language = data.get("source_language", "").lower()
         target_language = data.get("target_language", "").lower()
@@ -428,18 +434,18 @@ def translate_text():
         if not source_text:
             logger.warning("⚠️ Translation request with empty text")
             return jsonify({"error": "No text provided"}), 400
-        
+
         # Map language names to codes
         source_code = LANGUAGE_CODES.get(source_language, source_language)
         target_code = LANGUAGE_CODES.get(target_language, target_language)
-        
+
         logger.info(f"🔄 Translating from {source_language} to {target_language}: '{source_text}'")
-        
+
         # Special handling for pam-fil, fil-pam, pam-tgl and tgl-pam using the phi model
         use_phi_model = False
         actual_source_code = source_code
         actual_target_code = target_code
-        
+
         # Check if we need to use the phi model with fil replacement
         if (source_code == "pam" and target_code == "fil") or (source_code == "fil" and target_code == "pam"):
             use_phi_model = True
@@ -449,38 +455,38 @@ def translate_text():
         elif (source_code == "tgl" and target_code == "pam"):
             use_phi_model = True
             actual_source_code = "fil"  # Replace tgl with fil for the phi model
-            
+
         if use_phi_model:
             model_key = "phi"
-            
+
             # Check if we have the phi model
             if model_key not in translation_models or translation_models[model_key] is None:
                 logger.error(f"❌ Translation model for {model_key} not loaded")
                 return jsonify({"error": f"Translation model not available"}), 503
-                
+
             try:
                 # Get the phi model and tokenizer
                 model = translation_models[model_key]
                 tokenizer = translation_tokenizers[model_key]
-                
+
                 # Prepend target language token to input
                 input_text = f">>{actual_target_code}<< {source_text}"
-                
+
                 logger.info(f"🔄 Using phi model with input: '{input_text}'")
-                
+
                 # Tokenize the text
                 tokenized = tokenizer(input_text, return_tensors="pt", padding=True)
                 tokenized = {k: v.to(device) for k, v in tokenized.items()}
-                
+
                 # Generate translation
                 with torch.no_grad():
                     translated = model.generate(**tokenized)
-                
+
                 # Decode the translation
                 result = tokenizer.decode(translated[0], skip_special_tokens=True)
-                
+
                 logger.info(f"✅ Translation result: '{result}'")
-                
+
                 return jsonify({
                     "translated_text": result,
                     "source_language": source_language,
@@ -493,34 +499,35 @@ def translate_text():
         else:
             # Create the regular language pair key for other language pairs
             lang_pair = f"{source_code}-{target_code}"
-                
+
             # Check if we have a model for this language pair
             if lang_pair not in translation_models:
                 logger.warning(f"⚠️ No translation model available for {lang_pair}")
-                return jsonify({"error": f"Translation from {source_language} to {target_language} is not supported yet"}), 400
-            
+                return jsonify(
+                    {"error": f"Translation from {source_language} to {target_language} is not supported yet"}), 400
+
             if translation_models[lang_pair] is None or translation_tokenizers[lang_pair] is None:
                 logger.error(f"❌ Translation model for {lang_pair} not loaded")
                 return jsonify({"error": f"Translation model not available"}), 503
-                
+
             try:
                 # Regular translation process for other language pairs
                 model = translation_models[lang_pair]
                 tokenizer = translation_tokenizers[lang_pair]
-                
+
                 # Tokenize the text
                 tokenized = tokenizer(source_text, return_tensors="pt", padding=True)
                 tokenized = {k: v.to(device) for k, v in tokenized.items()}
-                
+
                 # Generate translation
                 with torch.no_grad():
                     translated = model.generate(**tokenized)
-                
+
                 # Decode the translation
                 result = tokenizer.decode(translated[0], skip_special_tokens=True)
-                
+
                 logger.info(f"✅ Translation result: '{result}'")
-                
+
                 return jsonify({
                     "translated_text": result,
                     "source_language": source_language,
@@ -530,30 +537,33 @@ def translate_text():
                 logger.error(f"❌ Translation processing failed: {str(e)}")
                 logger.debug(f"Stack trace: {traceback.format_exc()}")
                 return jsonify({"error": f"Translation processing failed: {str(e)}"}), 500
-                
+
     except Exception as e:
         logger.error(f"❌ Unhandled exception in translation endpoint: {str(e)}")
         logger.debug(f"Stack trace: {traceback.format_exc()}")
         return jsonify({"error": f"Internal server error: {str(e)}"}), 500
 
+
 # Add this function to your app.py
 def calculate_similarity(text1, text2):
     """Calculate text similarity percentage."""
+
     def clean_text(text):
         return text.lower()
-    
+
     clean1 = clean_text(text1)
     clean2 = clean_text(text2)
-    
+
     matcher = SequenceMatcher(None, clean1, clean2)
     return matcher.ratio() * 100
 
+
 @app.route("/evaluate", methods=["POST"])
 def evaluate_pronunciation():
     if asr_model is None or asr_processor is None:
         logger.error("❌ Evaluation endpoint called but ASR models aren't loaded")
         return jsonify({"error": "ASR model not available"}), 503
-        
+
     try:
         if "audio" not in request.files:
             logger.warning("⚠️ Evaluation request missing audio file")
@@ -570,17 +580,25 @@ def evaluate_pronunciation():
 
         # Construct full reference directory path
         reference_dir = os.path.join(REFERENCE_AUDIO_DIR, reference_locator)
+        logger.info(f"📁 Reference directory path: {reference_dir}")
+
         if not os.path.exists(reference_dir):
             logger.warning(f"⚠️ Reference directory not found: {reference_dir}")
             return jsonify({"error": f"Reference audio directory not found: {reference_locator}"}), 404
 
         reference_files = glob.glob(os.path.join(reference_dir, "*.wav"))
+        logger.info(f"📁 Reference files found: {len(reference_files)}")
+
         if not reference_files:
             logger.warning(f"⚠️ No reference audio files found in {reference_dir}")
             return jsonify({"error": f"No reference audio found for {reference_locator}"}), 404
 
+        # Log actual file paths for debugging
+        for ref_file in reference_files:
+            logger.debug(f"📁 Reference file: {ref_file}")
+
         lang_code = LANGUAGE_CODES.get(language, language)
-        logger.info(f"🔄 Evaluating pronunciation for reference: {reference_locator}")
+        logger.info(f"🔄 Evaluating pronunciation for reference: {reference_locator} with language code: {lang_code}")
 
         # Save the uploaded file temporarily
         with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_audio:
@@ -590,22 +608,31 @@ def evaluate_pronunciation():
 
         # Convert to WAV if necessary and ensure correct format
         try:
-            # Load audio with librosa for consistent processing
-            user_waveform, sr = librosa.load(user_audio_path, sr=SAMPLE_RATE, mono=True)
-            
+            logger.info(f"🔄 Processing user audio file")
+            # First try using pydub for consistent processing
+            audio = AudioSegment.from_file(user_audio_path)
+            audio = audio.set_frame_rate(SAMPLE_RATE).set_channels(1)
+
             # Save processed audio
             processed_path = os.path.join(OUTPUT_DIR, "processed_user_audio.wav")
-            sf.write(processed_path, user_waveform, SAMPLE_RATE)
+            audio.export(processed_path, format="wav")
             logger.debug(f"📁 Processed user audio saved to {processed_path}")
-            
+
+            # Load the processed audio for ASR
+            user_waveform, sr = torchaudio.load(processed_path)
+            user_waveform = user_waveform.squeeze().numpy()
+            logger.info(f"✅ User audio processed successfully: {sr}Hz, length: {len(user_waveform)} samples")
+
             # Update user_audio_path to processed file
             user_audio_path = processed_path
         except Exception as e:
             logger.error(f"❌ Audio processing failed: {str(e)}")
+            logger.debug(f"Stack trace: {traceback.format_exc()}")
             return jsonify({"error": f"Audio processing failed: {str(e)}"}), 500
 
         # Transcribe user audio
         try:
+            logger.info(f"🔄 Transcribing user audio")
             # Process audio for ASR
             inputs = asr_processor(
                 user_waveform,
@@ -614,14 +641,14 @@ def evaluate_pronunciation():
                 language=lang_code
             )
             inputs = {k: v.to(device) for k, v in inputs.items()}
-            
+
             # Perform ASR
             with torch.no_grad():
                 logits = asr_model(**inputs).logits
             ids = torch.argmax(logits, dim=-1)[0]
             user_transcription = asr_processor.decode(ids)
-            
-            logger.info(f"✅ User transcription: {user_transcription}")
+
+            logger.info(f"✅ User transcription: '{user_transcription}'")
         except Exception as e:
             logger.error(f"❌ ASR inference failed: {str(e)}")
             return jsonify({"error": f"ASR inference failed: {str(e)}"}), 500
@@ -631,13 +658,23 @@ def evaluate_pronunciation():
         best_score = 0
         best_reference = None
         best_transcription = None
-        
+
+        logger.info(f"🔄 Beginning comparison with {len(reference_files)} reference files")
+
         for ref_file in reference_files:
             try:
-                # Load reference audio
-                ref_waveform, _ = librosa.load(ref_file, sr=SAMPLE_RATE, mono=True)
-                
+                logger.info(f"🔄 Processing reference file: {os.path.basename(ref_file)}")
+
+                # Load reference audio using torchaudio instead of librosa
+                ref_waveform, ref_sr = torchaudio.load(ref_file)
+                if ref_sr != SAMPLE_RATE:
+                    logger.debug(f"🔄 Resampling reference audio from {ref_sr}Hz to {SAMPLE_RATE}Hz")
+                    ref_waveform = torchaudio.transforms.Resample(ref_sr, SAMPLE_RATE)(ref_waveform)
+                ref_waveform = ref_waveform.squeeze().numpy()
+                logger.debug(f"✅ Reference audio loaded: {len(ref_waveform)} samples")
+
                 # Transcribe reference audio
+                logger.debug(f"🔄 Transcribing reference audio")
                 inputs = asr_processor(
                     ref_waveform,
                     sampling_rate=SAMPLE_RATE,
@@ -645,41 +682,44 @@ def evaluate_pronunciation():
                     language=lang_code
                 )
                 inputs = {k: v.to(device) for k, v in inputs.items()}
-                
+
                 with torch.no_grad():
                     logits = asr_model(**inputs).logits
                 ids = torch.argmax(logits, dim=-1)[0]
                 ref_transcription = asr_processor.decode(ids)
-                
+                logger.info(f"✅ Reference transcription: '{ref_transcription}'")
+
                 # Calculate similarity
                 similarity = calculate_similarity(user_transcription, ref_transcription)
-                
+                logger.info(f"📊 Similarity with {os.path.basename(ref_file)}: {similarity:.2f}%")
+
                 results.append({
                     "reference_file": os.path.basename(ref_file),
                     "reference_text": ref_transcription,
                     "similarity_score": similarity
                 })
-                
+
                 if similarity > best_score:
                     best_score = similarity
                     best_reference = os.path.basename(ref_file)
                     best_transcription = ref_transcription
-                    
-                logger.debug(f"📊 Reference '{os.path.basename(ref_file)}': {similarity:.2f}%")
+                    logger.info(f"📊 New best match: {best_reference} with score {best_score:.2f}%")
             except Exception as e:
                 logger.error(f"❌ Error processing reference audio {ref_file}: {str(e)}")
-        
+                logger.debug(f"Stack trace: {traceback.format_exc()}")
+
         # Clean up temp files
         try:
             if os.path.exists(user_audio_path) and user_audio_path != processed_path:
                 os.unlink(user_audio_path)
+                logger.debug(f"🧹 Cleaned up temporary file: {user_audio_path}")
         except Exception as e:
             logger.warning(f"⚠️ Failed to clean up temp files: {str(e)}")
-        
+
         # Enhanced feedback based on score range
         is_correct = best_score >= 70.0
         feedback = ""
-        
+
         if best_score >= 90.0:
             feedback = "Perfect pronunciation! Excellent job!"
         elif best_score >= 80.0:
@@ -690,10 +730,13 @@ def evaluate_pronunciation():
             feedback = "Fair attempt. Try focusing on the syllables that differ from the sample."
         else:
             feedback = "Try again. Listen carefully to the sample pronunciation."
-        
+
+        logger.info(f"📊 Final evaluation results: score={best_score:.2f}%, is_correct={is_correct}")
+        logger.info(f"📝 Feedback: '{feedback}'")
+
         # Sort results by score descending
         results.sort(key=lambda x: x["similarity_score"], reverse=True)
-        
+
         return jsonify({
             "is_correct": is_correct,
             "score": best_score,
@@ -703,7 +746,7 @@ def evaluate_pronunciation():
             "reference_locator": reference_locator,
             "details": results
         })
-        
+
     except Exception as e:
         logger.error(f"❌ Unhandled exception in evaluation endpoint: {str(e)}")
         logger.debug(f"Stack trace: {traceback.format_exc()}")
@@ -723,10 +766,10 @@ def upload_reference_audio():
 
         # Validate reference word
         reference_patterns = [
-            "mayap_a_abak", "mayap_a_ugtu", "mayap_a_gatpanapun", 
+            "mayap_a_abak", "mayap_a_ugtu", "mayap_a_gatpanapun",
             "mayap_a_bengi", "komusta_ka"
         ]
-        
+
         if reference_word not in reference_patterns:
             logger.warning(f"⚠️ Invalid reference word: {reference_word}")
             return jsonify({"error": f"Invalid reference word. Available: {reference_patterns}"}), 400
@@ -771,21 +814,22 @@ def upload_reference_audio():
         logger.debug(f"Stack trace: {traceback.format_exc()}")
         return jsonify({"error": f"Internal server error: {str(e)}"}), 500
 
+
 def init_reference_audio():
     try:
         # Create the output directory first
         os.makedirs(OUTPUT_DIR, exist_ok=True)
         logger.info(f"📁 Created output directory: {OUTPUT_DIR}")
-        
+
         # Check if the reference audio directory exists in the repository
         if os.path.exists(REFERENCE_AUDIO_DIR):
             logger.info(f"✅ Found reference audio directory: {REFERENCE_AUDIO_DIR}")
-            
+
             # Log the contents to verify
-            pattern_dirs = [d for d in os.listdir(REFERENCE_AUDIO_DIR) 
-                          if os.path.isdir(os.path.join(REFERENCE_AUDIO_DIR, d))]
+            pattern_dirs = [d for d in os.listdir(REFERENCE_AUDIO_DIR)
+                            if os.path.isdir(os.path.join(REFERENCE_AUDIO_DIR, d))]
             logger.info(f"📁 Found reference patterns: {pattern_dirs}")
-            
+
             # Check each pattern directory for wav files
             for pattern_dir_name in pattern_dirs:
                 pattern_path = os.path.join(REFERENCE_AUDIO_DIR, pattern_dir_name)
@@ -796,6 +840,7 @@ def init_reference_audio():
     except Exception as e:
         logger.error(f"❌ Failed to set up reference audio directory: {str(e)}")
 
+
 # Add an initialization route that will be called before the first request
 @app.before_request
 def before_request():
@@ -804,12 +849,11 @@ def before_request():
         g.initialized = True
 
 
-
 if __name__ == "__main__":
     init_reference_audio()
     logger.info("🚀 Starting Speech API server")
     logger.info(f"📊 System status: ASR model: {'✅' if asr_model else '❌'}")
     for lang, model in tts_models.items():
         logger.info(f"📊 TTS model {lang}: {'✅' if model else '❌'}")
-    
+
     app.run(host="0.0.0.0", port=7860, debug=True)