diff --git a/tests/unit/test_competence_to_vwb_preview.py b/tests/unit/test_competence_to_vwb_preview.py new file mode 100644 index 000000000..a0f8ba0f1 --- /dev/null +++ b/tests/unit/test_competence_to_vwb_preview.py @@ -0,0 +1,105 @@ +from flask import Flask + +from visual_workflow_builder.backend.services.competence_vwb_preview import ( + competence_yaml_to_vwb_preview, + competence_yaml_to_vwb_steps, +) + + +def _by_type(preview): + return [step["action_type"] for step in preview["steps"]] + + +def test_competence_to_vwb_preview_key_win_r_steps(): + preview = competence_yaml_to_vwb_preview("key_win_r_wait_explorer_exe") + + assert preview["readonly"] is True + assert preview["write_back_enabled"] is False + assert preview["workflow"]["competence_id"] == "key_win_r_wait_explorer_exe" + assert _by_type(preview) == [ + "pause_for_human", + "keyboard_shortcut", + "wait_for_state", + "pause_for_human", + ] + + +def test_keyboard_shortcut_keys_are_preserved(): + steps = competence_yaml_to_vwb_steps("key_win_r_wait_explorer_exe") + shortcut = steps[1] + + assert shortcut["action_type"] == "keyboard_shortcut" + assert shortcut["parameters"]["keys"] == ["win", "r"] + assert shortcut["parameters"]["source"] == ( + "lea_competence:key_win_r_wait_explorer_exe" + ) + assert shortcut["parameters"]["source_method_id"] == "step_1_key_combo" + + +def test_wait_for_state_expected_state_is_preserved(): + steps = competence_yaml_to_vwb_steps("key_win_r_wait_explorer_exe") + wait_step = steps[2] + + assert wait_step["action_type"] == "wait_for_state" + params = wait_step["parameters"] + assert params["expected_state"]["window_title_in"] == ["Ex\u00e9cuter"] + assert params["expected_state"]["process_active"] == "explorer.exe" + assert params["timeout_ms"] == 5000 + assert params["poll_interval_ms"] == 250 + assert params["evidence_required"] == "window_or_process" + + +def test_pause_for_human_before_and_after_are_supervision_only(): + steps = competence_yaml_to_vwb_steps("key_win_r_wait_explorer_exe") + + before = steps[0] + after = steps[-1] + assert before["parameters"]["phase"] == "before" + assert before["parameters"]["verdict_required"] is False + assert after["parameters"]["phase"] == "after" + assert after["parameters"]["verdict_required"] is True + assert after["parameters"]["write_back_enabled"] is False + + +def test_adapter_is_generic_on_methods_not_hardcoded_to_win_r(): + steps = competence_yaml_to_vwb_steps("key_ctrl_s_wait_notepad_exe") + + assert [step["action_type"] for step in steps] == [ + "pause_for_human", + "keyboard_shortcut", + "wait_for_state", + "pause_for_human", + ] + assert steps[1]["parameters"]["keys"] == ["ctrl", "s"] + assert steps[2]["parameters"]["expected_state"]["window_title_in"] == [ + "Enregistrer sous" + ] + assert steps[2]["parameters"]["expected_state"]["process_active"] == "Notepad.exe" + + +def test_preview_endpoint_returns_read_only_workflow(): + from visual_workflow_builder.backend.catalog_routes_v2_vlm import ( + competence_preview_bp, + ) + + app = Flask(__name__) + app.register_blueprint(competence_preview_bp) + + with app.test_client() as client: + response = client.post( + "/api/vwb/competences/key_win_r_wait_explorer_exe/preview", + json={"supervised": True}, + ) + + assert response.status_code == 200 + data = response.get_json() + assert data["success"] is True + assert data["readonly"] is True + assert data["write_back_enabled"] is False + assert data["workflow"]["source"] == "lea_competence_preview" + assert [step["action_type"] for step in data["steps"]] == [ + "pause_for_human", + "keyboard_shortcut", + "wait_for_state", + "pause_for_human", + ] diff --git a/visual_workflow_builder/backend/app.py b/visual_workflow_builder/backend/app.py index 5e9235af1..5f0f048a2 100644 --- a/visual_workflow_builder/backend/app.py +++ b/visual_workflow_builder/backend/app.py @@ -180,8 +180,9 @@ except ImportError as e: # Catalogue VWB - actions VisionOnly # V2 avec VLM (Vision Language Model) pour détection intelligente try: - from catalog_routes_v2_vlm import catalog_bp, VLM_MODEL + from catalog_routes_v2_vlm import catalog_bp, competence_preview_bp, VLM_MODEL app.register_blueprint(catalog_bp) + app.register_blueprint(competence_preview_bp) print(f"✅ Blueprint catalog V2 VLM (Ollama {VLM_MODEL}) enregistré") except ImportError as e: print(f"⚠️ Blueprint catalog V2 VLM désactivé: {e}") diff --git a/visual_workflow_builder/backend/catalog_routes_v2_vlm.py b/visual_workflow_builder/backend/catalog_routes_v2_vlm.py index 1fe0d88f0..cc5c583cf 100644 --- a/visual_workflow_builder/backend/catalog_routes_v2_vlm.py +++ b/visual_workflow_builder/backend/catalog_routes_v2_vlm.py @@ -1073,6 +1073,11 @@ def find_anchor_multiscale(anchor_image_base64: str, scales: List[float] = None, # Créer le blueprint pour les routes catalogue catalog_bp = Blueprint('catalog', __name__, url_prefix='/api/vwb/catalog') +competence_preview_bp = Blueprint( + 'competence_preview', + __name__, + url_prefix='/api/vwb/competences', +) # Instance globale du ScreenCapturer (initialisée à la demande) _screen_capturer_instance = None @@ -1210,6 +1215,44 @@ def _execute_lea_competence_action( }) +@competence_preview_bp.route('//preview', methods=['POST']) +def preview_lea_competence_workflow(competence_id: str): + """Build a read-only VWB preview from a Lea competence YAML.""" + try: + data = request.get_json(silent=True) or {} + supervised = bool(data.get('supervised', True)) + + from visual_workflow_builder.backend.services.competence_vwb_preview import ( + competence_yaml_to_vwb_preview, + ) + + preview = competence_yaml_to_vwb_preview( + competence_id, + supervised=supervised, + ) + return jsonify({ + "success": True, + "competence_id": competence_id, + "preview": preview, + "workflow": preview["workflow"], + "steps": preview["steps"], + "warnings": preview["warnings"], + "readonly": True, + "write_back_enabled": False, + }) + except KeyError: + return jsonify({ + "success": False, + "error": f"Competence '{competence_id}' introuvable", + }), 404 + except Exception as e: + return jsonify({ + "success": False, + "error": f"Erreur preview competence: {str(e)}", + "traceback": traceback.format_exc(), + }), 500 + + def get_screen_capturer(): """ Obtient l'instance du ScreenCapturer (initialisation paresseuse). diff --git a/visual_workflow_builder/backend/services/competence_vwb_preview.py b/visual_workflow_builder/backend/services/competence_vwb_preview.py new file mode 100644 index 000000000..4c97da2e2 --- /dev/null +++ b/visual_workflow_builder/backend/services/competence_vwb_preview.py @@ -0,0 +1,254 @@ +"""Build read-only VWB previews from Lea competence YAML files.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional + +from core.competences.catalog import DEFAULT_COMPETENCE_ROOT, CompetenceSummary +from core.competences.replay import find_competence + + +def competence_yaml_to_vwb_preview( + competence_id: str, + *, + root: Path | str = DEFAULT_COMPETENCE_ROOT, + supervised: bool = True, + states: Optional[Iterable[str]] = None, +) -> Dict[str, Any]: + """Return a read-only VWB workflow preview for one persisted competence.""" + + competence = find_competence(competence_id, root=root, states=states) + warnings: List[str] = [] + steps: List[Dict[str, Any]] = [] + + if supervised: + steps.append(_pause_step(competence, phase="before", order=len(steps))) + + for method in competence.methods: + step = _method_to_vwb_step(competence, method, order=len(steps)) + if step is None: + warnings.append( + "Methode non supportee en preview VWB: " + f"{method.get('id') or method.get('kind') or 'unknown'}" + ) + continue + steps.append(step) + + if supervised: + steps.append(_pause_step(competence, phase="after", order=len(steps))) + + _compute_layout(steps) + + workflow = { + "id": f"preview_competence_{competence.id}", + "name": f"Lea preview - {competence.id}", + "description": competence.intent_fr, + "source": "lea_competence_preview", + "review_status": "preview", + "tags": ["lea", "competence", competence.learning_state], + "competence_id": competence.id, + "learning_state": competence.learning_state, + "source_path": competence.source_path, + "readonly": True, + "created_at": datetime.now(timezone.utc).isoformat(), + } + + return { + "workflow": workflow, + "steps": steps, + "warnings": warnings, + "readonly": True, + "write_back_enabled": False, + "db_write": False, + } + + +def competence_yaml_to_vwb_steps( + competence_id: str, + *, + root: Path | str = DEFAULT_COMPETENCE_ROOT, + supervised: bool = True, + states: Optional[Iterable[str]] = None, +) -> List[Dict[str, Any]]: + """Compatibility helper returning only VWB steps for a competence.""" + + return competence_yaml_to_vwb_preview( + competence_id, + root=root, + supervised=supervised, + states=states, + )["steps"] + + +def _method_to_vwb_step( + competence: CompetenceSummary, + method: Dict[str, Any], + *, + order: int, +) -> Optional[Dict[str, Any]]: + kind = str(method.get("kind") or method.get("primitive_ref") or "").strip() + params = method.get("parameters") if isinstance(method.get("parameters"), dict) else {} + method_id = str(method.get("id") or f"method_{order}") + source_params = _source_params(competence, method_id) + metadata = _source_metadata(competence, method, phase="method") + + if kind == "key_combo": + keys = params.get("keys") + if not isinstance(keys, list) or not keys: + return None + vwb_params = { + "keys": [str(key) for key in keys], + "hold_duration_ms": params.get("hold_duration_ms", 50), + **source_params, + } + return _step( + action_type="keyboard_shortcut", + order=order, + label=f"Raccourci : {'+'.join(vwb_params['keys'])}", + parameters=vwb_params, + metadata=metadata, + ) + + if kind in {"wait_state", "wait_for_state"}: + expected_state = params.get("expected_state") + if not isinstance(expected_state, dict) or not expected_state: + return None + vwb_params = { + "expected_state": expected_state, + "timeout_ms": int(params.get("timeout_ms", 5000)), + "poll_interval_ms": int(params.get("poll_interval_ms", 250)), + "evidence_required": params.get("evidence_required", "window_or_process"), + **source_params, + } + return _step( + action_type="wait_for_state", + order=order, + label="Attendre etat", + parameters=vwb_params, + metadata=metadata, + ) + + if kind in {"text_input", "text_input_focused"}: + text = params.get("text") + if text is None: + return None + return _step( + action_type="type_text", + order=order, + label="Saisir texte", + parameters={"text": str(text), **source_params}, + metadata=metadata, + ) + + if kind == "click_anchor": + vwb_params = { + key: value + for key, value in params.items() + if key in {"visual_anchor", "target_text", "target_role", "by_text"} + } + return _step( + action_type="click_anchor", + order=order, + label="Clic ancre", + parameters={**vwb_params, **source_params}, + metadata=metadata, + ) + + return None + + +def _pause_step( + competence: CompetenceSummary, + *, + phase: str, + order: int, +) -> Dict[str, Any]: + failure = competence.failure_message_template + if phase == "before": + message = ( + f"Prepare le test supervise de la competence '{competence.id}'. " + f"Intention: {competence.intent_fr}. " + f"Attendu: {failure.get('attendu', 'etat attendu non renseigne')}." + ) + verdict_required = False + else: + message = ( + f"Valide le resultat de la competence '{competence.id}'. " + f"Attendu: {failure.get('attendu', 'etat attendu non renseigne')}. " + "Indique si ce replay peut etre garde comme succes supervise." + ) + verdict_required = True + + parameters = { + "message": message, + "phase": phase, + "verdict_required": verdict_required, + "competence_id": competence.id, + "source": f"lea_competence:{competence.id}", + "write_back_enabled": False, + "intention": failure.get("intention", competence.intent_fr), + "attendu": failure.get("attendu", ""), + "demande": failure.get("demande", ""), + } + return _step( + action_type="pause_for_human", + order=order, + label="Pause supervision" if phase == "before" else "Verdict humain", + parameters=parameters, + metadata=_source_metadata(competence, {}, phase=phase), + ) + + +def _source_params(competence: CompetenceSummary, method_id: str) -> Dict[str, Any]: + return { + "competence_id": competence.id, + "source": f"lea_competence:{competence.id}", + "source_method_id": method_id, + "write_back_enabled": False, + } + + +def _source_metadata( + competence: CompetenceSummary, + method: Dict[str, Any], + *, + phase: str, +) -> Dict[str, Any]: + return { + "origin": "lea_competence_yaml", + "competence_id": competence.id, + "learning_state": competence.learning_state, + "source_path": competence.source_path, + "source_method_id": method.get("id"), + "primitive_ref": method.get("primitive_ref"), + "method_kind": method.get("kind"), + "phase": phase, + } + + +def _step( + *, + action_type: str, + order: int, + label: str, + parameters: Dict[str, Any], + metadata: Dict[str, Any], +) -> Dict[str, Any]: + return { + "id": f"preview_step_{order:02d}_{action_type}", + "action_type": action_type, + "order": order, + "position_x": 0, + "position_y": 0, + "label": label, + "parameters": parameters, + "metadata": metadata, + } + + +def _compute_layout(steps: List[Dict[str, Any]]) -> None: + for index, step in enumerate(steps): + step["position_x"] = 80 + index * 300 + step["position_y"] = 120