327 lines
10 KiB
Python
327 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test d'intégration du Mode Assisté avec suggestions de workflows.
|
|
Vérifie que l'Orchestrator détecte les workflows et crée des suggestions.
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent / "geniusia2"))
|
|
|
|
from core.logger import Logger
|
|
from core.config import get_config
|
|
from core.orchestrator import Orchestrator
|
|
from core.learning_manager import LearningManager
|
|
from core.embeddings_manager import EmbeddingsManager
|
|
from core.utils.vision_utils import VisionUtils
|
|
from core.llm_manager import LLMManager
|
|
|
|
|
|
def test_workflow_suggestion_integration():
|
|
"""Test que l'Orchestrator crée des suggestions de workflows"""
|
|
print("\n" + "="*60)
|
|
print("🧪 TEST: Intégration Mode Assisté + Workflows")
|
|
print("="*60)
|
|
|
|
logger = Logger()
|
|
config = get_config()
|
|
|
|
# Créer les composants
|
|
embeddings = EmbeddingsManager()
|
|
learning = LearningManager(embeddings, logger, config)
|
|
vision = VisionUtils()
|
|
llm = LLMManager(logger=logger, fallback_to_vision=True)
|
|
|
|
# Créer l'Orchestrator
|
|
orchestrator = Orchestrator(
|
|
learning_manager=learning,
|
|
vision_utils=vision,
|
|
llm_manager=llm,
|
|
logger=logger,
|
|
gui=None,
|
|
config=config
|
|
)
|
|
|
|
# Mettre en mode Assist
|
|
learning.set_mode("assist")
|
|
|
|
print("\n📋 Phase 1: Apprentissage d'un workflow...")
|
|
|
|
# Simuler 3 répétitions d'un workflow
|
|
for i in range(3):
|
|
base_time = datetime.now() - timedelta(hours=3-i)
|
|
|
|
actions = [
|
|
{
|
|
"action_type": "click",
|
|
"position": [100, 100],
|
|
"window": "Calculatrice",
|
|
"timestamp": base_time.isoformat(),
|
|
"description": "Ouvrir calculatrice"
|
|
},
|
|
{
|
|
"action_type": "type",
|
|
"text": "2+2",
|
|
"window": "Calculatrice",
|
|
"timestamp": (base_time + timedelta(seconds=1)).isoformat(),
|
|
"description": "Taper calcul"
|
|
},
|
|
{
|
|
"action_type": "click",
|
|
"position": [150, 150],
|
|
"window": "Calculatrice",
|
|
"timestamp": (base_time + timedelta(seconds=2)).isoformat(),
|
|
"description": "Cliquer égal"
|
|
}
|
|
]
|
|
|
|
for action in actions:
|
|
orchestrator.session_manager.add_action(action)
|
|
|
|
orchestrator.session_manager.force_finalize_session()
|
|
print(f" ✅ Répétition {i+1} enregistrée")
|
|
|
|
# Analyser les sessions pour détecter le workflow
|
|
sessions = orchestrator.session_manager.get_recent_sessions(10)
|
|
orchestrator.workflow_detector.analyze_sessions(sessions)
|
|
|
|
workflows = orchestrator.workflow_detector.get_workflows()
|
|
|
|
print(f"\n📊 Workflows détectés: {len(workflows)}")
|
|
|
|
if not workflows:
|
|
print(" ❌ Aucun workflow détecté")
|
|
return False
|
|
|
|
workflow = workflows[0]
|
|
print(f" ✅ Workflow: {workflow.name}")
|
|
print(f" Confiance: {workflow.confidence:.2%}")
|
|
print(f" Étapes: {len(workflow.steps)}")
|
|
|
|
print("\n📋 Phase 2: Simulation du début du workflow...")
|
|
|
|
# Simuler les 2 premières actions du workflow
|
|
base_time = datetime.now()
|
|
|
|
action1 = {
|
|
"action_type": "click",
|
|
"position": [100, 100],
|
|
"window": "Calculatrice",
|
|
"timestamp": base_time.isoformat(),
|
|
"description": "Ouvrir calculatrice"
|
|
}
|
|
orchestrator.session_manager.add_action(action1)
|
|
print(f" ✅ Action 1: {action1['description']}")
|
|
|
|
action2 = {
|
|
"action_type": "type",
|
|
"text": "2+2",
|
|
"window": "Calculatrice",
|
|
"timestamp": (base_time + timedelta(seconds=1)).isoformat(),
|
|
"description": "Taper calcul"
|
|
}
|
|
orchestrator.session_manager.add_action(action2)
|
|
print(f" ✅ Action 2: {action2['description']}")
|
|
|
|
print("\n🔍 Phase 3: Vérification de suggestion...")
|
|
|
|
# Vérifier s'il y a un match de workflow
|
|
workflow_match = orchestrator._check_workflow_match()
|
|
|
|
if workflow_match:
|
|
print(f"\n 🎉 Match trouvé !")
|
|
print(f" Workflow: {workflow_match.workflow_name}")
|
|
print(f" Confiance: {workflow_match.confidence:.2%}")
|
|
print(f" Étapes matchées: {workflow_match.matched_steps}/{workflow_match.total_steps}")
|
|
print(f" Étapes restantes: {len(workflow_match.remaining_steps)}")
|
|
|
|
# Créer une suggestion
|
|
suggestion = orchestrator.suggestion_manager.create_workflow_suggestion(workflow_match)
|
|
|
|
if suggestion:
|
|
print(f"\n 💡 Suggestion créée !")
|
|
print(f" Type: {suggestion['type']}")
|
|
print(f" Workflow: {suggestion['workflow_name']}")
|
|
print(f" Prochaine étape: {suggestion['next_steps_preview'][0]['action_type']}")
|
|
print(f" Confiance: {suggestion['confidence']:.2%}")
|
|
|
|
return True
|
|
else:
|
|
print(f"\n ⚠️ Match trouvé mais suggestion non créée (confiance trop faible ?)")
|
|
return False
|
|
else:
|
|
print(f"\n ❌ Aucun match de workflow trouvé")
|
|
return False
|
|
|
|
|
|
def test_orchestrator_check_for_suggestions():
|
|
"""Test que check_for_suggestions() détecte les workflows"""
|
|
print("\n" + "="*60)
|
|
print("🧪 TEST: check_for_suggestions() avec workflows")
|
|
print("="*60)
|
|
|
|
logger = Logger()
|
|
config = get_config()
|
|
|
|
# Créer les composants
|
|
embeddings = EmbeddingsManager()
|
|
learning = LearningManager(embeddings, logger, config)
|
|
vision = VisionUtils()
|
|
llm = LLMManager(logger=logger, fallback_to_vision=True)
|
|
|
|
# Créer l'Orchestrator
|
|
orchestrator = Orchestrator(
|
|
learning_manager=learning,
|
|
vision_utils=vision,
|
|
llm_manager=llm,
|
|
logger=logger,
|
|
gui=None,
|
|
config=config
|
|
)
|
|
|
|
# Mettre en mode Assist
|
|
learning.set_mode("assist")
|
|
|
|
print("\n📋 Préparation: Créer un workflow...")
|
|
|
|
# Créer un workflow manuellement
|
|
from core.workflow_detector import Workflow, WorkflowStep
|
|
|
|
workflow = Workflow(
|
|
workflow_id="test_workflow_001",
|
|
name="Test Workflow",
|
|
steps=[
|
|
WorkflowStep(
|
|
step_id=0,
|
|
action_type="click",
|
|
target_description="Action 1",
|
|
position=(100, 100),
|
|
window="TestApp"
|
|
),
|
|
WorkflowStep(
|
|
step_id=1,
|
|
action_type="type",
|
|
target_description="Action 2",
|
|
position=(0, 0),
|
|
window="TestApp"
|
|
),
|
|
WorkflowStep(
|
|
step_id=2,
|
|
action_type="click",
|
|
target_description="Action 3",
|
|
position=(200, 200),
|
|
window="TestApp"
|
|
)
|
|
],
|
|
repetitions=5,
|
|
confidence=0.95,
|
|
last_seen=datetime.now()
|
|
)
|
|
|
|
orchestrator.workflow_detector.workflows.append(workflow)
|
|
print(f" ✅ Workflow créé: {workflow.name}")
|
|
|
|
print("\n📋 Simulation: Début du workflow...")
|
|
|
|
# Simuler les 2 premières actions
|
|
base_time = datetime.now()
|
|
|
|
actions = [
|
|
{
|
|
"action_type": "click",
|
|
"position": [100, 100],
|
|
"window": "TestApp",
|
|
"timestamp": base_time.isoformat()
|
|
},
|
|
{
|
|
"action_type": "type",
|
|
"position": [0, 0],
|
|
"window": "TestApp",
|
|
"timestamp": (base_time + timedelta(seconds=1)).isoformat()
|
|
}
|
|
]
|
|
|
|
for action in actions:
|
|
orchestrator.session_manager.add_action(action)
|
|
|
|
print(f" ✅ {len(actions)} actions simulées")
|
|
|
|
print("\n🔍 Appel de check_for_suggestions()...")
|
|
|
|
# Appeler check_for_suggestions
|
|
orchestrator.check_for_suggestions()
|
|
|
|
# Vérifier s'il y a une suggestion
|
|
suggestion = orchestrator.suggestion_manager.get_current_suggestion()
|
|
|
|
if suggestion:
|
|
print(f"\n 🎉 Suggestion créée automatiquement !")
|
|
print(f" Type: {suggestion['type']}")
|
|
print(f" Workflow: {suggestion.get('workflow_name', 'N/A')}")
|
|
print(f" Confiance: {suggestion['confidence']:.2%}")
|
|
|
|
if suggestion['type'] == 'workflow':
|
|
print(f" Étape courante: {suggestion['current_step']}/{suggestion['total_steps']}")
|
|
print(f" Étapes restantes: {len(suggestion['remaining_steps'])}")
|
|
|
|
return True
|
|
else:
|
|
print(f"\n ❌ Aucune suggestion créée")
|
|
return False
|
|
|
|
|
|
def main():
|
|
"""Exécute tous les tests"""
|
|
print("\n" + "="*60)
|
|
print("🚀 TEST D'INTÉGRATION MODE ASSISTÉ")
|
|
print("="*60)
|
|
|
|
results = []
|
|
|
|
try:
|
|
# Test 1: Intégration complète
|
|
results.append(("Intégration workflow → suggestion", test_workflow_suggestion_integration()))
|
|
except Exception as e:
|
|
print(f"\n❌ ERREUR Test 1: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
results.append(("Intégration workflow → suggestion", False))
|
|
|
|
try:
|
|
# Test 2: check_for_suggestions()
|
|
results.append(("check_for_suggestions()", test_orchestrator_check_for_suggestions()))
|
|
except Exception as e:
|
|
print(f"\n❌ ERREUR Test 2: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
results.append(("check_for_suggestions()", False))
|
|
|
|
# Résumé
|
|
print("\n" + "="*60)
|
|
print("📊 RÉSUMÉ DES TESTS")
|
|
print("="*60)
|
|
|
|
for name, success in results:
|
|
status = "✅" if success else "❌"
|
|
print(f"{status} {name}")
|
|
|
|
total = len(results)
|
|
passed = sum(1 for _, success in results if success)
|
|
|
|
print(f"\n📈 Score: {passed}/{total} tests réussis")
|
|
|
|
if passed == total:
|
|
print("\n🎉 TOUS LES TESTS SONT PASSÉS !")
|
|
print("\n💡 Le Mode Assisté avec workflows est intégré !")
|
|
print("\n🚀 Prochaine étape: Améliorer la GUI pour afficher les suggestions")
|
|
return 0
|
|
else:
|
|
print(f"\n⚠️ {total - passed} test(s) échoué(s)")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|