| """Wiring for a real desktop: preflight checks, then one-call agent construction. |
| |
| This module exists because "make it work on my computer" is mostly a diagnosis |
| problem. Rather than failing deep inside an episode, :func:`doctor` reports exactly |
| which piece is missing - display, capture backend, input backend, OCR engine, model |
| runtime - and what to do about each. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import importlib.util |
| import os |
| import platform |
| from dataclasses import dataclass |
| from typing import Any, Callable, Iterable, Sequence |
|
|
| from tiny_computer_agent.act import DryRunExecutor, Executor, PyAutoGuiExecutor |
| from tiny_computer_agent.brain import Brain, PlannerBrain |
| from tiny_computer_agent.loop import AgentLoop, Step |
| from tiny_computer_agent.observe import GrabScreen, ScreenSource |
| from tiny_computer_agent.perception import ( |
| Detector, IndexedScreen, detector_available, make_detector, |
| ) |
| from tiny_computer_agent.gates import Gate |
| from tiny_computer_agent.safety import Policy |
| from tiny_computer_agent.skills import SkillBrain, SkillLibrary |
|
|
| DEFAULT_KILL_SWITCH = "/tmp/tiny-computer-agent.STOP" |
| BRAIN_NAMES = ("planner", "text-planner", "vlm") |
|
|
|
|
| @dataclass(frozen=True) |
| class Check: |
| """One preflight result: what was checked, whether it works, and how to fix it.""" |
|
|
| name: str |
| ok: bool |
| detail: str |
| fix: str = "" |
| required: bool = True |
|
|
| def render(self) -> str: |
| marker = "ok " if self.ok else ("FAIL" if self.required else "warn") |
| line = f" [{marker}] {self.name}: {self.detail}" |
| if not self.ok and self.fix: |
| line += f"\n fix: {self.fix}" |
| return line |
|
|
|
|
| def _installed(module: str) -> bool: |
| return importlib.util.find_spec(module) is not None |
|
|
|
|
| def display_available() -> tuple[bool, str]: |
| """Whether a windowing session looks reachable from this process.""" |
| system = platform.system() |
| if system in ("Darwin", "Windows"): |
| return True, f"{system} desktop session assumed" |
| display = os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY") |
| if display: |
| return True, f"DISPLAY={display}" |
| return False, "no DISPLAY or WAYLAND_DISPLAY set (headless container or SSH session)" |
|
|
|
|
| def run_checks(*, brain: str = "planner", model_id: str | None = None, |
| detector: str = "ocr", real_input: bool = False, |
| kill_switch_path: str | None = DEFAULT_KILL_SWITCH) -> list[Check]: |
| """Every precondition for driving a real desktop, in one list.""" |
| checks: list[Check] = [] |
|
|
| ok, detail = display_available() |
| checks.append(Check( |
| "display server", ok, detail, |
| fix="run this on the desktop session itself (a headless server cannot be driven)", |
| )) |
|
|
| checks.append(Check( |
| "pillow", _installed("PIL"), |
| "installed" if _installed("PIL") else "not installed", |
| fix="pip install 'tiny-computer-agent[desktop]'", |
| )) |
|
|
| has_mss = _installed("mss") |
| checks.append(Check( |
| "screen capture", has_mss or _installed("PIL"), |
| "mss (fast, multi-monitor)" if has_mss else "pillow ImageGrab fallback", |
| fix="pip install mss for faster multi-monitor capture", |
| required=False, |
| )) |
|
|
| has_pyautogui = _installed("pyautogui") |
| checks.append(Check( |
| "input backend (pyautogui)", has_pyautogui, |
| "installed" if has_pyautogui else "not installed", |
| fix="pip install 'tiny-computer-agent[desktop]'", |
| required=real_input, |
| )) |
|
|
| detector_ok, detector_detail = detector_available(detector) |
| checks.append(Check( |
| f"element detector ({detector})", detector_ok, detector_detail, |
| fix="pip install 'tiny-computer-agent[ocr]'", |
| )) |
|
|
| if brain == "planner": |
| checks.append(Check( |
| "model runtime", True, |
| "not needed: the planner is rule-based (no weights, no download)", |
| required=False, |
| )) |
| else: |
| from tiny_computer_agent.models import ( |
| DEFAULT_TEXT_MODEL, DEFAULT_VISION_MODEL, model_cached, model_dependencies, |
| ) |
|
|
| deps_ok, deps_detail = model_dependencies() |
| model = model_id or (DEFAULT_TEXT_MODEL if brain == "text-planner" else DEFAULT_VISION_MODEL) |
| checks.append(Check( |
| "model runtime", deps_ok, deps_detail, |
| fix="pip install 'tiny-computer-agent[vision]'", |
| )) |
| checks.append(Check( |
| f"model weights ({model})", |
| model_cached(model), |
| "cached locally" if model_cached(model) else "not downloaded yet", |
| fix=f"first run will download {model} from the Hub (one time)", |
| required=False, |
| )) |
|
|
| checks.append(Check( |
| "execution mode", |
| True, |
| "REAL mouse and keyboard input" if real_input else "dry run (nothing is clicked)", |
| fix="pass --real-input AND set TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT=1", |
| required=False, |
| )) |
|
|
| checks.append(Check( |
| "kill switch", True, |
| f"create {kill_switch_path} to abort the next action" |
| if kill_switch_path else "disabled", |
| required=False, |
| )) |
| return checks |
|
|
|
|
| def doctor(*, brain: str = "planner", model_id: str | None = None, |
| detector: str = "ocr", real_input: bool = False, |
| kill_switch_path: str | None = DEFAULT_KILL_SWITCH) -> str: |
| """A human-readable preflight report.""" |
| checks = run_checks(brain=brain, model_id=model_id, detector=detector, |
| real_input=real_input, kill_switch_path=kill_switch_path) |
| blocking = [c for c in checks if c.required and not c.ok] |
| lines = [ |
| f"tiny-computer-agent preflight ({platform.system()} {platform.machine()}, python {platform.python_version()})", |
| "", |
| *(c.render() for c in checks), |
| "", |
| ] |
| if blocking: |
| lines.append(f"NOT READY: {len(blocking)} blocking problem(s): " |
| + ", ".join(c.name for c in blocking)) |
| else: |
| lines.append("READY: the harness can drive this desktop.") |
| if not real_input: |
| lines.append("Dry run is on: nothing will be clicked. Read the plan first.") |
| return "\n".join(lines) |
|
|
|
|
| def build_agent(*, brain: Brain | str | None = None, screen: ScreenSource | None = None, |
| detector: Detector | str | None = None, dry_run: bool = True, |
| max_steps: int = 10, kill_switch_path: str | None = DEFAULT_KILL_SWITCH, |
| allow_grid_clicks: bool = False, |
| on_step: Callable[[Step], None] | None = None, |
| detector_options: dict[str, Any] | None = None, |
| skills: SkillLibrary | None = None, |
| gate: Gate | None = None) -> AgentLoop: |
| """Wire a real desktop into an :class:`AgentLoop` in one call. |
| |
| Defaults are the safe ones: dry run, a step budget of 10, and a kill-switch file |
| on disk that aborts the next action if it appears. ``screen``/``detector``/ |
| ``brain`` can be injected, which is how the tests drive this without a desktop. |
| |
| Pass ``skills`` and the brain becomes a :class:`~tiny_computer_agent.skills.SkillBrain` |
| that answers from the library first and only falls back to the model when no learned |
| skill applies; pass ``gate`` and the operator stands in front of every action. |
| """ |
| if detector is None or isinstance(detector, str): |
| detector = make_detector(detector or "ocr", **(detector_options or {})) |
| if screen is None: |
| screen = IndexedScreen(GrabScreen(), detector) |
| if brain is None or isinstance(brain, str): |
| name = brain or "planner" |
| if name == "planner": |
| brain = PlannerBrain() |
| else: |
| from tiny_computer_agent.models import make_model_brain |
|
|
| brain = make_model_brain(name) |
| if skills is not None: |
| brain = SkillBrain(skills, fallback=brain) |
| executor: Executor = DryRunExecutor() if dry_run else PyAutoGuiExecutor() |
| policy = Policy(dry_run=dry_run, step_budget=max_steps, |
| kill_switch_path=kill_switch_path, |
| allow_grid_clicks=allow_grid_clicks) |
| return AgentLoop(screen=screen, brain=brain, executor=executor, policy=policy, |
| on_step=on_step, gate=gate) |
|
|
