From d85aab6861a591ebcd1569e65225253f446e9eee Mon Sep 17 00:00:00 2001 From: abrichr Date: Sat, 25 Jul 2026 00:25:45 -0400 Subject: [PATCH] feat: capture Windows UIA action evidence --- README.md | 9 ++ openadapt_capture/__init__.py | 6 ++ openadapt_capture/capture.py | 25 ++++++ openadapt_capture/config.py | 2 + openadapt_capture/events.py | 35 ++++++++ openadapt_capture/processing.py | 3 + openadapt_capture/recorder.py | 10 ++- openadapt_capture/window/__init__.py | 15 ++++ openadapt_capture/window/_windows.py | 109 ++++++++++++++++++++++++ pyproject.toml | 2 + tests/test_structural_observation.py | 121 +++++++++++++++++++++++++++ 11 files changed, 335 insertions(+), 2 deletions(-) create mode 100644 tests/test_structural_observation.py diff --git a/README.md b/README.md index bd05684..9e5ee4a 100644 --- a/README.md +++ b/README.md @@ -163,6 +163,15 @@ with Recorder( input("Perform the task, then press Enter...") ``` +For native Windows applications, pass +`capture_structural_observations=True`. Each mouse-down action then carries a +versioned, JSON-safe UI Automation fingerprint through +`Action.structural_observation` (role, name, AutomationId, window, ancestry, +bounds, process, and supported native patterns). This evidence is stored in the +existing `recording.db`; it does not create another recorder or upload path. +Do not enable it for RDP or Citrix client windows: those sessions are observed +externally through pixels and OCR. + `owner` matches the application (macOS: window owner name; Windows: process executable name) and `title` optionally disambiguates among its windows; both are case-insensitive substrings, mirroring how `openadapt-flow`'s diff --git a/openadapt_capture/__init__.py b/openadapt_capture/__init__.py index 3fca6ff..9707fb4 100644 --- a/openadapt_capture/__init__.py +++ b/openadapt_capture/__init__.py @@ -56,6 +56,9 @@ MouseUpEvent, ScreenEvent, ScreenFrameEvent, + StructuralBounds, + StructuralNode, + StructuralObservationV1, ) # Event processing @@ -155,6 +158,9 @@ "MouseClickEvent", "MouseDoubleClickEvent", "MouseDragEvent", + "StructuralBounds", + "StructuralNode", + "StructuralObservationV1", # Keyboard events "KeyDownEvent", "KeyUpEvent", diff --git a/openadapt_capture/capture.py b/openadapt_capture/capture.py index 0df400e..49c7147 100644 --- a/openadapt_capture/capture.py +++ b/openadapt_capture/capture.py @@ -34,6 +34,7 @@ MouseMoveEvent, MouseScrollEvent, MouseUpEvent, + StructuralObservationV1, ) from openadapt_capture.processing import process_events @@ -43,6 +44,22 @@ from openadapt_capture.browser_events import BrowserEvent +def _parse_structural_observation(raw: object) -> StructuralObservationV1 | None: + """Parse only the versioned native-observation contract. + + Historical ``element_state`` JSON remains readable but is not silently + promoted to trusted structural evidence. + """ + if not isinstance(raw, dict): + return None + if raw.get("schema_version") != 1 or raw.get("observer") != "windows_uia": + return None + try: + return StructuralObservationV1.model_validate(raw) + except Exception: + return None + + def _convert_action_event(db_event) -> PydanticActionEvent | None: """Convert a SQLAlchemy ActionEvent to a Pydantic event. @@ -69,6 +86,9 @@ def _convert_action_event(db_event) -> PydanticActionEvent | None: x=db_event.mouse_x or 0, y=db_event.mouse_y or 0, button=button, + structural_observation=_parse_structural_observation( + getattr(db_event, "element_state", None) + ), ) elif db_event.mouse_pressed is False: return MouseUpEvent( @@ -369,6 +389,11 @@ def button(self) -> str | None: return btn.value if hasattr(btn, "value") else str(btn) return None + @property + def structural_observation(self) -> StructuralObservationV1 | None: + """Native accessibility evidence captured at action time, if present.""" + return getattr(self.event, "structural_observation", None) + @property def screenshot(self) -> "Image" | None: """Get the screenshot at the time of this action. diff --git a/openadapt_capture/config.py b/openadapt_capture/config.py index ba0e7c8..9c11932 100644 --- a/openadapt_capture/config.py +++ b/openadapt_capture/config.py @@ -99,6 +99,7 @@ class Settings(BaseSettings): "capture_audio": "RECORD_AUDIO", "capture_images": "RECORD_IMAGES", "capture_window_data": "RECORD_WINDOW_DATA", + "capture_structural_observations": "RECORD_READ_ACTIVE_ELEMENT_STATE", "capture_browser_events": "RECORD_BROWSER_EVENTS", "window_owner": "RECORD_WINDOW_OWNER", "window_title": "RECORD_WINDOW_TITLE", @@ -124,6 +125,7 @@ class RecordingConfig: capture_audio: bool | None = None capture_images: bool | None = None capture_window_data: bool | None = None + capture_structural_observations: bool | None = None capture_browser_events: bool | None = None # Window-scoped capture selectors (see window_capture.WindowTarget). window_owner: str | None = None diff --git a/openadapt_capture/events.py b/openadapt_capture/events.py index ac5ae5b..49a9fa4 100644 --- a/openadapt_capture/events.py +++ b/openadapt_capture/events.py @@ -62,6 +62,37 @@ class BaseEvent(BaseModel): model_config = {"use_enum_values": True} +class StructuralBounds(BaseModel): + """Element bounds in the native screen coordinate space.""" + + left: float + top: float + right: float + bottom: float + + +class StructuralNode(BaseModel): + """Stable, JSON-safe subset of a native accessibility node.""" + + role: str = "" + name: str = "" + automation_id: str = "" + class_name: str = "" + bounds: StructuralBounds | None = None + supported_patterns: list[str] = Field(default_factory=list) + + +class StructuralObservationV1(BaseModel): + """Versioned Windows UI Automation evidence captured with an action.""" + + schema_version: Literal[1] = 1 + observer: Literal["windows_uia"] = "windows_uia" + target: StructuralNode + ancestors: list[StructuralNode] = Field(default_factory=list) + window_name: str = "" + process_id: int | None = None + + # ============================================================================= # Mouse Events # ============================================================================= @@ -88,6 +119,7 @@ class MouseDownEvent(BaseEvent): x: float = Field(description="Mouse X position in pixels") y: float = Field(description="Mouse Y position in pixels") button: str = Field(description="Native mouse button name") + structural_observation: StructuralObservationV1 | None = None class MouseUpEvent(BaseEvent): @@ -203,6 +235,7 @@ class MouseClickEvent(BaseEvent): x: float = Field(description="Mouse X position in pixels") y: float = Field(description="Mouse Y position in pixels") button: str = Field(description="Native mouse button name") + structural_observation: StructuralObservationV1 | None = None children: list[MouseDownEvent | MouseUpEvent] = Field( default_factory=list, description="Child events that were merged" ) @@ -219,6 +252,7 @@ class MouseDoubleClickEvent(BaseEvent): x: float = Field(description="Mouse X position in pixels") y: float = Field(description="Mouse Y position in pixels") button: str = Field(description="Native mouse button name") + structural_observation: StructuralObservationV1 | None = None children: list[MouseDownEvent | MouseUpEvent] = Field( default_factory=list, description="Child events that were merged" ) @@ -237,6 +271,7 @@ class MouseDragEvent(BaseEvent): dx: float = Field(description="Horizontal displacement (end_x - start_x)") dy: float = Field(description="Vertical displacement (end_y - start_y)") button: str = Field(description="Native mouse button name") + structural_observation: StructuralObservationV1 | None = None children: list[MouseDownEvent | MouseMoveEvent | MouseUpEvent] = Field( default_factory=list, description="Child events that were merged" ) diff --git a/openadapt_capture/processing.py b/openadapt_capture/processing.py index 2dd5d3f..50ec50a 100644 --- a/openadapt_capture/processing.py +++ b/openadapt_capture/processing.py @@ -373,6 +373,7 @@ def calculate_distance(x1: float, y1: float, x2: float, y2: float) -> float: x=down.x, y=down.y, button=down.button, + structural_observation=down.structural_observation, children=[down, up, next_down, next_up], ) result.append(double_click) @@ -387,6 +388,7 @@ def calculate_distance(x1: float, y1: float, x2: float, y2: float) -> float: x=down.x, y=down.y, button=down.button, + structural_observation=down.structural_observation, children=[down, up], ) result.append(single_click) @@ -451,6 +453,7 @@ def calculate_distance(x1: float, y1: float, x2: float, y2: float) -> float: dx=event.x - down_event.x, dy=event.y - down_event.y, button=down_event.button, + structural_observation=down_event.structural_observation, children=[down_event] + moves + [event], ) result.append(drag) diff --git a/openadapt_capture/recorder.py b/openadapt_capture/recorder.py index 8843529..94a51fe 100644 --- a/openadapt_capture/recorder.py +++ b/openadapt_capture/recorder.py @@ -856,9 +856,13 @@ def trigger_action_event( x = action_event_args.get("mouse_x") y = action_event_args.get("mouse_y") if x is not None and y is not None: - if config.RECORD_READ_ACTIVE_ELEMENT_STATE: + if ( + config.RECORD_READ_ACTIVE_ELEMENT_STATE + and action_event_args.get("name") == "click" + and action_event_args.get("mouse_pressed") is True + ): # element lookup needs GLOBAL coordinates: translate afterwards. - element_state = window.get_active_element_state(x, y) + element_state = window.get_active_element_observation(x, y) else: element_state = {} action_event_args["element_state"] = element_state @@ -2176,6 +2180,7 @@ def __init__( capture_audio: bool | None = None, capture_images: bool | None = None, capture_window_data: bool | None = None, + capture_structural_observations: bool | None = None, capture_browser_events: bool | None = None, capture_full_video: bool | None = None, video_encoding: str | None = None, @@ -2209,6 +2214,7 @@ def __init__( capture_audio=capture_audio, capture_images=capture_images, capture_window_data=capture_window_data, + capture_structural_observations=capture_structural_observations, capture_browser_events=capture_browser_events, capture_full_video=capture_full_video, video_encoding=video_encoding, diff --git a/openadapt_capture/window/__init__.py b/openadapt_capture/window/__init__.py index eeb0a98..ebb7eb3 100644 --- a/openadapt_capture/window/__init__.py +++ b/openadapt_capture/window/__init__.py @@ -93,3 +93,18 @@ def get_active_element_state(x: int, y: int) -> dict | None: except Exception as exc: logger.warning(f"{exc=}") return None + + +def get_active_element_observation(x: int, y: int) -> dict | None: + """Return versioned structural evidence for the element at a point. + + Platforms without a versioned observer deliberately return ``None``; + legacy element-state dictionaries are not promoted to structural evidence. + """ + if impl is None or not hasattr(impl, "get_active_element_observation"): + return None + try: + return impl.get_active_element_observation(x, y) + except Exception as exc: + logger.warning(f"{exc=}") + return None diff --git a/openadapt_capture/window/_windows.py b/openadapt_capture/window/_windows.py index e5b85ab..9fa3d23 100644 --- a/openadapt_capture/window/_windows.py +++ b/openadapt_capture/window/_windows.py @@ -95,6 +95,115 @@ def get_active_element_state(x: int, y: int) -> dict: return properties +def _safe_text(value: object) -> str: + """Return a stable textual property without leaking object reprs.""" + if value is None: + return "" + if isinstance(value, str): + return value + return str(value) + + +def _element_info_value(element: object, name: str) -> object | None: + info = getattr(element, "element_info", None) + if info is None: + return None + try: + return getattr(info, name, None) + except Exception: + return None + + +def _safe_int(value: object) -> int | None: + try: + return int(value) if value is not None else None + except (TypeError, ValueError): + return None + + +def _supported_patterns(element: object) -> list[str]: + patterns = ( + ("invoke", "iface_invoke"), + ("value", "iface_value"), + ("selection_item", "iface_selection_item"), + ("toggle", "iface_toggle"), + ("expand_collapse", "iface_expand_collapse"), + ) + supported = [] + for label, attribute in patterns: + try: + if getattr(element, attribute, None) is not None: + supported.append(label) + except Exception: + continue + return supported + + +def _structural_node(element: object) -> dict: + try: + properties = element.get_properties() + except Exception: + try: + properties = get_properties(element) + except Exception: + properties = {} + + rectangle = properties.get("rectangle") + bounds = dictify_rect(rectangle) if rectangle is not None else None + texts = properties.get("texts") or [] + name = properties.get("name") or (texts[0] if texts else "") + role = properties.get("control_type") or _element_info_value( + element, "control_type" + ) + return { + "role": _safe_text(role), + "name": _safe_text(name), + "automation_id": _safe_text( + properties.get("automation_id") + or _element_info_value(element, "automation_id") + ), + "class_name": _safe_text( + properties.get("class_name") + or _element_info_value(element, "class_name") + ), + "bounds": bounds, + "supported_patterns": _supported_patterns(element), + } + + +def get_active_element_observation(x: int, y: int) -> dict: + """Capture a versioned, JSON-safe UIA fingerprint at action time.""" + active_window = get_active_window() + target = active_window.from_point(x, y) + + ancestors = [] + current = target + for _ in range(5): + try: + current = current.parent() + except Exception: + break + if current is None: + break + ancestors.append(_structural_node(current)) + if current == active_window: + break + + window_node = _structural_node(active_window) + process_id = _element_info_value(target, "process_id") + if process_id is None: + process_id = _element_info_value(active_window, "process_id") + + return { + "schema_version": 1, + "observer": "windows_uia", + "target": _structural_node(target), + "ancestors": ancestors, + "window_name": window_node["name"], + "process_id": _safe_int(process_id), + } + + def get_active_window() -> "pywinauto.application.WindowSpecification": """Get the active window object. diff --git a/pyproject.toml b/pyproject.toml index 5bb24ba..2e44cef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,6 +44,8 @@ dependencies = [ "matplotlib>=3.10.8", # Native macOS window/accessibility observation via permissive PyObjC. "pyobjc-framework-ApplicationServices>=12.2.1; sys_platform == 'darwin'", + # Native Windows window resolution and UI Automation observation. + "pywinauto>=0.6.8; sys_platform == 'win32'", ] [project.optional-dependencies] diff --git a/tests/test_structural_observation.py b/tests/test_structural_observation.py new file mode 100644 index 0000000..c452056 --- /dev/null +++ b/tests/test_structural_observation.py @@ -0,0 +1,121 @@ +"""Native structural evidence survives Capture's public action API.""" + +import time + +from openadapt_capture.capture import CaptureSession +from openadapt_capture.config import RecordingConfig, config, config_override +from openadapt_capture.db import create_db, crud +from openadapt_capture.window import _windows + +OBSERVATION = { + "schema_version": 1, + "observer": "windows_uia", + "target": { + "role": "Button", + "name": "Submit", + "automation_id": "submitButton", + "class_name": "Button", + "bounds": {"left": 10, "top": 20, "right": 110, "bottom": 50}, + "supported_patterns": ["invoke"], + }, + "ancestors": [], + "window_name": "Claims", + "process_id": 1234, +} + + +def _capture_with_click(tmp_path, element_state): + db_path = tmp_path / "recording.db" + engine, session_factory = create_db(str(db_path)) + session = session_factory() + recording = crud.insert_recording( + session, + { + "timestamp": time.time(), + "monitor_width": 1920, + "monitor_height": 1080, + "double_click_interval_seconds": 0.5, + "double_click_distance_pixels": 5, + "platform": "win32", + }, + ) + for offset, pressed in ((0.001, True), (0.002, False)): + crud.insert_action_event( + session, + recording, + recording.timestamp + offset, + { + "name": "click", + "mouse_x": 50, + "mouse_y": 35, + "mouse_button_name": "left", + "mouse_pressed": pressed, + "element_state": element_state if pressed else {}, + }, + ) + session.close() + engine.dispose() + return CaptureSession.load(tmp_path) + + +def test_versioned_uia_observation_round_trips_into_processed_click(tmp_path): + with _capture_with_click(tmp_path, OBSERVATION) as capture: + (action,) = list(capture.actions()) + + observation = action.structural_observation + assert observation is not None + assert observation.target.automation_id == "submitButton" + assert observation.window_name == "Claims" + + +def test_legacy_element_state_is_not_promoted_to_structural_evidence(tmp_path): + with _capture_with_click(tmp_path, {"control_type": "Button"}) as capture: + (action,) = list(capture.actions()) + + assert action.structural_observation is None + + +def test_windows_observer_emits_json_safe_fingerprint(monkeypatch): + class Rect: + left, top, right, bottom = 10, 20, 110, 50 + + class Info: + process_id = 1234 + automation_id = "submitButton" + class_name = "Button" + control_type = "Button" + + class Element: + element_info = Info() + iface_invoke = object() + + def __init__(self, name, parent=None): + self.name = name + self._parent = parent + + def get_properties(self): + return { + "texts": [self.name], + "control_type": "Button", + "rectangle": Rect(), + } + + def parent(self): + return self._parent + + window = Element("Claims") + target = Element("Submit", window) + window.from_point = lambda _x, _y: target + monkeypatch.setattr(_windows, "get_active_window", lambda: window) + + observation = _windows.get_active_element_observation(50, 35) + assert observation["target"]["automation_id"] == "submitButton" + assert observation["target"]["supported_patterns"] == ["invoke"] + assert observation["ancestors"][0]["name"] == "Claims" + + +def test_recording_config_exposes_structural_observer_toggle(): + original = config.RECORD_READ_ACTIVE_ELEMENT_STATE + with config_override(RecordingConfig(capture_structural_observations=True)): + assert config.RECORD_READ_ACTIVE_ELEMENT_STATE is True + assert config.RECORD_READ_ACTIVE_ELEMENT_STATE == original