added main.py

audio_preprocessing.py

@@ -0,0 +1,230 @@
+"""
+Audio Preprocessing Module for Respiratory Analysis
+Matches the exact preprocessing used during training
+"""
+
+import librosa
+import numpy as np
+import torch
+import warnings
+from typing import Union, Tuple, Dict
+import soundfile as sf
+
+warnings.filterwarnings('ignore')
+
+class RespiratoryAudioPreprocessor:
+    """
+    Audio preprocessor that matches training pipeline exactly
+    Converts raw audio files to mel-spectrograms for model inference
+    """
+    
+    def __init__(self, 
+                 target_sr: int = 22050,
+                 n_mels: int = 128,
+                 n_fft: int = 2048,
+                 hop_length: int = 512,
+                 win_length: int = None,
+                 window: str = 'hann',
+                 fmin: float = 0.0,
+                 fmax: float = None,
+                 power: float = 2.0,
+                 duration: float = 3.0):  # 3 seconds max duration
+        """
+        Initialize preprocessing parameters to match training
+        """
+        self.target_sr = target_sr
+        self.n_mels = n_mels
+        self.n_fft = n_fft
+        self.hop_length = hop_length
+        self.win_length = win_length
+        self.window = window
+        self.fmin = fmin
+        self.fmax = fmax or target_sr // 2
+        self.power = power
+        self.duration = duration
+        self.target_length = int(target_sr * duration)  # 3 seconds worth of samples
+        
+        # Expected output shape for your model
+        self.expected_shape = (1, 1, 128, 251)  # (batch, channels, height, width)
+        
+    def load_and_normalize_audio(self, audio_input: Union[str, np.ndarray, tuple]) -> np.ndarray:
+        """
+        Load audio file and normalize
+        """
+        try:
+            # Handle different input types
+            if isinstance(audio_input, str):
+                # File path
+                audio_data, sr = librosa.load(audio_input, sr=self.target_sr, duration=self.duration)
+            elif isinstance(audio_input, tuple):
+                # (sample_rate, audio_array) from Gradio
+                sr, audio_data = audio_input
+                
+                # Convert to float if needed
+                if audio_data.dtype != np.float32:
+                    if audio_data.dtype == np.int16:
+                        audio_data = audio_data.astype(np.float32) / 32767.0
+                    elif audio_data.dtype == np.int32:
+                        audio_data = audio_data.astype(np.float32) / 2147483647.0
+                    else:
+                        audio_data = audio_data.astype(np.float32)
+                
+                # Resample if needed
+                if sr != self.target_sr:
+                    audio_data = librosa.resample(audio_data, orig_sr=sr, target_sr=self.target_sr)
+                
+                # Trim to duration
+                if len(audio_data) > self.target_length:
+                    audio_data = audio_data[:self.target_length]
+                    
+            elif isinstance(audio_input, np.ndarray):
+                # Raw audio array
+                audio_data = audio_input
+                if len(audio_data) > self.target_length:
+                    audio_data = audio_data[:self.target_length]
+            else:
+                raise ValueError(f"Unsupported audio input type: {type(audio_input)}")
+            
+            # Pad if too short
+            if len(audio_data) < self.target_length:
+                audio_data = np.pad(audio_data, (0, self.target_length - len(audio_data)), 
+                                  mode='constant', constant_values=0)
+            
+            return audio_data
+            
+        except Exception as e:
+            raise RuntimeError(f"Failed to load audio: {str(e)}")
+    
+    def extract_mel_spectrogram(self, audio_data: np.ndarray) -> np.ndarray:
+        """
+        Extract mel spectrogram features matching training preprocessing
+        """
+        try:
+            # Extract mel spectrogram
+            mel_spec = librosa.feature.melspectrogram(
+                y=audio_data,
+                sr=self.target_sr,
+                n_mels=self.n_mels,
+                n_fft=self.n_fft,
+                hop_length=self.hop_length,
+                win_length=self.win_length,
+                window=self.window,
+                fmin=self.fmin,
+                fmax=self.fmax,
+                power=self.power
+            )
+            
+            # Convert to log scale (dB)
+            mel_spec_db = librosa.power_to_db(mel_spec, ref=np.max)
+            
+            return mel_spec_db
+            
+        except Exception as e:
+            raise RuntimeError(f"Failed to extract mel spectrogram: {str(e)}")
+    
+    def normalize_spectrogram(self, mel_spec: np.ndarray) -> np.ndarray:
+        """
+        Normalize mel spectrogram to match training normalization
+        This matches the normalization used in your training pipeline
+        """
+        # Mean and std normalization
+        mean = np.mean(mel_spec)
+        std = np.std(mel_spec)
+        
+        if std == 0:
+            normalized = mel_spec - mean
+        else:
+            normalized = (mel_spec - mean) / std
+        
+        # Clamp values to reasonable range (matching training)
+        normalized = np.clip(normalized, -5.0, 5.0)
+        
+        return normalized
+    
+    def resize_spectrogram(self, mel_spec: np.ndarray, target_width: int = 251) -> np.ndarray:
+        """
+        Resize spectrogram to target dimensions
+        """
+        current_height, current_width = mel_spec.shape
+        
+        if current_width == target_width:
+            return mel_spec
+        
+        # Use librosa's time stretching for width adjustment
+        if current_width < target_width:
+            # Pad if too narrow
+            pad_width = target_width - current_width
+            mel_spec = np.pad(mel_spec, ((0, 0), (0, pad_width)), mode='constant', constant_values=0)
+        else:
+            # Truncate if too wide
+            mel_spec = mel_spec[:, :target_width]
+        
+        return mel_spec
+    
+    def preprocess_audio(self, audio_input: Union[str, np.ndarray, tuple]) -> torch.Tensor:
+        """
+        Complete preprocessing pipeline from audio to model input tensor
+        """
+        try:
+            # Step 1: Load and normalize audio
+            audio_data = self.load_and_normalize_audio(audio_input)
+            
+            # Step 2: Extract mel spectrogram
+            mel_spec = self.extract_mel_spectrogram(audio_data)
+            
+            # Step 3: Normalize spectrogram
+            mel_spec_norm = self.normalize_spectrogram(mel_spec)
+            
+            # Step 4: Resize to target dimensions
+            mel_spec_resized = self.resize_spectrogram(mel_spec_norm)
+            
+            # Step 5: Convert to tensor and add batch + channel dimensions
+            tensor_input = torch.FloatTensor(mel_spec_resized)
+            tensor_input = tensor_input.unsqueeze(0).unsqueeze(0)  # Add batch and channel dims
+            
+            # Verify output shape
+            if tensor_input.shape != self.expected_shape:
+                raise RuntimeError(f"Output shape {tensor_input.shape} doesn't match expected {self.expected_shape}")
+            
+            return tensor_input
+            
+        except Exception as e:
+            raise RuntimeError(f"Preprocessing failed: {str(e)}")
+    
+    def get_preprocessing_info(self) -> Dict:
+        """
+        Get preprocessing configuration info
+        """
+        return {
+            'target_sr': self.target_sr,
+            'n_mels': self.n_mels,
+            'n_fft': self.n_fft,
+            'hop_length': self.hop_length,
+            'duration': self.duration,
+            'output_shape': self.expected_shape
+        }
+
+# Example usage and testing
+if __name__ == "__main__":
+    # Initialize preprocessor
+    preprocessor = RespiratoryAudioPreprocessor()
+    
+    # Test with dummy audio data
+    dummy_audio = np.random.randn(22050 * 2)  # 2 seconds of audio
+    
+    try:
+        # Preprocess
+        tensor_output = preprocessor.preprocess_audio(dummy_audio)
+        print(f"✅ Preprocessing successful!")
+        print(f"Output shape: {tensor_output.shape}")
+        print(f"Output dtype: {tensor_output.dtype}")
+        print(f"Output range: [{tensor_output.min():.3f}, {tensor_output.max():.3f}]")
+        
+        # Display preprocessing info
+        info = preprocessor.get_preprocessing_info()
+        print("\n📋 Preprocessing Configuration:")
+        for key, value in info.items():
+            print(f"  {key}: {value}")
+            
+    except Exception as e:
+        print(f"❌ Preprocessing failed: {e}")

