version corrigés avec parallel processing

Dockerfile

@@ -1,6 +1,5 @@
 # Read the doc: https://huggingface.co/docs/hub/spaces-sdks-docker
 # you will also find guides on how best to write your Dockerfile
-
 FROM python:3.9
 
 RUN useradd -m -u 1000 user

app.py

@@ -1,6 +1,7 @@
 from fastapi import FastAPI
 from dotenv import load_dotenv
 from tasks import text, image, audio
+from pathlib import Path
 
 # Load environment variables
 load_dotenv()
@@ -11,10 +12,22 @@ app = FastAPI(
 )
 
 # Include all routers
-app.include_router(text.router)
-app.include_router(image.router)
 app.include_router(audio.router)
 
+@app.get("/health")
+async def health_check():
+    try:
+        model_exists = Path("/app/models/audio_model.pkl").exists()
+        return {
+            "status": "healthy",
+            "model_loaded": model_exists
+        }
+    except Exception as e:
+        return {
+            "status": "unhealthy",
+            "error": str(e)
+        }
+
 @app.get("/")
 async def root():
     return {
@@ -22,4 +35,4 @@ async def root():
         "endpoints": {
             "audio": "/audio - Audio classification task"
         }
-    } 
+    }

models/audio_model.pkl

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:fc95e0a3e06625d1a666ead9869dc4b9307fb0e3cef4316264ec476b26b7de38
-size 925490
+oid sha256:39615d655f9a19dee949084eae2b12f4ba071ca867e19997b2794524a35f2947
+size 938130

tasks/audio.py

@@ -7,46 +7,110 @@ import pickle
 from pathlib import Path
 import numpy as np
 import librosa
-from sklearn.preprocessing import StandardScaler
-from dotenv import load_dotenv
+from concurrent.futures import ThreadPoolExecutor
+import multiprocessing
 from .utils.evaluation import AudioEvaluationRequest
 from .utils.emissions import tracker, clean_emissions_data, get_space_info
+from dotenv import load_dotenv
+import logging
 
-# Charger les variables d'environnement
+# Configuration
 load_dotenv()
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
-# Configuration du router
 router = APIRouter()
-
-DESCRIPTION = "Random Forest with Feature Engineering"
+DESCRIPTION = "Parallel Random Forest with Feature Engineering"
 ROUTE = "/audio"
-MODEL_PATH = Path(__file__).parent.parent / "models" / "audio_model.pkl"
-    
+MODEL_PATH = Path("/app/models/audio_model.pkl")
 SAMPLING_RATE = 12000
 N_MFCC = 13
+NUM_WORKERS = multiprocessing.cpu_count()
+BATCH_SIZE = 32
 
-def extract_features(audio_array):
-    """Feature engineering identical to the training phase."""
+def process_batch_parallel(batch):
+    """Process a batch of test samples in parallel"""
+    features = []
+    labels = []
+    
     try:
-        if not isinstance(audio_array, np.ndarray) or len(audio_array) == 0:
+        logger.info(f"Batch type: {type(batch)}")
+        
+        with ThreadPoolExecutor(max_workers=NUM_WORKERS) as executor:
+            futures = []
+            
+            # Itérer sur le dataset
+            for i in range(len(batch)):
+                audio = batch[i]['audio']
+                label = batch[i]['label']
+                
+                logger.info(f"Processing audio sample {i}")
+                logger.info(f"Audio type: {type(audio)}")
+                if hasattr(audio, 'array'):
+                    logger.info(f"Audio shape: {audio['array'].shape}")
+                
+                future = executor.submit(extract_features_parallel, audio)
+                futures.append((future, label))
+            
+            for idx, (future, label) in enumerate(futures):
+                try:
+                    feature = future.result()
+                    if feature is not None:
+                        logger.info(f"Successfully extracted features for sample {idx}")
+                        features.append(feature)
+                        labels.append(label)
+                    else:
+                        logger.warning(f"No features extracted for sample {idx}")
+                except Exception as e:
+                    logger.error(f"Feature extraction error for sample {idx}: {str(e)}")
+                    continue
+        
+        logger.info(f"Successfully processed {len(features)} samples out of {len(batch)}")
+        return features, labels
+        
+    except Exception as e:
+        logger.error(f"Batch processing error: {str(e)}")
+        return [], []
+
+
+def extract_features_parallel(audio_data):
+    """Optimized parallel feature extraction"""
+    try:
+        if isinstance(audio_data, dict):
+            if 'array' in audio_data:
+                audio_array = audio_data['array']
+            elif 'path' in audio_data:
+                # Si nous avons un chemin de fichier
+                y, sr = librosa.load(audio_data['path'], sr=SAMPLING_RATE)
+                audio_array = y
+            else:
+                logger.error("No array or path in audio data")
+                return None
+        else:
+            audio_array = audio_data
+
+        if len(audio_array) == 0:
+            logger.error("Empty audio array")
             return None
 
-        # Convert to mono if stereo
+        # Conversion en mono si stéréo
         y = np.mean(audio_array, axis=1) if audio_array.ndim > 1 else audio_array
