Spaces:
Paused
Paused
| """ | |
| app.py — tracer scan en tant que mini-app Gradio, à héberger sur un HuggingFace | |
| Space (SDK: gradio). Permet d'uploader n'importe quel .jsonl compatible TRACER, | |
| de choisir l'embedder / le target / le mode force, et de récupérer le rapport | |
| HTML interactif + un résumé terminal-like du balayage de targets. | |
| Déploiement : | |
| 1. huggingface.co/new-space -> SDK "Gradio" -> hardware "CPU basic" suffit | |
| 2. Uploader ce fichier (app.py) + requirements.txt à la racine du Space | |
| 3. Le Space build automatiquement et expose l'UI | |
| Aucun souci réseau/SSL attendu ici : le Space tourne sur l'infra HuggingFace, | |
| donc le téléchargement du modèle sentence-transformers se fait en local au | |
| Hub, sans proxy ni certificat tiers à valider. | |
| """ | |
| from __future__ import annotations | |
| import tempfile | |
| from pathlib import Path | |
| import gradio as gr | |
| import numpy as np | |
| import tracer | |
| from tracer.scanner import ThinDataError, load_scan_traces, scan_html | |
| from tracer.embeddings.index import embed_texts | |
| EMBED_MODEL_CHOICES = [ | |
| "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2", # défaut fit_tracer.py | |
| "BAAI/bge-m3", # exigé par la spec | |
| "all-MiniLM-L6-v2", # défaut CLI tracer (EN only) | |
| ] | |
| # Cache mémoire (process du Space) pour éviter de ré-embruter le même | |
| # fichier+modèle entre deux clics (target/force/viz-layout différents). | |
| _EMBED_CACHE: dict[str, np.ndarray] = {} | |
| def _cache_key(path: Path, model: str) -> str: | |
| import hashlib | |
| h = hashlib.sha256() | |
| h.update(path.read_bytes()) | |
| h.update(model.encode("utf-8")) | |
| return h.hexdigest() | |
| def _get_embeddings(path: Path, model: str, inputs: list[str], progress: gr.Progress | None = None) -> np.ndarray: | |
| key = _cache_key(path, model) | |
| if key in _EMBED_CACHE: | |
| return _EMBED_CACHE[key] | |
| if progress: | |
| progress(0.1, desc=f"Téléchargement / chargement de {model}...") | |
| X = embed_texts(inputs, model=model, show_progress=False) | |
| _EMBED_CACHE[key] = X | |
| return X | |
| def run_scan( | |
| files, | |
| embed_model: str, | |
| primary_target: float, | |
| sweep_targets_str: str, | |
| force_mode: str, | |
| price: float, | |
| monthly_calls: int, | |
| viz_layout: str, | |
| progress: gr.Progress = gr.Progress(), | |
| ): | |
| """Callback principal du bouton 'Lancer le scan'.""" | |
| if not files: | |
| return "⚠ Aucun fichier fourni.", "", [] | |
| try: | |
| sweep_targets = sorted( | |
| {float(t.strip()) for t in sweep_targets_str.split(",") if t.strip()} | |
| | {primary_target} | |
| ) | |
| except ValueError: | |
| return "⚠ Targets de balayage invalides (attendu : ex. '0.8, 0.85, 0.9, 0.95').", "", [] | |
| summary_lines: list[str] = [] | |
| html_reports: list[str] = [] | |
| html_files: list[str] = [] | |
| for i, file_obj in enumerate(files): | |
| path = Path(file_obj.name if hasattr(file_obj, "name") else file_obj) | |
| progress((i) / len(files), desc=f"Lecture de {path.name}...") | |
| try: | |
| inputs, labels = load_scan_traces(path) | |
| except Exception as exc: | |
| summary_lines.append(f"=== {path.name} ===\n⚠ Erreur de lecture : {exc}\n") | |
| continue | |
| n = len(inputs) | |
| n_classes = len(set(labels)) | |
| summary_lines.append(f"=== {path.name} ===") | |
| summary_lines.append(f"{n} traces utilisables · {n_classes} labels distincts") | |
| if force_mode == "auto": | |
| effective_force = n < 1000 | |
| if effective_force: | |
| summary_lines.append(f"⚠ {n} < 1000 traces : force appliqué automatiquement (résultat best-effort).") | |
| elif force_mode == "force": | |
| effective_force = True | |
| else: | |
| effective_force = False | |
| progress((i + 0.3) / len(files), desc=f"Embeddings ({embed_model})...") | |
| try: | |
| X = _get_embeddings(path, embed_model, inputs, progress=progress) | |
| except Exception as exc: | |
| summary_lines.append(f"⚠ Échec du calcul des embeddings : {exc}\n") | |
| continue | |
| progress((i + 0.6) / len(files), desc="Balayage de targets...") | |
| for t in sweep_targets: | |
| try: | |
| result = tracer.scan(path, target=t, embeddings=X, force=effective_force, viz_layout=viz_layout) | |
| summary_lines.append( | |
| f" target {t:.0%} certifiable = {result.certifiable_share:6.1%} " | |
| f"({result.n_clusters} cellules{' [forced]' if result.forced else ''})" | |
| ) | |
| except ThinDataError as exc: | |
| summary_lines.append(f" target {t:.0%} {exc}") | |
| progress((i + 0.9) / len(files), desc="Génération du rapport HTML...") | |
| try: | |
| result = tracer.scan( | |
| path, | |
| target=primary_target, | |
| embeddings=X, | |
| force=effective_force, | |
| teacher_price_per_1k=(price or None), | |
| monthly_calls=(monthly_calls or None), | |
| viz_layout=viz_layout, | |
| ) | |
| html = scan_html(result, source_name=path.name) | |
| tmp = tempfile.NamedTemporaryFile( | |
| delete=False, suffix=".html", prefix=f"{path.stem}_scan_" | |
| ) | |
| tmp.write(html.encode("utf-8")) | |
| tmp.close() | |
| html_files.append(tmp.name) | |
| html_reports.append(html) | |
| summary_lines.append(f" → rapport HTML généré ({path.name})") | |
| except ThinDataError as exc: | |
| summary_lines.append(f" ⚠ Rapport HTML non généré : {exc}") | |
| summary_lines.append("") | |
| terminal_summary = "\n".join(summary_lines) | |
| preview_html = html_reports[0] if html_reports else "<p>Aucun rapport généré.</p>" | |
| return terminal_summary, preview_html, html_files | |
| with gr.Blocks(title="TRACER scan — testeur de dataset") as demo: | |
| gr.Markdown( | |
| "# tracer scan — testeur de dataset\n" | |
| "Upload un ou plusieurs `.jsonl` compatibles TRACER (`input`+`teacher`, " | |
| "ou alias tolérés) pour voir leur routabilité probable **avant** de lancer " | |
| "`fit`. Tourne sur l'infra HuggingFace : le modèle d'embedding se " | |
| "télécharge directement depuis le Hub, sans souci réseau/proxy local." | |
| ) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| files_in = gr.File( | |
| label="Fichier(s) .jsonl", file_count="multiple", file_types=[".jsonl"] | |
| ) | |
| embed_model_in = gr.Dropdown( | |
| label="Embedder", | |
| choices=EMBED_MODEL_CHOICES, | |
| value=EMBED_MODEL_CHOICES[0], | |
| allow_custom_value=True, | |
| ) | |
| target_in = gr.Slider( | |
| label="Target primaire (rapport HTML détaillé)", | |
| minimum=0.5, maximum=0.99, step=0.01, value=0.95, | |
| ) | |
| sweep_in = gr.Textbox( | |
| label="Targets de balayage (séparés par des virgules)", | |
| value="0.80, 0.85, 0.90, 0.95", | |
| ) | |
| force_in = gr.Radio( | |
| label="Mode thin-data", | |
| choices=[("auto (force si <1000 traces)", "auto"), ("forcer", "force"), ("refuser", "no_force")], | |
| value="auto", | |
| ) | |
| with gr.Accordion("Options économiques (facultatif)", open=False): | |
| price_in = gr.Number(label="Prix teacher $/1k appels", value=None) | |
| monthly_in = gr.Number(label="Volume mensuel d'appels", value=None, precision=0) | |
| viz_in = gr.Radio( | |
| label="Layout 3D du rapport", choices=["pca", "umap", "tsne", "auto"], value="pca" | |
| ) | |
| run_btn = gr.Button("Lancer le scan", variant="primary") | |
| with gr.Column(scale=2): | |
| summary_out = gr.Textbox(label="Résumé (terminal-like)", lines=24) | |
| html_preview = gr.HTML(label="Aperçu du rapport (1er fichier)") | |
| html_download = gr.Files(label="Télécharger le(s) rapport(s) HTML") | |
| run_btn.click( | |
| run_scan, | |
| inputs=[files_in, embed_model_in, target_in, sweep_in, force_in, price_in, monthly_in, viz_in], | |
| outputs=[summary_out, html_preview, html_download], | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch(show_api=False) |