""" EMOTYC — Visualisation interactive des performances (FastAPI). Backend API avec : - Endpoints JSON pour configurations, métriques et instances - Fichiers statiques servis depuis static/ """ from __future__ import annotations from dataclasses import dataclass, field, asdict from pathlib import Path from fastapi import FastAPI, HTTPException, Query from fastapi.responses import HTMLResponse from fastapi.staticfiles import StaticFiles import numpy as np import pandas as pd # ═══════════════════════════════════════════════════════════════════════════ # CONSTANTS # ═══════════════════════════════════════════════════════════════════════════ BASE_DIR = Path(__file__).resolve().parent DATA_DIR = BASE_DIR / "data" ALL_LABELS = [ "Emo", "Comportementale", "Designee", "Montree", "Suggeree", "Base", "Complexe", "Admiration", "Autre", "Colere", "Culpabilite", "Degout", "Embarras", "Fierte", "Jalousie", "Joie", "Peur", "Surprise", "Tristesse", ] PRED_SUFFIX = "_pred_emotyc" DISPLAY_NAMES = { "Colere": "Colère", "Culpabilite": "Culpabilité", "Degout": "Dégoût", "Fierte": "Fierté", "Designee": "Désignée", "Montree": "Montrée", "Suggeree": "Suggérée", "Emo": "Émo", } OUTCOME_DISPLAY = { "tp": "Vrais Positifs (TP)", "fp": "Faux Positifs (FP)", "fn": "Faux Négatifs (FN)", "tn": "Vrais Négatifs (TN)", } CONFIGS: dict[str, str] = { "CyberAggAdo 200": "CyberAggAdo200.parquet", "CyberAggAdo Global — Contexte": "CyberAggAdoGlobal_Context.parquet", "CyberAggAdo Global — Sans Contexte": "CyberAggAdoGlobal_SansContexte.parquet", "TextToKids — Contexte": "TextToKids_Context.parquet", "TextToKids — Sans Contexte": "TextToKids_SansContexte.parquet", } def display_name(label: str) -> str: return DISPLAY_NAMES.get(label, label) # ═══════════════════════════════════════════════════════════════════════════ # DATA STRUCTURES # ═══════════════════════════════════════════════════════════════════════════ @dataclass class LabelMetrics: label: str display_name: str f1: float precision: float recall: float tp: int fp: int fn: int tn: int @dataclass class ConfigData: name: str df: pd.DataFrame labels: list[str] metrics: list[LabelMetrics] macro_f1: float case_index: dict[str, dict[str, list[int]]] = field(default_factory=dict) # ═══════════════════════════════════════════════════════════════════════════ # LOADING & COMPUTATION # ═══════════════════════════════════════════════════════════════════════════ def load_config(name: str, parquet_path: Path) -> ConfigData: df = pd.read_parquet(parquet_path) available = [] for label in ALL_LABELS: pred_col = f"{label}{PRED_SUFFIX}" if label in df.columns and pred_col in df.columns: available.append(label) if not available: raise ValueError(f"No valid label pairs found in {parquet_path.name}") metrics_list: list[LabelMetrics] = [] case_index: dict[str, dict[str, list[int]]] = {} for label in available: pred_col = f"{label}{PRED_SUFFIX}" gold = df[label].fillna(0).astype(int).values pred = df[pred_col].fillna(0).astype(int).values tp_mask = (gold == 1) & (pred == 1) fp_mask = (gold == 0) & (pred == 1) fn_mask = (gold == 1) & (pred == 0) tn_mask = (gold == 0) & (pred == 0) tp = int(tp_mask.sum()) fp = int(fp_mask.sum()) fn = int(fn_mask.sum()) tn = int(tn_mask.sum()) prec = tp / (tp + fp) if (tp + fp) > 0 else 0.0 rec = tp / (tp + fn) if (tp + fn) > 0 else 0.0 f1 = (2 * prec * rec / (prec + rec)) if (prec + rec) > 0 else 0.0 metrics_list.append(LabelMetrics( label=label, display_name=display_name(label), f1=round(f1, 3), precision=round(prec, 3), recall=round(rec, 3), tp=tp, fp=fp, fn=fn, tn=tn, )) case_index[label] = { "tp": np.where(tp_mask)[0].tolist(), "fp": np.where(fp_mask)[0].tolist(), "fn": np.where(fn_mask)[0].tolist(), "tn": np.where(tn_mask)[0].tolist(), } macro_f1 = round(float(np.mean([m.f1 for m in metrics_list])), 3) if metrics_list else 0.0 return ConfigData( name=name, df=df, labels=available, metrics=metrics_list, macro_f1=macro_f1, case_index=case_index, ) def load_all_configs() -> dict[str, ConfigData]: configs: dict[str, ConfigData] = {} for name, filename in CONFIGS.items(): path = DATA_DIR / filename if path.exists(): print(f"Chargement : {name} ({filename})") try: configs[name] = load_config(name, path) print(f" -> {len(configs[name].df)} lignes, {len(configs[name].labels)} labels") except Exception as e: print(f"Erreur lors du chargement de {path}: {e}") else: print(f"Fichier manquant : {path}") return configs print("=" * 60) print("EMOTYC — Chargement des configurations...") print("=" * 60) ALL_CONFIGS = load_all_configs() print(f"\n{len(ALL_CONFIGS)} configuration(s) chargee(s).\n") # ═══════════════════════════════════════════════════════════════════════════ # FASTAPI APP # ═══════════════════════════════════════════════════════════════════════════ app = FastAPI(title="EMOTYC Visualisation") app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static") @app.get("/", response_class=HTMLResponse) def serve_index(): """Serve the main HTML page.""" index_path = BASE_DIR / "static" / "index.html" return index_path.read_text(encoding="utf-8") @app.get("/api/configs") def get_configs(): """Return the list of available configuration names.""" return {"configs": list(ALL_CONFIGS.keys())} @app.get("/api/configs/{config_name}/metrics") def get_metrics(config_name: str): """Return the metrics for a given configuration.""" if config_name not in ALL_CONFIGS: raise HTTPException(status_code=404, detail="Configuration non trouvee") config = ALL_CONFIGS[config_name] return { "metrics": [asdict(m) for m in config.metrics], "macro_f1": config.macro_f1, } @app.get("/api/configs/{config_name}/instances") def get_instances(config_name: str, label: str = Query(...), outcome: str = Query(...)): """Return the TEXT instances for a specific label and outcome.""" if config_name not in ALL_CONFIGS: raise HTTPException(status_code=404, detail="Configuration non trouvee") config = ALL_CONFIGS[config_name] if label not in config.case_index: raise HTTPException(status_code=404, detail="Label introuvable") outcome_key = outcome.lower() if outcome_key not in ("tp", "fp", "tn", "fn"): raise HTTPException(status_code=400, detail="Outcome invalide. Doit etre tp, fp, tn ou fn") indices = config.case_index[label].get(outcome_key, []) if not indices: return { "title": f"{OUTCOME_DISPLAY.get(outcome_key, outcome_key)} — {display_name(label)} — 0 instance", "texts": [], } if "TEXT" not in config.df.columns: raise HTTPException(status_code=500, detail="Colonne TEXT introuvable dans le jeu de donnees") texts = config.df.iloc[indices]["TEXT"].fillna("").astype(str).tolist() count = len(indices) title = ( f"{OUTCOME_DISPLAY.get(outcome_key, outcome_key)} — " f"{display_name(label)} — " f"{count} instance{'s' if count > 1 else ''}" ) return {"title": title, "texts": texts} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)