from fastapi import APIRouter from datetime import datetime from datasets import load_dataset from sklearn.metrics import accuracy_score import random from .utils.evaluation import TextEvaluationRequest from .utils.emissions import tracker, clean_emissions_data, get_space_info import os import logging import numpy as np print(os.getcwd()) # from sentence_transformers import SentenceTransformer from xgboost import XGBClassifier import pickle import xgboost as xgb #logging logging.basicConfig(level=logging.INFO) logging.info("LAS ESTRELLAS!!!!!") router = APIRouter() DESCRIPTION = "Random Baseline" ROUTE = "/text" @router.post(ROUTE, tags=["Text Task"], description=DESCRIPTION) async def evaluate_text(request: TextEvaluationRequest): """ Evaluate text classification for climate disinformation detection. Current Model: Random Baseline - Makes random predictions from the label space (0-7) - Used as a baseline for comparison """ # Get space info username, space_url = get_space_info() # Define the label mapping LABEL_MAPPING = { "0_not_relevant": 0, "1_not_happening": 1, "2_not_human": 2, "3_not_bad": 3, "4_solutions_harmful_unnecessary": 4, "5_science_unreliable": 5, "6_proponents_biased": 6, "7_fossil_fuels_needed": 7 } # Load and prepare the dataset dataset = load_dataset(request.dataset_name) # Convert string labels to integers dataset = dataset.map(lambda x: {"label": LABEL_MAPPING[x["label"]]}) # 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") #-------------------------------------------------------------------------------------------- # Load a pre-trained Sentence-BERT model print("loading model") model = SentenceTransformer('sentence-transformers/all-MPNET-base-v2', device='cpu') #load the models with open("xgb_bin.pkl","rb") as f: xgb_bin = pickle.load(f) with open("xgb_multi.pkl","rb") as f: xgb_multi = pickle.load(f) logging.info("generating embedding") # Generate sentence embeddings sentence_embeddings = model.encode(test_dataset["quote"]) logging.info(" embedding done") X_train = sentence_embeddings.copy() y_train = np.array(test_dataset["label"].copy()) #binary y_train_binary = y_train.copy() y_train_binary[y_train_binary != 0] = 1 #multi class X_train_multi = X_train[y_train != 0] y_train_multi = y_train[y_train != 0] logging.info(f"Xtrain_multi_shape:{X_train_multi.shape}") logging.info(f"Xtrain shape:{X_train.shape}") #predictions y_pred_bin = xgb_bin.predict(X_train) y_pred_multi = xgb_multi.predict(X_train_multi.reshape(-1,768)) + 1 logging.info(f"y_pred_bin:{y_pred_bin.shape}") logging.info(f"y_pred_multi shape:{y_pred_multi.shape}") y_pred_bin[y_train_binary==1] = y_pred_multi #predictions = xgb.predict(embeddings) # Make random predictions (placeholder for actual model inference) true_labels = test_dataset["label"] #predictions = xgb.predict(embeddings) #-------------------------------------------------------------------------------------------- # YOUR MODEL INFERENCE STOPS HERE #-------------------------------------------------------------------------------------------- # Stop tracking emissions emissions_data = tracker.stop_task() # Calculate accuracy accuracy = accuracy_score(true_labels, y_pred_bin) logging.info(f"Accuracy : {accuracy}") # 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