Spaces:
Paused
Paused
| import hashlib | |
| import json | |
| import os | |
| import re | |
| import time | |
| import uuid | |
| from collections import Counter | |
| import torch | |
| from huggingface_hub import HfApi, hf_hub_download | |
| from huggingface_hub.utils import ( | |
| EntryNotFoundError, | |
| HfHubHTTPError, | |
| ) | |
| LEADERBOARD_DATASET_REPO = "puruchinera/top" | |
| LEADERBOARD_FILENAME = "leaderboard.json" | |
| MODEL_TYPE_CLASIFICACION = "clasificacion" | |
| MODEL_TYPE_SR = "sr" | |
| PATRON_MATRICULA = re.compile(r"^\d{2}-[A-Z]{4}-\d-\d{3}$") | |
| MENSAJE_ERROR_INTERNO = "Error: rokugogosango" | |
| MENSAJE_ERROR_NOMBRE = "El nombre completo debe incluir al menos nombre y apellido." | |
| MENSAJE_ERROR_MATRICULA = "La matr铆cula debe seguir el formato 00-AAAA-0-000." | |
| MAX_REINTENTOS_LEADERBOARD = 5 | |
| def cargar_etiquetas(): | |
| with open("etiquetas.txt", "r") as f: | |
| etiquetas = f.read().splitlines()[1:] | |
| num_clases = len(etiquetas) | |
| codigo = {etiqueta.lower(): i for i, etiqueta in enumerate(etiquetas)} | |
| return etiquetas, num_clases, codigo | |
| def multiclass_accuracy(predictions, labels): | |
| # Obt茅n las clases predichas (la clase con la mayor probabilidad) | |
| _, predicted_classes = torch.max(predictions, 1) | |
| # Compara las clases predichas con las etiquetas verdaderas | |
| correct_predictions = (predicted_classes == labels).sum().item() | |
| # Calcula la precisi贸n | |
| accuracy = correct_predictions / labels.size(0) | |
| return accuracy | |
| def calcular_psnr(predictions, targets): | |
| predictions = predictions.clamp(0, 1) | |
| targets = targets.clamp(0, 1) | |
| mse = torch.mean((predictions - targets) ** 2) | |
| if mse <= 0: | |
| return 100.0 | |
| return float((10 * torch.log10(torch.tensor(1.0, device=mse.device) / mse)).item()) | |
| def normalizar_tipo_modelo(model_type): | |
| value = (model_type or MODEL_TYPE_CLASIFICACION).strip().lower() | |
| if value in { | |
| MODEL_TYPE_SR, | |
| "superresolucion", | |
| "super-resolucion", | |
| "super resolution", | |
| }: | |
| return MODEL_TYPE_SR | |
| return MODEL_TYPE_CLASIFICACION | |
| def normalizar_nombre(nombre): | |
| return " ".join((nombre or "").strip().split()) | |
| def normalizar_matricula(matricula): | |
| return (matricula or "").strip().upper() | |
| def validar_nombre_completo(nombre, detailed=False): | |
| nombre = normalizar_nombre(nombre) | |
| if len(nombre.split()) < 2: | |
| return MENSAJE_ERROR_NOMBRE | |
| return None | |
| def validar_matricula(matricula, detailed=False): | |
| matricula = normalizar_matricula(matricula) | |
| if not PATRON_MATRICULA.fullmatch(matricula): | |
| return MENSAJE_ERROR_MATRICULA | |
| return None | |
| def validar_datos_estudiante(nombre, matricula, detailed=False): | |
| error_nombre = validar_nombre_completo(nombre, detailed=detailed) | |
| if error_nombre: | |
| return error_nombre | |
| return validar_matricula(matricula, detailed=detailed) | |
| def obtener_sha256(ruta_archivo): | |
| sha256 = hashlib.sha256() | |
| with open(ruta_archivo, "rb") as archivo: | |
| for bloque in iter(lambda: archivo.read(8192), b""): | |
| sha256.update(bloque) | |
| return sha256.hexdigest() | |
| def calcular_puntaje(metric_value, model_type=MODEL_TYPE_CLASIFICACION): | |
| model_type = normalizar_tipo_modelo(model_type) | |
| if model_type == MODEL_TYPE_SR: | |
| return max(0, int(metric_value - 10)) | |
| accuracy_pct = metric_value * 100 | |
| base = max(0, min(accuracy_pct - 55, 20)) | |
| extra = 10 if accuracy_pct >= 95 else 0 | |
| return int(base + extra) | |
| def formatear_commit_leaderboard(entry): | |
| registro = dict(entry) | |
| nombre = normalizar_nombre(registro.get("nombre", "")) | |
| matricula = normalizar_matricula(registro.get("matricula", "")) | |
| model_type = normalizar_tipo_modelo( | |
| registro.get("model_type", MODEL_TYPE_CLASIFICACION) | |
| ) | |
| if model_type == MODEL_TYPE_SR: | |
| metric = f"psnr={registro.get('psnr', 0.0):.2f}dB" | |
| else: | |
| accuracy_pct = registro.get("accuracy_pct") | |
| if accuracy_pct is None: | |
| accuracy_pct = registro.get("accuracy", 0.0) * 100 | |
| metric = f"accuracy={accuracy_pct:.2f}%" | |
| return f"Update leaderboard: {nombre} | {matricula} | {metric}" | |
| def _obtener_hf_token(): | |
| for env_var in ("HFKEY", "HF_TOKEN", "HUGGINGFACEHUB_API_TOKEN"): | |
| token = os.environ.get(env_var) | |
| if token: | |
| return token | |
| raise RuntimeError( | |
| "No se encontr贸 un token de Hugging Face en HFKEY, HF_TOKEN o " | |
| "HUGGINGFACEHUB_API_TOKEN." | |
| ) | |
| def _crear_cliente_hf(): | |
| return HfApi() | |
| def _subir_leaderboard(registros, commit_message, parent_commit=None): | |
| token = _obtener_hf_token() | |
| api = _crear_cliente_hf() | |
| payload = json.dumps(registros, indent=2, ensure_ascii=False).encode("utf-8") | |
| api.create_repo( | |
| repo_id=LEADERBOARD_DATASET_REPO, | |
| repo_type="dataset", | |
| private=True, | |
| exist_ok=True, | |
| token=token, | |
| ) | |
| api.upload_file( | |
| path_or_fileobj=payload, | |
| path_in_repo=LEADERBOARD_FILENAME, | |
| repo_id=LEADERBOARD_DATASET_REPO, | |
| repo_type="dataset", | |
| token=token, | |
| commit_message=commit_message, | |
| parent_commit=parent_commit, | |
| ) | |
| def _descargar_leaderboard_con_revision(): | |
| token = _obtener_hf_token() | |
| api = _crear_cliente_hf() | |
| api.create_repo( | |
| repo_id=LEADERBOARD_DATASET_REPO, | |
| repo_type="dataset", | |
| private=True, | |
| exist_ok=True, | |
| token=token, | |
| ) | |
| revision = api.repo_info( | |
| repo_id=LEADERBOARD_DATASET_REPO, | |
| repo_type="dataset", | |
| token=token, | |
| ).sha | |
| try: | |
| ruta_local = hf_hub_download( | |
| repo_id=LEADERBOARD_DATASET_REPO, | |
| filename=LEADERBOARD_FILENAME, | |
| repo_type="dataset", | |
| token=token, | |
| revision=revision, | |
| force_download=True, | |
| ) | |
| except EntryNotFoundError: | |
| return [], revision | |
| with open(ruta_local, "r", encoding="utf-8") as f: | |
| registros = json.load(f) | |
| if not isinstance(registros, list): | |
| raise ValueError( | |
| f"El archivo {LEADERBOARD_FILENAME} en el dataset " | |
| f"{LEADERBOARD_DATASET_REPO} debe contener una lista JSON." | |
| ) | |
| return registros, revision | |
| def _descargar_leaderboard(): | |
| registros, _ = _descargar_leaderboard_con_revision() | |
| return registros | |
| def _es_conflicto_de_commit(error): | |
| response = getattr(error, "response", None) | |
| return response is not None and response.status_code in (409, 412) | |
| def cargar_leaderboard(): | |
| registros = _descargar_leaderboard() | |
| normalizados = [] | |
| for entry in registros: | |
| registro = dict(entry) | |
| registro["model_type"] = normalizar_tipo_modelo( | |
| registro.get("model_type", MODEL_TYPE_CLASIFICACION) | |
| ) | |
| if ( | |
| registro["model_type"] == MODEL_TYPE_CLASIFICACION | |
| and "accuracy_pct" not in registro | |
| ): | |
| registro["accuracy_pct"] = registro.get("accuracy", 0) * 100 | |
| normalizados.append(registro) | |
| return _marcar_duplicados(normalizados) | |
| def _marcar_duplicados(registros): | |
| conteo = Counter( | |
| (normalizar_tipo_modelo(entry.get("model_type")), entry["sha256"]) | |
| for entry in registros | |
| ) | |
| for entry in registros: | |
| key = (normalizar_tipo_modelo(entry.get("model_type")), entry["sha256"]) | |
| entry["duplicado_count"] = conteo[key] | |
| entry["duplicado"] = conteo[key] > 1 | |
| return registros | |
| def filtrar_leaderboard_por_tipo(registros, model_type): | |
| model_type = normalizar_tipo_modelo(model_type) | |
| return [ | |
| entry | |
| for entry in registros | |
| if normalizar_tipo_modelo(entry.get("model_type")) == model_type | |
| ] | |
| def guardar_registro_leaderboard(entry): | |
| entry = dict(entry) | |
| entry.setdefault("id", str(uuid.uuid4())) | |
| entry["model_type"] = normalizar_tipo_modelo( | |
| entry.get("model_type", MODEL_TYPE_CLASIFICACION) | |
| ) | |
| for intento in range(MAX_REINTENTOS_LEADERBOARD): | |
| registros, parent_commit = _descargar_leaderboard_con_revision() | |
| # Hace idempotente el reintento si el servidor guard贸 el commit pero la | |
| # respuesta no lleg贸 al proceso que realiz贸 la evaluaci贸n. | |
| if any(registro.get("id") == entry["id"] for registro in registros): | |
| return _marcar_duplicados(registros) | |
| registros.append(entry) | |
| registros = _marcar_duplicados(registros) | |
| try: | |
| _subir_leaderboard( | |
| registros, | |
| formatear_commit_leaderboard(entry), | |
| parent_commit=parent_commit, | |
| ) | |
| return registros | |
| except HfHubHTTPError as error: | |
| if not _es_conflicto_de_commit(error): | |
| raise | |
| if intento + 1 == MAX_REINTENTOS_LEADERBOARD: | |
| raise RuntimeError( | |
| "No se pudo actualizar el leaderboard debido a escrituras " | |
| "concurrentes." | |
| ) from error | |
| time.sleep(0.1 * (2**intento)) | |
| raise RuntimeError("No se pudo actualizar el leaderboard.") | |
| def limpiar_leaderboard(): | |
| _subir_leaderboard([], "Reset leaderboard") | |