AnyBot/box/a11y

#!/usr/bin/env python3
"""The controls on the box's desktop, from its accessibility tree (AT-SPI), as JSON:
{"elements": [{"role", "name", "value", "x", "y", "w", "h", "active", "flags", "window"}]}
in screen pixels. Web pages are in it too (Chromium runs with renderer accessibility on).
AnyBot numbers them on the bot's picture; a click by number lands on the centre.

Adapted from XBot's box/box-a11y (which skips anything not "showing"; Chromium can say that of
its whole window). Usage: a11y [--limit N] [--budget SECONDS]
Exit 2 when AT-SPI itself is unreachable.
"""
import json
import os
import sys
import time
import warnings

warnings.filterwarnings("ignore", category=DeprecationWarning)

# Clicked or typed into (they get a number).
ACTIVE = {
    "push button", "toggle button", "radio button", "check box", "link", "menu item", "check menu item",
    "radio menu item", "tab", "page tab", "list item", "tree item", "combo box", "spin button", "slider",
    "entry", "password text", "menu",
}
# Listed when they carry words (headings, labels, cells...).
NAMED = {"heading", "label", "static", "text", "image", "table cell", "column header", "row header", "icon",
         "tool tip", "alert", "dialog", "paragraph"}
# The desktop's own parts are visible in the picture; the list is for the windows worked in.
SKIP_APPS = {"fluxbox", "a11y", "python3", "xterm-wrapper"}


def stacked_windows():
    """The desktop's shown windows, bottom to top: (x, y, w, h, pid)."""
    out = []
    try:
        ids = os.popen("xprop -root _NET_CLIENT_LIST_STACKING 2>/dev/null").read().split("#")[-1]
        for wid in [w.strip() for w in ids.split(",") if w.strip().startswith("0x")]:
            info = os.popen(f"xwininfo -id {wid} 2>/dev/null").read()
            if "IsViewable" not in info:
                continue
            num = lambda k: int(info.split(k)[1].split()[0])  # noqa: E731
            pid = os.popen(f"xprop -id {wid} _NET_WM_PID 2>/dev/null").read().split("=")[-1].strip()
            out.append((num("Absolute upper-left X:"), num("Absolute upper-left Y:"), num("Width:"), num("Height:"),
                        int(pid) if pid.isdigit() else 0))
    except Exception:  # noqa: BLE001
        pass
    return out


