"""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: # pragma: no cover - depends on the machine 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: # A perception failure must not kill a run that is otherwise fine. 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) # Which space did *this call* produce? A detector that chains several # sources (the "best" detector: accessibility tree, then OCR, then grid) # cannot answer that with a static flag, and answering it wrongly shifts # every click by the capture's scale factor. space = "frame" hook = getattr(self.detector, "coordinates_of_last_detect", None) if callable(hook): space = hook() if mapper is not None and space != "screen": # Frame pixels -> absolute screen pixels, so a click cannot drift. elements = tuple(mapper(element) for element in elements) text = observation.text if self.fill_text and not text: # Grid labels are coordinates, not visible text, so they are not page 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) # opt in (the "best" detector does): the accessibility tree is tried first, # because it is exact where OCR is a guess - and skipped when it is empty 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 # noqa: F401 except ImportError: return False, "rapidocr-onnxruntime is not installed (pip install 'tiny-computer-agent[ocr]')" try: import onnxruntime # noqa: F401 except ImportError: # pragma: no cover 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"): # Linux first, then Windows 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)