From 998b4b6c1dc19fe0b68c2b6a0ad4e60c0fedf822 Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Thu, 25 Jun 2026 06:35:50 +0800 Subject: [PATCH] Add ax_events: reactive UIA focus-change wait The accessibility recorder polls focus every ~250ms, missing fast transitions and reacting late. wait_for_focus_change blocks on the native AddFocusChangedEventHandler and returns the moment focus moves - zero-latency, miss-free, the accessibility-tree analogue of wait_for_window. The real event handler is registered/unregistered under a lock on the calling thread (COMObject + queue). Extends the backend ABC + Windows UIA backend via the fake-backend seam. --- WHATS_NEW.md | 6 ++ .../doc/new_features/v202_features_doc.rst | 37 +++++++++ .../Zh/doc/new_features/v202_features_doc.rst | 34 ++++++++ je_auto_control/__init__.py | 3 + .../gui/script_builder/command_schema.py | 6 ++ .../utils/accessibility/backends/base.py | 12 +++ .../accessibility/backends/windows_backend.py | 44 +++++++++++ je_auto_control/utils/ax_events/__init__.py | 4 + je_auto_control/utils/ax_events/ax_events.py | 25 ++++++ .../utils/executor/action_executor.py | 8 ++ .../utils/mcp_server/tools/_factories.py | 10 +++ .../utils/mcp_server/tools/_handlers.py | 6 ++ .../headless/test_ax_events_batch.py | 77 +++++++++++++++++++ 13 files changed, 272 insertions(+) create mode 100644 docs/source/Eng/doc/new_features/v202_features_doc.rst create mode 100644 docs/source/Zh/doc/new_features/v202_features_doc.rst create mode 100644 je_auto_control/utils/ax_events/__init__.py create mode 100644 je_auto_control/utils/ax_events/ax_events.py create mode 100644 test/unit_test/headless/test_ax_events_batch.py diff --git a/WHATS_NEW.md b/WHATS_NEW.md index 9b6390ec..c595e324 100644 --- a/WHATS_NEW.md +++ b/WHATS_NEW.md @@ -1,5 +1,11 @@ # What's New — AutoControl +## What's new (2026-06-25) — Reactive UIA Event Wait (focus change) + +Wait until focus lands on the dialog — a real, zero-latency UIA event, not polling. Full reference: [`docs/source/Eng/doc/new_features/v202_features_doc.rst`](docs/source/Eng/doc/new_features/v202_features_doc.rst). + +- **`wait_for_focus_change`** (`AC_wait_for_focus_change`): the accessibility recorder *polls* focus every ~250 ms, so it can miss a fast transition and reacts late. This blocks on the native `AddFocusChangedEventHandler` and returns the moment focus moves — the zero-latency, miss-free "wait until focus lands on the dialog" primitive, the accessibility-tree analogue of `wait_for_window` / `wait_for_image`. Returns the newly-focused element (or `None` on timeout). The real event subscription is registered/unregistered under a lock on the calling thread; dispatched through the injectable accessibility backend seam (headless-testable via a fake backend; real UIA in the Windows backend). No `PySide6`. + ## What's new (2026-06-25) — Container Selection + View Switching (Selection / MultipleView) Read what's selected in a listbox/grid, and switch Explorer-style views. Full reference: [`docs/source/Eng/doc/new_features/v201_features_doc.rst`](docs/source/Eng/doc/new_features/v201_features_doc.rst). diff --git a/docs/source/Eng/doc/new_features/v202_features_doc.rst b/docs/source/Eng/doc/new_features/v202_features_doc.rst new file mode 100644 index 00000000..12140666 --- /dev/null +++ b/docs/source/Eng/doc/new_features/v202_features_doc.rst @@ -0,0 +1,37 @@ +Reactive UIA Event Wait — Focus Change +====================================== + +The accessibility recorder *polls* the focused element every ~250 ms, so it can +miss a fast focus transition and reacts a quarter-second late. UIA exposes real +events: ``wait_for_focus_change`` blocks on the native +``AddFocusChangedEventHandler`` and returns the moment focus moves — the +zero-latency, miss-free "wait until focus lands on the dialog" primitive, the +accessibility-tree analogue of ``wait_for_window`` / ``wait_for_image``. + +It is a thin dispatch onto the injectable ``accessibility.backends.get_backend()`` +seam — headless-testable on any platform by injecting a fake backend; the real +event subscription (registered / unregistered under a lock, on the calling +thread) lives in the Windows backend. Imports no ``PySide6``. + +Headless API +------------ + +.. code-block:: python + + from je_auto_control import wait_for_focus_change, click_text + + click_text("Settings") + focused = wait_for_focus_change(timeout=3) + # {"name": "Search", "role": "ControlType_50004", "app_name": "app.exe", ...} + if focused is not None: + ... # focus has moved — the dialog / next field is ready + +Returns the newly-focused element as ``{name, role, app_name, bounds, …}``, or +``None`` if no focus change occurs within ``timeout`` seconds (default ``5``). + +Executor commands +----------------- + +``AC_wait_for_focus_change`` (``timeout``) returns ``{changed, element}``. It is +exposed as the read-only ``ac_wait_for_focus_change`` MCP tool and as a Script +Builder command under **Native UI**. diff --git a/docs/source/Zh/doc/new_features/v202_features_doc.rst b/docs/source/Zh/doc/new_features/v202_features_doc.rst new file mode 100644 index 00000000..22974c29 --- /dev/null +++ b/docs/source/Zh/doc/new_features/v202_features_doc.rst @@ -0,0 +1,34 @@ +反應式 UIA 事件等待——焦點變化 +============================== + +無障礙錄製器每約 250 毫秒*輪詢*一次焦點元素,因此可能錯過快速的焦點轉換,且慢上四分之一秒才反應。 +UIA 提供真正的事件:``wait_for_focus_change`` 阻塞於原生的 ``AddFocusChangedEventHandler``,並在 +焦點移動的當下回傳——這是零延遲、不漏失的「等到焦點落在對話框上」原語,是 ``wait_for_window`` / +``wait_for_image`` 在無障礙樹上的對應。 + +它是對可注入的 ``accessibility.backends.get_backend()`` 接縫的薄分派——可在任何平台透過注入 fake +backend 進行無頭測試;真正的事件訂閱(在呼叫執行緒上、以鎖註冊 / 取消註冊)位於 Windows 後端。 +不匯入 ``PySide6``。 + +無頭 API +-------- + +.. code-block:: python + + from je_auto_control import wait_for_focus_change, click_text + + click_text("Settings") + focused = wait_for_focus_change(timeout=3) + # {"name": "Search", "role": "ControlType_50004", "app_name": "app.exe", ...} + if focused is not None: + ... # 焦點已移動——對話框 / 下一個欄位已就緒 + +回傳新獲得焦點的元素 ``{name, role, app_name, bounds, …}``,若在 ``timeout`` 秒內(預設 ``5``) +沒有焦點變化則回傳 ``None``。 + +執行器指令 +---------- + +``AC_wait_for_focus_change``(``timeout``)回傳 ``{changed, element}``。以唯讀 +``ac_wait_for_focus_change`` MCP 工具及 Script Builder 指令(位於 **Native UI** 分類下) +形式提供。 diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index f514a937..6b8fcbc8 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -89,6 +89,8 @@ from je_auto_control.utils.selection_view import ( get_selection, list_views, set_view, ) +# Reactive UIA event waits (focus-changed) +from je_auto_control.utils.ax_events import wait_for_focus_change # Rich clipboard formats — RTF + CSV/TSV codecs and Windows get / set from je_auto_control.utils.clipboard_rich_formats import ( build_rtf, csv_to_rows, get_clipboard_csv, get_clipboard_rtf, rows_to_csv, @@ -1697,6 +1699,7 @@ def start_autocontrol_gui(*args, **kwargs): "window_interaction_state", "legacy_info", "legacy_default_action", "get_selection", "list_views", "set_view", + "wait_for_focus_change", "build_rtf", "rtf_to_text", "rows_to_csv", "csv_to_rows", "set_clipboard_rtf", "get_clipboard_rtf", "set_clipboard_csv", "get_clipboard_csv", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index c7aabaed..f5a8811a 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -1691,6 +1691,12 @@ def _add_native_control_specs(specs: List[CommandSpec]) -> None: fields=(FieldSpec("view", FieldType.STRING),) + fields, description="Switch a control to the named view (MultipleViewPattern).", )) + specs.append(CommandSpec( + "AC_wait_for_focus_change", "Native UI", "Wait for Focus Change", + fields=(FieldSpec("timeout", FieldType.FLOAT, optional=True, + default=5.0),), + description="Block until keyboard focus moves (real UIA focus event).", + )) specs.append(CommandSpec( "AC_get_control_text", "Native UI", "Get Control Text", fields=fields, diff --git a/je_auto_control/utils/accessibility/backends/base.py b/je_auto_control/utils/accessibility/backends/base.py index de085ab9..4b86ab54 100644 --- a/je_auto_control/utils/accessibility/backends/base.py +++ b/je_auto_control/utils/accessibility/backends/base.py @@ -281,6 +281,18 @@ def set_view(self, view: str = "", name: Optional[str] = None, """Switch a control to the named view (MultipleViewPattern); True on success.""" self._unsupported("set_view") + # --- reactive events (UIA event subscription) -------------------------- + + def wait_for_focus_change(self, timeout: float = 5.0, + ) -> Optional[Dict[str, Any]]: + """Block until the keyboard focus moves, then return the newly-focused + element ``{name, role, app_name, ...}`` (or None on ``timeout``). + + A zero-latency native wait (UIA AddFocusChangedEventHandler) — unlike the + polling recorder, it can't miss a fast focus transition. + """ + self._unsupported("wait_for_focus_change") + def _unsupported(self, operation: str): """Raise a clear error for an action this backend can't perform.""" raise AccessibilityNotAvailableError( diff --git a/je_auto_control/utils/accessibility/backends/windows_backend.py b/je_auto_control/utils/accessibility/backends/windows_backend.py index 8775bc5d..67598676 100644 --- a/je_auto_control/utils/accessibility/backends/windows_backend.py +++ b/je_auto_control/utils/accessibility/backends/windows_backend.py @@ -62,9 +62,11 @@ class WindowsAccessibilityBackend(AccessibilityBackend): name = "windows-uia" def __init__(self) -> None: + import threading self.available = _is_available() self._automation = None self._uia_module = None + self._event_lock = threading.Lock() def _ensure_automation(self): if self._automation is not None: @@ -420,6 +422,48 @@ def set_view(self, view="", name=None, role=None, app_name=None, return False return False + def _make_focus_handler(self, sink): + """Build a COM focus-changed event handler that puts events on ``sink``.""" + try: + import comtypes + interface = self._uia_module.IUIAutomationFocusChangedEventHandler + except (ImportError, AttributeError): + return None + + class _FocusHandler(comtypes.COMObject): + _com_interfaces_ = [interface] + + def IUIAutomationFocusChangedEventHandler_HandleFocusChangedEvent( + self, sender): # noqa: N802 # reason: comtypes callback name + element = _convert_uia(sender) + sink.put(element.to_dict() if element else {"focused": True}) + return 0 + + return _FocusHandler() + + def wait_for_focus_change(self, timeout=5.0) -> Optional[Dict[str, Any]]: + import queue + automation = self._ensure_automation() + events: "queue.Queue" = queue.Queue() + handler = self._make_focus_handler(events) + if handler is None: + return None + try: + with self._event_lock: + automation.AddFocusChangedEventHandler(None, handler) + except (OSError, AttributeError): + return None + try: + return events.get(timeout=float(timeout)) + except queue.Empty: + return None + finally: + with self._event_lock: + try: + automation.RemoveFocusChangedEventHandler(handler) + except (OSError, AttributeError): + pass + def get_table_headers(self, name=None, role=None, app_name=None, automation_id=None) -> Optional[Dict[str, Any]]: raw = self._find_raw(name, role, app_name, automation_id) diff --git a/je_auto_control/utils/ax_events/__init__.py b/je_auto_control/utils/ax_events/__init__.py new file mode 100644 index 00000000..d6cb48aa --- /dev/null +++ b/je_auto_control/utils/ax_events/__init__.py @@ -0,0 +1,4 @@ +"""Reactive UIA event waits (focus-changed) via the accessibility backend seam.""" +from je_auto_control.utils.ax_events.ax_events import wait_for_focus_change + +__all__ = ["wait_for_focus_change"] diff --git a/je_auto_control/utils/ax_events/ax_events.py b/je_auto_control/utils/ax_events/ax_events.py new file mode 100644 index 00000000..229228df --- /dev/null +++ b/je_auto_control/utils/ax_events/ax_events.py @@ -0,0 +1,25 @@ +"""Reactive UI Automation event waits (focus-changed). + +The accessibility recorder *polls* the focused element every ~250 ms, so it can +miss a fast focus transition and reacts a quarter-second late. UIA exposes real +events: ``wait_for_focus_change`` blocks on the native +``AddFocusChangedEventHandler`` and returns the moment focus moves — the +zero-latency, miss-free "wait until focus lands on the dialog" primitive, the +accessibility-tree analogue of ``wait_for_window`` / ``wait_for_image``. + +It is a thin dispatch onto the injectable ``accessibility.backends.get_backend()`` +seam — headless-testable on any platform by injecting a fake backend; the real +event subscription (registered / unregistered under a lock, on the calling +thread) lives in the Windows backend. Imports no ``PySide6``. +""" +from typing import Any, Dict, Optional + + +def wait_for_focus_change(*, timeout: float = 5.0) -> Optional[Dict[str, Any]]: + """Block until the keyboard focus moves, then return the newly-focused element. + + Returns the focused element as ``{name, role, app_name, bounds, ...}``, or + ``None`` if no focus change occurs within ``timeout`` seconds. + """ + from je_auto_control.utils.accessibility.backends import get_backend + return get_backend().wait_for_focus_change(float(timeout)) diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index 9f55c7eb..59dd56f7 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -2602,6 +2602,13 @@ def _set_view(view: str, name: Optional[str] = None, role: Optional[str] = None, automation_id=automation_id) +def _wait_for_focus_change(timeout: Any = 5.0) -> Dict[str, Any]: + """Adapter: block until the keyboard focus moves (UIA focus event).""" + from je_auto_control.utils.ax_events import wait_for_focus_change + element = wait_for_focus_change(timeout=float(timeout)) + return {"changed": element is not None, "element": element} + + def _get_control_text(name: Optional[str] = None, role: Optional[str] = None, app_name: Optional[str] = None, automation_id: Optional[str] = None) -> Dict[str, Any]: @@ -6591,6 +6598,7 @@ def __init__(self): "AC_get_selection": _get_selection, "AC_list_views": _list_views, "AC_set_view": _set_view, + "AC_wait_for_focus_change": _wait_for_focus_change, "AC_get_control_text": _get_control_text, "AC_find_control_text": _find_control_text, "AC_select_control_text": _select_control_text, diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index 7b91585c..de0c8f7c 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -1310,6 +1310,16 @@ def a11y_control_tools() -> List[MCPTool]: handler=h.set_view, annotations=DESTRUCTIVE, ), + MCPTool( + name="ac_wait_for_focus_change", + description=("Block until the keyboard focus moves (real UIA focus " + "event — zero-latency, miss-free, unlike polling). " + "Returns {changed, element} or {changed: false} on " + "'timeout' seconds (default 5)."), + input_schema=schema({"timeout": {"type": "number"}}), + handler=h.wait_for_focus_change, + annotations=READ_ONLY, + ), MCPTool( name="ac_get_control_text", description=("Read a control's full text via TextPattern: " diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index 2604c23a..26f98959 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -900,6 +900,12 @@ def set_view(view, name=None, role=None, app_name=None, automation_id=None): return _set_view(view, name, role, app_name, automation_id) +def wait_for_focus_change(timeout=5.0): + from je_auto_control.utils.executor.action_executor import ( + _wait_for_focus_change) + return _wait_for_focus_change(timeout) + + def get_selected_text(name=None, role=None, app_name=None, automation_id=None): from je_auto_control.utils.executor.action_executor import _get_selected_text return _get_selected_text(name, role, app_name, automation_id) diff --git a/test/unit_test/headless/test_ax_events_batch.py b/test/unit_test/headless/test_ax_events_batch.py new file mode 100644 index 00000000..688b48a0 --- /dev/null +++ b/test/unit_test/headless/test_ax_events_batch.py @@ -0,0 +1,77 @@ +"""Headless tests for reactive UIA focus-change waits (fake backend seam).""" +import je_auto_control as ac +from je_auto_control.utils.accessibility.backends import base as backend_base +from je_auto_control.utils.ax_events import wait_for_focus_change + +_FOCUSED = {"name": "Username", "role": "ControlType_50004", + "app_name": "login.exe", "bounds": [10, 20, 200, 24]} + + +class _FakeBackend(backend_base.AccessibilityBackend): + name = "fake" + available = True + + def __init__(self, result=None): + self.result = result + self.timeouts = [] + + def wait_for_focus_change(self, timeout=5.0): + self.timeouts.append(timeout) + return self.result + + +def _inject(monkeypatch, backend): + import je_auto_control.utils.accessibility.backends as backends + monkeypatch.setattr(backends, "_cached_backend", backend, raising=False) + + +def test_focus_change_returns_element(monkeypatch): + fake = _FakeBackend(dict(_FOCUSED)) + _inject(monkeypatch, fake) + assert wait_for_focus_change(timeout=2.0) == _FOCUSED + assert fake.timeouts == [2.0] + + +def test_focus_change_timeout_returns_none(monkeypatch): + _inject(monkeypatch, _FakeBackend(None)) + assert wait_for_focus_change(timeout=0.1) is None + + +def test_unsupported_backend_raises(monkeypatch): + from je_auto_control.utils.accessibility.element import ( + AccessibilityNotAvailableError) + _inject(monkeypatch, backend_base.AccessibilityBackend()) # all _unsupported + try: + wait_for_focus_change(timeout=0.1) + raised = False + except AccessibilityNotAvailableError: + raised = True + assert raised is True + + +# --- wiring --------------------------------------------------------------- + +def test_executor_adapter_wraps_element(monkeypatch): + _inject(monkeypatch, _FakeBackend(dict(_FOCUSED))) + from je_auto_control.utils.executor.action_executor import ( + _wait_for_focus_change) + out = _wait_for_focus_change(1.5) + assert out["changed"] is True and out["element"]["name"] == "Username" + _inject(monkeypatch, _FakeBackend(None)) + assert _wait_for_focus_change(0.1)["changed"] is False + + +def test_wiring(): + known = set(ac.executor.known_commands()) + assert "AC_wait_for_focus_change" in known + from je_auto_control.utils.mcp_server.tools import build_default_tool_registry + names = {t.name for t in build_default_tool_registry()} + assert "ac_wait_for_focus_change" in names + from je_auto_control.gui.script_builder.command_schema import _build_specs + specs = {s.command for s in _build_specs()} + assert "AC_wait_for_focus_change" in specs + + +def test_facade_export(): + assert hasattr(ac, "wait_for_focus_change") + assert "wait_for_focus_change" in ac.__all__