feat(vwb): preview lea competence workflows
This commit is contained in:
Dom
@@ -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
|
||||
Reference in New Issue
Block a user