+        
+        # Vérification de la longueur minimale
+        if len(y) < SAMPLING_RATE:
+            logger.warning("Audio too short, padding")
+            y = np.pad(y, (0, SAMPLING_RATE - len(y)))
+        
+        # Extraction des features
+        with ThreadPoolExecutor(max_workers=4) as executor:
+            futures = [
+                executor.submit(librosa.feature.mfcc, y=y, sr=SAMPLING_RATE, n_mfcc=N_MFCC),
+                executor.submit(librosa.feature.zero_crossing_rate, y),
+                executor.submit(librosa.feature.rms, y=y),
+                executor.submit(librosa.feature.spectral_centroid, y=y, sr=SAMPLING_RATE)
+            ]
+            mfccs, zcr, rms, spectral_centroid = [f.result() for f in futures]
 
-        # Extract MFCCs and additional features
-        mfccs = librosa.feature.mfcc(
-            y=y,
-            sr=SAMPLING_RATE,
-            n_mfcc=N_MFCC,
-            n_fft=2048,
-            hop_length=512
-        )
-        zcr = librosa.feature.zero_crossing_rate(y)
-        rms = librosa.feature.rms(y=y)
-        spectral_centroid = librosa.feature.spectral_centroid(y=y, sr=SAMPLING_RATE)
-
-        # Combine features into a single vector
         feature_vector = np.concatenate([
             np.mean(mfccs, axis=1),
             np.std(mfccs, axis=1),
@@ -58,72 +122,72 @@ def extract_features(audio_array):
         return feature_vector
 
     except Exception as e:
-        raise ValueError(f"Feature extraction error: {str(e)}")
+        logger.error(f"Feature extraction error: {str(e)}")
+        return None
+
 
 @router.post(ROUTE, tags=["Audio Task"], description=DESCRIPTION)
 async def evaluate_audio(request: AudioEvaluationRequest):
-    """
-    Evaluate audio classification for rainforest sound detection using Random Forest.
-    """
+    """Evaluate audio classification with parallel processing"""
     try:
-        # Get space information (username and URL)
+        logger.info("Starting audio evaluation...")
         username, space_url = get_space_info()
-
-        # Load dataset from Hugging Face
+        
+        logger.info(f"Loading dataset: {request.dataset_name}")
         dataset = load_dataset(
             request.dataset_name,
             token=os.getenv("HF_TOKEN")
         )
 
-        # Split dataset into train and test sets
+        logger.info("Splitting dataset...")
         train_test = dataset["train"].train_test_split(
             test_size=request.test_size,
             seed=request.test_seed
         )
         test_dataset = train_test["test"]
+        logger.info(f"Test dataset size: {len(test_dataset)}")
 
-        # Start emissions tracking for inference phase
         tracker.start()
         tracker.start_task("inference")
-
-        # Prepare test data using the same feature engineering as in training
+        
+        logger.info("Processing test data...")
         x_test = []
         true_labels = []
         
-        for sample in test_dataset:
-            features = extract_features(sample["audio"]["array"])
-            if features is not None:
-                x_test.append(features)
-                true_labels.append(sample["label"])
+        for i in range(0, len(test_dataset), BATCH_SIZE):
+            logger.info(f"Processing batch {i}/{len(test_dataset)}")
+            batch = test_dataset.select(range(i, min(i + BATCH_SIZE, len(test_dataset))))
+            try:
+                features, labels = process_batch_parallel(batch)
+                x_test.extend(features)
+                true_labels.extend(labels)
+            except Exception as e:
+                logger.error(f"Error processing batch: {str(e)}")
+                continue
 
         if len(x_test) == 0:
-            raise ValueError("No valid features could be extracted from the test dataset.")
+            raise ValueError("No valid features could be extracted")
 
-        x_test = np.array(x_test)
-
-        # Load the trained model and scaler from pickle file
+        logger.info("Loading model...")
+        if not MODEL_PATH.exists():
+            raise FileNotFoundError(f"Model not found at {MODEL_PATH}")
+            
         with open(MODEL_PATH, 'rb') as f:
             model_data = pickle.load(f)
 
         model = model_data['model']
         scaler = model_data['scaler']
 
-        # Scale the test data using the scaler from training phase
-        if scaler is not None:
-            x_test_scaled = scaler.transform(x_test)
-        else:
-            x_test_scaled = x_test
-
-        # Make predictions on the test set
+        logger.info("Making predictions...")
+        x_test = np.array(x_test)
+        x_test_scaled = scaler.transform(x_test) if scaler is not None else x_test
         predictions = model.predict(x_test_scaled)
 
-        # Stop emissions tracking and get data
         emissions_data = tracker.stop_task()
-
-        # Calculate accuracy score for evaluation
         accuracy = accuracy_score(true_labels, predictions)
+        
+        logger.info(f"Evaluation complete. Accuracy: {accuracy}")
 
-        # Prepare and return results as JSON response
         return {
             "username": username,
             "space_url": space_url,
@@ -142,6 +206,7 @@ async def evaluate_audio(request: AudioEvaluationRequest):
         }
 
     except Exception as e:
+        logger.error(f"Error in evaluate_audio: {str(e)}")
         raise HTTPException(
             status_code=500,
             detail=f"An error occurred during evaluation: {str(e)}"