Update app.py

app.py

@@ -4,6 +4,7 @@ import sys
 import logging
 import traceback
 
+
 # Configure logging
 logging.basicConfig(
     level=logging.INFO,
@@ -32,6 +33,10 @@ for env_var, path in cache_dirs.items():
 
 # Now import the rest of the libraries
 try:
+    import librosa
+    from difflib import SequenceMatcher
+    import glob
+    import numpy as np
     import torch
     from pydub import AudioSegment
     import tempfile
@@ -41,6 +46,8 @@ try:
     from flask_cors import CORS
     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)}")
@@ -162,6 +169,9 @@ for model_key, model_id in TRANSLATION_MODELS.items():
 # Constants
 SAMPLE_RATE = 16000
 OUTPUT_DIR = "/tmp/audio_outputs"
+# Update the constant
+REFERENCE_AUDIO_DIR = "/storage/reference_audio"
+
 try:
     os.makedirs(OUTPUT_DIR, exist_ok=True)
     logger.info(f"📁 Created output directory: {OUTPUT_DIR}")
@@ -498,6 +508,245 @@ def translate_text():
         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
+
+# Add this route to your Flask app
+@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")
+            return jsonify({"error": "No audio file uploaded"}), 400
+
+        audio_file = request.files["audio"]
+        reference_word = request.form.get("reference_word", "").strip()
+        language = request.form.get("language", "tagalog").lower()  # Default to tagalog for ASR
+
+        # Check if reference word is valid
+        reference_patterns = [
+            "mayap_a_abak", "mayap_a_ugtu", "mayap_a_gatpanapun", 
+            "mayap_a_bengi", "komusta_ka"
+        ]
+        
+        if not reference_word or 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
+
+        lang_code = LANGUAGE_CODES.get(language, language)
+        logger.info(f"🔄 Evaluating pronunciation of '{reference_word}' in {language}")
+
+        # Save the uploaded file temporarily
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_audio:
+            temp_audio.write(audio_file.read())
+            user_audio_path = temp_audio.name
+            logger.debug(f"📁 User audio saved to {user_audio_path}")
+
+        # 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)
+            
+            # Save processed audio
+            processed_path = os.path.join(OUTPUT_DIR, "processed_user_audio.wav")
+            sf.write(processed_path, user_waveform, SAMPLE_RATE)
+            logger.debug(f"📁 Processed user audio saved to {processed_path}")
+            
+            # Update user_audio_path to processed file
+            user_audio_path = processed_path
+        except Exception as e:
+            logger.error(f"❌ Audio processing failed: {str(e)}")
+            return jsonify({"error": f"Audio processing failed: {str(e)}"}), 500
+
+        # Find reference audio files
+        reference_dir = os.path.join(REFERENCE_AUDIO_DIR, reference_word)
+        if not os.path.exists(reference_dir):
+            logger.warning(f"⚠️ Reference directory not found: {reference_dir}")
+            return jsonify({"error": f"Reference audio for {reference_word} not found"}), 404
+            
+        reference_files = glob.glob(os.path.join(reference_dir, "*.wav"))
+        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_word}"}), 404
+            
+        logger.info(f"📊 Found {len(reference_files)} reference files for '{reference_word}'")
+
+        # Transcribe user audio
+        try:
+            # Process audio for ASR
+            inputs = asr_processor(
+                user_waveform,
+                sampling_rate=SAMPLE_RATE,
+                return_tensors="pt",
+                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}")
+        except Exception as e:
+            logger.error(f"❌ ASR inference failed: {str(e)}")
+            return jsonify({"error": f"ASR inference failed: {str(e)}"}), 500
+
+        # Compare with reference audios
+        results = []
+        best_score = 0
+        best_reference = None
+        
+        for ref_file in reference_files:
+            try:
+                # Load reference audio
+                ref_waveform, _ = librosa.load(ref_file, sr=SAMPLE_RATE, mono=True)
+                
+                # Transcribe reference audio
+                inputs = asr_processor(
+                    ref_waveform,
+                    sampling_rate=SAMPLE_RATE,
+                    return_tensors="pt",
+                    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)
+                
+                # Calculate similarity
+                similarity = calculate_similarity(user_transcription, ref_transcription)
+                
+                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)
+                    
+                logger.debug(f"📊 Reference '{os.path.basename(ref_file)}': {similarity:.2f}%")
+            except Exception as e:
+                logger.error(f"❌ Error processing reference audio {ref_file}: {str(e)}")
+        
+        # Clean up temp files
+        try:
+            if os.path.exists(user_audio_path) and user_audio_path != processed_path:
+                os.unlink(user_audio_path)
+        except Exception as e:
+            logger.warning(f"⚠️ Failed to clean up temp files: {str(e)}")
+        
+        # Decision on pronunciation correctness (70% threshold)
+        is_correct = best_score >= 70.0
+        feedback = "Great pronunciation!" if is_correct else "Try again! Listen to the sample"
+        
+        return jsonify({
+            "is_correct": is_correct,
+            "score": best_score,
+            "feedback": feedback,
+            "transcription": user_transcription,
+            "reference_word": reference_word,
+            "details": results
+        })
+        
+    except Exception as e:
+        logger.error(f"❌ Unhandled exception in evaluation endpoint: {str(e)}")
+        logger.debug(f"Stack trace: {traceback.format_exc()}")
+        return jsonify({"error": f"Internal server error: {str(e)}"}), 500
+
+@app.route("/upload_reference", methods=["POST"])
+def upload_reference_audio():
+    try:
+        if "audio" not in request.files:
+            logger.warning("⚠️ Reference upload missing audio file")
+            return jsonify({"error": "No audio file uploaded"}), 400
+
+        reference_word = request.form.get("reference_word", "").strip()
+        if not reference_word:
+            logger.warning("⚠️ Reference upload missing reference word")
+            return jsonify({"error": "No reference word provided"}), 400
+
+        # Validate reference word
+        reference_patterns = [
+            "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
+
+        # Create directory for reference pattern if it doesn't exist
+        pattern_dir = os.path.join(REFERENCE_AUDIO_DIR, reference_word)
+        os.makedirs(pattern_dir, exist_ok=True)
+
+        # Save the reference audio file
+        audio_file = request.files["audio"]
+        file_path = os.path.join(pattern_dir, secure_filename(audio_file.filename))
+        audio_file.save(file_path)
+
+        # Convert to WAV if not already in that format
+        if not file_path.lower().endswith('.wav'):
+            base_path = os.path.splitext(file_path)[0]
+            wav_path = f"{base_path}.wav"
+            try:
+                audio = AudioSegment.from_file(file_path)
+                audio = audio.set_frame_rate(SAMPLE_RATE).set_channels(1)
+                audio.export(wav_path, format="wav")
+                # Remove original file if conversion successful
+                os.unlink(file_path)
+                file_path = wav_path
+            except Exception as e:
+                logger.error(f"❌ Reference audio conversion failed: {str(e)}")
+                return jsonify({"error": f"Audio conversion failed: {str(e)}"}), 500
+
+        logger.info(f"✅ Reference audio saved successfully for {reference_word}: {file_path}")
+
+        # Count how many references we have now
+        references = glob.glob(os.path.join(pattern_dir, "*.wav"))
+        return jsonify({
+            "message": "Reference audio uploaded successfully",
+            "reference_word": reference_word,
+            "file": os.path.basename(file_path),
+            "total_references": len(references)
+        })
+
+    except Exception as e:
+        logger.error(f"❌ Unhandled exception in reference upload: {str(e)}")
+        logger.debug(f"Stack trace: {traceback.format_exc()}")
+        return jsonify({"error": f"Internal server error: {str(e)}"}), 500
+
+# Ensure directory exists
+@app.before_first_request
+def setup_reference_audio():
+    try:
+        os.makedirs(REFERENCE_AUDIO_DIR, exist_ok=True)
+        logger.info(f"📁 Created reference audio directory: {REFERENCE_AUDIO_DIR}")
+        
+        # Rest of your existing setup code...
+    except Exception as e:
+        logger.error(f"❌ Failed to set up reference audio directory: {str(e)}")
+
+
+
 if __name__ == "__main__":
     logger.info("🚀 Starting Speech API server")
     logger.info(f"📊 System status: ASR model: {'✅' if asr_model else '❌'}")