services/annotation_service.py

from huggingface_hub import HfApi
import gradio as gr
import pandas as pd
import os
from datetime import datetime, UTC
from datasets import Dataset, concatenate_datasets, load_dataset
from huggingface_hub import login
import json
import uuid
from datetime import datetime, UTC
import torch
from huggingface_hub import snapshot_download

from config.settings import *

HF_TOKEN = os.environ.get("MyJulySecretToken")  # store your token as a secret in Spaces
login(HF_TOKEN)

HF_DATASET_NAME = "PaulineDV/TTS_annotations_data" 

api = HfApi(token = HF_TOKEN)

def load_hf_dataset():
    """Load existing HF Dataset or create empty one if not exists."""
    try:
        ds = load_dataset(HF_DATASET_NAME, split="train")
    except:
        # Dataset does not exist yet
        df = pd.DataFrame(columns=["user_id", "gender", "audio_file", "score"])
        ds = Dataset.from_pandas(df)
        ds.push_to_hub(HF_DATASET_NAME, private=True)
    return ds

timestamp = datetime.now(UTC).isoformat()

def submit_pair_annotation_hf(
        user_id, 
        age_group, 
        gender, 
        native_language, 
        tts_experience, 
        device_type, 
        hidden_models,
        context,
        naturalness_A,
        intelligibility_A,
        context_A,
        mos_A,
        naturalness_B,
        intelligibility_B,
        context_B,
        mos_B,
        preferred_audio,
):

    short_id = str(uuid.uuid4()) [:8]
    
    annotation_A = {
        "user_id": user_id,
        "age_group": age_group,
        "gender": gender,
        "native_language": native_language,
        "tts_experience": tts_experience,
        "device_type": device_type,
        "context": context,
        "model_name": hidden_models[0],
        "naturalness": naturalness_A,
        "intelligibility": intelligibility_A,
        "context_score": context_A,
        "mos": mos_A,
        "preferred_audio": preferred_audio,
        "comparaison_id": short_id,
        "timestamp": datetime.now(UTC).isoformat()
    }

    annotation_B = {
        "user_id": user_id,
        "age_group": age_group,
        "gender": gender,
        "native_language": native_language,
        "tts_experience": tts_experience,
        "device_type": device_type,
        "context": context,
        "model_name": hidden_models[1],
        "naturalness": naturalness_B,
        "intelligibility": intelligibility_B,
        "context_score": context_B,
        "mos": mos_B,
        "preferred_audio": preferred_audio,
        "comparaison_id": short_id,
        "timestamp": datetime.now(UTC).isoformat()
    }


    file_name_A = (
        f"{hidden_models[0]}_{context}_{user_id}_{short_id}_A.json"
    )

    file_name_B = (
        f"{hidden_models[1]}_{context}_{user_id}_{short_id}_B.json"
    )

    with open(file_name_A, "w", encoding="utf-8") as f:
        json.dump(annotation_A, f, indent=2)

    with open(file_name_B, "w", encoding="utf-8") as f:
        json.dump(annotation_B, f, indent=2)

    api.upload_file(
        path_or_fileobj=file_name_A,
        path_in_repo=f"annotations/{file_name_A}",
        repo_id=HF_DATASET_NAME,
        repo_type="dataset"
    )

    api.upload_file(
        path_or_fileobj=file_name_B,
        path_in_repo=f"annotations/{file_name_B}",
        repo_id=HF_DATASET_NAME,
        repo_type="dataset"
    )
    
    os.remove(file_name_A)
    os.remove(file_name_B)



def submit_pair_annotation(
        user_id, 
        age_group, 
        gender, 
        native_language, 
        tts_experience, 
        device_type, 
        hidden_models,
        context,
        naturalness_A,
        intelligibility_A,
        context_A,
        mos_A,
        naturalness_B,
        intelligibility_B,
        context_B,
        mos_B,
        preferred_audio,
):
   try:

    submit_pair_annotation_hf(
        user_id, 
        age_group, 
        gender, 
        native_language, 
        tts_experience, 
        device_type, 
        hidden_models,
        context,
        naturalness_A,
        intelligibility_A,
        context_A,
        mos_A,
        naturalness_B,
        intelligibility_B,
        context_B,
        mos_B,
        preferred_audio,
    )

    return f"Annotations saved for {context}."

   except Exception as e:
       print("Submit error")
       print(type(e).__name__)
       print(e)

       return f"ERROR: {e}"