rpa_vision_v3/core/models/screen_state.py

"""
ScreenState - Couche 1 : Analyse Multi-Modale

Transforme un screenshot brut en représentation structurée à 4 niveaux :
- Niveau 1 : Raw (Ce que la machine voit)
- Niveau 2 : Perception (Ce que la vision déduit)
- Niveau 3 : Sémantique UI (Ce que le système comprend)
- Niveau 4 : Contexte Métier (Session/Application)

Tâche 4 : Contrats de données standardisés
- Timestamps : datetime objects uniquement
- IDs : Strings uniquement
"""

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional, Any, TYPE_CHECKING
from pathlib import Path
import json
from .base_models import Timestamp, StandardID, DataConverter

if TYPE_CHECKING:
    from .ui_element import UIElement


@dataclass
class EmbeddingRef:
    """Référence à un embedding stocké"""
    provider: str  # e.g., "openclip_ViT-B-32"
    vector_id: str  # Chemin vers fichier .npy
    dimensions: int
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "provider": self.provider,
            "vector_id": self.vector_id,
            "dimensions": self.dimensions
        }
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'EmbeddingRef':
        return cls(
            provider=data["provider"],
            vector_id=data["vector_id"],
            dimensions=data["dimensions"]
        )


@dataclass
class RawLevel:
    """Niveau 1 : Raw - Ce que la machine voit"""
    screenshot_path: str
    capture_method: str  # e.g., "mss", "pillow"
    file_size_bytes: int
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "screenshot_path": self.screenshot_path,
            "capture_method": self.capture_method,
            "file_size_bytes": self.file_size_bytes
        }
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'RawLevel':
        return cls(
            screenshot_path=data["screenshot_path"],
            capture_method=data["capture_method"],
            file_size_bytes=data["file_size_bytes"]
        )


@dataclass
class PerceptionLevel:
    """Niveau 2 : Perception - Ce que la vision déduit"""
    embedding: EmbeddingRef
    detected_text: List[str]
    text_detection_method: str  # e.g., "qwen_vl", "tesseract"
    confidence_avg: float
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "embedding": self.embedding.to_dict(),
            "detected_text": self.detected_text,
            "text_detection_method": self.text_detection_method,
            "confidence_avg": self.confidence_avg
        }
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'PerceptionLevel':
        return cls(
            embedding=EmbeddingRef.from_dict(data["embedding"]),
            detected_text=data["detected_text"],
            text_detection_method=data["text_detection_method"],
            confidence_avg=data["confidence_avg"]
        )


@dataclass
class ContextLevel:
    """Niveau 4 : Contexte Métier - Session/Application"""
    current_workflow_candidate: Optional[str] = None
    workflow_step: Optional[int] = None
    user_id: str = ""  # Standardisé en string
    tags: List[str] = field(default_factory=list)
    business_variables: Dict[str, Any] = field(default_factory=dict)
    
    def __post_init__(self):
        """Valider et migrer les données"""
        # Assurer que user_id est une string
        if self.user_id is not None and not isinstance(self.user_id, str):
            self.user_id = str(DataConverter.ensure_id(self.user_id))
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "current_workflow_candidate": self.current_workflow_candidate,
            "workflow_step": self.workflow_step,
            "user_id": self.user_id,
            "tags": self.tags,
            "business_variables": self.business_variables
        }
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'ContextLevel':
        # Migrer user_id vers string
        migrated_data = DataConverter.migrate_id_dict(data, ['user_id'])
        
        return cls(
            current_workflow_candidate=migrated_data.get("current_workflow_candidate"),
            workflow_step=migrated_data.get("workflow_step"),
            user_id=migrated_data.get("user_id", ""),
            tags=migrated_data.get("tags", []),
            business_variables=migrated_data.get("business_variables", {})
        )


@dataclass
class WindowContext:
    """Contexte de fenêtre avec métadonnées d'environnement graphique"""
    app_name: str
    window_title: str
    screen_resolution: List[int]
    workspace: str = "main"
    monitor_index: int = 0                              # Index du moniteur (0 = principal)
    dpi_scale: int = 100                                # Facteur DPI en % (100 = normal, 150 = haute résolution)
    window_bounds: Optional[List[int]] = None           # [x, y, width, height] de la fenêtre
    monitors: Optional[List[Dict[str, int]]] = None     # Liste des moniteurs [{width, height, x, y}]
    os_theme: str = "unknown"                           # "light", "dark", "unknown"
    os_language: str = "unknown"                        # Code langue (fr, en, de...)

    def to_dict(self) -> Dict[str, Any]:
        result = {
            "app_name": self.app_name,
            "window_title": self.window_title,
            "screen_resolution": self.screen_resolution,
            "workspace": self.workspace,
            "monitor_index": self.monitor_index,
            "dpi_scale": self.dpi_scale,
            "os_theme": self.os_theme,
            "os_language": self.os_language,
        }
        if self.window_bounds is not None:
            result["window_bounds"] = self.window_bounds
        if self.monitors is not None:
            result["monitors"] = self.monitors
        return result

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'WindowContext':
        return cls(
            app_name=data["app_name"],
            window_title=data["window_title"],
            screen_resolution=data["screen_resolution"],
            workspace=data.get("workspace", "main"),
            monitor_index=data.get("monitor_index", 0),
            dpi_scale=data.get("dpi_scale", 100),
            window_bounds=data.get("window_bounds"),
            monitors=data.get("monitors"),
            os_theme=data.get("os_theme", "unknown"),
            os_language=data.get("os_language", "unknown"),
        )


