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

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

from dotenv import load_dotenv
load_dotenv()


# Submission Code Preliminaries
from torchvision import models
from torch.utils.data import Dataset
import librosa
import torch
from tqdm import tqdm
from torch.utils.data import DataLoader
from torch import nn

model = models.alexnet(pretrained=True)
model.features[0] = nn.Conv2d(1, 64, kernel_size=7, stride=2, padding=3, bias=False)
model.classifier[-1] = nn.Linear(model.classifier[-1].in_features, 1)
model.load_state_dict(torch.load("chainsaw_model.pth"))


class AudioDataset(Dataset):
    def __init__(self, dataset, max_length):
        self.dataset = dataset
        self.max_length = max_length

    def __len__(self):
        return len(self.dataset)

    def __getitem__(self, idx):
        array = self.dataset[idx]["audio"]["array"]
        spectrogram = librosa.feature.melspectrogram(y=array, sr=16000)

        spectrogram = torch.tensor(spectrogram, dtype=torch.float32)
        # normalize the spectrogram
        spectrogram = (spectrogram - spectrogram.mean()) / spectrogram.std()
            
        label = torch.tensor(self.dataset[idx]["label"])
        if spectrogram.shape[1] > self.max_length:
            spectrogram = spectrogram[:, :self.max_length]
        if spectrogram.shape[1] < self.max_length:
            # duplicate the last frame until the spectrogram is the max length
            spectrogram = torch.cat([spectrogram, spectrogram[:, -1].unsqueeze(1).repeat(1, self.max_length - spectrogram.shape[1])], dim=1)
        spectrogram = spectrogram.unsqueeze(0)
        return spectrogram, label

router = APIRouter()

DESCRIPTION = "Random Baseline"
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"]
    test_dataset = test_dataset[:10]
    
    # 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.
    #--------------------------------------------------------------------------------------------   
    test_dataset = AudioDataset(test_dataset, max_length=100)
    print("Currently pinging test api. Please make sure you see this!")
    test_loader = DataLoader(test_dataset, batch_size=4, shuffle=False)
    predictions = []
    for batch in tqdm(test_loader):
        out = model(batch[0])
        out_logits = torch.sigmoid(out)
        predictions.extend(out_logits.round().tolist())
        
    # Make random predictions (placeholder for actual model inference)
    predictions = [x[0] for x in predictions]
    true_labels = test_dataset["label"]

    
    #--------------------------------------------------------------------------------------------
    # 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