a6eb4c168f
Aspect 1/4 de Léa (agent Windows) : rendre Léa humaine.
Nouveaux modules :
- agent_v1/ui/messages.py : 11 formatters (cible non trouvée, mauvaise fenêtre,
écran inchangé, connexion, workflow, retry, ralentissement, erreur générique)
- agent_v1/ui/activity_panel.py : panneau tkinter lazy avec état courant,
action, progression X/Y, temps écoulé, 7 états (OBSERVE/CHERCHE/AGIT/VERIFIE...)
Hiérarchie de notifications :
- INFO (4s, vert) — début workflow, étape en cours
- ATTENTION (7s, orange) — retry, ralentissement
- BLOCAGE (15s, rouge, persistent, bypass rate-limit) — cible introuvable, mauvaise fenêtre
Transformations de messages :
AVANT : "target_not_found: dans *bonjour, – Bloc-notes"
APRÈS : "Léa a besoin d'aide"
"Je ne trouve pas « bonjour » dans Bloc-notes.
Peux-tu cliquer dessus toi-même ? Je reprends ensuite."
Robustesse :
- Détection fenêtre Léa via regex word-boundaries (évite cléa.txt, leapfrog.exe)
- Centralisée dans messages.est_fenetre_lea() — source unique de vérité
- Noop stub universel via __getattr__ (plus besoin de lister les méthodes)
- Thread-safe (RLock + snapshots immutables)
- Fallback silencieux si tkinter/plyer absent
101 nouveaux tests, aucune régression.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
419 lines
14 KiB
Python
419 lines
14 KiB
Python
# agent_v1/ui/activity_panel.py
|
|
"""
|
|
Panel d'activité temps réel de Léa.
|
|
|
|
Affiche à l'utilisateur ce que Léa fait *maintenant* :
|
|
- État courant (Observe / Cherche / Agit / Vérifie / Bloquée)
|
|
- Action en cours (ex: "Clic sur Rechercher")
|
|
- Progression (ex: "3/15")
|
|
- Temps écoulé depuis le début du workflow
|
|
|
|
Contraintes :
|
|
- Fallback silencieux si tkinter absent (ne crash jamais)
|
|
- Thread-safe (mises à jour depuis les threads de replay)
|
|
- Pas de dépendance à PyQt5 (seulement tkinter, déjà utilisé par chat_window)
|
|
|
|
Utilisation :
|
|
panel = ActivityPanel()
|
|
panel.definir_workflow("Saisie patient", nb_etapes=15)
|
|
panel.mettre_a_jour(etat=EtatLea.AGIT, action="Clic sur Valider", etape=3)
|
|
panel.masquer()
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class EtatLea(Enum):
|
|
"""États macroscopiques de Léa pendant un replay."""
|
|
|
|
INACTIVE = ("inactive", "Prête", "#808080") # Gris
|
|
OBSERVE = ("observe", "Observe", "#4A90E2") # Bleu
|
|
CHERCHE = ("cherche", "Cherche", "#F5A623") # Orange
|
|
AGIT = ("agit", "Agit", "#7ED321") # Vert
|
|
VERIFIE = ("verifie", "Vérifie", "#9013FE") # Violet
|
|
BLOQUEE = ("bloquee", "Bloquée", "#D0021B") # Rouge
|
|
TERMINE = ("termine", "Terminé", "#50E3C2") # Turquoise
|
|
|
|
def __init__(self, code: str, libelle: str, couleur: str) -> None:
|
|
self.code = code
|
|
self.libelle = libelle
|
|
self.couleur = couleur
|
|
|
|
|
|
@dataclass
|
|
class EtatActivite:
|
|
"""Instantané de l'activité courante de Léa.
|
|
|
|
Utilisé par le panel et exposé par `ActivityPanel.snapshot()` pour les
|
|
tests (sans dépendre de tkinter).
|
|
"""
|
|
|
|
etat: EtatLea = EtatLea.INACTIVE
|
|
action_courante: str = ""
|
|
nom_workflow: str = ""
|
|
etape: int = 0
|
|
nb_etapes: int = 0
|
|
debut_timestamp: float = 0.0
|
|
dernier_message: str = ""
|
|
|
|
def temps_ecoule_s(self) -> float:
|
|
"""Temps écoulé depuis le début du workflow (secondes)."""
|
|
if self.debut_timestamp <= 0:
|
|
return 0.0
|
|
return max(0.0, time.time() - self.debut_timestamp)
|
|
|
|
def progression_texte(self) -> str:
|
|
"""Représentation textuelle de la progression (ex: '3/15')."""
|
|
if self.nb_etapes <= 0:
|
|
return ""
|
|
return f"{self.etape}/{self.nb_etapes}"
|
|
|
|
def temps_ecoule_texte(self) -> str:
|
|
"""Représentation humaine du temps écoulé (ex: '12s', '1m24s')."""
|
|
s = int(self.temps_ecoule_s())
|
|
if s < 60:
|
|
return f"{s}s"
|
|
return f"{s // 60}m{s % 60:02d}s"
|
|
|
|
def to_dict(self) -> dict:
|
|
"""Sérialiser pour le logging et les tests."""
|
|
return {
|
|
"etat": self.etat.code,
|
|
"etat_libelle": self.etat.libelle,
|
|
"action_courante": self.action_courante,
|
|
"nom_workflow": self.nom_workflow,
|
|
"etape": self.etape,
|
|
"nb_etapes": self.nb_etapes,
|
|
"progression": self.progression_texte(),
|
|
"temps_ecoule_s": round(self.temps_ecoule_s(), 1),
|
|
"dernier_message": self.dernier_message,
|
|
}
|
|
|
|
|
|
class ActivityPanel:
|
|
"""Panel d'activité de Léa.
|
|
|
|
Thread-safe. Le panel tkinter est créé à la demande (lazy) et uniquement
|
|
si tkinter est disponible. Toutes les méthodes sont safe à appeler même
|
|
si l'UI n'est pas dispo (fallback silencieux).
|
|
"""
|
|
|
|
def __init__(self, activer_ui: bool = True) -> None:
|
|
self._lock = threading.RLock()
|
|
self._etat = EtatActivite()
|
|
self._activer_ui = activer_ui
|
|
# UI tkinter (créée à la demande dans le thread UI)
|
|
self._tk_root = None
|
|
self._tk_labels: dict = {}
|
|
self._ui_disponible = None # Lazy : résolu au premier usage
|
|
self._listeners = [] # Callbacks pour les changements d'état
|
|
|
|
# ------------------------------------------------------------------
|
|
# API publique (thread-safe)
|
|
# ------------------------------------------------------------------
|
|
|
|
def definir_workflow(self, nom: str, nb_etapes: int = 0) -> None:
|
|
"""Démarrer le suivi d'un nouveau workflow."""
|
|
with self._lock:
|
|
self._etat = EtatActivite(
|
|
etat=EtatLea.OBSERVE,
|
|
nom_workflow=nom,
|
|
nb_etapes=nb_etapes,
|
|
debut_timestamp=time.time(),
|
|
)
|
|
self._notifier_changement()
|
|
self._rafraichir_ui()
|
|
logger.info(f"[ACTIVITY] Workflow démarré : {nom} ({nb_etapes} étapes)")
|
|
|
|
def mettre_a_jour(
|
|
self,
|
|
etat: Optional[EtatLea] = None,
|
|
action: Optional[str] = None,
|
|
etape: Optional[int] = None,
|
|
message: Optional[str] = None,
|
|
) -> None:
|
|
"""Mettre à jour l'état affiché.
|
|
|
|
Tous les paramètres sont optionnels — on ne met à jour que ce qui est
|
|
fourni. Les autres champs conservent leur valeur actuelle.
|
|
"""
|
|
with self._lock:
|
|
if etat is not None:
|
|
self._etat.etat = etat
|
|
if action is not None:
|
|
self._etat.action_courante = action
|
|
if etape is not None:
|
|
self._etat.etape = etape
|
|
if message is not None:
|
|
self._etat.dernier_message = message
|
|
|
|
self._notifier_changement()
|
|
self._rafraichir_ui()
|
|
|
|
def terminer(self, succes: bool = True) -> None:
|
|
"""Marquer le workflow comme terminé."""
|
|
with self._lock:
|
|
self._etat.etat = EtatLea.TERMINE if succes else EtatLea.BLOQUEE
|
|
if not succes:
|
|
self._etat.dernier_message = (
|
|
self._etat.dernier_message or "Léa a rendu la main"
|
|
)
|
|
self._notifier_changement()
|
|
self._rafraichir_ui()
|
|
|
|
def reinitialiser(self) -> None:
|
|
"""Remettre le panel en état inactif."""
|
|
with self._lock:
|
|
self._etat = EtatActivite()
|
|
self._notifier_changement()
|
|
self._rafraichir_ui()
|
|
|
|
def snapshot(self) -> EtatActivite:
|
|
"""Obtenir un instantané immuable de l'état courant (pour les tests)."""
|
|
with self._lock:
|
|
return EtatActivite(
|
|
etat=self._etat.etat,
|
|
action_courante=self._etat.action_courante,
|
|
nom_workflow=self._etat.nom_workflow,
|
|
etape=self._etat.etape,
|
|
nb_etapes=self._etat.nb_etapes,
|
|
debut_timestamp=self._etat.debut_timestamp,
|
|
dernier_message=self._etat.dernier_message,
|
|
)
|
|
|
|
def masquer(self) -> None:
|
|
"""Masquer le panel UI si affiché."""
|
|
if self._tk_root is not None:
|
|
try:
|
|
self._tk_root.withdraw()
|
|
except Exception:
|
|
pass
|
|
|
|
def afficher(self) -> None:
|
|
"""Afficher le panel UI si disponible."""
|
|
self._creer_ui_si_besoin()
|
|
if self._tk_root is not None:
|
|
try:
|
|
self._tk_root.deiconify()
|
|
except Exception:
|
|
pass
|
|
|
|
def on_change(self, callback) -> None:
|
|
"""Enregistrer un listener appelé à chaque changement d'état."""
|
|
with self._lock:
|
|
self._listeners.append(callback)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Gestion UI tkinter (lazy, fallback silencieux)
|
|
# ------------------------------------------------------------------
|
|
|
|
def _creer_ui_si_besoin(self) -> None:
|
|
"""Créer la fenêtre tkinter au premier usage (lazy)."""
|
|
if not self._activer_ui:
|
|
return
|
|
if self._tk_root is not None:
|
|
return
|
|
if self._ui_disponible is False:
|
|
return # Déjà testé et indisponible
|
|
|
|
try:
|
|
import tkinter as tk
|
|
except Exception as e:
|
|
logger.debug(f"[ACTIVITY] tkinter indisponible : {e}")
|
|
self._ui_disponible = False
|
|
return
|
|
|
|
try:
|
|
self._tk_root = tk.Toplevel() if _tk_root_existe() else tk.Tk()
|
|
self._tk_root.title("Léa — Activité")
|
|
self._tk_root.geometry("340x180+40+40")
|
|
self._tk_root.attributes("-topmost", True)
|
|
self._tk_root.resizable(False, False)
|
|
self._tk_root.configure(bg="#1E1E1E")
|
|
|
|
titre = tk.Label(
|
|
self._tk_root,
|
|
text="Léa",
|
|
font=("Segoe UI", 14, "bold"),
|
|
fg="#FFFFFF",
|
|
bg="#1E1E1E",
|
|
)
|
|
titre.pack(pady=(10, 2))
|
|
|
|
self._tk_labels["etat"] = tk.Label(
|
|
self._tk_root,
|
|
text="Prête",
|
|
font=("Segoe UI", 11),
|
|
fg="#808080",
|
|
bg="#1E1E1E",
|
|
)
|
|
self._tk_labels["etat"].pack()
|
|
|
|
self._tk_labels["action"] = tk.Label(
|
|
self._tk_root,
|
|
text="",
|
|
font=("Segoe UI", 10),
|
|
fg="#FFFFFF",
|
|
bg="#1E1E1E",
|
|
wraplength=300,
|
|
)
|
|
self._tk_labels["action"].pack(pady=(8, 2))
|
|
|
|
self._tk_labels["progression"] = tk.Label(
|
|
self._tk_root,
|
|
text="",
|
|
font=("Segoe UI", 9),
|
|
fg="#B0B0B0",
|
|
bg="#1E1E1E",
|
|
)
|
|
self._tk_labels["progression"].pack()
|
|
|
|
self._tk_labels["temps"] = tk.Label(
|
|
self._tk_root,
|
|
text="",
|
|
font=("Segoe UI", 9),
|
|
fg="#808080",
|
|
bg="#1E1E1E",
|
|
)
|
|
self._tk_labels["temps"].pack(pady=(4, 0))
|
|
|
|
self._tk_labels["message"] = tk.Label(
|
|
self._tk_root,
|
|
text="",
|
|
font=("Segoe UI", 9, "italic"),
|
|
fg="#B0B0B0",
|
|
bg="#1E1E1E",
|
|
wraplength=300,
|
|
)
|
|
self._tk_labels["message"].pack(pady=(6, 10))
|
|
|
|
# Masquer par défaut : on affiche seulement pendant un workflow
|
|
self._tk_root.withdraw()
|
|
self._ui_disponible = True
|
|
except Exception as e:
|
|
logger.debug(f"[ACTIVITY] Impossible de créer l'UI : {e}")
|
|
self._ui_disponible = False
|
|
self._tk_root = None
|
|
|
|
def _rafraichir_ui(self) -> None:
|
|
"""Mettre à jour les labels tkinter (safe si l'UI n'existe pas)."""
|
|
if not self._activer_ui or self._ui_disponible is False:
|
|
return
|
|
self._creer_ui_si_besoin()
|
|
if self._tk_root is None:
|
|
return
|
|
|
|
try:
|
|
with self._lock:
|
|
snap = self.snapshot()
|
|
|
|
# Utiliser after(0) pour rester dans le thread UI tkinter
|
|
def _update():
|
|
try:
|
|
self._tk_labels["etat"].config(
|
|
text=snap.etat.libelle,
|
|
fg=snap.etat.couleur,
|
|
)
|
|
if snap.action_courante:
|
|
self._tk_labels["action"].config(text=snap.action_courante)
|
|
else:
|
|
self._tk_labels["action"].config(text="")
|
|
|
|
prog = snap.progression_texte()
|
|
if prog and snap.nom_workflow:
|
|
self._tk_labels["progression"].config(
|
|
text=f"« {snap.nom_workflow} » — {prog}"
|
|
)
|
|
elif snap.nom_workflow:
|
|
self._tk_labels["progression"].config(
|
|
text=f"« {snap.nom_workflow} »"
|
|
)
|
|
else:
|
|
self._tk_labels["progression"].config(text="")
|
|
|
|
if snap.debut_timestamp > 0:
|
|
self._tk_labels["temps"].config(
|
|
text=f"⏱ {snap.temps_ecoule_texte()}"
|
|
)
|
|
else:
|
|
self._tk_labels["temps"].config(text="")
|
|
|
|
self._tk_labels["message"].config(text=snap.dernier_message)
|
|
|
|
# Afficher automatiquement si actif
|
|
if snap.etat != EtatLea.INACTIVE:
|
|
self._tk_root.deiconify()
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
self._tk_root.after(0, _update)
|
|
except Exception:
|
|
# Si le root a été détruit
|
|
self._tk_root = None
|
|
self._ui_disponible = False
|
|
except Exception as e:
|
|
logger.debug(f"[ACTIVITY] Erreur rafraîchissement UI : {e}")
|
|
|
|
def _notifier_changement(self) -> None:
|
|
"""Notifier tous les listeners du changement d'état."""
|
|
with self._lock:
|
|
listeners = list(self._listeners)
|
|
snap = self.snapshot()
|
|
|
|
for cb in listeners:
|
|
try:
|
|
cb(snap)
|
|
except Exception as e:
|
|
logger.debug(f"[ACTIVITY] Listener erreur : {e}")
|
|
|
|
|
|
def _tk_root_existe() -> bool:
|
|
"""Vérifier si un root tkinter existe déjà (pour créer un Toplevel)."""
|
|
try:
|
|
import tkinter as tk
|
|
|
|
default_root = getattr(tk, "_default_root", None)
|
|
return default_root is not None
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
# ============================================================================
|
|
# Singleton global (optionnel)
|
|
# ============================================================================
|
|
|
|
|
|
_INSTANCE_GLOBALE: Optional[ActivityPanel] = None
|
|
_LOCK_SINGLETON = threading.Lock()
|
|
|
|
|
|
def get_activity_panel(activer_ui: bool = True) -> ActivityPanel:
|
|
"""Obtenir l'instance globale du panel d'activité (lazy)."""
|
|
global _INSTANCE_GLOBALE
|
|
with _LOCK_SINGLETON:
|
|
if _INSTANCE_GLOBALE is None:
|
|
_INSTANCE_GLOBALE = ActivityPanel(activer_ui=activer_ui)
|
|
return _INSTANCE_GLOBALE
|
|
|
|
|
|
def reset_activity_panel() -> None:
|
|
"""Réinitialiser le singleton (utile pour les tests)."""
|
|
global _INSTANCE_GLOBALE
|
|
with _LOCK_SINGLETON:
|
|
if _INSTANCE_GLOBALE is not None:
|
|
try:
|
|
_INSTANCE_GLOBALE.masquer()
|
|
except Exception:
|
|
pass
|
|
_INSTANCE_GLOBALE = None
|