rpa_vision_v3/agent_v0/lea_ui/replay_integration.py

# agent_v0/lea_ui/replay_integration.py
"""
Integration du feedback visuel (overlay) dans la boucle de replay de l'Agent V1.

Ce module fournit un wrapper autour de ActionExecutorV1.execute_replay_action
qui affiche l'overlay AVANT chaque action et la marque comme terminee APRES.

Sequence pour chaque action :
  1. Afficher l'overlay avec la description de l'action (1.5s)
  2. Attendre que l'overlay ait ete vu par l'utilisateur
  3. Executer l'action
  4. Mettre a jour l'overlay (coche verte)
  5. Passer a l'action suivante
"""

from __future__ import annotations

import logging
import time
from typing import Any, Callable, Dict, Optional, Tuple

logger = logging.getLogger("lea_ui.replay_integration")

# Delai d'affichage de l'overlay avant execution (secondes)
PRE_ACTION_DELAY = 1.5
# Delai apres la coche verte (secondes)
POST_ACTION_DELAY = 0.5


class ReplayOverlayBridge:
    """Pont entre la boucle de replay et l'overlay.

    Fonctionne de maniere thread-safe : la boucle de replay tourne dans
    un thread daemon, et l'overlay est controle via des signaux Qt.

    L'overlay est optionnel — si non connecte, l'execution continue normalement.
    """

    def __init__(self) -> None:
        self._overlay = None
        self._show_callback: Optional[Callable] = None
        self._done_callback: Optional[Callable] = None
        self._hide_callback: Optional[Callable] = None
        self._enabled = False

        # Compteur de progression
        self._step_current = 0
        self._step_total = 0

    def connect_overlay(
        self,
        show_fn: Callable[[int, int, str, int, int, int], None],
        done_fn: Callable[[Optional[str]], None],
        hide_fn: Callable[[], None],
    ) -> None:
        """Connecter les callbacks de l'overlay.

        Args:
            show_fn: overlay.show_action(target_x, target_y, text, step, total, duration_ms)
            done_fn: overlay.show_done(text)
            hide_fn: overlay.hide_overlay()
        """
        self._show_callback = show_fn
        self._done_callback = done_fn
        self._hide_callback = hide_fn
        self._enabled = True
        logger.info("Overlay connecte au bridge de replay")

    def disconnect_overlay(self) -> None:
        """Deconnecter l'overlay."""
        self._show_callback = None
        self._done_callback = None
        self._hide_callback = None
        self._enabled = False

    def set_total_steps(self, total: int) -> None:
        """Definir le nombre total d'etapes du replay."""
        self._step_total = total
        self._step_current = 0

    def wrap_execute(
        self,
        action: Dict[str, Any],
        executor_fn: Callable[[Dict[str, Any]], Dict[str, Any]],
        screen_width: int = 1920,
        screen_height: int = 1080,
    ) -> Dict[str, Any]:
        """Wrapper autour de l'execution d'une action avec feedback overlay.

        Args:
            action: action normalisee (type, x_pct, y_pct, text, keys, ...)
            executor_fn: fonction d'execution (ex: ActionExecutorV1.execute_replay_action)
            screen_width: largeur de l'ecran en pixels
            screen_height: hauteur de l'ecran en pixels

        Returns:
            Resultat de l'execution (dict avec success, error, screenshot, ...)
        """
        self._step_current += 1

        if not self._enabled or not self._show_callback:
            # Pas d'overlay — execution directe
            return executor_fn(action)

        # --- 1. Afficher l'overlay ---
        action_text = self._describe_action(action)
        target_x, target_y = self._get_target_coords(action, screen_width, screen_height)

        try:
            self._show_callback(
                target_x, target_y,
                action_text,
                self._step_current,
                self._step_total,
                int(PRE_ACTION_DELAY * 1000),
            )
        except Exception as e:
            logger.warning("Erreur affichage overlay : %s", e)

        # --- 2. Attendre que l'utilisateur ait vu ---
        time.sleep(PRE_ACTION_DELAY)

        # --- 3. Executer l'action ---
        result = executor_fn(action)

        # --- 4. Marquer comme terminee ---
        if result.get("success"):
            done_text = f"{action_text} OK"
        else:
            done_text = f"{action_text} ECHEC"

        try:
            if self._done_callback:
                self._done_callback(done_text)
        except Exception as e:
            logger.warning("Erreur overlay done : %s", e)

        time.sleep(POST_ACTION_DELAY)

        # --- 5. Cacher si c'etait la derniere etape ---
        if self._step_current >= self._step_total and self._hide_callback:
            try:
                self._hide_callback()
            except Exception:
                pass

        return result

    def _describe_action(self, action: Dict[str, Any]) -> str:
        """Generer une description lisible d'une action."""
        action_type = action.get("type", "?")
        target_text = action.get("target_text", "")
        target_role = action.get("target_role", "")

        if action_type == "click":
            target = target_text or target_role or "cet element"
            return f"Je clique sur [{target}]"
        elif action_type == "type":
            text = action.get("text", "")
            preview = text[:25] + "..." if len(text) > 25 else text
            return f"Je tape : {preview}"
        elif action_type == "key_combo":
            keys = action.get("keys", [])
            return f"Combinaison : {'+'.join(keys)}"
        elif action_type == "scroll":
            return "Defilement"
        elif action_type == "wait":
            ms = action.get("duration_ms", 500)
            return f"Attente {ms}ms"
        else:
            return f"Action : {action_type}"

    def _get_target_coords(
        self, action: Dict[str, Any], sw: int, sh: int,
    ) -> Tuple[int, int]:
        """Calculer les coordonnees cible en pixels."""
        x_pct = action.get("x_pct", 0.5)
        y_pct = action.get("y_pct", 0.5)
        return int(x_pct * sw), int(y_pct * sh)


# Instance globale (singleton) pour l'integration
_bridge: Optional[ReplayOverlayBridge] = None


def get_replay_bridge() -> ReplayOverlayBridge:
    """Obtenir l'instance globale du bridge overlay/replay."""
    global _bridge
    if _bridge is None:
        _bridge = ReplayOverlayBridge()
    return _bridge