Spaces:
Running
Running
| """Interface Hugging Face Space : soumettre une tâche d'ingénierie et observer le bus JEV en temps réel.""" | |
| import os | |
| import gradio as gr | |
| from mouh.learning import autopilot | |
| from mouh.service import GENERATOR_CHOICES, stream_session | |
| EXAMPLES = { | |
| "Bug : add retourne a - b (rejeté)": ( | |
| "Écrire add(a, b) qui renvoie la somme.", | |
| "def add(a, b):\n return a - b\n", | |
| "from solution import add\n\n\ndef test_add():\n assert add(2, 3) == 5\n\n\ndef test_zero():\n assert add(0, 0) == 0\n", | |
| "", | |
| ), | |
| "Correct : add retourne a + b (certifié)": ( | |
| "Écrire add(a, b) qui renvoie la somme.", | |
| "def add(a, b):\n return a + b\n", | |
| "from solution import add\n\n\ndef test_add():\n assert add(2, 3) == 5\n\n\ndef test_zero():\n assert add(0, 0) == 0\n", | |
| '[{"name": "croissance", "declarations": {"x": "Int"}, "assumptions": [[">", "x", 0]], "goal": [">", ["+", "x", 1], 1]}]', | |
| ), | |
| } | |
| INTRO = """# M.O.U.H. — Moteur Hybride Optimisé Unifié | |
| Soumettez une tâche, un code candidat, des tests et (optionnellement) des obligations de preuve Z3. | |
| La boucle de vérification ne produit une sortie **que si** un certificat est délivré (score ≥ 0.85 et toutes les | |
| barrières franchies). Le tableau ci-dessous affiche chaque message du **bus JEV** au fur et à mesure.""" | |
| def run(instruction, code, tests, proofs, max_attempts, generator, record=True): | |
| last = None | |
| for s in stream_session(instruction, code, tests, proofs, int(max_attempts), generator, bool(record)): | |
| last = s | |
| yield s["rows"], s["status"], s["final_code"], s["report"] | |
| if last is None: | |
| yield [], "aucun résultat", "", "" | |
| def load_example(name): | |
| return EXAMPLES[name] | |
| with gr.Blocks(title="M.O.U.H.") as demo: | |
| gr.Markdown(INTRO) | |
| with gr.Accordion("Apprentissage automatique d'UNIQUE", open=True): | |
| live = gr.HTML(autopilot.render_progress(None)) | |
| timer = gr.Timer(15) | |
| timer.tick(lambda: autopilot.render_progress(autopilot.read_progress()), None, live) | |
| learn_status = gr.Markdown("Cliquez sur « Actualiser » pour lire l'état des jobs.") | |
| with gr.Row(): | |
| refresh = gr.Button("Actualiser") | |
| admin_key = gr.Textbox(label="Clé d'administration (UNIQUE_ADMIN_KEY)", type="password", scale=2) | |
| ingest = gr.Button("Alimenter maintenant (CPU)", variant="secondary") | |
| launch = gr.Button("Entraîner maintenant (GPU payant)", variant="secondary") | |
| launch_msg = gr.Markdown() | |
| refresh.click(lambda: autopilot.status_markdown(), None, learn_status) | |
| ingest.click(lambda k: autopilot.trigger_now(k, which="ingest"), admin_key, launch_msg) | |
| launch.click(autopilot.trigger_now, admin_key, launch_msg) | |
| with gr.Row(): | |
| with gr.Column(): | |
| example = gr.Dropdown(list(EXAMPLES), label="Exemple", value=None) | |
| instruction = gr.Textbox(label="Énoncé de la tâche", lines=2) | |
| code = gr.Code(label="Code candidat (solution.py)", language="python", lines=10) | |
| tests = gr.Code(label="Tests pytest (test_solution.py)", language="python", lines=10) | |
| proofs = gr.Code(label="Obligations de preuve Z3 (JSON, optionnel)", language="json", lines=6) | |
| with gr.Row(): | |
| attempts = gr.Slider(1, 8, value=4, step=1, label="Tentatives max") | |
| generator = gr.Dropdown(list(GENERATOR_CHOICES), value=GENERATOR_CHOICES[0], label="Générateur de corrections") | |
| record = gr.Checkbox(value=os.environ.get("MOUH_RECORD_DEFAULT", "1") == "1", | |
| label="Utiliser cette session pour entraîner UNIQUE (énoncé, tests et code " | |
| "sont enregistrés dans un dataset privé)") | |
| btn = gr.Button("Vérifier", variant="primary") | |
| with gr.Column(): | |
| status = gr.Textbox(label="Statut", interactive=False) | |
| bus = gr.Dataframe(headers=["#", "type", "émetteur → destinataire", "résumé"], interactive=False, | |
| wrap=True, label="Bus JEV (temps réel)") | |
| final = gr.Code(label="Sortie certifiée", language="python", interactive=False) | |
| report = gr.Code(label="Rapport", language="json", interactive=False) | |
| example.change(load_example, example, [instruction, code, tests, proofs]) | |
| btn.click(run, [instruction, code, tests, proofs, attempts, generator, record], [bus, status, final, report]) | |
| if __name__ == "__main__": | |
| autopilot.start_in_background() # crée le job planifié si UNIQUE_AUTOTRAIN=1 (une seule fois) | |
| demo.queue().launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", "7860"))) | |