| """Turning pixels into a labelled element index. |
| |
| This is the module that makes a *tiny* model viable. The model never has to guess |
| where anything is, because OCR has already turned the screen into a short list of |
| labels with coordinates - and the model only has to pick one. |
| |
| Filters matter as much as the engine. Raw OCR returns dialog prose, fragmentary |
| debris and stray punctuation, and every extra line of noise is another chance for a |
| small model to pick the wrong element, so the index is cleaned aggressively. |
| |
| Everything here is optional at import time: a missing OCR engine raises a clear |
| :class:`DetectorUnavailable` instead of breaking the harness. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import dataclass, field |
| from typing import Any, Protocol |
|
|
| from tiny_computer_agent.observe import Element, Observation, ScreenSource |
|
|
|
|
| class DetectorUnavailable(RuntimeError): |
| """Raised when an element detector is requested but cannot run on this machine.""" |
|
|
|
|
| class Detector(Protocol): |
| """Turns one frame into labelled, clickable elements.""" |
|
|
| name: str |
|
|
| def detect(self, image: Any) -> tuple[Element, ...]: ... |
|
|
|
|
| @dataclass |
| class OcrDetector: |
| """Indexes on-screen text with RapidOCR: ONNX, CPU, offline after install. |
| |
| ``max_label_chars`` is the important knob. Screen text includes paragraphs and |
| titles that are not widgets, and a model shown them will happily click prose - |
| which is exactly what a 0.5B planner did when dialog text sat in its menu. Short |
| strings are much more likely to be buttons and menu items, so the default is |
| deliberately tight. |
| """ |
|
|
| min_score: float = 0.6 |
| max_label_chars: int = 24 |
| min_label_chars: int = 2 |
| name: str = "ocr" |
| _engine: Any = field(default=None, init=False, repr=False) |
|
|
| def _loaded(self) -> Any: |
| if self._engine is None: |
| try: |
| from rapidocr_onnxruntime import RapidOCR |
| except ImportError as exc: |
| raise DetectorUnavailable( |
| "no OCR engine available: pip install 'tiny-computer-agent[ocr]'" |
| ) from exc |
| self._engine = RapidOCR() |
| return self._engine |
|
|
| def detect(self, image: Any) -> tuple[Element, ...]: |
| import numpy as np |
|
|
| array = np.asarray(image.convert("RGB") if hasattr(image, "convert") else image) |
| result, _ = self._loaded()(array) |
| elements: list[Element] = [] |
| for box, text, score in result or []: |
| element = self._to_element(box, text, float(score)) |
| if element is not None: |
| elements.append(element) |
| return tuple(elements) |
|
|
| def _to_element(self, box: Any, text: Any, score: float) -> Element | None: |
| """Convert one OCR box to an Element, or None if it is not a usable label.""" |
| label = " ".join(str(text).split()) |
| if score < self.min_score: |
| return None |
| if not self.min_label_chars <= len(label) <= self.max_label_chars: |
| return None |
| if not any(character.isalnum() for character in label): |
| return None |
| xs = [float(point[0]) for point in box] |
| ys = [float(point[1]) for point in box] |
| x, y = int(min(xs)), int(min(ys)) |
| return Element( |
| label=label, kind="text", |
| x=x, y=y, |
| width=int(max(xs)) - x, height=int(max(ys)) - y, |
| ) |
|
|
|
|
| @dataclass |
| class IndexedScreen: |
| """Composes pixel capture with a detector, so ``capture()`` yields an index. |
| |
| This is the wiring that turns the harness from a scripted demo into something |
| that can act on a real desktop. |
| """ |
|
|
| source: ScreenSource |
| detector: Detector |
| fill_text: bool = True |
|
|
| def capture(self) -> Observation: |
| observation = self.source.capture() |
| try: |
| elements = self.detector.detect(observation.frame) |
| except DetectorUnavailable: |
| raise |
| except Exception as exc: |
| |
| elements = () |
| observation = Observation( |
| index=observation.index, frame=observation.frame, |
| elements=(), text=f"perception error: {type(exc).__name__}: {exc}", |
| ) |
| return observation |
| mapper = getattr(self.source, "map_to_screen", None) |
| |
| |
| |
| |
| space = "frame" |
| hook = getattr(self.detector, "coordinates_of_last_detect", None) |
| if callable(hook): |
| space = hook() |
| if mapper is not None and space != "screen": |
| |
| elements = tuple(mapper(element) for element in elements) |
| text = observation.text |
| if self.fill_text and not text: |
| |
| text = " ".join(element.label for element in elements if element.kind != "grid") |
| return Observation( |
| index=observation.index, frame=observation.frame, elements=elements, text=text, |
| ) |
|
|
|
|
| @dataclass |
| class GridDetector: |
| """A numbered grid over the frame: click targets that need nothing installed. |
| |
| The guarantee that matters. OCR finds text, the accessibility tree finds widgets, |
| but a canvas app, a game, a photo, a custom-drawn toolbar - anything with no text |
| - leaves a text index empty, and an agent with an empty index has *no legal |
| moves*. A grid always has moves, and a numbered menu is something even a 256M |
| model can choose from reliably. |
| """ |
|
|
| rows: int = 8 |
| cols: int = 12 |
| name: str = "grid" |
|
|
| def __post_init__(self) -> None: |
| if self.rows < 1 or self.cols < 1: |
| raise ValueError("rows and cols must be at least 1") |
|
|
| def detect(self, image: Any) -> tuple[Element, ...]: |
| width, height = image.size |
| cell_w = width / self.cols |
| cell_h = height / self.rows |
| elements = [] |
| for row in range(self.rows): |
| for col in range(self.cols): |
| elements.append(Element( |
| label=f"cell {row * self.cols + col + 1}", |
| kind="grid", |
| x=round(col * cell_w), y=round(row * cell_h), |
| width=max(1, round(cell_w)), height=max(1, round(cell_h)), |
| )) |
| return tuple(elements) |
|
|
|
|
| @dataclass |
| class AutoDetector: |
| """OCR when it finds text, numbered cells when it does not. |
| |
| This is the default, so the harness is useful on a machine with nothing installed |
| beyond Pillow - and gets better automatically once an OCR engine appears. |
| """ |
|
|
| name: str = "auto" |
| ocr: Detector = field(default_factory=OcrDetector) |
| grid: Detector = field(default_factory=GridDetector) |
| |
| |
| accessibility: Detector | None = None |
|
|
| def __post_init__(self) -> None: |
| self._last_space = "frame" |
|
|
| def coordinates_of_last_detect(self) -> str: |
| """Whatever the branch that actually produced the last index spoke.""" |
| return self._last_space |
|
|
| def detect(self, image: Any) -> tuple[Element, ...]: |
| if self.accessibility is not None: |
| try: |
| elements = self.accessibility.detect(image) |
| except DetectorUnavailable: |
| elements = () |
| if elements: |
| self._last_space = "screen" |
| return elements |
| self._last_space = "frame" |
| try: |
| elements = self.ocr.detect(image) |
| except DetectorUnavailable: |
| return self.grid.detect(image) |
| return elements or self.grid.detect(image) |
|
|
|
|
| def _ocr_available() -> tuple[bool, str]: |
| try: |
| import rapidocr_onnxruntime |
| except ImportError: |
| return False, "rapidocr-onnxruntime is not installed (pip install 'tiny-computer-agent[ocr]')" |
| try: |
| import onnxruntime |
| except ImportError: |
| return False, "onnxruntime is not installed" |
| return True, "rapidocr-onnxruntime ready (CPU, offline)" |
|
|
|
|
| def detector_available(name: str = "auto") -> tuple[bool, str]: |
| """(available, detail) for preflight and the ``doctor`` command.""" |
| if name == "grid": |
| return True, "numbered grid cells (no dependencies)" |
| if name in ("atspi", "uia"): |
| from tiny_computer_agent.accessibility import accessibility_available |
|
|
| return accessibility_available(name) |
| if name == "best": |
| from tiny_computer_agent.accessibility import accessibility_available |
|
|
| for backend in ("atspi", "uia"): |
| reachable, detail = accessibility_available(backend) |
| if reachable: |
| return True, (f"{backend} tree when it has content, then OCR, then a " |
| f"numbered grid ({detail})") |
| return True, (f"OCR, then a numbered grid (no accessibility tree: {detail})") |
| if name == "auto": |
| available, detail = _ocr_available() |
| if available: |
| return True, f"OCR when it finds text, grid otherwise ({detail})" |
| return True, "grid only: no OCR engine installed (pip install 'tiny-computer-agent[ocr]')" |
| if name != "ocr": |
| return False, f"unknown detector {name!r}" |
| return _ocr_available() |
|
|
|
|
| def make_detector(name: str = "auto", **kwargs: Any) -> Detector: |
| """Build a detector by name, with an actionable error when it cannot run.""" |
| available, detail = detector_available(name) |
| if not available: |
| raise DetectorUnavailable(detail) |
| if name == "grid": |
| return GridDetector(**kwargs) |
| if name == "atspi": |
| from tiny_computer_agent.accessibility import AtspiDetector |
|
|
| return AtspiDetector(**kwargs) |
| if name == "uia": |
| from tiny_computer_agent.accessibility import UiaDetector |
|
|
| return UiaDetector(**kwargs) |
| if name == "best": |
| from tiny_computer_agent.accessibility import AtspiDetector, accessibility_available |
|
|
| from tiny_computer_agent.accessibility import AtspiDetector, UiaDetector |
|
|
| accessibility = None |
| if accessibility_available("atspi")[0]: |
| accessibility = AtspiDetector() |
| elif accessibility_available("uia")[0]: |
| accessibility = UiaDetector() |
| return AutoDetector(accessibility=accessibility, **kwargs) |
| if name == "auto": |
| if _ocr_available()[0]: |
| return AutoDetector(**kwargs) |
| return GridDetector(**kwargs) |
| return OcrDetector(**kwargs) |
|
|
