# agent_v1/main.py """ Point d'entree Agent V1 - Enrichi avec Intelligence de Contexte, Heartbeat et Replay. Boucles paralleles (threads daemon) : - _heartbeat_loop : capture periodique toutes les 5s - _command_watchdog_loop : surveillance du fichier command.json (legacy) - _replay_poll_loop : polling du serveur pour les actions de replay (P0-5) """ import sys import os import uuid import time import logging import threading from .config import ( SESSIONS_ROOT, AGENT_VERSION, SERVER_URL, MACHINE_ID, LOG_RETENTION_DAYS, SCREEN_RESOLUTION, DPI_SCALE, OS_THEME, API_TOKEN, MAX_SESSION_DURATION_S, STREAMING_ENDPOINT, ) from .core.captor import EventCaptorV1 from .core.executor import ActionExecutorV1 from .network.streamer import TraceStreamer from .ui.shared_state import AgentState from .ui.smart_tray import SmartTrayV1 from .ui.chat_window import ChatWindow from .ui.capture_server import CaptureServer from .session.storage import SessionStorage from .vision.capturer import VisionCapturer # Import optionnel du client serveur (pour le chat et les workflows) # Deux chemins : relatif (depuis agent_v0.agent_v1) ou absolu (depuis C:\rpa_vision\agent_v1) try: from ..lea_ui.server_client import LeaServerClient except (ImportError, ValueError): try: from lea_ui.server_client import LeaServerClient except ImportError: LeaServerClient = None # Configuration du logging — format structuré et lisible pour un TIM # Niveau de détail : INFO par défaut, DEBUG si RPA_AGENT_DEBUG=1 _log_level = logging.DEBUG if os.environ.get("RPA_AGENT_DEBUG") == "1" else logging.INFO logging.basicConfig( level=_log_level, format="%(asctime)s %(levelname)-7s %(name)-25s %(message)s", datefmt="%H:%M:%S", ) # Réduire le bruit de certaines libs for _noisy in ("urllib3", "requests.packages.urllib3", "PIL", "mss"): logging.getLogger(_noisy).setLevel(logging.WARNING) logger = logging.getLogger(__name__) # Intervalle de polling replay (secondes) REPLAY_POLL_INTERVAL = 1.0 class AgentV1: def __init__(self, user_id="demo_user"): self.user_id = user_id self.machine_id = MACHINE_ID self.session_id = None self.session_dir = None # Gestion du stockage local et nettoyage # Retention minimum 6 mois (Reglement IA, Article 12) self.storage = SessionStorage(SESSIONS_ROOT, retention_days=LOG_RETENTION_DAYS) threading.Thread(target=self._delayed_cleanup, daemon=True).start() self.vision = None self.streamer = None self.captor = None self.shot_counter = 0 self.running = False # Executeur partage entre watchdog et replay self._executor = None # Flag pour indiquer qu'un replay est en cours (eviter les conflits) self._replay_active = False # Etat partage entre systray et chat (source de verite unique) self._state = AgentState() self._state.set_on_start(self.start_session) self._state.set_on_stop(self.stop_session) # Client serveur pour le chat et les workflows # Plus de RPA_SERVER_HOST : le LeaServerClient derive tout de SERVER_URL self._server_client = None if LeaServerClient is not None: # Forcer le token API pour éviter les 401 # (le token est set par start.bat dans l'environnement) from .config import API_TOKEN as _token self._server_client = LeaServerClient() if _token and not self._server_client._api_token: self._server_client._api_token = _token logger.info("Token API forcé dans LeaServerClient") # Fenetre de chat Lea (tkinter natif) # Le host est derive de SERVER_URL (plus de RPA_SERVER_HOST) server_host = ( self._server_client.server_host if self._server_client is not None else "localhost" ) self._chat_window = ChatWindow( server_client=self._server_client, on_start_callback=self.start_session, server_host=server_host, chat_port=5004, shared_state=self._state, ) # Executeur pour le replay (doit exister avant le poll) self._executor = ActionExecutorV1() # Boucles permanentes (pas besoin de session active) self.running = True self._bg_vision = VisionCapturer(str(SESSIONS_ROOT / "_background")) threading.Thread(target=self._replay_poll_loop, daemon=True).start() threading.Thread(target=self._background_heartbeat_loop, daemon=True).start() # Mini-serveur HTTP pour captures a la demande (port 5006) self._capture_server = CaptureServer() self._capture_server.start() # Bannière de démarrage avec métadonnées système logger.info( f"Agent V1 v{AGENT_VERSION} | Machine={self.machine_id} | " f"Ecran={SCREEN_RESOLUTION[0]}x{SCREEN_RESOLUTION[1]} | " f"DPI={DPI_SCALE}% | Theme={OS_THEME} | " f"Serveur={SERVER_URL}" ) # UI Tray intelligent (remplace TrayAppV1, plus de PyQt5) self.ui = SmartTrayV1( self.start_session, self.stop_session, server_client=self._server_client, chat_window=self._chat_window, machine_id=self.machine_id, shared_state=self._state, ) def _delayed_cleanup(self): """Nettoyage en arrière-plan après 30s pour ne pas bloquer le démarrage.""" time.sleep(30) self.storage.run_auto_cleanup() def _auto_stop_loop(self): """Auto-stop de l'enregistrement après MAX_SESSION_DURATION_S. L'utilisateur peut oublier d'arrêter. On notifie à 50 min, puis on arrête automatiquement à 60 min (configurable). """ warn_before = 600 # Prévenir 10 min avant la fin warned = False while self.running and self.session_id: elapsed = time.time() - self._session_start_time remaining = MAX_SESSION_DURATION_S - elapsed # Notification 10 min avant la fin if not warned and remaining <= warn_before: warned = True mins = int(remaining / 60) logger.info(f"Auto-stop dans {mins} min") try: from .ui.notifications import NotificationManager NotificationManager().notify( "Léa", f"L'enregistrement s'arrêtera automatiquement dans {mins} minutes.", ) except Exception: pass # Auto-stop if remaining <= 0: logger.info( f"Auto-stop : session {self.session_id} après " f"{int(elapsed)}s ({int(elapsed/60)} min)" ) try: from .ui.notifications import NotificationManager NotificationManager().notify( "Léa", f"Enregistrement terminé automatiquement après " f"{int(elapsed/60)} minutes. Merci !", ) except Exception: pass # Arrêter via l'état partagé (synchronise systray + chat) if self._state is not None: self._state.stop_recording() else: self.stop_session() break time.sleep(30) # Vérifier toutes les 30s def start_session(self, workflow_name): self.session_id = f"sess_{time.strftime('%Y%m%dT%H%M%S')}_{uuid.uuid4().hex[:6]}" self.session_dir = self.storage.get_session_dir(self.session_id) self.vision = VisionCapturer(str(self.session_dir)) self.streamer = TraceStreamer(self.session_id, machine_id=self.machine_id) self.captor = EventCaptorV1(self._on_event_bridge) # Initialiser l'executeur partage self._executor = ActionExecutorV1() self.shot_counter = 0 self.running = True self._replay_active = False self.streamer.start() self.captor.start() # Heartbeat Contextuel (Toutes les 5s par defaut) threading.Thread(target=self._heartbeat_loop, daemon=True).start() # Auto-stop : arrêter l'enregistrement après MAX_SESSION_DURATION_S # L'utilisateur peut oublier d'arrêter — on le fait automatiquement self._session_start_time = time.time() threading.Thread(target=self._auto_stop_loop, daemon=True).start() # Watchdog de Commandes (GHOST Replay — legacy fichier) threading.Thread(target=self._command_watchdog_loop, daemon=True).start() # Note: la boucle de polling replay est déjà lancée dans __init__ (ligne 102) # Ne PAS en relancer une ici — deux threads poll simultanés causent # une race condition où les actions sont consommées mais pas exécutées. logger.info(f"Session {self.session_id} ({workflow_name}) sur machine {self.machine_id} en cours...") def _command_watchdog_loop(self): """Surveille un fichier de commande pour executer des ordres visuels (legacy).""" import json import platform from .config import BASE_DIR # Chemin du fichier de commande selon l'OS if platform.system() == "Windows": cmd_path = "C:\\rpa_vision\\command.json" else: cmd_path = str(BASE_DIR / "command.json") while self.running and self.session_id: # Ne pas traiter les commandes fichier pendant un replay serveur if self._replay_active: time.sleep(1) continue if os.path.exists(cmd_path): try: with open(cmd_path, "r") as f: order = json.load(f) os.remove(cmd_path) # On consomme l'ordre if self._executor: self._executor.execute_normalized_order(order) except Exception as e: logger.error(f"Erreur Watchdog: {e}") time.sleep(1) def _replay_poll_loop(self): """ Boucle de polling pour les actions de replay depuis le serveur (P0-5). Tourne en parallele du heartbeat et du watchdog. Poll GET /replay/next toutes les REPLAY_POLL_INTERVAL secondes. Quand une action est recue, l'execute via l'executor et rapporte le resultat. """ msg = ( f"[REPLAY] Boucle replay demarree — poll toutes les " f"{REPLAY_POLL_INTERVAL}s sur {SERVER_URL}" ) print(msg) logger.info(msg) poll_count = 0 while self.running: if not self._executor: time.sleep(REPLAY_POLL_INTERVAL) continue # TOUJOURS utiliser un session_id stable pour le replay. # L'enregistrement et le replay sont indépendants : le serveur # envoie les actions sur agent_{user_id}, pas sur la session # d'enregistrement (sess_xxx). poll_session = f"agent_{self.user_id}" # Log periodique pour confirmer que la boucle tourne (toutes les 60s) poll_count += 1 if poll_count % int(60 / REPLAY_POLL_INTERVAL) == 0: print( f"[REPLAY] Poll #{poll_count} — session={poll_session} " f"— serveur={SERVER_URL}" ) try: # Tenter de recuperer et executer une action had_action = self._executor.poll_and_execute( session_id=poll_session, server_url=SERVER_URL, machine_id=self.machine_id, ) if had_action: if not self._replay_active: self._replay_active = True self.ui.set_replay_active(True) self._state.set_replay_active(True) # Si une action a ete executee, poll plus rapidement # pour enchainer les actions du workflow time.sleep(0.2) else: # Pas d'action en attente — utiliser le backoff de l'executor # (augmente si le serveur est indisponible, reset a 1s sinon) if self._replay_active: print("[REPLAY] Replay termine — retour en mode capture") logger.info("Replay termine — retour en mode capture") self._replay_active = False self.ui.set_replay_active(False) self._state.set_replay_active(False) poll_delay = getattr(self._executor, '_poll_backoff', REPLAY_POLL_INTERVAL) time.sleep(max(poll_delay, REPLAY_POLL_INTERVAL)) except Exception as e: print(f"[REPLAY] ERREUR boucle replay : {e}") logger.error(f"Erreur replay poll loop : {e}") self._replay_active = False self._state.set_replay_active(False) poll_delay = getattr(self._executor, '_poll_backoff', REPLAY_POLL_INTERVAL) time.sleep(max(poll_delay, REPLAY_POLL_INTERVAL)) _last_bg_hash: str = "" def _background_heartbeat_loop(self): """Heartbeat permanent — envoie un screenshot toutes les 5s au serveur. Tourne même sans session active, pour que le VWB puisse capturer Windows. """ import requests as req bg_session = f"bg_{self.machine_id}" logger.info(f"[HEARTBEAT] Boucle permanente démarrée (session={bg_session})") while self.running: try: # Ne pas envoyer pendant un enregistrement (le heartbeat session s'en charge) if self.session_id: time.sleep(5) continue full_path = self._bg_vision.capture_full_context("heartbeat") if not full_path: time.sleep(5) continue # Dédup : skip si écran identique img_hash = self._quick_hash(full_path) if img_hash and img_hash == self._last_bg_hash: time.sleep(5) continue self._last_bg_hash = img_hash # Envoyer au streaming server (via STREAMING_ENDPOINT unifié) headers = {"Authorization": f"Bearer {API_TOKEN}"} if API_TOKEN else {} with open(full_path, 'rb') as f: req.post( f"{STREAMING_ENDPOINT}/image", params={ "session_id": bg_session, "shot_id": f"heartbeat_{int(time.time())}", "machine_id": self.machine_id, }, headers=headers, files={"file": ("screenshot.png", f, "image/png")}, timeout=10, allow_redirects=False, ) except Exception as e: logger.debug(f"[HEARTBEAT] Erreur: {e}") time.sleep(5) def stop_session(self): # Sauvegarder le session_id avant de l'annuler (pour les logs) ended_session_id = self.session_id # Arrêter la capture d'abord (plus d'events entrants) if self.captor: self.captor.stop() # Attendre que les events en cours de traitement dans _on_event_bridge # aient le temps d'être envoyés au streamer (capture duale + push) import time time.sleep(1.5) # Maintenant arrêter le streamer (drain queue + finalize) if self.streamer: self.streamer.stop() logger.info(f"Session {ended_session_id} terminée.") # Reset le session_id APRÈS le stop complet du streamer self.session_id = None # Reset le backoff de l'executor pour reprendre le polling immédiatement if self._executor: self._executor._poll_backoff = self._executor._poll_backoff_min self._executor._server_available = True if hasattr(self._executor, '_last_conn_error_logged'): self._executor._last_conn_error_logged = False # NE PAS mettre self.running = False ici ! # self.running contrôle la boucle _replay_poll_loop (permanente). # Seule la sortie du programme doit le mettre à False. # Les boucles _heartbeat_loop et _command_watchdog_loop vérifieront # self.session_id pour savoir si elles doivent fonctionner. logger.info( f"Session arrêtée — replay poll actif avec session=" f"agent_{self.user_id}" ) _last_heartbeat_hash: str = "" def _heartbeat_loop(self): """Capture périodique pour donner du contexte au stagiaire. Déduplication : n'envoie que si l'écran a changé. Tourne tant que session_id est défini (= enregistrement actif). Enrichi avec le titre de la fenêtre active pour contextualisation. """ while self.running and self.session_id: try: full_path = self.vision.capture_full_context("heartbeat") if full_path: # Hash rapide pour détecter les changements d'écran img_hash = self._quick_hash(full_path) if img_hash != self._last_heartbeat_hash: self._last_heartbeat_hash = img_hash self.streamer.push_image(full_path, f"heartbeat_{int(time.time())}") heartbeat_event = { "type": "heartbeat", "image": full_path, "timestamp": time.time(), "machine_id": self.machine_id, } # Ajouter le titre de la fenêtre active (léger, pas de crop) window_title = self.vision.get_active_window_title() if window_title: heartbeat_event["active_window_title"] = window_title self.streamer.push_event(heartbeat_event) except Exception as e: logger.error(f"Heartbeat error: {e}") time.sleep(5) @staticmethod def _quick_hash(image_path: str) -> str: """Hash perceptuel rapide (16x16 niveaux de gris).""" try: from PIL import Image import hashlib img = Image.open(image_path).resize((16, 16)).convert('L') return hashlib.md5(img.tobytes()).hexdigest() except Exception: return "" def _on_event_bridge(self, event): """Pont intelligent avec capture duale et post-action monitoring.""" if not self.session_id: return # Injecter l'identifiant machine dans chaque événement (multi-machine) event["machine_id"] = self.machine_id # Injecter le contexte fenêtre dans chaque événement (nécessaire # pour que le serveur maintienne last_window_info) if self.captor and self.captor.last_window: event["window"] = self.captor.last_window # Capture Proactive sur changement de fenêtre if event["type"] == "window_focus_change": full_path = self.vision.capture_full_context("focus_change") event["screenshot_context"] = full_path self.streamer.push_image(full_path, f"focus_{int(time.time())}") # Capture Interactive (Dual + Fenêtre active) if event["type"] in ["mouse_click", "key_combo"]: self.shot_counter += 1 shot_id = f"shot_{self.shot_counter:04d}" pos = event.get("pos", (0, 0)) capture_info = self.vision.capture_dual(pos[0], pos[1], shot_id) event["screenshot_id"] = shot_id event["vision_info"] = capture_info # Enrichir l'event avec les métadonnées de la fenêtre active # (titre, rect, coordonnées clic relatives, taille fenêtre) window_capture = capture_info.get("window_capture") if window_capture: event["window_capture"] = { "title": window_capture.get("window_title", ""), "app_name": window_capture.get("app_name", ""), "rect": window_capture.get("window_rect"), "click_relative": window_capture.get("click_in_window"), "window_size": window_capture.get("window_size"), "click_inside_window": window_capture.get("click_inside_window", True), } self._stream_capture_info(capture_info, shot_id) # POST-ACTION : Capture du résultat après 1s (pour voir le résultat du clic) threading.Timer(1.0, self._capture_result, args=(shot_id,)).start() self.ui.update_stats(self.shot_counter) self._state.update_actions_count(self.shot_counter) print(f"📸 Action capturée : {event['type']}") self.streamer.push_event(event) def _capture_result(self, base_shot_id: str): """Capture l'état de l'écran 1s après l'action pour voir l'effet.""" if not self.running: return res_path = self.vision.capture_full_context(f"result_of_{base_shot_id}") self.streamer.push_image(res_path, f"res_{base_shot_id}") self.streamer.push_event({"type": "action_result", "base_shot_id": base_shot_id, "image": res_path}) def _stream_capture_info(self, capture_info, shot_id): if "full" in capture_info: self.streamer.push_image(capture_info["full"], f"{shot_id}_full") if "crop" in capture_info: self.streamer.push_image(capture_info["crop"], f"{shot_id}_crop") # Streamer l'image de la fenêtre active si disponible window_capture = capture_info.get("window_capture") if window_capture and "window_image" in window_capture: self.streamer.push_image( window_capture["window_image"], f"{shot_id}_window" ) def run(self): self.ui.run() def main(): agent = AgentV1() agent.run() if __name__ == "__main__": main()