app.py

"""
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)