def main():
    limit, budget = 400, 3.0
    args = sys.argv[1:]
    for i, a in enumerate(args):
        if a == "--limit":
            limit = int(args[i + 1])
        if a == "--budget":
            budget = float(args[i + 1])
    if not os.environ.get("DBUS_SESSION_BUS_ADDRESS"):
        try:
            for line in open("/tmp/anybot-dbus.env", encoding="utf-8"):
                k, _, v = line.strip().partition("=")
                if k == "DBUS_SESSION_BUS_ADDRESS" and v:
                    os.environ[k] = v
        except OSError:
            pass
    os.environ.setdefault("DISPLAY", ":1")
    try:
        import gi
        gi.require_version("Atspi", "2.0")
        from gi.repository import Atspi
        Atspi.init()
        desktop = Atspi.get_desktop(0)
        apps = desktop.get_child_count()
    except Exception as exc:  # noqa: BLE001
        print(f"AT-SPI unreachable: {exc}", file=sys.stderr)
        return 2

    sw, sh = 0, 0
    try:
        geo = os.popen("xdotool getdisplaygeometry 2>/dev/null").read().split()
        sw, sh = int(geo[0]), int(geo[1])
    except Exception:  # noqa: BLE001
        pass
    deadline = time.monotonic() + budget
    out = []
    windows = stacked_windows()

    def state(node):
        try:
            return node.get_state_set()
        except Exception:  # noqa: BLE001
            return None

    def extents(node):
        try:
            r = node.get_component_iface().get_extents(Atspi.CoordType.SCREEN)
            return int(r.x), int(r.y), int(r.width), int(r.height)
        except Exception:  # noqa: BLE001
            return None

    def cut(b, clip):
        """The part of box b inside clip (both x, y, w, h), or None."""
        x1, y1 = max(b[0], clip[0]), max(b[1], clip[1])
        x2, y2 = min(b[0] + b[2], clip[0] + clip[2]), min(b[1] + b[3], clip[1] + clip[3])
        return (x1, y1, x2 - x1, y2 - y1) if x2 > x1 and y2 > y1 else None

    def covered(cx, cy, pid):
        """Another app's window is the top one at this spot (a terminal over the browser)."""
        for (x, y, w, h, wpid) in reversed(windows):
            if x <= cx < x + w and y <= cy < y + h:
                return bool(pid) and bool(wpid) and wpid != pid
        return False

    def walk(node, depth, window, clip, pid):
        if time.monotonic() > deadline or len(out) >= limit or depth > 45:
            return
        try:
            role = node.get_role_name()
        except Exception:  # noqa: BLE001
            return
        st = state(node)
        showing = bool(st and st.contains(Atspi.StateType.SHOWING))
        visible = bool(st and st.contains(Atspi.StateType.VISIBLE))
        # A page that isn't showing is a tab in the background: none of it is on the screen.
        if role in ("document web", "document frame") and not showing:
            return
        # Otherwise "showing" isn't trusted: Chromium can call its whole window not showing (it
        # took itself for minimized under this window manager) while its page is on the screen.
        # Visible with a place on the screen is what counts; containers are walked through.
        box = extents(node)
        # What is really seen of it: inside the screen, its window and the page's visible area
        # (a link scrolled up under the toolbar was numbered over the Back button).
        seen = cut(box, clip) if box is not None and box[2] > 0 and box[3] > 0 else None
        if depth > 1 and (not visible or seen is None):
            return
        if seen is not None and (role in ("frame", "dialog", "window") or role in ("document web", "document frame")):
            clip = seen
        try:
            name = (node.get_name() or "").strip()
        except Exception:  # noqa: BLE001
            name = ""
        if role in ("frame", "dialog", "window") and name:
            window = name
        editable = bool(st and st.contains(Atspi.StateType.EDITABLE))
        value = ""
        if (role in ("entry", "password text", "text", "spin button", "slider") or editable) and role != "password text":
            try:
                t = node.get_text_iface()
                if t is not None:
                    value = (t.get_text(0, min(t.get_character_count(), 200)) or "").strip()
            except Exception:  # noqa: BLE001
                pass
        active = role in ACTIVE or (editable and role in ("text", "paragraph", "section", "document web"))
        if depth > 1 and seen is not None and (active or (role in NAMED and (name or value))) \
                and not covered(seen[0] + seen[2] // 2, seen[1] + seen[3] // 2, pid):
            flags = []
            if st is not None:
                for s, word in ((Atspi.StateType.FOCUSED, "focused"), (Atspi.StateType.CHECKED, "checked"),
                                (Atspi.StateType.SELECTED, "selected")):
                    if st.contains(s):
                        flags.append(word)
                if not st.contains(Atspi.StateType.ENABLED):
                    flags.append("disabled")
            x, y, w, h = seen
            out.append({"role": role, "name": name[:120], "value": value[:200], "x": x, "y": y, "w": w, "h": h,
                        "active": active, "flags": flags, "window": window})
        try:
            n = node.get_child_count()
        except Exception:  # noqa: BLE001
            return
        for i in range(min(n, 2000)):
            try:
                c = node.get_child_at_index(i)
            except Exception:  # noqa: BLE001
                continue
            if c is not None:
                walk(c, depth + 1, window, clip, pid)

    for i in range(apps):
        try:
            app = desktop.get_child_at_index(i)
            name = app.get_name() or ""
        except Exception:  # noqa: BLE001
            continue
        if name in SKIP_APPS:
            continue
        try:
            pid = app.get_process_id()
        except Exception:  # noqa: BLE001
            pid = 0
        walk(app, 0, name, (0, 0, sw or 100000, sh or 100000), pid)
    json.dump({"elements": out, "truncated": len(out) >= limit or time.monotonic() > deadline}, sys.stdout, ensure_ascii=False)
    return 0


if __name__ == "__main__":
    sys.exit(main())