Geniusia_v2/test_task_replay.py

#!/usr/bin/env python3
"""
Test du système de rejeu de tâches.
"""

import asyncio
import sys
from pathlib import Path

# Ajouter le chemin du projet
sys.path.insert(0, str(Path(__file__).parent / "geniusia2"))

from core.task_replay import TaskReplayEngine
from core.learning_manager import LearningManager
from core.embeddings_manager import EmbeddingsManager
from core.utils.vision_utils import VisionUtils
from core.utils.input_utils import InputUtils
from core.logger import Logger
from core.config import get_config


async def test_list_tasks():
    """Test de listage des tâches disponibles."""
    print("=== Test: Liste des tâches disponibles ===\n")
    
    # Charger la configuration
    config = get_config()
    
    # Initialiser les composants
    logger = Logger()
    embeddings_manager = EmbeddingsManager(logger=logger)
    learning_manager = LearningManager(
        embeddings_manager,
        logger,
        config,
        profiles_path="geniusia2/data/user_profiles"
    )
    vision_utils = VisionUtils(config)
    input_utils = InputUtils(logger, config)
    
    # Créer le moteur de rejeu
    replay_engine = TaskReplayEngine(
        learning_manager,
        embeddings_manager,
        vision_utils,
        input_utils,
        logger,
        config
    )
    
    # Lister les tâches
    tasks = replay_engine.list_available_tasks()
    
    print(f"Tâches disponibles: {len(tasks)}\n")
    
    for task in tasks:
        print(f"📋 {task['task_name']}")
        print(f"   ID: {task['task_id']}")
        print(f"   Observations: {task['observation_count']}")
        print(f"   Confiance: {task['confidence']:.2%}")
        print()
    
    return tasks


async def test_replay_task(task_id: str):
    """Test de rejeu d'une tâche."""
    print(f"\n=== Test: Rejeu de la tâche {task_id} ===\n")
    
    # Charger la configuration
    config = get_config()
    
    # Initialiser les composants
    logger = Logger()
    embeddings_manager = EmbeddingsManager(logger=logger)
    learning_manager = LearningManager(
        embeddings_manager,
        logger,
        config,
        profiles_path="geniusia2/data/user_profiles"
    )
    vision_utils = VisionUtils(config)
    input_utils = InputUtils(logger, config)
    
    # Créer le moteur de rejeu
    replay_engine = TaskReplayEngine(
        learning_manager,
        embeddings_manager,
        vision_utils,
        input_utils,
        logger,
        config
    )
    
    print("⏳ Démarrage du rejeu dans 3 secondes...")
    print("   (Préparez l'interface si nécessaire)")
    await asyncio.sleep(3)
    
    # Rejouer la tâche
    results = await replay_engine.replay_task(task_id, interactive=False)
    
    print("\n📊 Résultats du rejeu:")
    print(f"   Succès: {'✅' if results['success'] else '❌'}")
    print(f"   Actions exécutées: {results['executed_actions']}/{results['total_actions']}")
    print(f"   Actions échouées: {results['failed_actions']}")
    
    print("\n📝 Détails des actions:")
    for action in results['actions']:
        status = "✅" if action['success'] else "❌"
        print(f"   {status} Étape {action['step']}: {action.get('action_type', 'unknown')}")
        if 'location' in action:
            loc = action['location']
            print(f"      Position: ({loc['x']}, {loc['y']}) - Confiance: {loc['confidence']:.2%}")
        if 'error' in action:
            print(f"      Erreur: {action['error']}")
    
    return results


async def test_replay_with_monitoring(task_id: str):
    """Test de rejeu avec monitoring en temps réel."""
    print(f"\n=== Test: Rejeu avec monitoring - {task_id} ===\n")
    
    # Charger la configuration
    config = get_config()
    
    # Initialiser les composants
    logger = Logger()
    embeddings_manager = EmbeddingsManager(logger=logger)
    learning_manager = LearningManager(
        embeddings_manager,
        logger,
        config,
        profiles_path="geniusia2/data/user_profiles"
    )
    vision_utils = VisionUtils(config)
    input_utils = InputUtils(logger, config)
    
    # Créer le moteur de rejeu
    replay_engine = TaskReplayEngine(
        learning_manager,
        embeddings_manager,
        vision_utils,
        input_utils,
        logger,
        config
    )
    
    # Callback pour monitoring
    def on_step(step_result):
        status_icon = {
            "success": "✅",
            "failed": "❌",
            "not_found": "🔍",
            "pending": "⏳"
        }.get(step_result["status"], "❓")
        
        print(f"{status_icon} Étape {step_result['step']}: {step_result['description']} - {step_result['status']}")
    
    print("⏳ Démarrage du rejeu dans 3 secondes...")
    await asyncio.sleep(3)
    
    # Rejouer avec monitoring
    results = await replay_engine.replay_task_with_monitoring(
        task_id,
        on_step_completed=on_step
    )
    
    print(f"\n📊 Résultat final: {'✅ Succès' if results['success'] else '❌ Échec'}")
    
    return results


async def main():
    """Fonction principale."""
    print("🎮 Test du système de rejeu de tâches\n")
    
    # 1. Lister les tâches disponibles
    tasks = await test_list_tasks()
    
    if not tasks:
        print("❌ Aucune tâche disponible pour le rejeu.")
        print("   Utilisez d'abord le système de capture pour apprendre des tâches.")
        return
    
    # 2. Demander quelle tâche rejouer
    print("\n" + "="*60)
    print("Quelle tâche voulez-vous rejouer ?")
    for i, task in enumerate(tasks, 1):
        print(f"{i}. {task['task_name']} ({task['task_id']})")
    
    try:
        choice = input("\nNuméro de la tâche (ou 'q' pour quitter): ").strip()
        
        if choice.lower() == 'q':
            print("Au revoir!")
            return
        
        task_index = int(choice) - 1
        if task_index < 0 or task_index >= len(tasks):
            print("❌ Choix invalide")
            return
        
        selected_task = tasks[task_index]
        task_id = selected_task['task_id']
        
        # 3. Rejouer la tâche avec monitoring
        await test_replay_with_monitoring(task_id)
    
    except KeyboardInterrupt:
        print("\n\n⚠️  Interruption par l'utilisateur")
    except Exception as e:
        print(f"\n❌ Erreur: {e}")
        import traceback
        traceback.print_exc()


if __name__ == "__main__":
    asyncio.run(main())