#!/usr/bin/env python3 """ Lightweight Aphasia Classification App Optimized for Hugging Face Spaces with lazy loading and fallbacks """ from flask import Flask, request, render_template_string, jsonify import os import tempfile import logging import json import threading import time from pathlib import Path # Set up logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) app = Flask(__name__) app.config['MAX_CONTENT_LENGTH'] = 50 * 1024 * 1024 # 50MB max (reduced) print("🚀 Starting Lightweight Aphasia Classification System") # Global state MODULES = {} MODELS_LOADED = False LOADING_STATUS = "Starting up..." def lazy_import_modules(): """Import modules only when needed""" global MODULES, MODELS_LOADED, LOADING_STATUS if MODELS_LOADED: return True try: LOADING_STATUS = "Loading audio processing..." logger.info("Importing utils_audio...") from utils_audio import convert_to_wav MODULES['convert_to_wav'] = convert_to_wav logger.info("✓ Audio processing loaded") LOADING_STATUS = "Loading speech analysis..." logger.info("Importing to_cha...") from to_cha import to_cha_from_wav MODULES['to_cha_from_wav'] = to_cha_from_wav logger.info("✓ Speech analysis loaded") LOADING_STATUS = "Loading data conversion..." logger.info("Importing cha_json...") from cha_json import cha_to_json_file MODULES['cha_to_json_file'] = cha_to_json_file logger.info("✓ Data conversion loaded") LOADING_STATUS = "Loading AI model..." logger.info("Importing output...") from output import predict_from_chajson MODULES['predict_from_chajson'] = predict_from_chajson logger.info("✓ AI model loaded") MODELS_LOADED = True LOADING_STATUS = "Ready!" logger.info("🎉 All modules loaded successfully!") return True except Exception as e: logger.error(f"Failed to load modules: {e}") LOADING_STATUS = f"Error: {str(e)}" return False def background_loader(): """Load modules in background thread""" logger.info("Starting background module loading...") lazy_import_modules() # Start loading modules in background loading_thread = threading.Thread(target=background_loader, daemon=True) loading_thread.start() # HTML Template (simplified) HTML_TEMPLATE = """ 🧠 Aphasia Classification

🧠 Aphasia Classification

AI-powered speech analysis for aphasia identification

🔄 System Status

{{ status_message }}

📁 Upload Audio File

Upload speech audio for aphasia classification


Supported: MP3, WAV, M4A (max 50MB)

🔄 Processing Audio...

This may take 2-5 minutes. Please be patient.

""" @app.route('/') def index(): """Main page""" return render_template_string(HTML_TEMPLATE, status_message=LOADING_STATUS) @app.route('/status') def status(): """Status check endpoint""" return jsonify({ 'ready': MODELS_LOADED, 'status': LOADING_STATUS, 'modules_loaded': len(MODULES) }) @app.route('/analyze', methods=['POST']) def analyze_audio(): """Process uploaded audio - only if models are loaded""" try: # Check if system is ready if not MODELS_LOADED: return jsonify({ 'success': False, 'error': f'System still loading: {LOADING_STATUS}' }) # Check file upload if 'audio' not in request.files: return jsonify({'success': False, 'error': 'No audio file uploaded'}) audio_file = request.files['audio'] if audio_file.filename == '': return jsonify({'success': False, 'error': 'No file selected'}) # Save uploaded file with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(audio_file.filename)[1]) as tmp_file: audio_file.save(tmp_file.name) temp_path = tmp_file.name try: logger.info("🎵 Starting audio processing...") # Step 1: Convert to WAV logger.info("Converting to WAV...") wav_path = MODULES['convert_to_wav'](temp_path, sr=16000, mono=True) # Step 2: Generate CHA logger.info("Generating CHA file...") cha_path = MODULES['to_cha_from_wav'](wav_path, lang="eng") # Step 3: Convert to JSON logger.info("Converting to JSON...") json_path, _ = MODULES['cha_to_json_file'](cha_path) # Step 4: Classification logger.info("Running classification...") results = MODULES['predict_from_chajson'](".", json_path, output_file=None) # Cleanup for temp_file in [temp_path, wav_path, cha_path, json_path]: try: os.unlink(temp_file) except: pass # Format results if "predictions" in results and results["predictions"]: pred = results["predictions"][0] classification = pred["prediction"]["predicted_class"] confidence = pred["prediction"]["confidence_percentage"] description = pred["class_description"]["name"] severity = pred["additional_predictions"]["predicted_severity_level"] fluency = pred["additional_predictions"]["fluency_rating"] result_text = f"""🧠 APHASIA CLASSIFICATION RESULTS 🎯 Classification: {classification} 📊 Confidence: {confidence} 📋 Type: {description} 📈 Severity: {severity}/3 🗣️ Fluency: {fluency} 📊 Top 3 Probabilities:""" prob_dist = pred["probability_distribution"] for i, (atype, info) in enumerate(list(prob_dist.items())[:3], 1): result_text += f"\n{i}. {atype}: {info['percentage']}" result_text += f""" 📝 Description: {pred["class_description"]["description"]} ✅ Processing completed successfully! """ return jsonify({'success': True, 'result': result_text}) else: return jsonify({'success': False, 'error': 'No predictions generated'}) except Exception as e: # Cleanup on error try: os.unlink(temp_path) except: pass raise e except Exception as e: logger.error(f"Processing error: {e}") return jsonify({'success': False, 'error': str(e)}) if __name__ == '__main__': port = int(os.environ.get('PORT', 7860)) print(f"🚀 Starting on port {port}") print("🔄 Models loading in background...") app.run(host='0.0.0.0', port=port, debug=False, threaded=True)