""" core/grounding/element_signature.py — Signatures d'éléments UI apprises Chaque élément cliqué avec succès enrichit sa signature : - texte OCR, type, position relative, voisins contextuels - nombre de succès/échecs, confiance moyenne - variantes observées (résolutions, positions) Les signatures sont stockées en SQLite pour un lookup rapide. Pattern identique à TargetMemoryStore (validé en prod). Utilisation : from core.grounding.element_signature import SignatureStore store = SignatureStore() # Après un clic réussi store.record_success("btn_valider", "notepad_1920x1080", element, confidence=0.92) # Au replay sig = store.lookup("btn_valider", "notepad_1920x1080") if sig: print(f"Signature connue : {sig['text']} position={sig['relative_position']}") """ from __future__ import annotations import hashlib import json import os import sqlite3 import threading import time from typing import Any, Dict, List, Optional from core.grounding.fast_types import DetectedUIElement # Chemin par défaut de la DB _DEFAULT_DB = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "data", "learning", "element_signatures.db", ) class SignatureStore: """Stockage SQLite des signatures d'éléments UI appris.""" def __init__(self, db_path: str = _DEFAULT_DB): self.db_path = db_path self._lock = threading.Lock() self._ensure_db() def _ensure_db(self): """Crée la DB et la table si nécessaire.""" os.makedirs(os.path.dirname(self.db_path), exist_ok=True) with sqlite3.connect(self.db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS signatures ( target_key TEXT NOT NULL, screen_context TEXT NOT NULL, text TEXT DEFAULT '', element_type TEXT DEFAULT 'element', relative_position TEXT DEFAULT '', neighbors TEXT DEFAULT '[]', success_count INTEGER DEFAULT 0, fail_count INTEGER DEFAULT 0, avg_confidence REAL DEFAULT 0.0, last_seen TEXT DEFAULT '', variants TEXT DEFAULT '[]', PRIMARY KEY (target_key, screen_context) ) """) conn.execute(""" CREATE INDEX IF NOT EXISTS idx_target_key ON signatures(target_key) """) # ------------------------------------------------------------------ # Lookup # ------------------------------------------------------------------ def lookup(self, target_key: str, screen_context: str = "") -> Optional[Dict[str, Any]]: """Cherche une signature connue. Args: target_key: Clé unique de la cible (hash du texte + description). screen_context: Contexte d'écran (hash titre fenêtre + résolution). Returns: Dict avec les champs de la signature, ou None. """ with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row # Chercher avec le contexte exact d'abord row = conn.execute( "SELECT * FROM signatures WHERE target_key = ? AND screen_context = ?", (target_key, screen_context), ).fetchone() # Fallback : chercher sans contexte (toutes les variantes) if row is None and screen_context: row = conn.execute( "SELECT * FROM signatures WHERE target_key = ? ORDER BY success_count DESC LIMIT 1", (target_key,), ).fetchone() if row is None: return None return { "target_key": row["target_key"], "screen_context": row["screen_context"], "text": row["text"], "element_type": row["element_type"], "relative_position": row["relative_position"], "neighbors": json.loads(row["neighbors"]), "success_count": row["success_count"], "fail_count": row["fail_count"], "avg_confidence": row["avg_confidence"], "last_seen": row["last_seen"], "variants": json.loads(row["variants"]), } # ------------------------------------------------------------------ # Enregistrement # ------------------------------------------------------------------ def record_success( self, target_key: str, screen_context: str, element: DetectedUIElement, confidence: float, ): """Enregistre un succès — crée ou enrichit la signature.""" with self._lock: existing = self.lookup(target_key, screen_context) now = time.strftime("%Y-%m-%dT%H:%M:%S") if existing: # Enrichir la signature existante n = existing["success_count"] new_avg = (existing["avg_confidence"] * n + confidence) / (n + 1) # Ajouter la variante si position différente variants = existing["variants"] variant = { "position": element.relative_position, "center": list(element.center), "confidence": confidence, "timestamp": now, } variants.append(variant) # Garder les 20 dernières variantes max variants = variants[-20:] # Mettre à jour les voisins (union) neighbors = list(set(existing["neighbors"] + element.neighbors))[:10] with sqlite3.connect(self.db_path) as conn: conn.execute(""" UPDATE signatures SET success_count = success_count + 1, avg_confidence = ?, last_seen = ?, neighbors = ?, variants = ?, relative_position = ? WHERE target_key = ? AND screen_context = ? """, ( new_avg, now, json.dumps(neighbors), json.dumps(variants), element.relative_position, target_key, screen_context, )) else: # Créer une nouvelle signature with sqlite3.connect(self.db_path) as conn: conn.execute(""" INSERT INTO signatures (target_key, screen_context, text, element_type, relative_position, neighbors, success_count, fail_count, avg_confidence, last_seen, variants) VALUES (?, ?, ?, ?, ?, ?, 1, 0, ?, ?, ?) """, ( target_key, screen_context, element.ocr_text, element.element_type, element.relative_position, json.dumps(element.neighbors[:10]), confidence, now, json.dumps([{ "position": element.relative_position, "center": list(element.center), "confidence": confidence, "timestamp": now, }]), )) print(f"📝 [Signature] '{target_key}' {'enrichie' if existing else 'créée'} " f"(conf={confidence:.2f}, ctx='{screen_context[:30]}')") def record_failure(self, target_key: str, screen_context: str): """Enregistre un échec pour une signature.""" with self._lock: with sqlite3.connect(self.db_path) as conn: conn.execute(""" UPDATE signatures SET fail_count = fail_count + 1, last_seen = ? WHERE target_key = ? AND screen_context = ? """, (time.strftime("%Y-%m-%dT%H:%M:%S"), target_key, screen_context)) # ------------------------------------------------------------------ # Utilitaires # ------------------------------------------------------------------ @staticmethod def make_target_key(text: str, description: str = "") -> str: """Génère une clé unique pour une cible.""" raw = f"{text.lower().strip()}|{description.lower().strip()}" return hashlib.md5(raw.encode()).hexdigest()[:16] @staticmethod def make_screen_context(window_title: str, resolution: tuple = (0, 0)) -> str: """Génère un contexte d'écran.""" raw = f"{window_title.lower().strip()}|{resolution[0]}x{resolution[1]}" return hashlib.md5(raw.encode()).hexdigest()[:12] def get_stats(self) -> Dict[str, Any]: """Statistiques de la base de signatures.""" with sqlite3.connect(self.db_path) as conn: total = conn.execute("SELECT COUNT(*) FROM signatures").fetchone()[0] reliable = conn.execute( "SELECT COUNT(*) FROM signatures WHERE success_count >= 3 AND fail_count = 0" ).fetchone()[0] return { "total_signatures": total, "reliable": reliable, "db_path": self.db_path, }