# MIT License # Copyright (c) 2024 """Data schemas for recording and DSL components.""" from __future__ import annotations from typing import Any, Dict, List, Literal, Optional from pydantic import BaseModel, Field class Rect(BaseModel): left: int top: int right: int bottom: int @property def width(self) -> int: return self.right - self.left @property def height(self) -> int: return self.bottom - self.top class WindowInfo(BaseModel): hwnd: Optional[int] = None title: Optional[str] = None process_name: Optional[str] = None rect: Optional[Rect] = None class UISelector(BaseModel): automation_id: Optional[str] = None name: Optional[str] = None class_name: Optional[str] = None control_type: Optional[str] = None bounding_rect: Optional[Rect] = None class FramePaths(BaseModel): full: Optional[str] = None crop_mouse: Optional[str] = None crop_element: Optional[str] = None class MouseInfo(BaseModel): x: int y: int button: Optional[str] = None action: Optional[str] = None class UITreeNode(BaseModel): name: Optional[str] automation_id: Optional[str] class_name: Optional[str] control_type: Optional[str] depth: int EventType = Literal["mouse_click", "text_input", "window_change"] class EventRecord(BaseModel): ts: float event_type: EventType window: Optional[WindowInfo] = None mouse: Optional[MouseInfo] = None text: Optional[str] = None uia: Optional[UISelector] = None frame_paths: Optional[FramePaths] = None video_time_offset_ms: Optional[int] = Field(None, alias="video_time_offset_ms") ui_snapshot: Optional[str] = None class UISnapshot(BaseModel): selector: Optional[UISelector] = None tree: List[UITreeNode] = Field(default_factory=list) class SessionManifest(BaseModel): session_id: str start_time: float end_time: float resolution: Optional[str] = None fps: int screen: int video_path: str events_path: str frames_dir: str frames_crops_dir: str ui_snapshots_dir: str # DSL schemas (kept for executor/infer workflow) ------------------------ class DSLAction(BaseModel): action: Literal["click", "type", "set_value", "assert_exists", "wait_for"] target: Dict[str, Any] = Field(default_factory=dict) text: Optional[str] = None params: Dict[str, Any] = Field(default_factory=dict) retry_policy: Optional[Dict[str, Any]] = None waits: Optional[Dict[str, Any]] = None class DSLBlock(BaseModel): name: str steps: List[Any] = Field(default_factory=list) if_condition: Optional[str] = None else_steps: Optional[List[Any]] = None for_each: Optional[str] = None class DSLSpec(BaseModel): params: Dict[str, Any] = Field(default_factory=dict) steps: List[Any] assertions: List[str] = Field(default_factory=list) retry_policy: Dict[str, Any] = Field(default_factory=lambda: {"max_attempts": 2, "interval": 1.0}) waits: Dict[str, Any] = Field(default_factory=lambda: {"appear": 5.0, "disappear": 5.0})