token/vm/accessibility.py

"""Bounded AT-SPI collection runs in a disposable process to contain D-Bus stalls."""
import json
import subprocess
import sys
from pathlib import Path


def collect(width, height):
    try:
        result = subprocess.run(
            [sys.executable, str(Path(__file__).resolve()), str(width), str(height)],
            capture_output=True, text=True, timeout=3, check=True,
        )
        return json.loads(result.stdout)
    except (subprocess.SubprocessError, ValueError) as error:
        return {"targets": [], "warning": f"Accessibility unavailable ({type(error).__name__})."}


def resolve_target(saved, current, observation_id, target_id, width, height):
    if saved.get("observationId") != observation_id or saved.get("size") != [width, height]:
        raise ValueError("Stale target observation; capture again")
    target = next((item for item in saved["targets"] if item["id"] == target_id), None)
    if target is None:
        raise ValueError("Unknown target; capture again")
    matching = next((item for item in current["targets"] if item["path"] == target["path"]), None)
    if matching is None or any(matching[key] != target[key] for key in ("name", "role", "bounds", "app")):
        raise ValueError("Target changed or disappeared; capture again")
    x, y, w, h = target["bounds"]
    return x + w // 2, y + h // 2


def badge_position(bounds, badge_width, badge_height, image_width, image_height):
    x, y, w, h = bounds
    # Keep row labels within their own row, never above it in the previous item.
    left = min(x + 2, max(0, image_width - badge_width))
    top = min(y + max(0, (h - badge_height) // 2), max(0, image_height - badge_height))
    return left, top


def annotate(image, targets):
    from PIL import ImageDraw, ImageFont
    marked = image.copy()
    draw = ImageDraw.Draw(marked)
    font = ImageFont.load_default(size=14)
    for target in targets:
        x, y, w, h = target["bounds"]
        draw.rectangle((x, y, x + w - 1, y + h - 1), outline="#ff00cc", width=2)
    # Draw badges last so another element's outline cannot cross out a number.
    for target in targets:
        label = str(target["id"])
        box = draw.textbbox((0, 0), label, font=font)
        label_width, label_height = box[2] - box[0] + 6, box[3] - box[1] + 6
        left, top = badge_position(target["bounds"], label_width, label_height, image.width, image.height)
        draw.rectangle((left, top, left + label_width - 1, top + label_height - 1), fill="#ffff00", outline="black")
        draw.text((left + 3 - box[0], top + 3 - box[1]), label, fill="black", font=font)
    return marked


def scan(width, height):
    import pyatspi
    desktop = pyatspi.Registry.getDesktop(0)
    targets = []
    visited = 0
    truncated = False

    def walk(node, path, app, depth):
        nonlocal visited, truncated
        visited += 1
        if visited > 1500 or len(targets) >= 100 or depth > 30:
            truncated = True
            return
        try:
            state = node.getState()
            if not state.contains(pyatspi.STATE_SHOWING):
                return
            role = node.getRoleName()
            actionable = state.contains(pyatspi.STATE_FOCUSABLE)
            try:
                actionable = actionable or node.queryAction().nActions > 0
            except NotImplementedError:
                pass
            if actionable and state.contains(pyatspi.STATE_ENABLED):
                rect = node.queryComponent().getExtents(pyatspi.DESKTOP_COORDS)
                x, y = max(0, rect.x), max(0, rect.y)
                right, bottom = min(width, rect.x + rect.width), min(height, rect.y + rect.height)
                if right > x and bottom > y:
                    states = []
                    for name, constant in (
                        ("focused", pyatspi.STATE_FOCUSED),
                        ("editable", pyatspi.STATE_EDITABLE),
                        ("enabled", pyatspi.STATE_ENABLED),
                        ("checked", pyatspi.STATE_CHECKED),
                        ("active", pyatspi.STATE_ACTIVE),
                        ("selected", pyatspi.STATE_SELECTED),
                        ("selectable", pyatspi.STATE_SELECTABLE),
                        ("expanded", pyatspi.STATE_EXPANDED),
                        ("collapsed", pyatspi.STATE_COLLAPSED),
                        ("busy", pyatspi.STATE_BUSY),
                    ):
                        try:
                            if state.contains(constant):
                                states.append(name)
                        except Exception:
                            pass
                    value = None
                    try:
                        text_val = node.queryText().getText(0, -1)
                        if text_val is not None:
                            value = text_val[:200]
                    except Exception:
                        pass
                    if value is None:
                        try:
                            value = str(node.queryValue().currentValue)
                        except Exception:
                            pass
                    target = {
                        "id": str(len(targets) + 1), "path": path, "app": app,
                        "name": (node.name or "")[:200], "role": role,
                        "bounds": [x, y, right - x, bottom - y], "states": states,
                    }
                    if value is not None and (value != target["name"] or "editable" in states):
                        target["value"] = value
                    targets.append(target)
            for index in range(min(node.childCount, 1500)):
                if visited >= 1500 or len(targets) >= 100:
                    truncated = True
                    break
                walk(node[index], path + [index], app, depth + 1)
        except Exception:
            # Individual applications can disappear or expose incomplete interfaces.
            return

    # Restrict marks to active windows, avoiding targets in covered background windows.
    for app_index in range(min(desktop.childCount, 100)):
        try:
            app = desktop[app_index]
            for window_index in range(min(app.childCount, 100)):
                window = app[window_index]
                if window.getState().contains(pyatspi.STATE_ACTIVE):
                    walk(window, [app_index, window_index], app.name or "", 0)
        except Exception:
            continue
    warning = "Accessibility target list truncated." if truncated else None
    if not targets:
        warning = "No actionable targets in an active accessible window."
    return {"targets": targets, "warning": warning}


if __name__ == "__main__":
    print(json.dumps(scan(int(sys.argv[1]), int(sys.argv[2]))))