#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Démonstration Complète du Système de Capture d'Écran Réelle - RPA Vision V3 Auteur : Dom, Alice, Kiro - 8 janvier 2026 Script de démonstration pour tester toutes les fonctionnalités du système de capture d'écran réelle. """ import time import json import requests import sys import os from typing import Dict, List, Any # Ajouter le chemin du projet sys.path.append(os.path.dirname(__file__)) from visual_workflow_builder.backend.services.real_screen_capture import RealScreenCaptureService class RealScreenCaptureDemo: """Démonstration du système de capture d'écran réelle""" def __init__(self): self.service = RealScreenCaptureService() self.api_base_url = "http://localhost:5002/api/real-demo" def print_header(self, title: str): """Afficher un en-tête formaté""" print(f"\n{'='*60}") print(f" {title}") print(f"{'='*60}") def print_step(self, step: str): """Afficher une étape""" print(f"\n🔹 {step}") def print_success(self, message: str): """Afficher un message de succès""" print(f"✅ {message}") def print_error(self, message: str): """Afficher un message d'erreur""" print(f"❌ {message}") def print_info(self, message: str): """Afficher une information""" print(f"ℹ️ {message}") def demo_service_direct(self): """Démonstration du service direct (sans API)""" self.print_header("DÉMONSTRATION SERVICE DIRECT") try: self.print_step("Initialisation du service") monitors = self.service.get_monitors() self.print_success(f"Service initialisé - {len(monitors)} moniteurs détectés") for monitor in monitors: print(f" 📺 Moniteur {monitor['id']}: {monitor['width']}x{monitor['height']}") self.print_step("Sélection du moniteur principal") if len(monitors) > 0: success = self.service.select_monitor(0) if success: self.print_success("Moniteur 0 sélectionné") else: self.print_error("Échec de la sélection du moniteur") self.print_step("Démarrage de la capture (intervalle: 1s)") success = self.service.start_capture(interval=1.0) if success: self.print_success("Capture démarrée") else: self.print_error("Échec du démarrage de la capture") return self.print_step("Capture et détection pendant 10 secondes") for i in range(10): time.sleep(1) status = self.service.get_status() elements = self.service.get_detected_elements() screenshot = self.service.get_current_screenshot_base64() print(f" Seconde {i+1:2d}: " f"{len(elements):2d} éléments détectés, " f"Screenshot: {'✓' if screenshot else '✗'}") # Afficher quelques éléments détectés if elements and i % 3 == 0: # Tous les 3 secondes print(" 📋 Éléments récents:") for elem in elements[:3]: # Afficher les 3 premiers bbox = elem.get('bbox', {}) print(f" - {elem.get('type', 'unknown')}: " f"({bbox.get('x', 0)}, {bbox.get('y', 0)}) " f"conf={elem.get('confidence', 0):.2f}") self.print_step("Arrêt de la capture") success = self.service.stop_capture() if success: self.print_success("Capture arrêtée") else: self.print_error("Échec de l'arrêt de la capture") # Statistiques finales final_status = self.service.get_status() self.print_info(f"Statistiques finales:") print(f" - Éléments détectés: {final_status['elements_detected']}") print(f" - Moniteurs disponibles: {final_status['monitors_count']}") print(f" - Capture active: {final_status['is_capturing']}") except Exception as e: self.print_error(f"Erreur lors de la démonstration: {e}") finally: self.service.cleanup() def demo_api_endpoints(self): """Démonstration des endpoints API""" self.print_header("DÉMONSTRATION API REST") try: self.print_step("Test de connectivité API") response = requests.get(f"{self.api_base_url}/capture/status", timeout=5) if response.status_code == 200: self.print_success("API accessible") else: self.print_error(f"API non accessible (status: {response.status_code})") return except requests.exceptions.RequestException as e: self.print_error(f"Impossible de se connecter à l'API: {e}") self.print_info("Assurez-vous que le serveur backend est démarré sur le port 5002") return try: # Test des moniteurs self.print_step("Récupération des moniteurs via API") response = requests.get(f"{self.api_base_url}/monitors") if response.status_code == 200: data = response.json() monitors = data['monitors'] self.print_success(f"{len(monitors)} moniteurs récupérés") for monitor in monitors: print(f" 📺 Moniteur {monitor['id']}: {monitor['width']}x{monitor['height']}") # Sélection de moniteur if len(monitors) > 0: self.print_step("Sélection du moniteur via API") response = requests.post(f"{self.api_base_url}/monitors/0/select") if response.status_code == 200: self.print_success("Moniteur sélectionné via API") # Démarrage de la capture self.print_step("Démarrage de la capture via API") response = requests.post( f"{self.api_base_url}/capture/start", json={'interval': 1.5} ) if response.status_code == 200: self.print_success("Capture démarrée via API") else: self.print_error(f"Échec du démarrage: {response.text}") return # Surveillance pendant quelques secondes self.print_step("Surveillance de la capture via API") for i in range(6): time.sleep(1) # Statut response = requests.get(f"{self.api_base_url}/capture/status") if response.status_code == 200: status = response.json()['status'] # Éléments response = requests.get(f"{self.api_base_url}/elements") elements_count = 0 if response.status_code == 200: elements_count = response.json()['count'] print(f" Seconde {i+1}: " f"Capture: {'✓' if status['is_capturing'] else '✗'}, " f"Éléments: {elements_count}") # Test de screenshot self.print_step("Récupération du screenshot via API") response = requests.get(f"{self.api_base_url}/capture/screenshot") if response.status_code == 200: screenshot_data = response.json() self.print_success("Screenshot récupéré") self.print_info(f"Éléments dans le screenshot: {len(screenshot_data['elements'])}") # Afficher quelques éléments for i, elem in enumerate(screenshot_data['elements'][:3]): bbox = elem['bbox'] print(f" {i+1}. {elem['type']}: " f"pos=({bbox['x']}, {bbox['y']}) " f"size={bbox['width']}x{bbox['height']} " f"conf={elem['confidence']:.2f}") # Arrêt de la capture self.print_step("Arrêt de la capture via API") response = requests.post(f"{self.api_base_url}/capture/stop") if response.status_code == 200: self.print_success("Capture arrêtée via API") except Exception as e: self.print_error(f"Erreur lors des tests API: {e}") def demo_interaction_simulation(self): """Démonstration des interactions simulées""" self.print_header("DÉMONSTRATION INTERACTIONS SIMULÉES") try: # Test de clic par coordonnées self.print_step("Test de clic par coordonnées") response = requests.post( f"{self.api_base_url}/interact/click", json={'x': 100, 'y': 100} ) if response.status_code == 200: result = response.json() self.print_success(f"Clic simulé: {result['message']}") else: self.print_info("Clic non effectué (pyautogui non disponible ou erreur)") # Test de saisie self.print_step("Test de saisie de texte") response = requests.post( f"{self.api_base_url}/interact/type", json={'text': 'Test RPA Vision V3 - Capture Réelle'} ) if response.status_code == 200: result = response.json() self.print_success(f"Saisie simulée: {result['message']}") else: self.print_info("Saisie non effectuée (pyautogui non disponible ou erreur)") # Test d'arrêt d'urgence self.print_step("Test d'arrêt d'urgence") response = requests.post(f"{self.api_base_url}/safety/emergency-stop") if response.status_code == 200: self.print_success("Arrêt d'urgence testé") except Exception as e: self.print_error(f"Erreur lors des tests d'interaction: {e}") def demo_workflow_execution(self): """Démonstration d'exécution de workflow simple""" self.print_header("DÉMONSTRATION WORKFLOW SIMPLE") try: # Workflow simple : clic + saisie + attente workflow_actions = [ {'type': 'click', 'x': 200, 'y': 200}, {'type': 'wait', 'duration': 0.5}, {'type': 'type', 'text': 'Bonjour RPA Vision V3'}, {'type': 'wait', 'duration': 0.5}, {'type': 'click', 'x': 300, 'y': 300} ] self.print_step("Exécution d'un workflow simple") self.print_info("Actions du workflow:") for i, action in enumerate(workflow_actions): print(f" {i+1}. {action['type']}: {action}") response = requests.post( f"{self.api_base_url}/workflow/execute", json={'actions': workflow_actions} ) if response.status_code == 200: result = response.json() self.print_success("Workflow exécuté") summary = result['summary'] print(f" 📊 Résumé: {summary['successful_actions']}/{summary['total_actions']} " f"actions réussies ({summary['success_rate']:.1%})") # Détails des résultats for res in result['results']: status = "✅" if res['success'] else "❌" print(f" {status} Action {res['action_index']+1} ({res['type']}): " f"{res.get('message', res.get('error', 'N/A'))}") else: self.print_error(f"Échec du workflow: {response.text}") except Exception as e: self.print_error(f"Erreur lors de l'exécution du workflow: {e}") def run_complete_demo(self): """Exécuter la démonstration complète""" self.print_header("DÉMONSTRATION COMPLÈTE - SYSTÈME DE CAPTURE RÉELLE") print("RPA Vision V3 - Système de Capture d'Écran et Interaction Réelle") print("Auteur : Dom, Alice, Kiro - 8 janvier 2026") try: # 1. Service direct self.demo_service_direct() # Pause entre les démos self.print_info("Pause de 2 secondes entre les démonstrations...") time.sleep(2) # 2. API REST self.demo_api_endpoints() # Pause time.sleep(1) # 3. Interactions self.demo_interaction_simulation() # Pause time.sleep(1) # 4. Workflow self.demo_workflow_execution() # Résumé final self.print_header("DÉMONSTRATION TERMINÉE") self.print_success("Toutes les démonstrations ont été exécutées") self.print_info("Fonctionnalités testées:") print(" ✅ Service de capture d'écran réelle") print(" ✅ Détection d'éléments UI en temps réel") print(" ✅ API REST complète") print(" ✅ Interactions simulées (clic, saisie)") print(" ✅ Exécution de workflows simples") print(" ✅ Contrôles de sécurité") self.print_info("Le système de capture d'écran réelle est opérationnel ! 🚀") except KeyboardInterrupt: self.print_info("Démonstration interrompue par l'utilisateur") except Exception as e: self.print_error(f"Erreur générale: {e}") finally: # Nettoyage final try: self.service.cleanup() requests.post(f"{self.api_base_url}/safety/emergency-stop", timeout=2) except: pass def main(): """Fonction principale""" print("🎯 Démarrage de la démonstration du système de capture d'écran réelle") # Vérifications préliminaires try: import mss print("✅ MSS disponible pour la capture d'écran") except ImportError: print("❌ MSS non disponible - capture d'écran limitée") try: import pyautogui print("✅ PyAutoGUI disponible pour les interactions") except ImportError: print("⚠️ PyAutoGUI non disponible - interactions simulées seulement") # Lancer la démonstration demo = RealScreenCaptureDemo() demo.run_complete_demo() if __name__ == "__main__": main()