| """Execution: the only module in the harness that touches the machine.""" |
|
|
| from __future__ import annotations |
|
|
| import time |
| from dataclasses import dataclass |
| from typing import Protocol |
|
|
| from tiny_computer_agent.actions import Action |
| from tiny_computer_agent.observe import Observation |
|
|
|
|
| @dataclass(frozen=True) |
| class ActionResult: |
| action: Action |
| ok: bool |
| note: str = "" |
|
|
| def summary(self) -> str: |
| marker = "ok" if self.ok else "FAILED" |
| return f"{marker}: {self.note}" |
|
|
|
|
| class InputUnavailable(RuntimeError): |
| """Raised when real input is requested but pyautogui cannot drive the machine.""" |
|
|
|
|
| def resolve_point( |
| action: Action, observation: Observation | None = None |
| ) -> tuple[int, int] | None: |
| """Turn a click into pixels, using the element index for label clicks.""" |
| if action.kind != "click": |
| return None |
| if action.x is not None and action.y is not None: |
| return (action.x, action.y) |
| if observation is not None and action.label: |
| element = observation.find(action.label) |
| if element is not None: |
| return element.center |
| return None |
|
|
|
|
| class Executor(Protocol): |
| dry_run: bool |
|
|
| def execute( |
| self, action: Action, observation: Observation | None = None |
| ) -> ActionResult: ... |
|
|
|
|
| class DryRunExecutor: |
| """Records actions instead of performing them. The default everywhere. |
| |
| Lets you read an entire episode — including which pixel each label resolved |
| to — before a single event reaches the machine. |
| """ |
|
|
| dry_run = True |
|
|
| def __init__(self) -> None: |
| self.performed: list[Action] = [] |
| self.log: list[str] = [] |
|
|
| def execute( |
| self, action: Action, observation: Observation | None = None |
| ) -> ActionResult: |
| note = f"(dry-run) would {action.describe()}" |
| ok = True |
| if action.kind == "click": |
| point = resolve_point(action, observation) |
| if point is not None: |
| note += f" -> pixel {point}" |
| elif action.label: |
| |
| |
| ok = False |
| note += " (label absent from the element index; live input refuses this)" |
| self.performed.append(action) |
| self.log.append(note) |
| return ActionResult(action=action, ok=ok, note=note) |
|
|
|
|
| class PyAutoGuiExecutor: |
| """Real mouse and keyboard input. |
| |
| Refuses to click when a label cannot be resolved to an element: a tiny model |
| guessing pixel coordinates is exactly how you click "Delete account". |
| """ |
|
|
| dry_run = False |
|
|
| def __init__(self, *, pause: float = 0.1, typing_interval: float = 0.02, |
| scroll_ticks: int = 100) -> None: |
| self.pause = pause |
| self.typing_interval = typing_interval |
| self.scroll_ticks = scroll_ticks |
|
|
| @staticmethod |
| def _pyautogui(): |
| try: |
| import pyautogui |
| except Exception as exc: |
| raise InputUnavailable( |
| "real input needs pyautogui and a live display: pip install \'tiny-computer-agent[desktop]\'" |
| ) from exc |
| return pyautogui |
|
|
| def execute( |
| self, action: Action, observation: Observation | None = None |
| ) -> ActionResult: |
| if action.kind == "done": |
| return ActionResult(action=action, ok=True, note="goal reported met") |
|
|
| point: tuple[int, int] | None = None |
| if action.kind == "click": |
| point = resolve_point(action, observation) |
| if point is None: |
| return ActionResult( |
| action=action, |
| ok=False, |
| note=f"no element matching {action.label!r}; refusing to guess pixels", |
| ) |
|
|
| pyautogui = self._pyautogui() |
| try: |
| if action.kind == "click": |
| pyautogui.click(*point) |
| note = f"clicked {point}" |
| elif action.kind == "type": |
| pyautogui.write(action.text or "", interval=self.typing_interval) |
| note = f"typed {len(action.text or '')} characters" |
| elif action.kind == "key": |
| chord = [k.strip() for k in (action.keys or "").split("+") if k.strip()] |
| pyautogui.hotkey(*chord) |
| note = f"pressed {action.keys}" |
| elif action.kind == "scroll": |
| pyautogui.scroll(action.amount * self.scroll_ticks) |
| note = f"scrolled {action.amount} ticks" |
| elif action.kind == "wait": |
| time.sleep(action.seconds) |
| note = f"waited {action.seconds}s" |
| else: |
| return ActionResult(action=action, ok=False, note=f"unhandled kind {action.kind!r}") |
| except Exception as exc: |
| return ActionResult(action=action, ok=False, note=f"{type(exc).__name__}: {exc}") |
| if self.pause: |
| time.sleep(self.pause) |
| return ActionResult(action=action, ok=True, note=note) |
|
|