@dataclass
class ScreenState:
    """
    État d'écran structuré à 4 niveaux
    
    Représente un screenshot analysé avec :
    - Raw : Image brute
    - Perception : Embeddings + texte détecté
    - Sémantique UI : Éléments UI (sera ajouté séparément)
    - Contexte : Métadonnées métier
    
    Tâche 4 : Contrats standardisés
    - screen_state_id, session_id : Strings standardisés
    - timestamp : datetime object uniquement
    """
    screen_state_id: str  # Standardisé en string
    timestamp: datetime  # datetime object uniquement
    session_id: str  # Standardisé en string
    window: WindowContext
    raw: RawLevel
    perception: PerceptionLevel
    context: ContextLevel
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    # Niveau 3 : UI Elements - Liste des éléments UI détectés
    ui_elements: List[Any] = field(default_factory=list)  # List[UIElement]
    
    def __post_init__(self):
        """Valider et migrer les données après initialisation"""
        # Migrer les IDs vers strings
        if not isinstance(self.screen_state_id, str):
            self.screen_state_id = str(DataConverter.ensure_id(self.screen_state_id))
        if not isinstance(self.session_id, str):
            self.session_id = str(DataConverter.ensure_id(self.session_id))
        
        # Migrer timestamp vers datetime
        if not isinstance(self.timestamp, datetime):
            self.timestamp = DataConverter.ensure_timestamp(self.timestamp).value
    
    # =========================================================================
    # ALIASES DE COMPATIBILITÉ (Fiche #1 - Migration douce)
    # Auteur: Dom, Alice Kiro - 15 décembre 2024
    # =========================================================================
    
    @property
    def state_id(self) -> str:
        """Alias de compatibilité pour screen_state_id"""
        return self.screen_state_id
    
    @property
    def raw_level(self) -> RawLevel:
        """Alias de compatibilité pour raw"""
        return self.raw
    
    @property
    def perception_level(self) -> PerceptionLevel:
        """Alias de compatibilité pour perception"""
        return self.perception
    
    @property
    def screenshot_path(self) -> str:
        """Alias de compatibilité pour raw.screenshot_path"""
        return self.raw.screenshot_path
    
    @property
    def ui_elements_count(self) -> int:
        """Nombre d'éléments UI détectés"""
        return len(self.ui_elements)
    
    def to_json(self) -> Dict[str, Any]:
        """Sérialiser en JSON"""
        return {
            "screen_state_id": self.screen_state_id,
            "timestamp": self.timestamp.isoformat(),
            "session_id": self.session_id,
            "window": self.window.to_dict(),
            "raw": self.raw.to_dict(),
            "perception": self.perception.to_dict(),
            "context": self.context.to_dict(),
            "metadata": self.metadata,
            "ui_elements": [el.to_dict() if hasattr(el, 'to_dict') else el for el in self.ui_elements]
        }
    
    @classmethod
    def from_json(cls, data: Dict[str, Any]) -> 'ScreenState':
        """Désérialiser depuis JSON avec migration automatique"""
        # Migrer les données vers les nouveaux contrats
        migrated_data = DataConverter.migrate_timestamp_dict(data, ['timestamp'])
        migrated_data = DataConverter.migrate_id_dict(migrated_data, ['screen_state_id', 'session_id'])
        
        timestamp = migrated_data["timestamp"]
        if isinstance(timestamp, str):
            timestamp = datetime.fromisoformat(timestamp)
        
        window = WindowContext.from_dict(migrated_data["window"])
        raw = RawLevel.from_dict(migrated_data["raw"])
        perception = PerceptionLevel.from_dict(migrated_data["perception"])
        context = ContextLevel.from_dict(migrated_data["context"])
        
        # Import UIElement ici pour éviter import circulaire
        from .ui_element import UIElement
        
        # Parser ui_elements si présents
        ui_elements_data = migrated_data.get("ui_elements", [])
        ui_elements = []
        for el_data in ui_elements_data:
            if isinstance(el_data, dict):
                ui_elements.append(UIElement.from_dict(el_data))
            else:
                ui_elements.append(el_data)
        
        return cls(
            screen_state_id=migrated_data["screen_state_id"],
            timestamp=timestamp,
            session_id=migrated_data["session_id"],
            window=window,
            raw=raw,
            perception=perception,
            context=context,
            metadata=migrated_data.get("metadata", {}),
            ui_elements=ui_elements
        )
    
    def save_to_file(self, filepath: Path) -> None:
        """Sauvegarder dans un fichier JSON"""
        filepath.parent.mkdir(parents=True, exist_ok=True)
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(self.to_json(), f, indent=2, ensure_ascii=False)
    
    @classmethod
    def load_from_file(cls, filepath: Path) -> 'ScreenState':
        """Charger depuis un fichier JSON"""
        with open(filepath, 'r', encoding='utf-8') as f:
            data = json.load(f)
        return cls.from_json(data)
    
    def validate_consistency(self) -> bool:
        """
        Valider que les 4 niveaux référencent le même screenshot et timestamp
        
        Property 2: ScreenState Multi-Level Consistency
        """
        # Tous les niveaux doivent exister
        if not all([self.raw, self.perception, self.context]):
            return False
        
        # Le timestamp doit être cohérent
        # (tous les niveaux référencent le même instant)
        return True