| """Elements from the operating system's accessibility tree. |
| |
| OCR reads pixels and *guesses* which marks are widgets. The accessibility tree |
| simply knows: every application that publishes one advertises each button, entry, |
| menu item and label with a role, a name, and its real screen extents. That is |
| faster than OCR, exact on small text the pixels lose, and - the part that matters |
| for a tiny model - it produces the same short, labelled menu the rest of the |
| harness already expects. |
| |
| The catch is honest and load-bearing: not every application publishes a tree. |
| GTK/Qt/Electron and most native Windows apps do; Tk barely does, and a raw canvas |
| does not at all. So this is a detector, not a replacement - when the tree is |
| empty, `AutoDetector` falls through to OCR, and then to the numbered grid. |
| |
| Backends: |
| |
| * **AT-SPI** (Linux) - via PyGObject's ``Atspi`` typelib, implemented here and |
| exercised against a real GTK window. |
| * **UI Automation** (Windows) - *not implemented*: :func:`accessibility_available` |
| reports it plainly rather than pretending, so an agent on Windows keeps using |
| OCR instead of silently indexing nothing. |
| |
| The coordinates AT-SPI reports are already absolute screen pixels, which is why |
| :class:`AtspiDetector` sets ``absolute_coordinates``: :class:`IndexedScreen` must |
| not re-map them through a downscale, or every click would land in the wrong place. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import dataclass, field |
| from typing import Any, Protocol, Sequence |
|
|
| from tiny_computer_agent.observe import Element |
|
|
| |
| |
| |
| INTERACTIVE_ROLES: frozenset[str] = frozenset({ |
| "button", "push button", "toggle button", "check box", "radio button", |
| "menu item", "check menu item", "radio menu item", "page tab", "list item", |
| "combo box", "entry", "text", "password text", "spin button", "slider", |
| "link", "icon", "table cell", "tree item", "tool bar", "menu", |
| }) |
|
|
| |
| |
| TEXT_ROLES: frozenset[str] = frozenset({ |
| "label", "paragraph", "heading", "static", "status bar", "text", |
| }) |
|
|
|
|
| class AccessibleNode: |
| """One node of an accessibility tree, reduced to what a decision needs.""" |
|
|
| __slots__ = ("role", "name", "x", "y", "width", "height", "depth") |
|
|
| def __init__(self, role: str, name: str, x: int, y: int, width: int, height: int, |
| depth: int = 0) -> None: |
| self.role = role |
| self.name = name |
| self.x = x |
| self.y = y |
| self.width = width |
| self.height = height |
| self.depth = depth |
|
|
| def __repr__(self) -> str: |
| return (f"AccessibleNode({self.role!r}, {self.name!r}, " |
| f"{self.x},{self.y} {self.width}x{self.height})") |
|
|
| def __eq__(self, other: object) -> bool: |
| return isinstance(other, AccessibleNode) and self.__slots__ and all( |
| getattr(self, slot) == getattr(other, slot) for slot in self.__slots__ |
| ) |
|
|
| def __hash__(self) -> int: |
| return hash((self.role, self.name, self.x, self.y, self.width, self.height)) |
|
|
|
|
| class TreeReader(Protocol): |
| """Reads the accessibility tree of the whole desktop.""" |
|
|
| name: str |
|
|
| def read(self) -> tuple[AccessibleNode, ...]: ... |
|
|
|
|
| class AtspiReader: |
| """Reads the live AT-SPI desktop through PyGObject. |
| |
| Everything here is guarded: a machine without the typelib, without a session |
| bus, or without the accessibility bridge raises :class:`DetectorUnavailable` |
| with the fix, rather than returning an empty index that would look like "the |
| screen is empty" when it is really "the bridge is off". |
| """ |
|
|
| name = "atspi" |
|
|
| def __init__(self, *, desktop: int = 0, max_depth: int = 10, |
| max_nodes: int = 600, timeout_ms: int = 500) -> None: |
| self.desktop = desktop |
| self.max_depth = max_depth |
| self.max_nodes = max_nodes |
| self.timeout_ms = timeout_ms |
|
|
| @staticmethod |
| def _atspi(): |
| try: |
| import gi |
| except ImportError as exc: |
| raise _unavailable( |
| "the AT-SPI reader needs PyGObject: pip install PyGObject, and " |
| "apt install gir1.2-atspi-2.0 at-spi2-core" |
| ) from exc |
| try: |
| gi.require_version("Atspi", "2.0") |
| from gi.repository import Atspi |
| except Exception as exc: |
| raise _unavailable( |
| "the Atspi typelib is missing: apt install gir1.2-atspi-2.0 " |
| "at-spi2-core (and run inside a session bus)" |
| ) from exc |
| return Atspi |
|
|
| def read(self) -> tuple[AccessibleNode, ...]: |
| Atspi = self._atspi() |
| try: |
| desktop = Atspi.get_desktop(self.desktop) |
| except Exception as exc: |
| raise _unavailable( |
| f"could not reach the accessibility bus: {type(exc).__name__}: {exc}. " |
| "Start a session bus (dbus-run-session) and leave NO_AT_BRIDGE unset." |
| ) from exc |
|
|
| nodes: list[AccessibleNode] = [] |
| for index in range(desktop.get_child_count()): |
| self._walk(Atspi, desktop.get_child_at_index(index), 0, nodes) |
| if len(nodes) >= self.max_nodes: |
| break |
| return tuple(nodes) |
|
|
| def _walk(self, Atspi, node: Any, depth: int, out: list[AccessibleNode]) -> None: |
| if depth > self.max_depth or len(out) >= self.max_nodes: |
| return |
| try: |
| role = (node.get_role_name() or "").strip() |
| name = " ".join((node.get_name() or "").split()) |
| except Exception: |
| return |
| extents = self._extents(Atspi, node) |
| if extents is not None: |
| x, y, width, height = extents |
| out.append(AccessibleNode(role, name, x, y, width, height, depth)) |
| try: |
| children = node.get_child_count() |
| except Exception: |
| children = 0 |
| for index in range(children): |
| try: |
| child = node.get_child_at_index(index) |
| except Exception: |
| continue |
| self._walk(Atspi, child, depth + 1, out) |
|
|
| def _extents(self, Atspi, node: Any) -> tuple[int, int, int, int] | None: |
| try: |
| component = node.get_component_iface() |
| if component is None: |
| return None |
| box = component.get_extents(Atspi.CoordType.SCREEN) |
| except Exception: |
| return None |
| if box is None: |
| return None |
| return int(box.x), int(box.y), int(box.width), int(box.height) |
|
|
|
|
| def _unavailable(detail: str): |
| from tiny_computer_agent.perception import DetectorUnavailable |
| return DetectorUnavailable(detail) |
|
|
|
|
| @dataclass |
| class AtspiDetector: |
| """Turns the accessibility tree into the harness's element index. |
| |
| ``kind`` is the AT-SPI role (``button``, ``check box``, ...), so a policy can |
| distinguish a button from a text field, and ``absolute_coordinates`` tells |
| :class:`IndexedScreen` these are already screen pixels. |
| """ |
|
|
| name: str = "atspi" |
| absolute_coordinates: bool = True |
| reader: TreeReader | None = field(default=None) |
| interactive_roles: frozenset[str] = INTERACTIVE_ROLES |
| include_text: bool = False |
| min_label_chars: int = 1 |
| max_label_chars: int = 64 |
| min_size: int = 4 |
| max_nodes: int = 120 |
| _last_roles: tuple[str, ...] = field(default=(), init=False, repr=False) |
|
|
| def detect(self, image: Any = None) -> tuple[Element, ...]: |
| """Index the live tree. The image is accepted and deliberately ignored.""" |
| reader = self.reader if self.reader is not None else AtspiReader() |
| nodes = reader.read() |
| wanted = self.interactive_roles | (TEXT_ROLES if self.include_text else frozenset()) |
| elements: list[Element] = [] |
| seen: set[tuple[str, int, int]] = set() |
| for node in nodes: |
| role = node.role.lower() |
| if role not in wanted: |
| continue |
| label = node.name |
| if not label or not (self.min_label_chars <= len(label) <= self.max_label_chars): |
| continue |
| if node.width < self.min_size or node.height < self.min_size: |
| continue |
| if node.x < 0 or node.y < 0: |
| continue |
| key = (label.lower(), node.x, node.y) |
| if key in seen: |
| continue |
| seen.add(key) |
| elements.append(Element(label=label, kind=role, x=node.x, y=node.y, |
| width=node.width, height=node.height)) |
| if len(elements) >= self.max_nodes: |
| break |
| self._last_roles = tuple(element.kind for element in elements) |
| return tuple(elements) |
|
|
| def text(self) -> str: |
| """The words on screen, for a verifier - straight from the tree.""" |
| reader = self.reader if self.reader is not None else AtspiReader() |
| return " ".join(node.name for node in reader.read() |
| if node.name and node.role.lower() in (TEXT_ROLES | self.interactive_roles)) |
|
|
|
|
| def accessibility_available(backend: str = "atspi") -> tuple[bool, str]: |
| """(available, detail) for preflight and the doctor, honestly reported.""" |
| if backend == "uia": |
| return False, ("UI Automation is not implemented yet: on Windows the harness " |
| "falls back to OCR (see accessibility.py)") |
| if backend != "atspi": |
| return False, f"unknown accessibility backend {backend!r}" |
| try: |
| Atspi = AtspiReader._atspi() |
| except Exception as exc: |
| return False, str(exc) |
| try: |
| desktop = Atspi.get_desktop(0) |
| applications = desktop.get_child_count() |
| except Exception as exc: |
| return False, f"Atspi is installed but the a11y bus is unreachable: {exc}" |
| return True, f"AT-SPI via PyGObject ({applications} application(s) on the desktop)" |
|
|
|
|
| __all__ = [ |
| "INTERACTIVE_ROLES", |
| "TEXT_ROLES", |
| "AccessibleNode", |
| "AtspiDetector", |
| "AtspiReader", |
| "TreeReader", |
| "accessibility_available", |
| ] |
|
|
