tasks/audio.py

from fastapi import APIRouter
from datetime import datetime
from datasets import load_dataset
from sklearn.metrics import accuracy_score
import random
import os
#import torch
#from torch.utils.data import DataLoader
#from /app/tasks/Model_Loader.py import M5, load_model

from .utils.evaluation import AudioEvaluationRequest
from .utils.emissions import tracker, clean_emissions_data, get_space_info

from dotenv import load_dotenv
load_dotenv()

router = APIRouter()

DESCRIPTION = "Quantized M5"
ROUTE = "/audio"



@router.post(ROUTE, tags=["Audio Task"],
             description=DESCRIPTION)
async def evaluate_audio(request: AudioEvaluationRequest):
    """
    Evaluate audio classification for rainforest sound detection.
    
    Current Model: Random Baseline
    - Makes random predictions from the label space (0-1)
    - Used as a baseline for comparison
    """
    # Get space info
    username, space_url = get_space_info()

    # Define the label mapping
    LABEL_MAPPING = {
        "chainsaw": 0,
        "environment": 1
    }
    # Load and prepare the dataset
    # Because the dataset is gated, we need to use the HF_TOKEN environment variable to authenticate
    dataset = load_dataset(request.dataset_name,token=os.getenv("HF_TOKEN"))
    
    # Split dataset
    train_test = dataset["train"].train_test_split(test_size=request.test_size, seed=request.test_seed)
    test_dataset = train_test["test"]
    
    # Start tracking emissions
    tracker.start()
    tracker.start_task("inference")
    
    #--------------------------------------------------------------------------------------------
    # YOUR MODEL INFERENCE CODE HERE
    # Update the code below to replace the random baseline by your model inference within the inference pass where the energy consumption and emissions are tracked.
    #--------------------------------------------------------------------------------------------   
    
    # Make random predictions (placeholder for actual model inference)
    #model_path = "quantized_teacher_m5_static.pth"
    #model, device = load_model(model_path)

    # def preprocess_audio(example, target_length=32000):
    #     """
    #     Convert dataset into tensors:
    #     - Convert to tensor
    #     - Normalize waveform
    #     - Pad/truncate to `target_length`
    #     """
    #     waveform = torch.tensor(example["audio"]["array"], dtype=torch.float32).unsqueeze(0)  # Add batch dim
    
    #     # Normalize waveform
    #     waveform = (waveform - waveform.mean()) / (waveform.std() + 1e-6)
    
    #     # Pad or truncate to fixed length
    #     if waveform.shape[1] < target_length:
    #         pad = torch.zeros(1, target_length - waveform.shape[1])
    #         waveform = torch.cat((waveform, pad), dim=1)  # Pad
    #     else:
    #         waveform = waveform[:, :target_length]  # Truncate
    
    #     label = torch.tensor(example["label"], dtype=torch.long)  # Ensure int64
    # return {"waveform": waveform, "label": label}



    # train_test = train_test.map(preprocess_audio, batched=True)
    # test_dataset = train_test.map(preprocess_audio)

    # train_loader = DataLoader(train_test, batch_size=32, shuffle=True)

    
    true_labels = train_dataset["label"]
    predictions = []
   
    predictions = [random.randint(0, 1) for _ in range(len(true_labels))]

    # with torch.no_grad():
    #     for waveforms, labels in train_loader:
    #         waveforms, labels = waveforms.to(device), labels.to(device)
            
    #         outputs = model(waveforms)
    #         predicted_label = torch.argmax(F.softmax(outputs, dim=1), dim=1)
    #         true_labels.extend(labels.cpu().numpy())
    #         predicted_labels.extend(predicted_label.cpu().numpy())
    
    #--------------------------------------------------------------------------------------------
    # YOUR MODEL INFERENCE STOPS HERE
    #--------------------------------------------------------------------------------------------   
    
    # Stop tracking emissions
    emissions_data = tracker.stop_task()
    
    # Calculate accuracy
    accuracy = accuracy_score(true_labels, predictions)
    
    # Prepare results dictionary
    results = {
        "username": username,
        "space_url": space_url,
        "submission_timestamp": datetime.now().isoformat(),
        "model_description": DESCRIPTION,
        "accuracy": float(accuracy),
        "energy_consumed_wh": emissions_data.energy_consumed * 1000,
        "emissions_gco2eq": emissions_data.emissions * 1000,
        "emissions_data": clean_emissions_data(emissions_data),
        "api_route": ROUTE,
        "dataset_config": {
            "dataset_name": request.dataset_name,
            "test_size": request.test_size,
            "test_seed": request.test_seed
        }
    }
    
    return results