| """Command line entry point. |
| |
| python -m tiny_computer_agent # scripted demo, dry-run |
| python -m tiny_computer_agent --doctor # what can this machine do? |
| python -m tiny_computer_agent --image example # replay a screenshot, dry-run |
| python -m tiny_computer_agent --image shot.png # replay your own screenshot |
| python -m tiny_computer_agent --live # your real screen, dry-run |
| python -m tiny_computer_agent --live --real-input "click 'OK'" |
| |
| Real mouse and keyboard input additionally requires |
| ``TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT=1``. Dry run is the default, so you can read a |
| whole episode before anything touches your pointer. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import os |
| import platform |
| import sys |
|
|
| from tiny_computer_agent.act import DryRunExecutor, PyAutoGuiExecutor |
| from tiny_computer_agent.brain import MockBrain, PlannerBrain |
| from tiny_computer_agent.demo import fake_desktop |
| from tiny_computer_agent.desktop import ( |
| BRAIN_NAMES, DEFAULT_KILL_SWITCH, Check, display_available, run_checks, |
| ) |
| from tiny_computer_agent.loop import AgentLoop, Step |
| from tiny_computer_agent.observe import ( |
| Element, FakeScreen, GrabScreen, ImageScreen, ScreenUnavailable, |
| ) |
| from tiny_computer_agent.perception import IndexedScreen, detector_available, make_detector |
| from tiny_computer_agent.safety import Policy |
|
|
| ALLOW_REAL_INPUT_ENV = "TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT" |
| EXAMPLE_IMAGE = "example" |
|
|
|
|
| def demo_screen() -> FakeScreen: |
| """A three-frame desktop: a dialog, then an editor, then a saved file.""" |
| return FakeScreen( |
| [ |
| (Element("Terminal", x=0, y=0, width=40, height=24), |
| Element("Files", x=60, y=0, width=40, height=24)), |
| (Element("Name", kind="input", x=100, y=120, width=200, height=24), |
| Element("Save", x=320, y=120, width=80, height=24)), |
| (Element("Save", x=320, y=120, width=80, height=24),), |
| ], |
| text="Untitled document - unsaved changes", |
| ) |
|
|
|
|
| def build_parser() -> argparse.ArgumentParser: |
| parser = argparse.ArgumentParser( |
| prog="tiny-computer-agent", |
| description="Drive a computer with a tiny local model (dry-run by default).", |
| ) |
| parser.add_argument("goal", nargs="?", default="click 'Save' to save the document") |
| parser.add_argument("--max-steps", type=int, default=6) |
| parser.add_argument("--script", default=None, |
| help="JSON list of actions to replay instead of using a brain") |
| parser.add_argument("--doctor", action="store_true", |
| help="report what this machine can and cannot do, then exit") |
| parser.add_argument("--image", default=None, |
| help="replay a screenshot instead of the screen; " |
| f"'{EXAMPLE_IMAGE}' uses the built-in one") |
| parser.add_argument("--brain", default="planner", choices=BRAIN_NAMES, |
| help="planner (rule-based, no weights) | text-planner | vlm") |
| parser.add_argument("--detector", default="auto", choices=("auto", "ocr", "grid"), |
| help="how to build the element index (auto = OCR, grid fallback)") |
| parser.add_argument("--live", action="store_true", |
| help="capture your real screen instead of the scripted demo") |
| parser.add_argument("--real-input", action="store_true", |
| help="actually move the mouse and type (requires the allow env var)") |
| parser.add_argument("--monitor", type=int, default=1, help="1 is the primary monitor") |
| parser.add_argument("--scale", type=float, default=None, |
| help="screen pixels per captured pixel (HiDPI, secondary monitors)") |
| parser.add_argument("--no-ocr", action="store_true", |
| help="skip OCR and index a numbered grid instead") |
| parser.add_argument("--max-width", type=int, default=1600, |
| help="downscale captures wider than this before indexing") |
| parser.add_argument("--verify-text", default=None, metavar="TEXT", |
| help="stop with success once this text appears on screen " |
| "(otherwise a successful action that removes its own " |
| "target is indistinguishable from a failure)") |
| parser.add_argument("--allow-grid", action="store_true", |
| help="permit clicks on numbered grid cells with --real-input") |
| parser.add_argument("--kill-switch", default=DEFAULT_KILL_SWITCH, |
| help="create this file to abort the run before the next action") |
| return parser |
|
|
|
|
| def report(checks: list[Check]) -> int: |
| """Print a preflight report; non-zero when something required is missing.""" |
| blocking = [c for c in checks if c.required and not c.ok] |
| lines = [ |
| f"tiny-computer-agent preflight ({platform.system()} {platform.machine()}, " |
| f"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.") |
| print("\n".join(lines)) |
| return 1 if blocking else 0 |
|
|
|
|
| def _probe_live(args: argparse.Namespace) -> Check: |
| """Actually grab one frame, because only a real capture proves it works.""" |
| try: |
| observation = GrabScreen( |
| monitor=args.monitor, max_width=args.max_width, scale=args.scale |
| ).capture() |
| frame = observation.frame |
| return Check("live capture", True, f"captured {frame.width}x{frame.height} pixels") |
| except ScreenUnavailable as exc: |
| return Check("live capture", False, str(exc), |
| fix="run this on the desktop session itself, not a headless shell") |
|
|
|
|
| def _build_screen(args: argparse.Namespace): |
| """(screen, description). Raises ScreenUnavailable / OSError on failure.""" |
| detector_name = "grid" if args.no_ocr else args.detector |
| if args.image is not None: |
| source = ImageScreen(fake_desktop() if args.image == EXAMPLE_IMAGE else args.image) |
| return IndexedScreen(source, make_detector(detector_name)), f"screenshot replay: {args.image}" |
| if args.live: |
| source = GrabScreen(monitor=args.monitor, max_width=args.max_width, scale=args.scale) |
| return IndexedScreen(source, make_detector(detector_name)), f"real desktop (monitor {args.monitor})" |
| return demo_screen(), "scripted demo" |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| args = build_parser().parse_args(argv) |
| detector_name = "grid" if args.no_ocr else args.detector |
|
|
| if args.doctor: |
| checks = run_checks(brain=args.brain, detector=detector_name, |
| real_input=args.real_input, kill_switch_path=args.kill_switch) |
| if args.live: |
| checks.append(_probe_live(args)) |
| return report(checks) |
|
|
| if args.real_input and os.environ.get(ALLOW_REAL_INPUT_ENV) != "1": |
| print(f"refusing --real-input: set {ALLOW_REAL_INPUT_ENV}=1 to confirm you " |
| f"understand this will drive your actual mouse and keyboard.", file=sys.stderr) |
| return 2 |
|
|
| if args.brain == "vlm" and not (args.live or args.image): |
| print("the vlm brain needs --live (or --image) so it has pixels to look at; " |
| "use --brain planner to exercise the harness offline.", file=sys.stderr) |
| return 2 |
|
|
| if args.live: |
| available, detail = display_available() |
| if not available: |
| print(f"live capture needs a display session: {detail}") |
| print(f"refusing to run live: {detail}. Try --image to replay a screenshot, " |
| f"or --doctor to diagnose.", file=sys.stderr) |
| return 2 |
|
|
| try: |
| screen, description = _build_screen(args) |
| except ScreenUnavailable as exc: |
| print(f"cannot see the screen: {exc}", file=sys.stderr) |
| return 3 |
| except Exception as exc: |
| print(f"could not start the screen: {type(exc).__name__}: {exc}", file=sys.stderr) |
| return 2 |
|
|
| if args.script: |
| try: |
| script = json.loads(args.script) |
| except json.JSONDecodeError as exc: |
| print(f"--script is not valid JSON: {exc}", file=sys.stderr) |
| return 2 |
| brain = MockBrain(script) |
| elif args.brain == "planner": |
| brain = PlannerBrain() |
| else: |
| from tiny_computer_agent.models import make_model_brain |
|
|
| try: |
| brain = make_model_brain(args.brain) |
| except Exception as exc: |
| print(f"could not load the {args.brain} brain: {type(exc).__name__}: {exc}", |
| file=sys.stderr) |
| return 2 |
|
|
| policy = Policy( |
| dry_run=not args.real_input, |
| step_budget=args.max_steps, |
| kill_switch_path=args.kill_switch, |
| allow_grid_clicks=args.allow_grid, |
| ) |
| executor = PyAutoGuiExecutor() if args.real_input else DryRunExecutor() |
|
|
| mode = "REAL INPUT (mouse and keyboard live)" if args.real_input else "dry-run" |
| print(f"goal: {args.goal}") |
| print(f"mode: {mode} | screen: {description} | brain: {args.brain} | " |
| f"kill switch: create {args.kill_switch} to stop") |
| if args.image is not None or args.live: |
| _, index_detail = detector_available(detector_name) |
| print(f"index: {detector_name} - {index_detail}") |
| if args.image is not None: |
| |
| try: |
| preview = screen.capture() |
| except ScreenUnavailable as exc: |
| print(f"cannot see the screen: {exc}", file=sys.stderr) |
| return 3 |
| print(f"elements detected: {len(preview.elements)}") |
| print() |
|
|
| warned = False |
|
|
| verify = None |
| if args.verify_text: |
| _needle = args.verify_text.lower() |
|
|
| def verify(observation) -> bool: |
| return _needle in observation.text.lower() |
|
|
| def on_step(step: Step) -> None: |
| nonlocal warned |
| elements = step.observation.elements |
| if not warned and elements and all(e.kind == "grid" for e in elements): |
| warned = True |
| _, index_detail = detector_available(detector_name) |
| print(f"warning: no text was found on screen, so the index is " |
| f"{len(elements)} numbered grid cells and the agent is choosing " |
| f"blind ({index_detail})", file=sys.stderr) |
| print(" " + step.line()) |
|
|
| try: |
| episode = AgentLoop(screen=screen, brain=brain, executor=executor, |
| policy=policy, on_step=on_step, verify=verify).run(args.goal) |
| except ScreenUnavailable as exc: |
| print(f"cannot see the screen: {exc}", file=sys.stderr) |
| print("run `python -m tiny_computer_agent --doctor --live` for details.", file=sys.stderr) |
| return 3 |
|
|
| print() |
| print(episode.render()) |
| return 0 if episode.success else 1 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
