File size: 5,206 Bytes
e187916 57356d8 e187916 57356d8 e187916 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """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:
# Fail exactly where live input would, so a dry run never gives
# false confidence about a plan the executor would refuse.
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: # pragma: no cover - depends on the machine
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: # unreachable: Action already validated the vocabulary
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)
|