main.py

@@ -0,0 +1,413 @@
+"""
+FastAPI Backend for Respiratory Symptom Analysis
+Deployed on HuggingFace Spaces for use with Netlify frontend
+Updated for optimized_model_cpu folder structure
+"""
+
+from fastapi import FastAPI, File, UploadFile, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+import torch
+import json
+import numpy as np
+import tempfile
+import os
+from pathlib import Path
+from typing import Dict, List, Any
+import time
+import warnings
+
+# Import your preprocessing module
+from audio_preprocessing import RespiratoryAudioPreprocessor
+
+warnings.filterwarnings('ignore')
+
+# Initialize FastAPI app
+app = FastAPI(
+    title="🫁 Respiratory Symptom Analysis API",
+    description="AI-powered respiratory symptom detection from cough audio",
+    version="2.0.0",
+    docs_url="/docs",
+    redoc_url="/redoc"
+)
+
+# Add CORS middleware for Netlify frontend
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Configure this for your Netlify domain in production
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+class RespiratoryAnalysisService:
+    """
+    Service class for respiratory symptom analysis
+    """
+    
+    def __init__(self, 
+                 model_path: str = "optimized_model_cpu/model_torchscript.pt", 
+                 config_path: str = "optimized_model_cpu/model_config.json"):
+        """Initialize the service with model and configuration"""
+        self.model_path = model_path
+        self.config_path = config_path
+        self.model = None
+        self.config = None
+        self.preprocessor = None
+        
+        # Load model and configuration
+        self.load_model_and_config()
+        self.setup_preprocessor()
+        
+    def load_model_and_config(self):
+        """Load the optimized model and configuration with fallback options"""
+        try:
+            # Load configuration
+            if Path(self.config_path).exists():
+                with open(self.config_path, 'r') as f:
+                    self.config = json.load(f)
+                print(f"✅ Configuration loaded from {self.config_path}")
+            else:
+                raise FileNotFoundError(f"Config file not found: {self.config_path}")
+            
+            # Try loading models in priority order
+            model_files_to_try = [
+                ("optimized_model_cpu/model_torchscript.pt", "TorchScript"),
+                ("optimized_model_cpu/model_quantized.pt", "Quantized PyTorch"),
+                ("optimized_model_cpu/model_pytorch.pt", "Regular PyTorch")
+            ]
+            
+            model_loaded = False
+            for model_file, model_type in model_files_to_try:
+                if Path(model_file).exists():
+                    try:
+                        if "torchscript" in model_file.lower():
+                            # Load TorchScript model
+                            self.model = torch.jit.load(model_file, map_location='cpu')
+                            print(f"✅ {model_type} model loaded from {model_file}")
+                        else:
+                            # Load regular PyTorch model
+                            self.model = torch.load(model_file, map_location='cpu')
+                            print(f"✅ {model_type} model loaded from {model_file}")
+                        
+                        self.model.eval()
+                        model_loaded = True
+                        break
+                        
+                    except Exception as e:
+                        print(f"⚠️ Failed to load {model_type} model: {str(e)}")
+                        continue
+                else:
+                    print(f"⚠️ Model file not found: {model_file}")
+            
+            if not model_loaded:
+                raise RuntimeError("Failed to load any model file")
+            
+            # Set CPU optimization
+            if 'optimization_settings' in self.config:
+                torch.set_num_threads(self.config['optimization_settings'].get('torch_threads', 4))
+            else:
+                torch.set_num_threads(4)  # Default
+                
+        except Exception as e:
+            raise RuntimeError(f"Failed to load model/config: {str(e)}")
+    
+    def setup_preprocessor(self):
+        """Initialize audio preprocessor"""
+        self.preprocessor = RespiratoryAudioPreprocessor()
+        print("✅ Audio preprocessor initialized")
+    
+    def predict_symptoms(self, audio_file_path: str) -> Dict[str, Any]:
+        """Predict respiratory symptoms from audio file"""
+        try:
+            start_time = time.time()
+            
+            # Preprocess audio
+            tensor_input = self.preprocessor.preprocess_audio(audio_file_path)
+            preprocessing_time = time.time() - start_time
+            
+            # Run inference
+            inference_start = time.time()
+            with torch.no_grad():
+                outputs = self.model(tensor_input)
+                
+            inference_time = time.time() - inference_start
+            
+            # Parse outputs based on model type
+            if isinstance(outputs, dict):
+                # New model format with dictionary output
+                probabilities = outputs['probabilities'].squeeze().numpy()
+                predictions = outputs['predictions'].squeeze().numpy()
+            else:
+                # Handle legacy model formats
+                if isinstance(outputs, tuple):
+                    logits = outputs[0].squeeze()  # Take first output (symptom logits)
+                else:
+                    logits = outputs.squeeze()
+                
+                probabilities = torch.sigmoid(logits).numpy()
+                
+                # Apply thresholds
+                threshold_tensor = torch.tensor([
+                    self.config['confidence_thresholds'][symptom] 
+                    for symptom in self.config['target_symptoms']
+                ])
+                predictions = (torch.sigmoid(logits) >= threshold_tensor).float().numpy()
+            
+            # Format results
+            results = self.format_results(probabilities, predictions)
+            
+            # Add timing and model info
+            results['processing_info'] = {
+                'preprocessing_time_ms': round(preprocessing_time * 1000, 1),
+                'inference_time_ms': round(inference_time * 1000, 1),
+                'total_time_ms': round((preprocessing_time + inference_time) * 1000, 1),
+                'model_path': self.model_path,
+                'model_type': type(self.model).__name__
+            }
+            
+            return results
+            
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=f"Prediction failed: {str(e)}")
+    
+    def format_results(self, probabilities: np.ndarray, predictions: np.ndarray) -> Dict[str, Any]:
+        """Format prediction results for API response"""
+        results = {
+            'detected_symptoms': [],
+            'all_symptoms': {},
+            'summary': {},
+            'recommendations': []
+        }
+        
+        # Process each symptom
+        for i, symptom in enumerate(self.config['target_symptoms']):
+            prob = float(probabilities[i])
+            pred = bool(predictions[i])
+            display_name = self.config['symptom_display_names'][symptom]
+            threshold = self.config['confidence_thresholds'][symptom]
+            
+            # All symptoms with details
+            results['all_symptoms'][symptom] = {
+                'display_name': display_name,
+                'confidence': prob,
+                'detected': pred,
+                'threshold': threshold,
+                'color': self.config['symptom_colors'][symptom]
+            }
+            
+            # Detected symptoms only
+            if pred:
+                results['detected_symptoms'].append({
+                    'symptom': symptom,
+                    'display_name': display_name,
+                    'confidence': prob,
+                    'color': self.config['symptom_colors'][symptom]
+                })
+        
+        # Sort detected symptoms by confidence
+        results['detected_symptoms'].sort(key=lambda x: x['confidence'], reverse=True)
+        
+        # Generate summary
+        results['summary'] = {
+            'total_detected': len(results['detected_symptoms']),
+            'highest_confidence': results['detected_symptoms'][0]['confidence'] if results['detected_symptoms'] else 0.0,
+            'status': 'symptoms_detected' if results['detected_symptoms'] else 'no_symptoms'
+        }
+        
+        # Generate recommendations
+        if len(results['detected_symptoms']) == 0:
+            results['recommendations'] = [
+                "No significant respiratory symptoms detected.",
+                "Continue monitoring your health.",
+                "This screening is for informational purposes only."
+            ]
+        elif len(results['detected_symptoms']) == 1:
+            symptom_name = results['detected_symptoms'][0]['display_name']
+            results['recommendations'] = [
+                f"Detected: {symptom_name}",
+                "Consider monitoring symptoms and consult healthcare provider if symptoms persist.",
+                "This AI screening should not replace professional medical advice."
+            ]
+        else:
+            symptom_names = [s['display_name'] for s in results['detected_symptoms']]
+            results['recommendations'] = [
+                f"Multiple symptoms detected: {', '.join(symptom_names)}",
+                "Please consult a healthcare provider for proper evaluation.",
+                "This AI screening should not replace professional medical advice."
+            ]
+        
+        return results
+
+# Initialize service with error handling
+print("🚀 Initializing Respiratory Analysis Service...")
+try:
+    service = RespiratoryAnalysisService()
+    print("✅ Service initialized successfully!")
+except Exception as e:
+    print(f"❌ Service initialization failed: {str(e)}")
+    # Create a dummy service for debugging
+    service = None
+
+# API Routes
+@app.get("/")
+async def root():
+    """Root endpoint with API information"""
+    if service is None:
+        return {
+            "service": "Respiratory Symptom Analysis API",
+            "version": "2.0.0",
+            "status": "error - service not initialized",
+            "error": "Model loading failed"
+        }
+    
+    return {
+        "service": "Respiratory Symptom Analysis API",
+        "version": "2.0.0",
+        "status": "active",
+        "endpoints": {
+            "analyze": "/analyze",
+            "health": "/health",
+            "info": "/info",
+            "docs": "/docs"
+        },
+        "supported_symptoms": list(service.config['target_symptoms']),
+        "model_info": {
+            "version": service.config['model_version'],
+            "optimization": "CPU-optimized with quantization",
+            "model_path": service.model_path
+        }
+    }
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint"""
+    return {
+        "status": "healthy" if service is not None else "unhealthy",
+        "timestamp": time.time(),
+        "model_loaded": service.model is not None if service else False,
+        "config_loaded": service.config is not None if service else False,
+        "model_files_available": {
+            "torchscript": Path("optimized_model_cpu/model_torchscript.pt").exists(),
+            "quantized": Path("optimized_model_cpu/model_quantized.pt").exists(),
+            "pytorch": Path("optimized_model_cpu/model_pytorch.pt").exists(),
+            "config": Path("optimized_model_cpu/model_config.json").exists()
+        }
+    }
+
+@app.get("/info")
+async def get_info():
+    """Get model and service information"""
+    if service is None:
+        return {"error": "Service not initialized"}
+    
+    return {
+        "model_info": {
+            "version": service.config.get('model_version', '2.0'),
+            "target_symptoms": service.config['target_symptoms'],
+            "symptom_display_names": service.config['symptom_display_names'],
+            "confidence_thresholds": service.config['confidence_thresholds'],
+            "optimization_settings": service.config.get('optimization_settings', {})
+        },
+        "preprocessing_info": service.preprocessor.get_preprocessing_info(),
+        "supported_formats": ["wav", "mp3", "flac", "ogg", "m4a"],
+        "max_duration": "30 seconds",
+        "api_version": "2.0.0",
+        "model_path": service.model_path
+    }
+
+@app.post("/analyze")
+async def analyze_audio(audio_file: UploadFile = File(...)):
+    """
+    Analyze audio file for respiratory symptoms
+    
+    Parameters:
+    - audio_file: Audio file (WAV, MP3, FLAC, etc.)
+    
+    Returns:
+    - JSON response with symptom predictions and confidence scores
+    """
+    if service is None:
+        raise HTTPException(status_code=503, detail="Service not available - model loading failed")
+    
+    # Validate file type
+    allowed_types = ['audio/wav', 'audio/mpeg', 'audio/flac', 'audio/ogg', 'audio/x-m4a', 'audio/mp4']
+    if audio_file.content_type not in allowed_types:
+        raise HTTPException(
+            status_code=400, 
+            detail=f"Unsupported audio format: {audio_file.content_type}. Supported: {allowed_types}"
+        )
+    
+    # Validate file size (max 10MB)
+    max_size = 10 * 1024 * 1024  # 10MB
+    content = await audio_file.read()
+    if len(content) > max_size:
+        raise HTTPException(
+            status_code=400, 
+            detail="Audio file too large. Maximum size: 10MB"
+        )
+    
+    try:
+        # Save uploaded file temporarily
+        file_extension = audio_file.filename.split('.')[-1] if audio_file.filename else 'wav'
+        with tempfile.NamedTemporaryFile(delete=False, suffix=f".{file_extension}") as temp_file:
+            temp_file.write(content)
+            temp_file_path = temp_file.name
+        
+        # Analyze audio
+        results = service.predict_symptoms(temp_file_path)
+        
+        # Clean up temporary file
+        os.unlink(temp_file_path)
+        
+        # Return results
+        return JSONResponse(
+            status_code=200,
+            content={
+                "success": True,
+                "data": results,
+                "metadata": {
+                    "filename": audio_file.filename,
+                    "file_size_bytes": len(content),
+                    "content_type": audio_file.content_type,
+                    "timestamp": time.time()
+                }
+            }
+        )
+        
+    except HTTPException:
+        # Re-raise HTTP exceptions
+        raise
+    except Exception as e:
+        # Clean up temporary file if it exists
+        if 'temp_file_path' in locals():
+            try:
+                os.unlink(temp_file_path)
+            except:
+                pass
+        
+        raise HTTPException(status_code=500, detail=f"Analysis failed: {str(e)}")
+
+# Error handler
+@app.exception_handler(Exception)
+async def global_exception_handler(request, exc):
+    """Global exception handler"""
+    return JSONResponse(
+        status_code=500,
+        content={
+            "success": False,
+            "error": "Internal server error",
+            "detail": str(exc) if app.debug else "An unexpected error occurred"
+        }
+    )
+
+if __name__ == "__main__":
+    import uvicorn
+    
+    # Run the API server
+    uvicorn.run(
+        "main:app",
+        host="0.0.0.0",
+        port=7860,  # HuggingFace Spaces default port
+        reload=False
+    )